Compare commits

..

1 Commits

Author SHA1 Message Date
Kit Langton 40ca8a6fc4 feat: unwrap share namespaces to flat exports + barrel 2026-04-15 23:14:55 -04:00
59 changed files with 5347 additions and 5335 deletions
+2 -2
View File
@@ -42,9 +42,9 @@ import { ModelID, ProviderID } from "../provider/schema"
import { Agent as AgentModule } from "../agent/agent"
import { AppRuntime } from "@/effect/app-runtime"
import { Installation } from "@/installation"
import { MessageV2 } from "@/session"
import { MessageV2 } from "@/session/message-v2"
import { Config } from "@/config"
import { Todo } from "@/session"
import { Todo } from "@/session/todo"
import { z } from "zod"
import { LoadAPIKeyError } from "ai"
import type { AssistantMessage, Event, OpencodeClient, SessionMessageResponse, ToolPart } from "@opencode-ai/sdk/v2"
+1 -1
View File
@@ -4,7 +4,7 @@ import { Effect } from "effect"
import { Agent } from "../../../agent/agent"
import { Provider } from "../../../provider"
import { Session } from "../../../session"
import type { MessageV2 } from "../../../session"
import type { MessageV2 } from "../../../session/message-v2"
import { MessageID, PartID } from "../../../session/schema"
import { ToolRegistry } from "../../../tool/registry"
import { Instance } from "../../../project/instance"
+1 -1
View File
@@ -1,6 +1,6 @@
import type { Argv } from "yargs"
import { Session } from "../../session"
import { MessageV2 } from "../../session"
import { MessageV2 } from "../../session/message-v2"
import { SessionID } from "../../session/schema"
import { cmd } from "./cmd"
import { bootstrap } from "../bootstrap"
+3 -3
View File
@@ -21,14 +21,14 @@ import { cmd } from "./cmd"
import { ModelsDev } from "../../provider/models"
import { Instance } from "@/project/instance"
import { bootstrap } from "../bootstrap"
import { SessionShare } from "@/share/session"
import { SessionShare } from "@/share"
import { Session } from "../../session"
import type { SessionID } from "../../session/schema"
import { MessageID, PartID } from "../../session/schema"
import { Provider } from "../../provider"
import { Bus } from "../../bus"
import { MessageV2 } from "../../session"
import { SessionPrompt } from "@/session"
import { MessageV2 } from "../../session/message-v2"
import { SessionPrompt } from "@/session/prompt"
import { AppRuntime } from "@/effect/app-runtime"
import { Git } from "@/git"
import { setTimeout as sleep } from "node:timers/promises"
+2 -2
View File
@@ -1,13 +1,13 @@
import type { Argv } from "yargs"
import type { Session as SDKSession, Message, Part } from "@opencode-ai/sdk/v2"
import { Session } from "../../session"
import { MessageV2 } from "../../session"
import { MessageV2 } from "../../session/message-v2"
import { cmd } from "./cmd"
import { bootstrap } from "../bootstrap"
import { Database } from "../../storage/db"
import { SessionTable, MessageTable, PartTable } from "../../session/session.sql"
import { Instance } from "../../project/instance"
import { ShareNext } from "../../share/share-next"
import { ShareNext } from "../../share"
import { EOL } from "os"
import { Filesystem } from "../../util/filesystem"
import { AppRuntime } from "@/effect/app-runtime"
@@ -85,7 +85,7 @@ import { useTuiConfig } from "../../context/tui-config"
import { getScrollAcceleration } from "../../util/scroll"
import { TuiPluginRuntime } from "../../plugin"
import { DialogGoUpsell } from "../../component/dialog-go-upsell"
import { SessionRetry } from "@/session"
import { SessionRetry } from "@/session/retry"
addDefaultParsers(parsers.parsers)
+12 -12
View File
@@ -22,17 +22,17 @@ import { Skill } from "@/skill"
import { Discovery } from "@/skill/discovery"
import { Question } from "@/question"
import { Permission } from "@/permission"
import { Todo } from "@/session"
import { Todo } from "@/session/todo"
import { Session } from "@/session"
import { SessionStatus } from "@/session"
import { SessionRunState } from "@/session"
import { SessionProcessor } from "@/session"
import { SessionCompaction } from "@/session"
import { SessionRevert } from "@/session"
import { SessionSummary } from "@/session"
import { SessionPrompt } from "@/session"
import { Instruction } from "@/session"
import { LLM } from "@/session"
import { SessionStatus } from "@/session/status"
import { SessionRunState } from "@/session/run-state"
import { SessionProcessor } from "@/session/processor"
import { SessionCompaction } from "@/session/compaction"
import { SessionRevert } from "@/session/revert"
import { SessionSummary } from "@/session/summary"
import { SessionPrompt } from "@/session/prompt"
import { Instruction } from "@/session/instruction"
import { LLM } from "@/session/llm"
import { LSP } from "@/lsp"
import { MCP } from "@/mcp"
import { McpAuth } from "@/mcp/auth"
@@ -45,8 +45,8 @@ import { Vcs } from "@/project/vcs"
import { Worktree } from "@/worktree"
import { Pty } from "@/pty"
import { Installation } from "@/installation"
import { ShareNext } from "@/share/share-next"
import { SessionShare } from "@/share/session"
import { ShareNext } from "@/share"
import { SessionShare } from "@/share"
export const AppLayer = Layer.mergeAll(
AppFileSystem.defaultLayer,
@@ -5,7 +5,7 @@ import { Plugin } from "@/plugin"
import { LSP } from "@/lsp"
import { FileWatcher } from "@/file/watcher"
import { Format } from "@/format"
import { ShareNext } from "@/share/share-next"
import { ShareNext } from "@/share"
import { File } from "@/file"
import { Vcs } from "@/project/vcs"
import { Snapshot } from "@/snapshot"
@@ -5,7 +5,7 @@ import { iife } from "@/util/iife"
import { Log } from "../../util/log"
import { setTimeout as sleep } from "node:timers/promises"
import { CopilotModels } from "./models"
import { MessageV2 } from "@/session"
import { MessageV2 } from "@/session/message-v2"
const log = Log.create({ service: "plugin.copilot" })
+1 -1
View File
@@ -10,7 +10,7 @@ import { Command } from "../command"
import { Instance } from "./instance"
import { Log } from "@/util/log"
import { FileWatcher } from "@/file/watcher"
import { ShareNext } from "@/share/share-next"
import { ShareNext } from "@/share"
import * as Effect from "effect/Effect"
export const InstanceBootstrap = Effect.gen(function* () {
@@ -4,15 +4,15 @@ import { describeRoute, validator, resolver } from "hono-openapi"
import { SessionID, MessageID, PartID } from "@/session/schema"
import z from "zod"
import { Session } from "../../session"
import { MessageV2 } from "../../session"
import { SessionPrompt } from "../../session"
import { SessionRunState } from "@/session"
import { SessionCompaction } from "../../session"
import { SessionRevert } from "../../session"
import { SessionShare } from "@/share/session"
import { SessionStatus } from "@/session"
import { SessionSummary } from "@/session"
import { Todo } from "../../session"
import { MessageV2 } from "../../session/message-v2"
import { SessionPrompt } from "../../session/prompt"
import { SessionRunState } from "@/session/run-state"
import { SessionCompaction } from "../../session/compaction"
import { SessionRevert } from "../../session/revert"
import { SessionShare } from "@/share"
import { SessionStatus } from "@/session/status"
import { SessionSummary } from "@/session/summary"
import { Todo } from "../../session/todo"
import { Effect } from "effect"
import { AppRuntime } from "../../effect/app-runtime"
import { Agent } from "../../agent/agent"
+336 -334
View File
@@ -3,11 +3,11 @@ import { Bus } from "@/bus"
import { Session } from "."
import { SessionID, MessageID, PartID } from "./schema"
import { Provider } from "../provider"
import { MessageV2 } from "."
import { MessageV2 } from "./message-v2"
import z from "zod"
import { Token } from "../util/token"
import { Log } from "../util/log"
import { SessionProcessor } from "."
import { SessionProcessor } from "./processor"
import { Agent } from "@/agent/agent"
import { Plugin } from "@/plugin"
import { Config } from "@/config"
@@ -17,172 +17,173 @@ import { Effect, Layer, Context } from "effect"
import { InstanceState } from "@/effect"
import { isOverflow as overflow } from "./overflow"
const log = Log.create({ service: "session.compaction" })
export namespace SessionCompaction {
const log = Log.create({ service: "session.compaction" })
export const Event = {
Compacted: BusEvent.define(
"session.compacted",
z.object({
sessionID: SessionID.zod,
}),
),
}
export const Event = {
Compacted: BusEvent.define(
"session.compacted",
z.object({
sessionID: SessionID.zod,
}),
),
}
export const PRUNE_MINIMUM = 20_000
export const PRUNE_PROTECT = 40_000
const PRUNE_PROTECTED_TOOLS = ["skill"]
export const PRUNE_MINIMUM = 20_000
export const PRUNE_PROTECT = 40_000
const PRUNE_PROTECTED_TOOLS = ["skill"]
export interface Interface {
readonly isOverflow: (input: {
tokens: MessageV2.Assistant["tokens"]
model: Provider.Model
}) => Effect.Effect<boolean>
readonly prune: (input: { sessionID: SessionID }) => Effect.Effect<void>
readonly process: (input: {
parentID: MessageID
messages: MessageV2.WithParts[]
sessionID: SessionID
auto: boolean
overflow?: boolean
}) => Effect.Effect<"continue" | "stop">
readonly create: (input: {
sessionID: SessionID
agent: string
model: { providerID: ProviderID; modelID: ModelID }
auto: boolean
overflow?: boolean
}) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionCompaction") {}
export const layer: Layer.Layer<
Service,
never,
| Bus.Service
| Config.Service
| Session.Service
| Agent.Service
| Plugin.Service
| SessionProcessor.Service
| Provider.Service
> = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const config = yield* Config.Service
const session = yield* Session.Service
const agents = yield* Agent.Service
const plugin = yield* Plugin.Service
const processors = yield* SessionProcessor.Service
const provider = yield* Provider.Service
const isOverflow = Effect.fn("SessionCompaction.isOverflow")(function* (input: {
export interface Interface {
readonly isOverflow: (input: {
tokens: MessageV2.Assistant["tokens"]
model: Provider.Model
}) {
return overflow({ cfg: yield* config.get(), tokens: input.tokens, model: input.model })
})
// goes backwards through parts until there are PRUNE_PROTECT tokens worth of tool
// calls, then erases output of older tool calls to free context space
const prune = Effect.fn("SessionCompaction.prune")(function* (input: { sessionID: SessionID }) {
const cfg = yield* config.get()
if (cfg.compaction?.prune === false) return
log.info("pruning")
const msgs = yield* session
.messages({ sessionID: input.sessionID })
.pipe(Effect.catchIf(NotFoundError.isInstance, () => Effect.succeed(undefined)))
if (!msgs) return
let total = 0
let pruned = 0
const toPrune: MessageV2.ToolPart[] = []
let turns = 0
loop: for (let msgIndex = msgs.length - 1; msgIndex >= 0; msgIndex--) {
const msg = msgs[msgIndex]
if (msg.info.role === "user") turns++
if (turns < 2) continue
if (msg.info.role === "assistant" && msg.info.summary) break loop
for (let partIndex = msg.parts.length - 1; partIndex >= 0; partIndex--) {
const part = msg.parts[partIndex]
if (part.type === "tool")
if (part.state.status === "completed") {
if (PRUNE_PROTECTED_TOOLS.includes(part.tool)) continue
if (part.state.time.compacted) break loop
const estimate = Token.estimate(part.state.output)
total += estimate
if (total > PRUNE_PROTECT) {
pruned += estimate
toPrune.push(part)
}
}
}
}
log.info("found", { pruned, total })
if (pruned > PRUNE_MINIMUM) {
for (const part of toPrune) {
if (part.state.status === "completed") {
part.state.time.compacted = Date.now()
yield* session.updatePart(part)
}
}
log.info("pruned", { count: toPrune.length })
}
})
const processCompaction = Effect.fn("SessionCompaction.process")(function* (input: {
}) => Effect.Effect<boolean>
readonly prune: (input: { sessionID: SessionID }) => Effect.Effect<void>
readonly process: (input: {
parentID: MessageID
messages: MessageV2.WithParts[]
sessionID: SessionID
auto: boolean
overflow?: boolean
}) {
const parent = input.messages.findLast((m) => m.info.id === input.parentID)
if (!parent || parent.info.role !== "user") {
throw new Error(`Compaction parent must be a user message: ${input.parentID}`)
}
const userMessage = parent.info
}) => Effect.Effect<"continue" | "stop">
readonly create: (input: {
sessionID: SessionID
agent: string
model: { providerID: ProviderID; modelID: ModelID }
auto: boolean
overflow?: boolean
}) => Effect.Effect<void>
}
let messages = input.messages
let replay:
| {
info: MessageV2.User
parts: MessageV2.Part[]
}
| undefined
if (input.overflow) {
const idx = input.messages.findIndex((m) => m.info.id === input.parentID)
for (let i = idx - 1; i >= 0; i--) {
const msg = input.messages[i]
if (msg.info.role === "user" && !msg.parts.some((p) => p.type === "compaction")) {
replay = { info: msg.info, parts: msg.parts }
messages = input.messages.slice(0, i)
break
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionCompaction") {}
export const layer: Layer.Layer<
Service,
never,
| Bus.Service
| Config.Service
| Session.Service
| Agent.Service
| Plugin.Service
| SessionProcessor.Service
| Provider.Service
> = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const config = yield* Config.Service
const session = yield* Session.Service
const agents = yield* Agent.Service
const plugin = yield* Plugin.Service
const processors = yield* SessionProcessor.Service
const provider = yield* Provider.Service
const isOverflow = Effect.fn("SessionCompaction.isOverflow")(function* (input: {
tokens: MessageV2.Assistant["tokens"]
model: Provider.Model
}) {
return overflow({ cfg: yield* config.get(), tokens: input.tokens, model: input.model })
})
// goes backwards through parts until there are PRUNE_PROTECT tokens worth of tool
// calls, then erases output of older tool calls to free context space
const prune = Effect.fn("SessionCompaction.prune")(function* (input: { sessionID: SessionID }) {
const cfg = yield* config.get()
if (cfg.compaction?.prune === false) return
log.info("pruning")
const msgs = yield* session
.messages({ sessionID: input.sessionID })
.pipe(Effect.catchIf(NotFoundError.isInstance, () => Effect.succeed(undefined)))
if (!msgs) return
let total = 0
let pruned = 0
const toPrune: MessageV2.ToolPart[] = []
let turns = 0
loop: for (let msgIndex = msgs.length - 1; msgIndex >= 0; msgIndex--) {
const msg = msgs[msgIndex]
if (msg.info.role === "user") turns++
if (turns < 2) continue
if (msg.info.role === "assistant" && msg.info.summary) break loop
for (let partIndex = msg.parts.length - 1; partIndex >= 0; partIndex--) {
const part = msg.parts[partIndex]
if (part.type === "tool")
if (part.state.status === "completed") {
if (PRUNE_PROTECTED_TOOLS.includes(part.tool)) continue
if (part.state.time.compacted) break loop
const estimate = Token.estimate(part.state.output)
total += estimate
if (total > PRUNE_PROTECT) {
pruned += estimate
toPrune.push(part)
}
}
}
}
const hasContent =
replay && messages.some((m) => m.info.role === "user" && !m.parts.some((p) => p.type === "compaction"))
if (!hasContent) {
replay = undefined
messages = input.messages
}
}
const agent = yield* agents.get("compaction")
const model = agent.model
? yield* provider.getModel(agent.model.providerID, agent.model.modelID)
: yield* provider.getModel(userMessage.model.providerID, userMessage.model.modelID)
// Allow plugins to inject context or replace compaction prompt.
const compacting = yield* plugin.trigger(
"experimental.session.compacting",
{ sessionID: input.sessionID },
{ context: [], prompt: undefined },
)
const defaultPrompt = `Provide a detailed prompt for continuing our conversation above.
log.info("found", { pruned, total })
if (pruned > PRUNE_MINIMUM) {
for (const part of toPrune) {
if (part.state.status === "completed") {
part.state.time.compacted = Date.now()
yield* session.updatePart(part)
}
}
log.info("pruned", { count: toPrune.length })
}
})
const processCompaction = Effect.fn("SessionCompaction.process")(function* (input: {
parentID: MessageID
messages: MessageV2.WithParts[]
sessionID: SessionID
auto: boolean
overflow?: boolean
}) {
const parent = input.messages.findLast((m) => m.info.id === input.parentID)
if (!parent || parent.info.role !== "user") {
throw new Error(`Compaction parent must be a user message: ${input.parentID}`)
}
const userMessage = parent.info
let messages = input.messages
let replay:
| {
info: MessageV2.User
parts: MessageV2.Part[]
}
| undefined
if (input.overflow) {
const idx = input.messages.findIndex((m) => m.info.id === input.parentID)
for (let i = idx - 1; i >= 0; i--) {
const msg = input.messages[i]
if (msg.info.role === "user" && !msg.parts.some((p) => p.type === "compaction")) {
replay = { info: msg.info, parts: msg.parts }
messages = input.messages.slice(0, i)
break
}
}
const hasContent =
replay && messages.some((m) => m.info.role === "user" && !m.parts.some((p) => p.type === "compaction"))
if (!hasContent) {
replay = undefined
messages = input.messages
}
}
const agent = yield* agents.get("compaction")
const model = agent.model
? yield* provider.getModel(agent.model.providerID, agent.model.modelID)
: yield* provider.getModel(userMessage.model.providerID, userMessage.model.modelID)
// Allow plugins to inject context or replace compaction prompt.
const compacting = yield* plugin.trigger(
"experimental.session.compacting",
{ sessionID: input.sessionID },
{ context: [], prompt: undefined },
)
const defaultPrompt = `Provide a detailed prompt for continuing our conversation above.
Focus on information that would be helpful for continuing the conversation, including what we did, what we're doing, which files we're working on, and what we're going to do next.
The summary that you construct will be used so that another agent can read it and continue the work.
Do not call any tools. Respond only with the summary text.
@@ -212,199 +213,200 @@ When constructing the summary, try to stick to this template:
[Construct a structured list of relevant files that have been read, edited, or created that pertain to the task at hand. If all the files in a directory are relevant, include the path to the directory.]
---`
const prompt = compacting.prompt ?? [defaultPrompt, ...compacting.context].join("\n\n")
const msgs = structuredClone(messages)
yield* plugin.trigger("experimental.chat.messages.transform", {}, { messages: msgs })
const modelMessages = yield* MessageV2.toModelMessagesEffect(msgs, model, { stripMedia: true })
const ctx = yield* InstanceState.context
const msg: MessageV2.Assistant = {
id: MessageID.ascending(),
role: "assistant",
parentID: input.parentID,
sessionID: input.sessionID,
mode: "compaction",
agent: "compaction",
variant: userMessage.model.variant,
summary: true,
path: {
cwd: ctx.directory,
root: ctx.worktree,
},
cost: 0,
tokens: {
output: 0,
input: 0,
reasoning: 0,
cache: { read: 0, write: 0 },
},
modelID: model.id,
providerID: model.providerID,
time: {
created: Date.now(),
},
}
yield* session.updateMessage(msg)
const processor = yield* processors.create({
assistantMessage: msg,
sessionID: input.sessionID,
model,
})
const result = yield* processor.process({
user: userMessage,
agent,
sessionID: input.sessionID,
tools: {},
system: [],
messages: [
...modelMessages,
{
role: "user",
content: [{ type: "text", text: prompt }],
const prompt = compacting.prompt ?? [defaultPrompt, ...compacting.context].join("\n\n")
const msgs = structuredClone(messages)
yield* plugin.trigger("experimental.chat.messages.transform", {}, { messages: msgs })
const modelMessages = yield* MessageV2.toModelMessagesEffect(msgs, model, { stripMedia: true })
const ctx = yield* InstanceState.context
const msg: MessageV2.Assistant = {
id: MessageID.ascending(),
role: "assistant",
parentID: input.parentID,
sessionID: input.sessionID,
mode: "compaction",
agent: "compaction",
variant: userMessage.model.variant,
summary: true,
path: {
cwd: ctx.directory,
root: ctx.worktree,
},
],
model,
})
cost: 0,
tokens: {
output: 0,
input: 0,
reasoning: 0,
cache: { read: 0, write: 0 },
},
modelID: model.id,
providerID: model.providerID,
time: {
created: Date.now(),
},
}
yield* session.updateMessage(msg)
const processor = yield* processors.create({
assistantMessage: msg,
sessionID: input.sessionID,
model,
})
const result = yield* processor.process({
user: userMessage,
agent,
sessionID: input.sessionID,
tools: {},
system: [],
messages: [
...modelMessages,
{
role: "user",
content: [{ type: "text", text: prompt }],
},
],
model,
})
if (result === "compact") {
processor.message.error = new MessageV2.ContextOverflowError({
message: replay
? "Conversation history too large to compact - exceeds model context limit"
: "Session too large to compact - context exceeds model limit even after stripping media",
}).toObject()
processor.message.finish = "error"
yield* session.updateMessage(processor.message)
return "stop"
}
if (result === "continue" && input.auto) {
if (replay) {
const original = replay.info
const replayMsg = yield* session.updateMessage({
id: MessageID.ascending(),
role: "user",
sessionID: input.sessionID,
time: { created: Date.now() },
agent: original.agent,
model: original.model,
format: original.format,
tools: original.tools,
system: original.system,
})
for (const part of replay.parts) {
if (part.type === "compaction") continue
const replayPart =
part.type === "file" && MessageV2.isMedia(part.mime)
? { type: "text" as const, text: `[Attached ${part.mime}: ${part.filename ?? "file"}]` }
: part
yield* session.updatePart({
...replayPart,
id: PartID.ascending(),
messageID: replayMsg.id,
sessionID: input.sessionID,
})
}
if (result === "compact") {
processor.message.error = new MessageV2.ContextOverflowError({
message: replay
? "Conversation history too large to compact - exceeds model context limit"
: "Session too large to compact - context exceeds model limit even after stripping media",
}).toObject()
processor.message.finish = "error"
yield* session.updateMessage(processor.message)
return "stop"
}
if (!replay) {
const info = yield* provider.getProvider(userMessage.model.providerID)
if (
(yield* plugin.trigger(
"experimental.compaction.autocontinue",
{
sessionID: input.sessionID,
agent: userMessage.agent,
model: yield* provider.getModel(userMessage.model.providerID, userMessage.model.modelID),
provider: {
source: info.source,
info,
options: info.options,
},
message: userMessage,
overflow: input.overflow === true,
},
{ enabled: true },
)).enabled
) {
const continueMsg = yield* session.updateMessage({
if (result === "continue" && input.auto) {
if (replay) {
const original = replay.info
const replayMsg = yield* session.updateMessage({
id: MessageID.ascending(),
role: "user",
sessionID: input.sessionID,
time: { created: Date.now() },
agent: userMessage.agent,
model: userMessage.model,
})
const text =
(input.overflow
? "The previous request exceeded the provider's size limit due to large media attachments. The conversation was compacted and media files were removed from context. If the user was asking about attached images or files, explain that the attachments were too large to process and suggest they try again with smaller or fewer files.\n\n"
: "") +
"Continue if you have next steps, or stop and ask for clarification if you are unsure how to proceed."
yield* session.updatePart({
id: PartID.ascending(),
messageID: continueMsg.id,
sessionID: input.sessionID,
type: "text",
// Internal marker for auto-compaction followups so provider plugins
// can distinguish them from manual post-compaction user prompts.
// This is not a stable plugin contract and may change or disappear.
metadata: { compaction_continue: true },
synthetic: true,
text,
time: {
start: Date.now(),
end: Date.now(),
},
agent: original.agent,
model: original.model,
format: original.format,
tools: original.tools,
system: original.system,
})
for (const part of replay.parts) {
if (part.type === "compaction") continue
const replayPart =
part.type === "file" && MessageV2.isMedia(part.mime)
? { type: "text" as const, text: `[Attached ${part.mime}: ${part.filename ?? "file"}]` }
: part
yield* session.updatePart({
...replayPart,
id: PartID.ascending(),
messageID: replayMsg.id,
sessionID: input.sessionID,
})
}
}
if (!replay) {
const info = yield* provider.getProvider(userMessage.model.providerID)
if (
(yield* plugin.trigger(
"experimental.compaction.autocontinue",
{
sessionID: input.sessionID,
agent: userMessage.agent,
model: yield* provider.getModel(userMessage.model.providerID, userMessage.model.modelID),
provider: {
source: info.source,
info,
options: info.options,
},
message: userMessage,
overflow: input.overflow === true,
},
{ enabled: true },
)).enabled
) {
const continueMsg = yield* session.updateMessage({
id: MessageID.ascending(),
role: "user",
sessionID: input.sessionID,
time: { created: Date.now() },
agent: userMessage.agent,
model: userMessage.model,
})
const text =
(input.overflow
? "The previous request exceeded the provider's size limit due to large media attachments. The conversation was compacted and media files were removed from context. If the user was asking about attached images or files, explain that the attachments were too large to process and suggest they try again with smaller or fewer files.\n\n"
: "") +
"Continue if you have next steps, or stop and ask for clarification if you are unsure how to proceed."
yield* session.updatePart({
id: PartID.ascending(),
messageID: continueMsg.id,
sessionID: input.sessionID,
type: "text",
// Internal marker for auto-compaction followups so provider plugins
// can distinguish them from manual post-compaction user prompts.
// This is not a stable plugin contract and may change or disappear.
metadata: { compaction_continue: true },
synthetic: true,
text,
time: {
start: Date.now(),
end: Date.now(),
},
})
}
}
}
}
if (processor.message.error) return "stop"
if (result === "continue") yield* bus.publish(Event.Compacted, { sessionID: input.sessionID })
return result
})
const create = Effect.fn("SessionCompaction.create")(function* (input: {
sessionID: SessionID
agent: string
model: { providerID: ProviderID; modelID: ModelID }
auto: boolean
overflow?: boolean
}) {
const msg = yield* session.updateMessage({
id: MessageID.ascending(),
role: "user",
model: input.model,
sessionID: input.sessionID,
agent: input.agent,
time: { created: Date.now() },
if (processor.message.error) return "stop"
if (result === "continue") yield* bus.publish(Event.Compacted, { sessionID: input.sessionID })
return result
})
yield* session.updatePart({
id: PartID.ascending(),
messageID: msg.id,
sessionID: msg.sessionID,
type: "compaction",
auto: input.auto,
overflow: input.overflow,
const create = Effect.fn("SessionCompaction.create")(function* (input: {
sessionID: SessionID
agent: string
model: { providerID: ProviderID; modelID: ModelID }
auto: boolean
overflow?: boolean
}) {
const msg = yield* session.updateMessage({
id: MessageID.ascending(),
role: "user",
model: input.model,
sessionID: input.sessionID,
agent: input.agent,
time: { created: Date.now() },
})
yield* session.updatePart({
id: PartID.ascending(),
messageID: msg.id,
sessionID: msg.sessionID,
type: "compaction",
auto: input.auto,
overflow: input.overflow,
})
})
})
return Service.of({
isOverflow,
prune,
process: processCompaction,
create,
})
}),
)
return Service.of({
isOverflow,
prune,
process: processCompaction,
create,
})
}),
)
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(Provider.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(SessionProcessor.defaultLayer),
Layer.provide(Agent.defaultLayer),
Layer.provide(Plugin.defaultLayer),
Layer.provide(Bus.layer),
Layer.provide(Config.defaultLayer),
),
)
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(Provider.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(SessionProcessor.defaultLayer),
Layer.provide(Agent.defaultLayer),
Layer.provide(Plugin.defaultLayer),
Layer.provide(Bus.layer),
Layer.provide(Config.defaultLayer),
),
)
}
-14
View File
@@ -1,15 +1 @@
export * as Session from "./session"
export * as SessionRunState from "./run-state"
export * as SystemPrompt from "./system"
export * as Message from "./message"
export * as SessionRetry from "./retry"
export * as SessionProcessor from "./processor"
export * as SessionRevert from "./revert"
export * as Instruction from "./instruction"
export * as SessionSummary from "./summary"
export * as Todo from "./todo"
export * as LLM from "./llm"
export * as SessionStatus from "./status"
export * as SessionCompaction from "./compaction"
export * as SessionPrompt from "./prompt"
export * as MessageV2 from "./message-v2"
+164 -162
View File
@@ -10,7 +10,7 @@ import { withTransientReadRetry } from "@/util/effect-http-client"
import { Global } from "../global"
import { Instance } from "../project/instance"
import { Log } from "../util/log"
import type { MessageV2 } from "."
import type { MessageV2 } from "./message-v2"
import type { MessageID } from "./schema"
const log = Log.create({ service: "instruction" })
@@ -50,192 +50,194 @@ function extract(messages: MessageV2.WithParts[]) {
return paths
}
export interface Interface {
readonly clear: (messageID: MessageID) => Effect.Effect<void>
readonly systemPaths: () => Effect.Effect<Set<string>, AppFileSystem.Error>
readonly system: () => Effect.Effect<string[], AppFileSystem.Error>
readonly find: (dir: string) => Effect.Effect<string | undefined, AppFileSystem.Error>
readonly resolve: (
messages: MessageV2.WithParts[],
filepath: string,
messageID: MessageID,
) => Effect.Effect<{ filepath: string; content: string }[], AppFileSystem.Error>
}
export namespace Instruction {
export interface Interface {
readonly clear: (messageID: MessageID) => Effect.Effect<void>
readonly systemPaths: () => Effect.Effect<Set<string>, AppFileSystem.Error>
readonly system: () => Effect.Effect<string[], AppFileSystem.Error>
readonly find: (dir: string) => Effect.Effect<string | undefined, AppFileSystem.Error>
readonly resolve: (
messages: MessageV2.WithParts[],
filepath: string,
messageID: MessageID,
) => Effect.Effect<{ filepath: string; content: string }[], AppFileSystem.Error>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/Instruction") {}
export class Service extends Context.Service<Service, Interface>()("@opencode/Instruction") {}
export const layer: Layer.Layer<Service, never, AppFileSystem.Service | Config.Service | HttpClient.HttpClient> =
Layer.effect(
Service,
Effect.gen(function* () {
const cfg = yield* Config.Service
const fs = yield* AppFileSystem.Service
const http = HttpClient.filterStatusOk(withTransientReadRetry(yield* HttpClient.HttpClient))
export const layer: Layer.Layer<Service, never, AppFileSystem.Service | Config.Service | HttpClient.HttpClient> =
Layer.effect(
Service,
Effect.gen(function* () {
const cfg = yield* Config.Service
const fs = yield* AppFileSystem.Service
const http = HttpClient.filterStatusOk(withTransientReadRetry(yield* HttpClient.HttpClient))
const state = yield* InstanceState.make(
Effect.fn("Instruction.state")(() =>
Effect.succeed({
// Track which instruction files have already been attached for a given assistant message.
claims: new Map<MessageID, Set<string>>(),
}),
),
)
const relative = Effect.fnUntraced(function* (instruction: string) {
if (!Flag.OPENCODE_DISABLE_PROJECT_CONFIG) {
return yield* fs
.globUp(instruction, Instance.directory, Instance.worktree)
.pipe(Effect.catch(() => Effect.succeed([] as string[])))
}
if (!Flag.OPENCODE_CONFIG_DIR) {
log.warn(
`Skipping relative instruction "${instruction}" - no OPENCODE_CONFIG_DIR set while project config is disabled`,
)
return []
}
return yield* fs
.globUp(instruction, Flag.OPENCODE_CONFIG_DIR, Flag.OPENCODE_CONFIG_DIR)
.pipe(Effect.catch(() => Effect.succeed([] as string[])))
})
const read = Effect.fnUntraced(function* (filepath: string) {
return yield* fs.readFileString(filepath).pipe(Effect.catch(() => Effect.succeed("")))
})
const fetch = Effect.fnUntraced(function* (url: string) {
const res = yield* http.execute(HttpClientRequest.get(url)).pipe(
Effect.timeout(5000),
Effect.catch(() => Effect.succeed(null)),
const state = yield* InstanceState.make(
Effect.fn("Instruction.state")(() =>
Effect.succeed({
// Track which instruction files have already been attached for a given assistant message.
claims: new Map<MessageID, Set<string>>(),
}),
),
)
if (!res) return ""
const body = yield* res.arrayBuffer.pipe(Effect.catch(() => Effect.succeed(new ArrayBuffer(0))))
return new TextDecoder().decode(body)
})
const clear = Effect.fn("Instruction.clear")(function* (messageID: MessageID) {
const s = yield* InstanceState.get(state)
s.claims.delete(messageID)
})
const relative = Effect.fnUntraced(function* (instruction: string) {
if (!Flag.OPENCODE_DISABLE_PROJECT_CONFIG) {
return yield* fs
.globUp(instruction, Instance.directory, Instance.worktree)
.pipe(Effect.catch(() => Effect.succeed([] as string[])))
}
if (!Flag.OPENCODE_CONFIG_DIR) {
log.warn(
`Skipping relative instruction "${instruction}" - no OPENCODE_CONFIG_DIR set while project config is disabled`,
)
return []
}
return yield* fs
.globUp(instruction, Flag.OPENCODE_CONFIG_DIR, Flag.OPENCODE_CONFIG_DIR)
.pipe(Effect.catch(() => Effect.succeed([] as string[])))
})
const systemPaths = Effect.fn("Instruction.systemPaths")(function* () {
const config = yield* cfg.get()
const paths = new Set<string>()
const read = Effect.fnUntraced(function* (filepath: string) {
return yield* fs.readFileString(filepath).pipe(Effect.catch(() => Effect.succeed("")))
})
// The first project-level match wins so we don't stack AGENTS.md/CLAUDE.md from every ancestor.
if (!Flag.OPENCODE_DISABLE_PROJECT_CONFIG) {
for (const file of FILES) {
const matches = yield* fs.findUp(file, Instance.directory, Instance.worktree)
if (matches.length > 0) {
matches.forEach((item) => paths.add(path.resolve(item)))
const fetch = Effect.fnUntraced(function* (url: string) {
const res = yield* http.execute(HttpClientRequest.get(url)).pipe(
Effect.timeout(5000),
Effect.catch(() => Effect.succeed(null)),
)
if (!res) return ""
const body = yield* res.arrayBuffer.pipe(Effect.catch(() => Effect.succeed(new ArrayBuffer(0))))
return new TextDecoder().decode(body)
})
const clear = Effect.fn("Instruction.clear")(function* (messageID: MessageID) {
const s = yield* InstanceState.get(state)
s.claims.delete(messageID)
})
const systemPaths = Effect.fn("Instruction.systemPaths")(function* () {
const config = yield* cfg.get()
const paths = new Set<string>()
// The first project-level match wins so we don't stack AGENTS.md/CLAUDE.md from every ancestor.
if (!Flag.OPENCODE_DISABLE_PROJECT_CONFIG) {
for (const file of FILES) {
const matches = yield* fs.findUp(file, Instance.directory, Instance.worktree)
if (matches.length > 0) {
matches.forEach((item) => paths.add(path.resolve(item)))
break
}
}
}
for (const file of globalFiles()) {
if (yield* fs.existsSafe(file)) {
paths.add(path.resolve(file))
break
}
}
}
for (const file of globalFiles()) {
if (yield* fs.existsSafe(file)) {
paths.add(path.resolve(file))
break
if (config.instructions) {
for (const raw of config.instructions) {
if (raw.startsWith("https://") || raw.startsWith("http://")) continue
const instruction = raw.startsWith("~/") ? path.join(os.homedir(), raw.slice(2)) : raw
const matches = yield* (
path.isAbsolute(instruction)
? fs.glob(path.basename(instruction), {
cwd: path.dirname(instruction),
absolute: true,
include: "file",
})
: relative(instruction)
).pipe(Effect.catch(() => Effect.succeed([] as string[])))
matches.forEach((item) => paths.add(path.resolve(item)))
}
}
}
if (config.instructions) {
for (const raw of config.instructions) {
if (raw.startsWith("https://") || raw.startsWith("http://")) continue
const instruction = raw.startsWith("~/") ? path.join(os.homedir(), raw.slice(2)) : raw
const matches = yield* (
path.isAbsolute(instruction)
? fs.glob(path.basename(instruction), {
cwd: path.dirname(instruction),
absolute: true,
include: "file",
})
: relative(instruction)
).pipe(Effect.catch(() => Effect.succeed([] as string[])))
matches.forEach((item) => paths.add(path.resolve(item)))
return paths
})
const system = Effect.fn("Instruction.system")(function* () {
const config = yield* cfg.get()
const paths = yield* systemPaths()
const urls = (config.instructions ?? []).filter(
(item) => item.startsWith("https://") || item.startsWith("http://"),
)
const files = yield* Effect.forEach(Array.from(paths), read, { concurrency: 8 })
const remote = yield* Effect.forEach(urls, fetch, { concurrency: 4 })
return [
...Array.from(paths).flatMap((item, i) => (files[i] ? [`Instructions from: ${item}\n${files[i]}`] : [])),
...urls.flatMap((item, i) => (remote[i] ? [`Instructions from: ${item}\n${remote[i]}`] : [])),
]
})
const find = Effect.fn("Instruction.find")(function* (dir: string) {
for (const file of FILES) {
const filepath = path.resolve(path.join(dir, file))
if (yield* fs.existsSafe(filepath)) return filepath
}
}
})
return paths
})
const resolve = Effect.fn("Instruction.resolve")(function* (
messages: MessageV2.WithParts[],
filepath: string,
messageID: MessageID,
) {
const sys = yield* systemPaths()
const already = extract(messages)
const results: { filepath: string; content: string }[] = []
const s = yield* InstanceState.get(state)
const system = Effect.fn("Instruction.system")(function* () {
const config = yield* cfg.get()
const paths = yield* systemPaths()
const urls = (config.instructions ?? []).filter(
(item) => item.startsWith("https://") || item.startsWith("http://"),
)
const target = path.resolve(filepath)
const root = path.resolve(Instance.directory)
let current = path.dirname(target)
const files = yield* Effect.forEach(Array.from(paths), read, { concurrency: 8 })
const remote = yield* Effect.forEach(urls, fetch, { concurrency: 4 })
// Walk upward from the file being read and attach nearby instruction files once per message.
while (current.startsWith(root) && current !== root) {
const found = yield* find(current)
if (!found || found === target || sys.has(found) || already.has(found)) {
current = path.dirname(current)
continue
}
return [
...Array.from(paths).flatMap((item, i) => (files[i] ? [`Instructions from: ${item}\n${files[i]}`] : [])),
...urls.flatMap((item, i) => (remote[i] ? [`Instructions from: ${item}\n${remote[i]}`] : [])),
]
})
let set = s.claims.get(messageID)
if (!set) {
set = new Set()
s.claims.set(messageID, set)
}
if (set.has(found)) {
current = path.dirname(current)
continue
}
const find = Effect.fn("Instruction.find")(function* (dir: string) {
for (const file of FILES) {
const filepath = path.resolve(path.join(dir, file))
if (yield* fs.existsSafe(filepath)) return filepath
}
})
set.add(found)
const content = yield* read(found)
if (content) {
results.push({ filepath: found, content: `Instructions from: ${found}\n${content}` })
}
const resolve = Effect.fn("Instruction.resolve")(function* (
messages: MessageV2.WithParts[],
filepath: string,
messageID: MessageID,
) {
const sys = yield* systemPaths()
const already = extract(messages)
const results: { filepath: string; content: string }[] = []
const s = yield* InstanceState.get(state)
const target = path.resolve(filepath)
const root = path.resolve(Instance.directory)
let current = path.dirname(target)
// Walk upward from the file being read and attach nearby instruction files once per message.
while (current.startsWith(root) && current !== root) {
const found = yield* find(current)
if (!found || found === target || sys.has(found) || already.has(found)) {
current = path.dirname(current)
continue
}
let set = s.claims.get(messageID)
if (!set) {
set = new Set()
s.claims.set(messageID, set)
}
if (set.has(found)) {
current = path.dirname(current)
continue
}
return results
})
set.add(found)
const content = yield* read(found)
if (content) {
results.push({ filepath: found, content: `Instructions from: ${found}\n${content}` })
}
return Service.of({ clear, systemPaths, system, find, resolve })
}),
)
current = path.dirname(current)
}
return results
})
return Service.of({ clear, systemPaths, system, find, resolve })
}),
export const defaultLayer = layer.pipe(
Layer.provide(Config.defaultLayer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(FetchHttpClient.layer),
)
export const defaultLayer = layer.pipe(
Layer.provide(Config.defaultLayer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(FetchHttpClient.layer),
)
export function loaded(messages: MessageV2.WithParts[]) {
return extract(messages)
export function loaded(messages: MessageV2.WithParts[]) {
return extract(messages)
}
}
+395 -393
View File
@@ -9,9 +9,9 @@ import { ProviderTransform } from "@/provider/transform"
import { Config } from "@/config"
import { Instance } from "@/project/instance"
import type { Agent } from "@/agent/agent"
import type { MessageV2 } from "."
import type { MessageV2 } from "./message-v2"
import { Plugin } from "@/plugin"
import { SystemPrompt } from "."
import { SystemPrompt } from "./system"
import { Flag } from "@/flag/flag"
import { Permission } from "@/permission"
import { PermissionID } from "@/permission/schema"
@@ -24,428 +24,430 @@ import { EffectBridge } from "@/effect"
import * as Option from "effect/Option"
import * as OtelTracer from "@effect/opentelemetry/Tracer"
const log = Log.create({ service: "llm" })
export const OUTPUT_TOKEN_MAX = ProviderTransform.OUTPUT_TOKEN_MAX
type Result = Awaited<ReturnType<typeof streamText>>
export namespace LLM {
const log = Log.create({ service: "llm" })
export const OUTPUT_TOKEN_MAX = ProviderTransform.OUTPUT_TOKEN_MAX
type Result = Awaited<ReturnType<typeof streamText>>
export type StreamInput = {
user: MessageV2.User
sessionID: string
parentSessionID?: string
model: Provider.Model
agent: Agent.Info
permission?: Permission.Ruleset
system: string[]
messages: ModelMessage[]
small?: boolean
tools: Record<string, Tool>
retries?: number
toolChoice?: "auto" | "required" | "none"
}
export type StreamInput = {
user: MessageV2.User
sessionID: string
parentSessionID?: string
model: Provider.Model
agent: Agent.Info
permission?: Permission.Ruleset
system: string[]
messages: ModelMessage[]
small?: boolean
tools: Record<string, Tool>
retries?: number
toolChoice?: "auto" | "required" | "none"
}
export type StreamRequest = StreamInput & {
abort: AbortSignal
}
export type StreamRequest = StreamInput & {
abort: AbortSignal
}
export type Event = Result["fullStream"] extends AsyncIterable<infer T> ? T : never
export type Event = Result["fullStream"] extends AsyncIterable<infer T> ? T : never
export interface Interface {
readonly stream: (input: StreamInput) => Stream.Stream<Event, unknown>
}
export interface Interface {
readonly stream: (input: StreamInput) => Stream.Stream<Event, unknown>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/LLM") {}
export class Service extends Context.Service<Service, Interface>()("@opencode/LLM") {}
const live: Layer.Layer<
Service,
never,
Auth.Service | Config.Service | Provider.Service | Plugin.Service | Permission.Service
> = Layer.effect(
Service,
Effect.gen(function* () {
const auth = yield* Auth.Service
const config = yield* Config.Service
const provider = yield* Provider.Service
const plugin = yield* Plugin.Service
const perm = yield* Permission.Service
const live: Layer.Layer<
Service,
never,
Auth.Service | Config.Service | Provider.Service | Plugin.Service | Permission.Service
> = Layer.effect(
Service,
Effect.gen(function* () {
const auth = yield* Auth.Service
const config = yield* Config.Service
const provider = yield* Provider.Service
const plugin = yield* Plugin.Service
const perm = yield* Permission.Service
const run = Effect.fn("LLM.run")(function* (input: StreamRequest) {
const l = log
.clone()
.tag("providerID", input.model.providerID)
.tag("modelID", input.model.id)
.tag("sessionID", input.sessionID)
.tag("small", (input.small ?? false).toString())
.tag("agent", input.agent.name)
.tag("mode", input.agent.mode)
l.info("stream", {
modelID: input.model.id,
providerID: input.model.providerID,
})
const run = Effect.fn("LLM.run")(function* (input: StreamRequest) {
const l = log
.clone()
.tag("providerID", input.model.providerID)
.tag("modelID", input.model.id)
.tag("sessionID", input.sessionID)
.tag("small", (input.small ?? false).toString())
.tag("agent", input.agent.name)
.tag("mode", input.agent.mode)
l.info("stream", {
modelID: input.model.id,
providerID: input.model.providerID,
})
const [language, cfg, item, info] = yield* Effect.all(
[
provider.getLanguage(input.model),
config.get(),
provider.getProvider(input.model.providerID),
auth.get(input.model.providerID),
],
{ concurrency: "unbounded" },
)
const [language, cfg, item, info] = yield* Effect.all(
[
provider.getLanguage(input.model),
config.get(),
provider.getProvider(input.model.providerID),
auth.get(input.model.providerID),
],
{ concurrency: "unbounded" },
)
// TODO: move this to a proper hook
const isOpenaiOauth = item.id === "openai" && info?.type === "oauth"
// TODO: move this to a proper hook
const isOpenaiOauth = item.id === "openai" && info?.type === "oauth"
const system: string[] = []
system.push(
[
// use agent prompt otherwise provider prompt
...(input.agent.prompt ? [input.agent.prompt] : SystemPrompt.provider(input.model)),
// any custom prompt passed into this call
...input.system,
// any custom prompt from last user message
...(input.user.system ? [input.user.system] : []),
]
.filter((x) => x)
.join("\n"),
)
const system: string[] = []
system.push(
[
// use agent prompt otherwise provider prompt
...(input.agent.prompt ? [input.agent.prompt] : SystemPrompt.provider(input.model)),
// any custom prompt passed into this call
...input.system,
// any custom prompt from last user message
...(input.user.system ? [input.user.system] : []),
]
.filter((x) => x)
.join("\n"),
)
const header = system[0]
yield* plugin.trigger(
"experimental.chat.system.transform",
{ sessionID: input.sessionID, model: input.model },
{ system },
)
// rejoin to maintain 2-part structure for caching if header unchanged
if (system.length > 2 && system[0] === header) {
const rest = system.slice(1)
system.length = 0
system.push(header, rest.join("\n"))
}
const header = system[0]
yield* plugin.trigger(
"experimental.chat.system.transform",
{ sessionID: input.sessionID, model: input.model },
{ system },
)
// rejoin to maintain 2-part structure for caching if header unchanged
if (system.length > 2 && system[0] === header) {
const rest = system.slice(1)
system.length = 0
system.push(header, rest.join("\n"))
}
const variant =
!input.small && input.model.variants && input.user.model.variant
? input.model.variants[input.user.model.variant]
: {}
const base = input.small
? ProviderTransform.smallOptions(input.model)
: ProviderTransform.options({
model: input.model,
sessionID: input.sessionID,
providerOptions: item.options,
})
const options: Record<string, any> = pipe(
base,
mergeDeep(input.model.options),
mergeDeep(input.agent.options),
mergeDeep(variant),
)
if (isOpenaiOauth) {
options.instructions = system.join("\n")
}
const variant =
!input.small && input.model.variants && input.user.model.variant
? input.model.variants[input.user.model.variant]
: {}
const base = input.small
? ProviderTransform.smallOptions(input.model)
: ProviderTransform.options({
model: input.model,
sessionID: input.sessionID,
providerOptions: item.options,
})
const options: Record<string, any> = pipe(
base,
mergeDeep(input.model.options),
mergeDeep(input.agent.options),
mergeDeep(variant),
)
if (isOpenaiOauth) {
options.instructions = system.join("\n")
}
const isWorkflow = language instanceof GitLabWorkflowLanguageModel
const messages = isOpenaiOauth
? input.messages
: isWorkflow
const isWorkflow = language instanceof GitLabWorkflowLanguageModel
const messages = isOpenaiOauth
? input.messages
: [
...system.map(
(x): ModelMessage => ({
role: "system",
content: x,
}),
),
...input.messages,
]
: isWorkflow
? input.messages
: [
...system.map(
(x): ModelMessage => ({
role: "system",
content: x,
}),
),
...input.messages,
]
const params = yield* plugin.trigger(
"chat.params",
{
sessionID: input.sessionID,
agent: input.agent.name,
model: input.model,
provider: item,
message: input.user,
},
{
temperature: input.model.capabilities.temperature
? (input.agent.temperature ?? ProviderTransform.temperature(input.model))
: undefined,
topP: input.agent.topP ?? ProviderTransform.topP(input.model),
topK: ProviderTransform.topK(input.model),
maxOutputTokens: ProviderTransform.maxOutputTokens(input.model),
options,
},
)
const params = yield* plugin.trigger(
"chat.params",
{
sessionID: input.sessionID,
agent: input.agent.name,
model: input.model,
provider: item,
message: input.user,
},
{
temperature: input.model.capabilities.temperature
? (input.agent.temperature ?? ProviderTransform.temperature(input.model))
: undefined,
topP: input.agent.topP ?? ProviderTransform.topP(input.model),
topK: ProviderTransform.topK(input.model),
maxOutputTokens: ProviderTransform.maxOutputTokens(input.model),
options,
},
)
const { headers } = yield* plugin.trigger(
"chat.headers",
{
sessionID: input.sessionID,
agent: input.agent.name,
model: input.model,
provider: item,
message: input.user,
},
{
headers: {},
},
)
const { headers } = yield* plugin.trigger(
"chat.headers",
{
sessionID: input.sessionID,
agent: input.agent.name,
model: input.model,
provider: item,
message: input.user,
},
{
headers: {},
},
)
const tools = resolveTools(input)
const tools = resolveTools(input)
// LiteLLM and some Anthropic proxies require the tools parameter to be present
// when message history contains tool calls, even if no tools are being used.
// Add a dummy tool that is never called to satisfy this validation.
// This is enabled for:
// 1. Providers with "litellm" in their ID or API ID (auto-detected)
// 2. Providers with explicit "litellmProxy: true" option (opt-in for custom gateways)
const isLiteLLMProxy =
item.options?.["litellmProxy"] === true ||
input.model.providerID.toLowerCase().includes("litellm") ||
input.model.api.id.toLowerCase().includes("litellm")
// LiteLLM and some Anthropic proxies require the tools parameter to be present
// when message history contains tool calls, even if no tools are being used.
// Add a dummy tool that is never called to satisfy this validation.
// This is enabled for:
// 1. Providers with "litellm" in their ID or API ID (auto-detected)
// 2. Providers with explicit "litellmProxy: true" option (opt-in for custom gateways)
const isLiteLLMProxy =
item.options?.["litellmProxy"] === true ||
input.model.providerID.toLowerCase().includes("litellm") ||
input.model.api.id.toLowerCase().includes("litellm")
// LiteLLM/Bedrock rejects requests where the message history contains tool
// calls but no tools param is present. When there are no active tools (e.g.
// during compaction), inject a stub tool to satisfy the validation requirement.
// The stub description explicitly tells the model not to call it.
if (
(isLiteLLMProxy || input.model.providerID.includes("github-copilot")) &&
Object.keys(tools).length === 0 &&
hasToolCalls(input.messages)
) {
tools["_noop"] = tool({
description: "Do not call this tool. It exists only for API compatibility and must never be invoked.",
inputSchema: jsonSchema({
type: "object",
properties: {
reason: { type: "string", description: "Unused" },
},
}),
execute: async () => ({ output: "", title: "", metadata: {} }),
})
}
// Wire up toolExecutor for DWS workflow models so that tool calls
// from the workflow service are executed via opencode's tool system
// and results sent back over the WebSocket.
if (language instanceof GitLabWorkflowLanguageModel) {
const workflowModel = language as GitLabWorkflowLanguageModel & {
sessionID?: string
sessionPreapprovedTools?: string[]
approvalHandler?: (approvalTools: { name: string; args: string }[]) => Promise<{ approved: boolean }>
}
workflowModel.sessionID = input.sessionID
workflowModel.systemPrompt = system.join("\n")
workflowModel.toolExecutor = async (toolName, argsJson, _requestID) => {
const t = tools[toolName]
if (!t || !t.execute) {
return { result: "", error: `Unknown tool: ${toolName}` }
}
try {
const result = await t.execute!(JSON.parse(argsJson), {
toolCallId: _requestID,
messages: input.messages,
abortSignal: input.abort,
})
const output = typeof result === "string" ? result : (result?.output ?? JSON.stringify(result))
return {
result: output,
metadata: typeof result === "object" ? result?.metadata : undefined,
title: typeof result === "object" ? result?.title : undefined,
}
} catch (e: any) {
return { result: "", error: e.message ?? String(e) }
}
}
const ruleset = Permission.merge(input.agent.permission ?? [], input.permission ?? [])
workflowModel.sessionPreapprovedTools = Object.keys(tools).filter((name) => {
const match = ruleset.findLast((rule) => Wildcard.match(name, rule.permission))
return !match || match.action !== "ask"
})
const bridge = yield* EffectBridge.make()
const approvedToolsForSession = new Set<string>()
workflowModel.approvalHandler = Instance.bind(async (approvalTools) => {
const uniqueNames = [...new Set(approvalTools.map((t: { name: string }) => t.name))] as string[]
// Auto-approve tools that were already approved in this session
// (prevents infinite approval loops for server-side MCP tools)
if (uniqueNames.every((name) => approvedToolsForSession.has(name))) {
return { approved: true }
}
const id = PermissionID.ascending()
let reply: Permission.Reply | undefined
let unsub: (() => void) | undefined
try {
unsub = Bus.subscribe(Permission.Event.Replied, (evt) => {
if (evt.properties.requestID === id) reply = evt.properties.reply
})
const toolPatterns = approvalTools.map((t: { name: string; args: string }) => {
try {
const parsed = JSON.parse(t.args) as Record<string, unknown>
const title = (parsed?.title ?? parsed?.name ?? "") as string
return title ? `${t.name}: ${title}` : t.name
} catch {
return t.name
}
})
const uniquePatterns = [...new Set(toolPatterns)] as string[]
await bridge.promise(
perm.ask({
id,
sessionID: SessionID.make(input.sessionID),
permission: "workflow_tool_approval",
patterns: uniquePatterns,
metadata: { tools: approvalTools },
always: uniquePatterns,
ruleset: [],
}),
)
for (const name of uniqueNames) approvedToolsForSession.add(name)
workflowModel.sessionPreapprovedTools = [...(workflowModel.sessionPreapprovedTools ?? []), ...uniqueNames]
return { approved: true }
} catch {
return { approved: false }
} finally {
unsub?.()
}
})
}
const tracer = cfg.experimental?.openTelemetry
? Option.getOrUndefined(yield* Effect.serviceOption(OtelTracer.OtelTracer))
: undefined
return streamText({
onError(error) {
l.error("stream error", {
error,
// LiteLLM/Bedrock rejects requests where the message history contains tool
// calls but no tools param is present. When there are no active tools (e.g.
// during compaction), inject a stub tool to satisfy the validation requirement.
// The stub description explicitly tells the model not to call it.
if (
(isLiteLLMProxy || input.model.providerID.includes("github-copilot")) &&
Object.keys(tools).length === 0 &&
hasToolCalls(input.messages)
) {
tools["_noop"] = tool({
description: "Do not call this tool. It exists only for API compatibility and must never be invoked.",
inputSchema: jsonSchema({
type: "object",
properties: {
reason: { type: "string", description: "Unused" },
},
}),
execute: async () => ({ output: "", title: "", metadata: {} }),
})
},
async experimental_repairToolCall(failed) {
const lower = failed.toolCall.toolName.toLowerCase()
if (lower !== failed.toolCall.toolName && tools[lower]) {
l.info("repairing tool call", {
tool: failed.toolCall.toolName,
repaired: lower,
}
// Wire up toolExecutor for DWS workflow models so that tool calls
// from the workflow service are executed via opencode's tool system
// and results sent back over the WebSocket.
if (language instanceof GitLabWorkflowLanguageModel) {
const workflowModel = language as GitLabWorkflowLanguageModel & {
sessionID?: string
sessionPreapprovedTools?: string[]
approvalHandler?: (approvalTools: { name: string; args: string }[]) => Promise<{ approved: boolean }>
}
workflowModel.sessionID = input.sessionID
workflowModel.systemPrompt = system.join("\n")
workflowModel.toolExecutor = async (toolName, argsJson, _requestID) => {
const t = tools[toolName]
if (!t || !t.execute) {
return { result: "", error: `Unknown tool: ${toolName}` }
}
try {
const result = await t.execute!(JSON.parse(argsJson), {
toolCallId: _requestID,
messages: input.messages,
abortSignal: input.abort,
})
const output = typeof result === "string" ? result : (result?.output ?? JSON.stringify(result))
return {
result: output,
metadata: typeof result === "object" ? result?.metadata : undefined,
title: typeof result === "object" ? result?.title : undefined,
}
} catch (e: any) {
return { result: "", error: e.message ?? String(e) }
}
}
const ruleset = Permission.merge(input.agent.permission ?? [], input.permission ?? [])
workflowModel.sessionPreapprovedTools = Object.keys(tools).filter((name) => {
const match = ruleset.findLast((rule) => Wildcard.match(name, rule.permission))
return !match || match.action !== "ask"
})
const bridge = yield* EffectBridge.make()
const approvedToolsForSession = new Set<string>()
workflowModel.approvalHandler = Instance.bind(async (approvalTools) => {
const uniqueNames = [...new Set(approvalTools.map((t: { name: string }) => t.name))] as string[]
// Auto-approve tools that were already approved in this session
// (prevents infinite approval loops for server-side MCP tools)
if (uniqueNames.every((name) => approvedToolsForSession.has(name))) {
return { approved: true }
}
const id = PermissionID.ascending()
let reply: Permission.Reply | undefined
let unsub: (() => void) | undefined
try {
unsub = Bus.subscribe(Permission.Event.Replied, (evt) => {
if (evt.properties.requestID === id) reply = evt.properties.reply
})
const toolPatterns = approvalTools.map((t: { name: string; args: string }) => {
try {
const parsed = JSON.parse(t.args) as Record<string, unknown>
const title = (parsed?.title ?? parsed?.name ?? "") as string
return title ? `${t.name}: ${title}` : t.name
} catch {
return t.name
}
})
const uniquePatterns = [...new Set(toolPatterns)] as string[]
await bridge.promise(
perm.ask({
id,
sessionID: SessionID.make(input.sessionID),
permission: "workflow_tool_approval",
patterns: uniquePatterns,
metadata: { tools: approvalTools },
always: uniquePatterns,
ruleset: [],
}),
)
for (const name of uniqueNames) approvedToolsForSession.add(name)
workflowModel.sessionPreapprovedTools = [...(workflowModel.sessionPreapprovedTools ?? []), ...uniqueNames]
return { approved: true }
} catch {
return { approved: false }
} finally {
unsub?.()
}
})
}
const tracer = cfg.experimental?.openTelemetry
? Option.getOrUndefined(yield* Effect.serviceOption(OtelTracer.OtelTracer))
: undefined
return streamText({
onError(error) {
l.error("stream error", {
error,
})
},
async experimental_repairToolCall(failed) {
const lower = failed.toolCall.toolName.toLowerCase()
if (lower !== failed.toolCall.toolName && tools[lower]) {
l.info("repairing tool call", {
tool: failed.toolCall.toolName,
repaired: lower,
})
return {
...failed.toolCall,
toolName: lower,
}
}
return {
...failed.toolCall,
toolName: lower,
}
}
return {
...failed.toolCall,
input: JSON.stringify({
tool: failed.toolCall.toolName,
error: failed.error.message,
}),
toolName: "invalid",
}
},
temperature: params.temperature,
topP: params.topP,
topK: params.topK,
providerOptions: ProviderTransform.providerOptions(input.model, params.options),
activeTools: Object.keys(tools).filter((x) => x !== "invalid"),
tools,
toolChoice: input.toolChoice,
maxOutputTokens: params.maxOutputTokens,
abortSignal: input.abort,
headers: {
...(input.model.providerID.startsWith("opencode")
? {
"x-opencode-project": Instance.project.id,
"x-opencode-session": input.sessionID,
"x-opencode-request": input.user.id,
"x-opencode-client": Flag.OPENCODE_CLIENT,
}
: {
"x-session-affinity": input.sessionID,
...(input.parentSessionID ? { "x-parent-session-id": input.parentSessionID } : {}),
"User-Agent": `opencode/${Installation.VERSION}`,
input: JSON.stringify({
tool: failed.toolCall.toolName,
error: failed.error.message,
}),
...input.model.headers,
...headers,
},
maxRetries: input.retries ?? 0,
messages,
model: wrapLanguageModel({
model: language,
middleware: [
{
specificationVersion: "v3" as const,
async transformParams(args) {
if (args.type === "stream") {
// @ts-expect-error
args.params.prompt = ProviderTransform.message(args.params.prompt, input.model, options)
}
return args.params
},
},
],
}),
experimental_telemetry: {
isEnabled: cfg.experimental?.openTelemetry,
functionId: "session.llm",
tracer,
metadata: {
userId: cfg.username ?? "unknown",
sessionId: input.sessionID,
toolName: "invalid",
}
},
},
})
})
const stream: Interface["stream"] = (input) =>
Stream.scoped(
Stream.unwrap(
Effect.gen(function* () {
const ctrl = yield* Effect.acquireRelease(
Effect.sync(() => new AbortController()),
(ctrl) => Effect.sync(() => ctrl.abort()),
)
const result = yield* run({ ...input, abort: ctrl.signal })
return Stream.fromAsyncIterable(result.fullStream, (e) => (e instanceof Error ? e : new Error(String(e))))
temperature: params.temperature,
topP: params.topP,
topK: params.topK,
providerOptions: ProviderTransform.providerOptions(input.model, params.options),
activeTools: Object.keys(tools).filter((x) => x !== "invalid"),
tools,
toolChoice: input.toolChoice,
maxOutputTokens: params.maxOutputTokens,
abortSignal: input.abort,
headers: {
...(input.model.providerID.startsWith("opencode")
? {
"x-opencode-project": Instance.project.id,
"x-opencode-session": input.sessionID,
"x-opencode-request": input.user.id,
"x-opencode-client": Flag.OPENCODE_CLIENT,
}
: {
"x-session-affinity": input.sessionID,
...(input.parentSessionID ? { "x-parent-session-id": input.parentSessionID } : {}),
"User-Agent": `opencode/${Installation.VERSION}`,
}),
...input.model.headers,
...headers,
},
maxRetries: input.retries ?? 0,
messages,
model: wrapLanguageModel({
model: language,
middleware: [
{
specificationVersion: "v3" as const,
async transformParams(args) {
if (args.type === "stream") {
// @ts-expect-error
args.params.prompt = ProviderTransform.message(args.params.prompt, input.model, options)
}
return args.params
},
},
],
}),
),
)
experimental_telemetry: {
isEnabled: cfg.experimental?.openTelemetry,
functionId: "session.llm",
tracer,
metadata: {
userId: cfg.username ?? "unknown",
sessionId: input.sessionID,
},
},
})
})
return Service.of({ stream })
}),
)
const stream: Interface["stream"] = (input) =>
Stream.scoped(
Stream.unwrap(
Effect.gen(function* () {
const ctrl = yield* Effect.acquireRelease(
Effect.sync(() => new AbortController()),
(ctrl) => Effect.sync(() => ctrl.abort()),
)
export const layer = live.pipe(Layer.provide(Permission.defaultLayer))
const result = yield* run({ ...input, abort: ctrl.signal })
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(Config.defaultLayer),
Layer.provide(Provider.defaultLayer),
Layer.provide(Plugin.defaultLayer),
),
)
return Stream.fromAsyncIterable(result.fullStream, (e) => (e instanceof Error ? e : new Error(String(e))))
}),
),
)
function resolveTools(input: Pick<StreamInput, "tools" | "agent" | "permission" | "user">) {
const disabled = Permission.disabled(
Object.keys(input.tools),
Permission.merge(input.agent.permission, input.permission ?? []),
return Service.of({ stream })
}),
)
return Record.filter(input.tools, (_, k) => input.user.tools?.[k] !== false && !disabled.has(k))
}
// Check if messages contain any tool-call content
// Used to determine if a dummy tool should be added for LiteLLM proxy compatibility
export function hasToolCalls(messages: ModelMessage[]): boolean {
for (const msg of messages) {
if (!Array.isArray(msg.content)) continue
for (const part of msg.content) {
if (part.type === "tool-call" || part.type === "tool-result") return true
}
export const layer = live.pipe(Layer.provide(Permission.defaultLayer))
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(Config.defaultLayer),
Layer.provide(Provider.defaultLayer),
Layer.provide(Plugin.defaultLayer),
),
)
function resolveTools(input: Pick<StreamInput, "tools" | "agent" | "permission" | "user">) {
const disabled = Permission.disabled(
Object.keys(input.tools),
Permission.merge(input.agent.permission, input.permission ?? []),
)
return Record.filter(input.tools, (_, k) => input.user.tools?.[k] !== false && !disabled.has(k))
}
// Check if messages contain any tool-call content
// Used to determine if a dummy tool should be added for LiteLLM proxy compatibility
export function hasToolCalls(messages: ModelMessage[]): boolean {
for (const msg of messages) {
if (!Array.isArray(msg.content)) continue
for (const part of msg.content) {
if (part.type === "tool-call" || part.type === "tool-result") return true
}
}
return false
}
return false
}
File diff suppressed because it is too large Load Diff
+171 -169
View File
@@ -3,187 +3,189 @@ import { SessionID } from "./schema"
import { ModelID, ProviderID } from "../provider/schema"
import { NamedError } from "@opencode-ai/shared/util/error"
export const OutputLengthError = NamedError.create("MessageOutputLengthError", z.object({}))
export const AuthError = NamedError.create(
"ProviderAuthError",
z.object({
providerID: z.string(),
message: z.string(),
}),
)
export namespace Message {
export const OutputLengthError = NamedError.create("MessageOutputLengthError", z.object({}))
export const AuthError = NamedError.create(
"ProviderAuthError",
z.object({
providerID: z.string(),
message: z.string(),
}),
)
export const ToolCall = z
.object({
state: z.literal("call"),
step: z.number().optional(),
toolCallId: z.string(),
toolName: z.string(),
args: z.custom<Required<unknown>>(),
})
.meta({
ref: "ToolCall",
})
export type ToolCall = z.infer<typeof ToolCall>
export const ToolCall = z
.object({
state: z.literal("call"),
step: z.number().optional(),
toolCallId: z.string(),
toolName: z.string(),
args: z.custom<Required<unknown>>(),
})
.meta({
ref: "ToolCall",
})
export type ToolCall = z.infer<typeof ToolCall>
export const ToolPartialCall = z
.object({
state: z.literal("partial-call"),
step: z.number().optional(),
toolCallId: z.string(),
toolName: z.string(),
args: z.custom<Required<unknown>>(),
})
.meta({
ref: "ToolPartialCall",
})
export type ToolPartialCall = z.infer<typeof ToolPartialCall>
export const ToolPartialCall = z
.object({
state: z.literal("partial-call"),
step: z.number().optional(),
toolCallId: z.string(),
toolName: z.string(),
args: z.custom<Required<unknown>>(),
})
.meta({
ref: "ToolPartialCall",
})
export type ToolPartialCall = z.infer<typeof ToolPartialCall>
export const ToolResult = z
.object({
state: z.literal("result"),
step: z.number().optional(),
toolCallId: z.string(),
toolName: z.string(),
args: z.custom<Required<unknown>>(),
result: z.string(),
})
.meta({
ref: "ToolResult",
})
export type ToolResult = z.infer<typeof ToolResult>
export const ToolResult = z
.object({
state: z.literal("result"),
step: z.number().optional(),
toolCallId: z.string(),
toolName: z.string(),
args: z.custom<Required<unknown>>(),
result: z.string(),
})
.meta({
ref: "ToolResult",
})
export type ToolResult = z.infer<typeof ToolResult>
export const ToolInvocation = z.discriminatedUnion("state", [ToolCall, ToolPartialCall, ToolResult]).meta({
ref: "ToolInvocation",
})
export type ToolInvocation = z.infer<typeof ToolInvocation>
export const ToolInvocation = z.discriminatedUnion("state", [ToolCall, ToolPartialCall, ToolResult]).meta({
ref: "ToolInvocation",
})
export type ToolInvocation = z.infer<typeof ToolInvocation>
export const TextPart = z
.object({
type: z.literal("text"),
text: z.string(),
})
.meta({
ref: "TextPart",
})
export type TextPart = z.infer<typeof TextPart>
export const TextPart = z
.object({
type: z.literal("text"),
text: z.string(),
})
.meta({
ref: "TextPart",
})
export type TextPart = z.infer<typeof TextPart>
export const ReasoningPart = z
.object({
type: z.literal("reasoning"),
text: z.string(),
providerMetadata: z.record(z.string(), z.any()).optional(),
})
.meta({
ref: "ReasoningPart",
})
export type ReasoningPart = z.infer<typeof ReasoningPart>
export const ReasoningPart = z
.object({
type: z.literal("reasoning"),
text: z.string(),
providerMetadata: z.record(z.string(), z.any()).optional(),
})
.meta({
ref: "ReasoningPart",
})
export type ReasoningPart = z.infer<typeof ReasoningPart>
export const ToolInvocationPart = z
.object({
type: z.literal("tool-invocation"),
toolInvocation: ToolInvocation,
})
.meta({
ref: "ToolInvocationPart",
})
export type ToolInvocationPart = z.infer<typeof ToolInvocationPart>
export const ToolInvocationPart = z
.object({
type: z.literal("tool-invocation"),
toolInvocation: ToolInvocation,
})
.meta({
ref: "ToolInvocationPart",
})
export type ToolInvocationPart = z.infer<typeof ToolInvocationPart>
export const SourceUrlPart = z
.object({
type: z.literal("source-url"),
sourceId: z.string(),
url: z.string(),
title: z.string().optional(),
providerMetadata: z.record(z.string(), z.any()).optional(),
})
.meta({
ref: "SourceUrlPart",
})
export type SourceUrlPart = z.infer<typeof SourceUrlPart>
export const SourceUrlPart = z
.object({
type: z.literal("source-url"),
sourceId: z.string(),
url: z.string(),
title: z.string().optional(),
providerMetadata: z.record(z.string(), z.any()).optional(),
})
.meta({
ref: "SourceUrlPart",
})
export type SourceUrlPart = z.infer<typeof SourceUrlPart>
export const FilePart = z
.object({
type: z.literal("file"),
mediaType: z.string(),
filename: z.string().optional(),
url: z.string(),
})
.meta({
ref: "FilePart",
})
export type FilePart = z.infer<typeof FilePart>
export const FilePart = z
.object({
type: z.literal("file"),
mediaType: z.string(),
filename: z.string().optional(),
url: z.string(),
})
.meta({
ref: "FilePart",
})
export type FilePart = z.infer<typeof FilePart>
export const StepStartPart = z
.object({
type: z.literal("step-start"),
})
.meta({
ref: "StepStartPart",
})
export type StepStartPart = z.infer<typeof StepStartPart>
export const StepStartPart = z
.object({
type: z.literal("step-start"),
})
.meta({
ref: "StepStartPart",
})
export type StepStartPart = z.infer<typeof StepStartPart>
export const MessagePart = z
.discriminatedUnion("type", [TextPart, ReasoningPart, ToolInvocationPart, SourceUrlPart, FilePart, StepStartPart])
.meta({
ref: "MessagePart",
})
export type MessagePart = z.infer<typeof MessagePart>
export const MessagePart = z
.discriminatedUnion("type", [TextPart, ReasoningPart, ToolInvocationPart, SourceUrlPart, FilePart, StepStartPart])
.meta({
ref: "MessagePart",
})
export type MessagePart = z.infer<typeof MessagePart>
export const Info = z
.object({
id: z.string(),
role: z.enum(["user", "assistant"]),
parts: z.array(MessagePart),
metadata: z
.object({
time: z.object({
created: z.number(),
completed: z.number().optional(),
}),
error: z
.discriminatedUnion("name", [AuthError.Schema, NamedError.Unknown.Schema, OutputLengthError.Schema])
.optional(),
sessionID: SessionID.zod,
tool: z.record(
z.string(),
z
export const Info = z
.object({
id: z.string(),
role: z.enum(["user", "assistant"]),
parts: z.array(MessagePart),
metadata: z
.object({
time: z.object({
created: z.number(),
completed: z.number().optional(),
}),
error: z
.discriminatedUnion("name", [AuthError.Schema, NamedError.Unknown.Schema, OutputLengthError.Schema])
.optional(),
sessionID: SessionID.zod,
tool: z.record(
z.string(),
z
.object({
title: z.string(),
snapshot: z.string().optional(),
time: z.object({
start: z.number(),
end: z.number(),
}),
})
.catchall(z.any()),
),
assistant: z
.object({
title: z.string(),
snapshot: z.string().optional(),
time: z.object({
start: z.number(),
end: z.number(),
system: z.string().array(),
modelID: ModelID.zod,
providerID: ProviderID.zod,
path: z.object({
cwd: z.string(),
root: z.string(),
}),
cost: z.number(),
summary: z.boolean().optional(),
tokens: z.object({
input: z.number(),
output: z.number(),
reasoning: z.number(),
cache: z.object({
read: z.number(),
write: z.number(),
}),
}),
})
.catchall(z.any()),
),
assistant: z
.object({
system: z.string().array(),
modelID: ModelID.zod,
providerID: ProviderID.zod,
path: z.object({
cwd: z.string(),
root: z.string(),
}),
cost: z.number(),
summary: z.boolean().optional(),
tokens: z.object({
input: z.number(),
output: z.number(),
reasoning: z.number(),
cache: z.object({
read: z.number(),
write: z.number(),
}),
}),
})
.optional(),
snapshot: z.string().optional(),
})
.meta({ ref: "MessageMetadata" }),
})
.meta({
ref: "Message",
})
export type Info = z.infer<typeof Info>
.optional(),
snapshot: z.string().optional(),
})
.meta({ ref: "MessageMetadata" }),
})
.meta({
ref: "Message",
})
export type Info = z.infer<typeof Info>
}
+1 -1
View File
@@ -1,7 +1,7 @@
import type { Config } from "@/config"
import type { Provider } from "@/provider"
import { ProviderTransform } from "@/provider/transform"
import type { MessageV2 } from "."
import type { MessageV2 } from "./message-v2"
const COMPACTION_BUFFER = 20_000
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -1,7 +1,7 @@
import { NotFoundError, eq, and } from "../storage/db"
import { SyncEvent } from "@/sync"
import { Session } from "."
import { MessageV2 } from "."
import { MessageV2 } from "./message-v2"
import { SessionTable, MessageTable, PartTable } from "./session.sql"
import { Log } from "../util/log"
File diff suppressed because it is too large Load Diff
+105 -103
View File
@@ -1,123 +1,125 @@
import type { NamedError } from "@opencode-ai/shared/util/error"
import { Cause, Clock, Duration, Effect, Schedule } from "effect"
import { MessageV2 } from "."
import { MessageV2 } from "./message-v2"
import { iife } from "@/util/iife"
export type Err = ReturnType<NamedError["toObject"]>
export namespace SessionRetry {
export type Err = ReturnType<NamedError["toObject"]>
// This exported message is shared with the TUI upsell detector. Matching on a
// literal error string kind of sucks, but it is the simplest for now.
export const GO_UPSELL_MESSAGE = "Free usage exceeded, subscribe to Go https://opencode.ai/go"
// This exported message is shared with the TUI upsell detector. Matching on a
// literal error string kind of sucks, but it is the simplest for now.
export const GO_UPSELL_MESSAGE = "Free usage exceeded, subscribe to Go https://opencode.ai/go"
export const RETRY_INITIAL_DELAY = 2000
export const RETRY_BACKOFF_FACTOR = 2
export const RETRY_MAX_DELAY_NO_HEADERS = 30_000 // 30 seconds
export const RETRY_MAX_DELAY = 2_147_483_647 // max 32-bit signed integer for setTimeout
export const RETRY_INITIAL_DELAY = 2000
export const RETRY_BACKOFF_FACTOR = 2
export const RETRY_MAX_DELAY_NO_HEADERS = 30_000 // 30 seconds
export const RETRY_MAX_DELAY = 2_147_483_647 // max 32-bit signed integer for setTimeout
function cap(ms: number) {
return Math.min(ms, RETRY_MAX_DELAY)
}
function cap(ms: number) {
return Math.min(ms, RETRY_MAX_DELAY)
}
export function delay(attempt: number, error?: MessageV2.APIError) {
if (error) {
const headers = error.data.responseHeaders
if (headers) {
const retryAfterMs = headers["retry-after-ms"]
if (retryAfterMs) {
const parsedMs = Number.parseFloat(retryAfterMs)
if (!Number.isNaN(parsedMs)) {
return cap(parsedMs)
export function delay(attempt: number, error?: MessageV2.APIError) {
if (error) {
const headers = error.data.responseHeaders
if (headers) {
const retryAfterMs = headers["retry-after-ms"]
if (retryAfterMs) {
const parsedMs = Number.parseFloat(retryAfterMs)
if (!Number.isNaN(parsedMs)) {
return cap(parsedMs)
}
}
}
const retryAfter = headers["retry-after"]
if (retryAfter) {
const parsedSeconds = Number.parseFloat(retryAfter)
if (!Number.isNaN(parsedSeconds)) {
// convert seconds to milliseconds
return cap(Math.ceil(parsedSeconds * 1000))
const retryAfter = headers["retry-after"]
if (retryAfter) {
const parsedSeconds = Number.parseFloat(retryAfter)
if (!Number.isNaN(parsedSeconds)) {
// convert seconds to milliseconds
return cap(Math.ceil(parsedSeconds * 1000))
}
// Try parsing as HTTP date format
const parsed = Date.parse(retryAfter) - Date.now()
if (!Number.isNaN(parsed) && parsed > 0) {
return cap(Math.ceil(parsed))
}
}
// Try parsing as HTTP date format
const parsed = Date.parse(retryAfter) - Date.now()
if (!Number.isNaN(parsed) && parsed > 0) {
return cap(Math.ceil(parsed))
return cap(RETRY_INITIAL_DELAY * Math.pow(RETRY_BACKOFF_FACTOR, attempt - 1))
}
}
return cap(Math.min(RETRY_INITIAL_DELAY * Math.pow(RETRY_BACKOFF_FACTOR, attempt - 1), RETRY_MAX_DELAY_NO_HEADERS))
}
export function retryable(error: Err) {
// context overflow errors should not be retried
if (MessageV2.ContextOverflowError.isInstance(error)) return undefined
if (MessageV2.APIError.isInstance(error)) {
const status = error.data.statusCode
// 5xx errors are transient server failures and should always be retried,
// even when the provider SDK doesn't explicitly mark them as retryable.
if (!error.data.isRetryable && !(status !== undefined && status >= 500)) return undefined
if (error.data.responseBody?.includes("FreeUsageLimitError")) return GO_UPSELL_MESSAGE
return error.data.message.includes("Overloaded") ? "Provider is overloaded" : error.data.message
}
// Check for rate limit patterns in plain text error messages
const msg = error.data?.message
if (typeof msg === "string") {
const lower = msg.toLowerCase()
if (
lower.includes("rate increased too quickly") ||
lower.includes("rate limit") ||
lower.includes("too many requests")
) {
return msg
}
}
const json = iife(() => {
try {
if (typeof error.data?.message === "string") {
const parsed = JSON.parse(error.data.message)
return parsed
}
return JSON.parse(error.data.message)
} catch {
return undefined
}
})
if (!json || typeof json !== "object") return undefined
const code = typeof json.code === "string" ? json.code : ""
return cap(RETRY_INITIAL_DELAY * Math.pow(RETRY_BACKOFF_FACTOR, attempt - 1))
if (json.type === "error" && json.error?.type === "too_many_requests") {
return "Too Many Requests"
}
}
return cap(Math.min(RETRY_INITIAL_DELAY * Math.pow(RETRY_BACKOFF_FACTOR, attempt - 1), RETRY_MAX_DELAY_NO_HEADERS))
}
export function retryable(error: Err) {
// context overflow errors should not be retried
if (MessageV2.ContextOverflowError.isInstance(error)) return undefined
if (MessageV2.APIError.isInstance(error)) {
const status = error.data.statusCode
// 5xx errors are transient server failures and should always be retried,
// even when the provider SDK doesn't explicitly mark them as retryable.
if (!error.data.isRetryable && !(status !== undefined && status >= 500)) return undefined
if (error.data.responseBody?.includes("FreeUsageLimitError")) return GO_UPSELL_MESSAGE
return error.data.message.includes("Overloaded") ? "Provider is overloaded" : error.data.message
}
// Check for rate limit patterns in plain text error messages
const msg = error.data?.message
if (typeof msg === "string") {
const lower = msg.toLowerCase()
if (
lower.includes("rate increased too quickly") ||
lower.includes("rate limit") ||
lower.includes("too many requests")
) {
return msg
if (code.includes("exhausted") || code.includes("unavailable")) {
return "Provider is overloaded"
}
}
const json = iife(() => {
try {
if (typeof error.data?.message === "string") {
const parsed = JSON.parse(error.data.message)
return parsed
}
return JSON.parse(error.data.message)
} catch {
return undefined
if (json.type === "error" && typeof json.error?.code === "string" && json.error.code.includes("rate_limit")) {
return "Rate Limited"
}
})
if (!json || typeof json !== "object") return undefined
const code = typeof json.code === "string" ? json.code : ""
return undefined
}
if (json.type === "error" && json.error?.type === "too_many_requests") {
return "Too Many Requests"
export function policy(opts: {
parse: (error: unknown) => Err
set: (input: { attempt: number; message: string; next: number }) => Effect.Effect<void>
}) {
return Schedule.fromStepWithMetadata(
Effect.succeed((meta: Schedule.InputMetadata<unknown>) => {
const error = opts.parse(meta.input)
const message = retryable(error)
if (!message) return Cause.done(meta.attempt)
return Effect.gen(function* () {
const wait = delay(meta.attempt, MessageV2.APIError.isInstance(error) ? error : undefined)
const now = yield* Clock.currentTimeMillis
yield* opts.set({ attempt: meta.attempt, message, next: now + wait })
return [meta.attempt, Duration.millis(wait)] as [number, Duration.Duration]
})
}),
)
}
if (code.includes("exhausted") || code.includes("unavailable")) {
return "Provider is overloaded"
}
if (json.type === "error" && typeof json.error?.code === "string" && json.error.code.includes("rate_limit")) {
return "Rate Limited"
}
return undefined
}
export function policy(opts: {
parse: (error: unknown) => Err
set: (input: { attempt: number; message: string; next: number }) => Effect.Effect<void>
}) {
return Schedule.fromStepWithMetadata(
Effect.succeed((meta: Schedule.InputMetadata<unknown>) => {
const error = opts.parse(meta.input)
const message = retryable(error)
if (!message) return Cause.done(meta.attempt)
return Effect.gen(function* () {
const wait = delay(meta.attempt, MessageV2.APIError.isInstance(error) ? error : undefined)
const now = yield* Clock.currentTimeMillis
yield* opts.set({ attempt: meta.attempt, message, next: now + wait })
return [meta.attempt, Duration.millis(wait)] as [number, Duration.Duration]
})
}),
)
}
+135 -133
View File
@@ -6,154 +6,156 @@ import { Storage } from "@/storage/storage"
import { SyncEvent } from "../sync"
import { Log } from "../util/log"
import { Session } from "."
import { MessageV2 } from "."
import { MessageV2 } from "./message-v2"
import { SessionID, MessageID, PartID } from "./schema"
import { SessionRunState } from "."
import { SessionSummary } from "."
import { SessionRunState } from "./run-state"
import { SessionSummary } from "./summary"
const log = Log.create({ service: "session.revert" })
export namespace SessionRevert {
const log = Log.create({ service: "session.revert" })
export const RevertInput = z.object({
sessionID: SessionID.zod,
messageID: MessageID.zod,
partID: PartID.zod.optional(),
})
export type RevertInput = z.infer<typeof RevertInput>
export const RevertInput = z.object({
sessionID: SessionID.zod,
messageID: MessageID.zod,
partID: PartID.zod.optional(),
})
export type RevertInput = z.infer<typeof RevertInput>
export interface Interface {
readonly revert: (input: RevertInput) => Effect.Effect<Session.Info>
readonly unrevert: (input: { sessionID: SessionID }) => Effect.Effect<Session.Info>
readonly cleanup: (session: Session.Info) => Effect.Effect<void>
}
export interface Interface {
readonly revert: (input: RevertInput) => Effect.Effect<Session.Info>
readonly unrevert: (input: { sessionID: SessionID }) => Effect.Effect<Session.Info>
readonly cleanup: (session: Session.Info) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRevert") {}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRevert") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const sessions = yield* Session.Service
const snap = yield* Snapshot.Service
const storage = yield* Storage.Service
const bus = yield* Bus.Service
const summary = yield* SessionSummary.Service
const state = yield* SessionRunState.Service
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const sessions = yield* Session.Service
const snap = yield* Snapshot.Service
const storage = yield* Storage.Service
const bus = yield* Bus.Service
const summary = yield* SessionSummary.Service
const state = yield* SessionRunState.Service
const revert = Effect.fn("SessionRevert.revert")(function* (input: RevertInput) {
yield* state.assertNotBusy(input.sessionID)
const all = yield* sessions.messages({ sessionID: input.sessionID })
let lastUser: MessageV2.User | undefined
const session = yield* sessions.get(input.sessionID)
const revert = Effect.fn("SessionRevert.revert")(function* (input: RevertInput) {
yield* state.assertNotBusy(input.sessionID)
const all = yield* sessions.messages({ sessionID: input.sessionID })
let lastUser: MessageV2.User | undefined
const session = yield* sessions.get(input.sessionID)
let rev: Session.Info["revert"]
const patches: Snapshot.Patch[] = []
for (const msg of all) {
if (msg.info.role === "user") lastUser = msg.info
const remaining = []
for (const part of msg.parts) {
if (rev) {
if (part.type === "patch") patches.push(part)
let rev: Session.Info["revert"]
const patches: Snapshot.Patch[] = []
for (const msg of all) {
if (msg.info.role === "user") lastUser = msg.info
const remaining = []
for (const part of msg.parts) {
if (rev) {
if (part.type === "patch") patches.push(part)
continue
}
if (!rev) {
if ((msg.info.id === input.messageID && !input.partID) || part.id === input.partID) {
const partID = remaining.some((item) => ["text", "tool"].includes(item.type)) ? input.partID : undefined
rev = {
messageID: !partID && lastUser ? lastUser.id : msg.info.id,
partID,
}
}
remaining.push(part)
}
}
}
if (!rev) return session
rev.snapshot = session.revert?.snapshot ?? (yield* snap.track())
if (session.revert?.snapshot) yield* snap.restore(session.revert.snapshot)
yield* snap.revert(patches)
if (rev.snapshot) rev.diff = yield* snap.diff(rev.snapshot as string)
const range = all.filter((msg) => msg.info.id >= rev!.messageID)
const diffs = yield* summary.computeDiff({ messages: range })
yield* storage.write(["session_diff", input.sessionID], diffs).pipe(Effect.ignore)
yield* bus.publish(Session.Event.Diff, { sessionID: input.sessionID, diff: diffs })
yield* sessions.setRevert({
sessionID: input.sessionID,
revert: rev,
summary: {
additions: diffs.reduce((sum, x) => sum + x.additions, 0),
deletions: diffs.reduce((sum, x) => sum + x.deletions, 0),
files: diffs.length,
},
})
return yield* sessions.get(input.sessionID)
})
const unrevert = Effect.fn("SessionRevert.unrevert")(function* (input: { sessionID: SessionID }) {
log.info("unreverting", input)
yield* state.assertNotBusy(input.sessionID)
const session = yield* sessions.get(input.sessionID)
if (!session.revert) return session
if (session.revert.snapshot) yield* snap.restore(session.revert!.snapshot!)
yield* sessions.clearRevert(input.sessionID)
return yield* sessions.get(input.sessionID)
})
const cleanup = Effect.fn("SessionRevert.cleanup")(function* (session: Session.Info) {
if (!session.revert) return
const sessionID = session.id
const msgs = yield* sessions.messages({ sessionID })
const messageID = session.revert.messageID
const remove = [] as MessageV2.WithParts[]
let target: MessageV2.WithParts | undefined
for (const msg of msgs) {
if (msg.info.id < messageID) continue
if (msg.info.id > messageID) {
remove.push(msg)
continue
}
if (!rev) {
if ((msg.info.id === input.messageID && !input.partID) || part.id === input.partID) {
const partID = remaining.some((item) => ["text", "tool"].includes(item.type)) ? input.partID : undefined
rev = {
messageID: !partID && lastUser ? lastUser.id : msg.info.id,
partID,
}
}
remaining.push(part)
if (session.revert.partID) {
target = msg
continue
}
}
}
if (!rev) return session
rev.snapshot = session.revert?.snapshot ?? (yield* snap.track())
if (session.revert?.snapshot) yield* snap.restore(session.revert.snapshot)
yield* snap.revert(patches)
if (rev.snapshot) rev.diff = yield* snap.diff(rev.snapshot as string)
const range = all.filter((msg) => msg.info.id >= rev!.messageID)
const diffs = yield* summary.computeDiff({ messages: range })
yield* storage.write(["session_diff", input.sessionID], diffs).pipe(Effect.ignore)
yield* bus.publish(Session.Event.Diff, { sessionID: input.sessionID, diff: diffs })
yield* sessions.setRevert({
sessionID: input.sessionID,
revert: rev,
summary: {
additions: diffs.reduce((sum, x) => sum + x.additions, 0),
deletions: diffs.reduce((sum, x) => sum + x.deletions, 0),
files: diffs.length,
},
})
return yield* sessions.get(input.sessionID)
})
const unrevert = Effect.fn("SessionRevert.unrevert")(function* (input: { sessionID: SessionID }) {
log.info("unreverting", input)
yield* state.assertNotBusy(input.sessionID)
const session = yield* sessions.get(input.sessionID)
if (!session.revert) return session
if (session.revert.snapshot) yield* snap.restore(session.revert!.snapshot!)
yield* sessions.clearRevert(input.sessionID)
return yield* sessions.get(input.sessionID)
})
const cleanup = Effect.fn("SessionRevert.cleanup")(function* (session: Session.Info) {
if (!session.revert) return
const sessionID = session.id
const msgs = yield* sessions.messages({ sessionID })
const messageID = session.revert.messageID
const remove = [] as MessageV2.WithParts[]
let target: MessageV2.WithParts | undefined
for (const msg of msgs) {
if (msg.info.id < messageID) continue
if (msg.info.id > messageID) {
remove.push(msg)
continue
}
if (session.revert.partID) {
target = msg
continue
for (const msg of remove) {
SyncEvent.run(MessageV2.Event.Removed, {
sessionID,
messageID: msg.info.id,
})
}
remove.push(msg)
}
for (const msg of remove) {
SyncEvent.run(MessageV2.Event.Removed, {
sessionID,
messageID: msg.info.id,
})
}
if (session.revert.partID && target) {
const partID = session.revert.partID
const idx = target.parts.findIndex((part) => part.id === partID)
if (idx >= 0) {
const removeParts = target.parts.slice(idx)
target.parts = target.parts.slice(0, idx)
for (const part of removeParts) {
SyncEvent.run(MessageV2.Event.PartRemoved, {
sessionID,
messageID: target.info.id,
partID: part.id,
})
if (session.revert.partID && target) {
const partID = session.revert.partID
const idx = target.parts.findIndex((part) => part.id === partID)
if (idx >= 0) {
const removeParts = target.parts.slice(idx)
target.parts = target.parts.slice(0, idx)
for (const part of removeParts) {
SyncEvent.run(MessageV2.Event.PartRemoved, {
sessionID,
messageID: target.info.id,
partID: part.id,
})
}
}
}
}
yield* sessions.clearRevert(sessionID)
})
yield* sessions.clearRevert(sessionID)
})
return Service.of({ revert, unrevert, cleanup })
}),
)
return Service.of({ revert, unrevert, cleanup })
}),
)
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(SessionRunState.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(Snapshot.defaultLayer),
Layer.provide(Storage.defaultLayer),
Layer.provide(Bus.layer),
Layer.provide(SessionSummary.defaultLayer),
),
)
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(SessionRunState.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(Snapshot.defaultLayer),
Layer.provide(Storage.defaultLayer),
Layer.provide(Bus.layer),
Layer.provide(SessionSummary.defaultLayer),
),
)
}
+92 -90
View File
@@ -2,105 +2,107 @@ import { InstanceState } from "@/effect"
import { Runner } from "@/effect/runner"
import { Effect, Layer, Scope, Context } from "effect"
import { Session } from "."
import { MessageV2 } from "."
import { MessageV2 } from "./message-v2"
import { SessionID } from "./schema"
import { SessionStatus } from "."
import { SessionStatus } from "./status"
export interface Interface {
readonly assertNotBusy: (sessionID: SessionID) => Effect.Effect<void>
readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
readonly ensureRunning: (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
work: Effect.Effect<MessageV2.WithParts>,
) => Effect.Effect<MessageV2.WithParts>
readonly startShell: (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
work: Effect.Effect<MessageV2.WithParts>,
) => Effect.Effect<MessageV2.WithParts>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRunState") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const status = yield* SessionStatus.Service
const state = yield* InstanceState.make(
Effect.fn("SessionRunState.state")(function* () {
const scope = yield* Scope.Scope
const runners = new Map<SessionID, Runner<MessageV2.WithParts>>()
yield* Effect.addFinalizer(
Effect.fnUntraced(function* () {
yield* Effect.forEach(runners.values(), (runner) => runner.cancel, {
concurrency: "unbounded",
discard: true,
})
runners.clear()
}),
)
return { runners, scope }
}),
)
const runner = Effect.fn("SessionRunState.runner")(function* (
export namespace SessionRunState {
export interface Interface {
readonly assertNotBusy: (sessionID: SessionID) => Effect.Effect<void>
readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
readonly ensureRunning: (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
) {
const data = yield* InstanceState.get(state)
const existing = data.runners.get(sessionID)
if (existing) return existing
const next = Runner.make<MessageV2.WithParts>(data.scope, {
onIdle: Effect.gen(function* () {
data.runners.delete(sessionID)
yield* status.set(sessionID, { type: "idle" })
work: Effect.Effect<MessageV2.WithParts>,
) => Effect.Effect<MessageV2.WithParts>
readonly startShell: (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
work: Effect.Effect<MessageV2.WithParts>,
) => Effect.Effect<MessageV2.WithParts>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRunState") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const status = yield* SessionStatus.Service
const state = yield* InstanceState.make(
Effect.fn("SessionRunState.state")(function* () {
const scope = yield* Scope.Scope
const runners = new Map<SessionID, Runner<MessageV2.WithParts>>()
yield* Effect.addFinalizer(
Effect.fnUntraced(function* () {
yield* Effect.forEach(runners.values(), (runner) => runner.cancel, {
concurrency: "unbounded",
discard: true,
})
runners.clear()
}),
)
return { runners, scope }
}),
onBusy: status.set(sessionID, { type: "busy" }),
onInterrupt,
busy: () => {
throw new Session.BusyError(sessionID)
},
)
const runner = Effect.fn("SessionRunState.runner")(function* (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
) {
const data = yield* InstanceState.get(state)
const existing = data.runners.get(sessionID)
if (existing) return existing
const next = Runner.make<MessageV2.WithParts>(data.scope, {
onIdle: Effect.gen(function* () {
data.runners.delete(sessionID)
yield* status.set(sessionID, { type: "idle" })
}),
onBusy: status.set(sessionID, { type: "busy" }),
onInterrupt,
busy: () => {
throw new Session.BusyError(sessionID)
},
})
data.runners.set(sessionID, next)
return next
})
data.runners.set(sessionID, next)
return next
})
const assertNotBusy = Effect.fn("SessionRunState.assertNotBusy")(function* (sessionID: SessionID) {
const data = yield* InstanceState.get(state)
const existing = data.runners.get(sessionID)
if (existing?.busy) throw new Session.BusyError(sessionID)
})
const assertNotBusy = Effect.fn("SessionRunState.assertNotBusy")(function* (sessionID: SessionID) {
const data = yield* InstanceState.get(state)
const existing = data.runners.get(sessionID)
if (existing?.busy) throw new Session.BusyError(sessionID)
})
const cancel = Effect.fn("SessionRunState.cancel")(function* (sessionID: SessionID) {
const data = yield* InstanceState.get(state)
const existing = data.runners.get(sessionID)
if (!existing || !existing.busy) {
yield* status.set(sessionID, { type: "idle" })
return
}
yield* existing.cancel
})
const cancel = Effect.fn("SessionRunState.cancel")(function* (sessionID: SessionID) {
const data = yield* InstanceState.get(state)
const existing = data.runners.get(sessionID)
if (!existing || !existing.busy) {
yield* status.set(sessionID, { type: "idle" })
return
}
yield* existing.cancel
})
const ensureRunning = Effect.fn("SessionRunState.ensureRunning")(function* (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
work: Effect.Effect<MessageV2.WithParts>,
) {
return yield* (yield* runner(sessionID, onInterrupt)).ensureRunning(work)
})
const ensureRunning = Effect.fn("SessionRunState.ensureRunning")(function* (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
work: Effect.Effect<MessageV2.WithParts>,
) {
return yield* (yield* runner(sessionID, onInterrupt)).ensureRunning(work)
})
const startShell = Effect.fn("SessionRunState.startShell")(function* (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
work: Effect.Effect<MessageV2.WithParts>,
) {
return yield* (yield* runner(sessionID, onInterrupt)).startShell(work)
})
const startShell = Effect.fn("SessionRunState.startShell")(function* (
sessionID: SessionID,
onInterrupt: Effect.Effect<MessageV2.WithParts>,
work: Effect.Effect<MessageV2.WithParts>,
) {
return yield* (yield* runner(sessionID, onInterrupt)).startShell(work)
})
return Service.of({ assertNotBusy, cancel, ensureRunning, startShell })
}),
)
return Service.of({ assertNotBusy, cancel, ensureRunning, startShell })
}),
)
export const defaultLayer = layer.pipe(Layer.provide(SessionStatus.defaultLayer))
export const defaultLayer = layer.pipe(Layer.provide(SessionStatus.defaultLayer))
}
+1 -1
View File
@@ -1,6 +1,6 @@
import { sqliteTable, text, integer, index, primaryKey } from "drizzle-orm/sqlite-core"
import { ProjectTable } from "../project/project.sql"
import type { MessageV2 } from "."
import type { MessageV2 } from "./message-v2"
import type { SessionEntry } from "../v2/session-entry"
import type { Snapshot } from "../snapshot"
import type { Permission } from "../permission"
+1 -1
View File
@@ -16,7 +16,7 @@ import { ProjectTable } from "../project/project.sql"
import { Storage } from "@/storage/storage"
import { Log } from "../util/log"
import { updateSchema } from "../util/update-schema"
import { MessageV2 } from "."
import { MessageV2 } from "./message-v2"
import { Instance } from "../project/instance"
import { InstanceState } from "@/effect"
import { Snapshot } from "@/snapshot"
+78 -76
View File
@@ -5,82 +5,84 @@ import { SessionID } from "./schema"
import { Effect, Layer, Context } from "effect"
import z from "zod"
export const Info = z
.union([
z.object({
type: z.literal("idle"),
}),
z.object({
type: z.literal("retry"),
attempt: z.number(),
message: z.string(),
next: z.number(),
}),
z.object({
type: z.literal("busy"),
}),
])
.meta({
ref: "SessionStatus",
})
export type Info = z.infer<typeof Info>
export namespace SessionStatus {
export const Info = z
.union([
z.object({
type: z.literal("idle"),
}),
z.object({
type: z.literal("retry"),
attempt: z.number(),
message: z.string(),
next: z.number(),
}),
z.object({
type: z.literal("busy"),
}),
])
.meta({
ref: "SessionStatus",
})
export type Info = z.infer<typeof Info>
export const Event = {
Status: BusEvent.define(
"session.status",
z.object({
sessionID: SessionID.zod,
status: Info,
export const Event = {
Status: BusEvent.define(
"session.status",
z.object({
sessionID: SessionID.zod,
status: Info,
}),
),
// deprecated
Idle: BusEvent.define(
"session.idle",
z.object({
sessionID: SessionID.zod,
}),
),
}
export interface Interface {
readonly get: (sessionID: SessionID) => Effect.Effect<Info>
readonly list: () => Effect.Effect<Map<SessionID, Info>>
readonly set: (sessionID: SessionID, status: Info) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionStatus") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const state = yield* InstanceState.make(
Effect.fn("SessionStatus.state")(() => Effect.succeed(new Map<SessionID, Info>())),
)
const get = Effect.fn("SessionStatus.get")(function* (sessionID: SessionID) {
const data = yield* InstanceState.get(state)
return data.get(sessionID) ?? { type: "idle" as const }
})
const list = Effect.fn("SessionStatus.list")(function* () {
return new Map(yield* InstanceState.get(state))
})
const set = Effect.fn("SessionStatus.set")(function* (sessionID: SessionID, status: Info) {
const data = yield* InstanceState.get(state)
yield* bus.publish(Event.Status, { sessionID, status })
if (status.type === "idle") {
yield* bus.publish(Event.Idle, { sessionID })
data.delete(sessionID)
return
}
data.set(sessionID, status)
})
return Service.of({ get, list, set })
}),
),
// deprecated
Idle: BusEvent.define(
"session.idle",
z.object({
sessionID: SessionID.zod,
}),
),
)
export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
}
export interface Interface {
readonly get: (sessionID: SessionID) => Effect.Effect<Info>
readonly list: () => Effect.Effect<Map<SessionID, Info>>
readonly set: (sessionID: SessionID, status: Info) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionStatus") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const state = yield* InstanceState.make(
Effect.fn("SessionStatus.state")(() => Effect.succeed(new Map<SessionID, Info>())),
)
const get = Effect.fn("SessionStatus.get")(function* (sessionID: SessionID) {
const data = yield* InstanceState.get(state)
return data.get(sessionID) ?? { type: "idle" as const }
})
const list = Effect.fn("SessionStatus.list")(function* () {
return new Map(yield* InstanceState.get(state))
})
const set = Effect.fn("SessionStatus.set")(function* (sessionID: SessionID, status: Info) {
const data = yield* InstanceState.get(state)
yield* bus.publish(Event.Status, { sessionID, status })
if (status.type === "idle") {
yield* bus.publish(Event.Idle, { sessionID })
data.delete(sessionID)
return
}
data.set(sessionID, status)
})
return Service.of({ get, list, set })
}),
)
export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
+137 -135
View File
@@ -4,160 +4,162 @@ import { Bus } from "@/bus"
import { Snapshot } from "@/snapshot"
import { Storage } from "@/storage/storage"
import { Session } from "."
import { MessageV2 } from "."
import { MessageV2 } from "./message-v2"
import { SessionID, MessageID } from "./schema"
function unquoteGitPath(input: string) {
if (!input.startsWith('"')) return input
if (!input.endsWith('"')) return input
const body = input.slice(1, -1)
const bytes: number[] = []
export namespace SessionSummary {
function unquoteGitPath(input: string) {
if (!input.startsWith('"')) return input
if (!input.endsWith('"')) return input
const body = input.slice(1, -1)
const bytes: number[] = []
for (let i = 0; i < body.length; i++) {
const char = body[i]!
if (char !== "\\") {
bytes.push(char.charCodeAt(0))
continue
}
const next = body[i + 1]
if (!next) {
bytes.push("\\".charCodeAt(0))
continue
}
if (next >= "0" && next <= "7") {
const chunk = body.slice(i + 1, i + 4)
const match = chunk.match(/^[0-7]{1,3}/)
if (!match) {
bytes.push(next.charCodeAt(0))
i++
for (let i = 0; i < body.length; i++) {
const char = body[i]!
if (char !== "\\") {
bytes.push(char.charCodeAt(0))
continue
}
bytes.push(parseInt(match[0], 8))
i += match[0].length
continue
const next = body[i + 1]
if (!next) {
bytes.push("\\".charCodeAt(0))
continue
}
if (next >= "0" && next <= "7") {
const chunk = body.slice(i + 1, i + 4)
const match = chunk.match(/^[0-7]{1,3}/)
if (!match) {
bytes.push(next.charCodeAt(0))
i++
continue
}
bytes.push(parseInt(match[0], 8))
i += match[0].length
continue
}
const escaped =
next === "n"
? "\n"
: next === "r"
? "\r"
: next === "t"
? "\t"
: next === "b"
? "\b"
: next === "f"
? "\f"
: next === "v"
? "\v"
: next === "\\" || next === '"'
? next
: undefined
bytes.push((escaped ?? next).charCodeAt(0))
i++
}
const escaped =
next === "n"
? "\n"
: next === "r"
? "\r"
: next === "t"
? "\t"
: next === "b"
? "\b"
: next === "f"
? "\f"
: next === "v"
? "\v"
: next === "\\" || next === '"'
? next
: undefined
bytes.push((escaped ?? next).charCodeAt(0))
i++
return Buffer.from(bytes).toString()
}
return Buffer.from(bytes).toString()
}
export interface Interface {
readonly summarize: (input: { sessionID: SessionID; messageID: MessageID }) => Effect.Effect<void>
readonly diff: (input: { sessionID: SessionID; messageID?: MessageID }) => Effect.Effect<Snapshot.FileDiff[]>
readonly computeDiff: (input: { messages: MessageV2.WithParts[] }) => Effect.Effect<Snapshot.FileDiff[]>
}
export interface Interface {
readonly summarize: (input: { sessionID: SessionID; messageID: MessageID }) => Effect.Effect<void>
readonly diff: (input: { sessionID: SessionID; messageID?: MessageID }) => Effect.Effect<Snapshot.FileDiff[]>
readonly computeDiff: (input: { messages: MessageV2.WithParts[] }) => Effect.Effect<Snapshot.FileDiff[]>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionSummary") {}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionSummary") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const sessions = yield* Session.Service
const snapshot = yield* Snapshot.Service
const storage = yield* Storage.Service
const bus = yield* Bus.Service
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const sessions = yield* Session.Service
const snapshot = yield* Snapshot.Service
const storage = yield* Storage.Service
const bus = yield* Bus.Service
const computeDiff = Effect.fn("SessionSummary.computeDiff")(function* (input: {
messages: MessageV2.WithParts[]
}) {
let from: string | undefined
let to: string | undefined
for (const item of input.messages) {
if (!from) {
for (const part of item.parts) {
if (part.type === "step-start" && part.snapshot) {
from = part.snapshot
break
const computeDiff = Effect.fn("SessionSummary.computeDiff")(function* (input: {
messages: MessageV2.WithParts[]
}) {
let from: string | undefined
let to: string | undefined
for (const item of input.messages) {
if (!from) {
for (const part of item.parts) {
if (part.type === "step-start" && part.snapshot) {
from = part.snapshot
break
}
}
}
for (const part of item.parts) {
if (part.type === "step-finish" && part.snapshot) to = part.snapshot
}
}
for (const part of item.parts) {
if (part.type === "step-finish" && part.snapshot) to = part.snapshot
}
}
if (from && to) return yield* snapshot.diffFull(from, to)
return []
})
const summarize = Effect.fn("SessionSummary.summarize")(function* (input: {
sessionID: SessionID
messageID: MessageID
}) {
const all = yield* sessions.messages({ sessionID: input.sessionID })
if (!all.length) return
const diffs = yield* computeDiff({ messages: all })
yield* sessions.setSummary({
sessionID: input.sessionID,
summary: {
additions: diffs.reduce((sum, x) => sum + x.additions, 0),
deletions: diffs.reduce((sum, x) => sum + x.deletions, 0),
files: diffs.length,
},
if (from && to) return yield* snapshot.diffFull(from, to)
return []
})
yield* storage.write(["session_diff", input.sessionID], diffs).pipe(Effect.ignore)
yield* bus.publish(Session.Event.Diff, { sessionID: input.sessionID, diff: diffs })
const messages = all.filter(
(m) => m.info.id === input.messageID || (m.info.role === "assistant" && m.info.parentID === input.messageID),
)
const target = messages.find((m) => m.info.id === input.messageID)
if (!target || target.info.role !== "user") return
const msgDiffs = yield* computeDiff({ messages })
target.info.summary = { ...target.info.summary, diffs: msgDiffs }
yield* sessions.updateMessage(target.info)
})
const summarize = Effect.fn("SessionSummary.summarize")(function* (input: {
sessionID: SessionID
messageID: MessageID
}) {
const all = yield* sessions.messages({ sessionID: input.sessionID })
if (!all.length) return
const diff = Effect.fn("SessionSummary.diff")(function* (input: { sessionID: SessionID; messageID?: MessageID }) {
const diffs = yield* storage
.read<Snapshot.FileDiff[]>(["session_diff", input.sessionID])
.pipe(Effect.catch(() => Effect.succeed([] as Snapshot.FileDiff[])))
const next = diffs.map((item) => {
const file = unquoteGitPath(item.file)
if (file === item.file) return item
return { ...item, file }
const diffs = yield* computeDiff({ messages: all })
yield* sessions.setSummary({
sessionID: input.sessionID,
summary: {
additions: diffs.reduce((sum, x) => sum + x.additions, 0),
deletions: diffs.reduce((sum, x) => sum + x.deletions, 0),
files: diffs.length,
},
})
yield* storage.write(["session_diff", input.sessionID], diffs).pipe(Effect.ignore)
yield* bus.publish(Session.Event.Diff, { sessionID: input.sessionID, diff: diffs })
const messages = all.filter(
(m) => m.info.id === input.messageID || (m.info.role === "assistant" && m.info.parentID === input.messageID),
)
const target = messages.find((m) => m.info.id === input.messageID)
if (!target || target.info.role !== "user") return
const msgDiffs = yield* computeDiff({ messages })
target.info.summary = { ...target.info.summary, diffs: msgDiffs }
yield* sessions.updateMessage(target.info)
})
const changed = next.some((item, i) => item.file !== diffs[i]?.file)
if (changed) yield* storage.write(["session_diff", input.sessionID], next).pipe(Effect.ignore)
return next
})
return Service.of({ summarize, diff, computeDiff })
}),
)
const diff = Effect.fn("SessionSummary.diff")(function* (input: { sessionID: SessionID; messageID?: MessageID }) {
const diffs = yield* storage
.read<Snapshot.FileDiff[]>(["session_diff", input.sessionID])
.pipe(Effect.catch(() => Effect.succeed([] as Snapshot.FileDiff[])))
const next = diffs.map((item) => {
const file = unquoteGitPath(item.file)
if (file === item.file) return item
return { ...item, file }
})
const changed = next.some((item, i) => item.file !== diffs[i]?.file)
if (changed) yield* storage.write(["session_diff", input.sessionID], next).pipe(Effect.ignore)
return next
})
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(Session.defaultLayer),
Layer.provide(Snapshot.defaultLayer),
Layer.provide(Storage.defaultLayer),
Layer.provide(Bus.layer),
),
)
return Service.of({ summarize, diff, computeDiff })
}),
)
export const DiffInput = z.object({
sessionID: SessionID.zod,
messageID: MessageID.zod.optional(),
})
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
Layer.provide(Session.defaultLayer),
Layer.provide(Snapshot.defaultLayer),
Layer.provide(Storage.defaultLayer),
Layer.provide(Bus.layer),
),
)
export const DiffInput = z.object({
sessionID: SessionID.zod,
messageID: MessageID.zod.optional(),
})
}
+63 -61
View File
@@ -16,67 +16,69 @@ import type { Agent } from "@/agent/agent"
import { Permission } from "@/permission"
import { Skill } from "@/skill"
export function provider(model: Provider.Model) {
if (model.api.id.includes("gpt-4") || model.api.id.includes("o1") || model.api.id.includes("o3"))
return [PROMPT_BEAST]
if (model.api.id.includes("gpt")) {
if (model.api.id.includes("codex")) {
return [PROMPT_CODEX]
export namespace SystemPrompt {
export function provider(model: Provider.Model) {
if (model.api.id.includes("gpt-4") || model.api.id.includes("o1") || model.api.id.includes("o3"))
return [PROMPT_BEAST]
if (model.api.id.includes("gpt")) {
if (model.api.id.includes("codex")) {
return [PROMPT_CODEX]
}
return [PROMPT_GPT]
}
return [PROMPT_GPT]
if (model.api.id.includes("gemini-")) return [PROMPT_GEMINI]
if (model.api.id.includes("claude")) return [PROMPT_ANTHROPIC]
if (model.api.id.toLowerCase().includes("trinity")) return [PROMPT_TRINITY]
if (model.api.id.toLowerCase().includes("kimi")) return [PROMPT_KIMI]
return [PROMPT_DEFAULT]
}
if (model.api.id.includes("gemini-")) return [PROMPT_GEMINI]
if (model.api.id.includes("claude")) return [PROMPT_ANTHROPIC]
if (model.api.id.toLowerCase().includes("trinity")) return [PROMPT_TRINITY]
if (model.api.id.toLowerCase().includes("kimi")) return [PROMPT_KIMI]
return [PROMPT_DEFAULT]
export interface Interface {
readonly environment: (model: Provider.Model) => string[]
readonly skills: (agent: Agent.Info) => Effect.Effect<string | undefined>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SystemPrompt") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const skill = yield* Skill.Service
return Service.of({
environment(model) {
const project = Instance.project
return [
[
`You are powered by the model named ${model.api.id}. The exact model ID is ${model.providerID}/${model.api.id}`,
`Here is some useful information about the environment you are running in:`,
`<env>`,
` Working directory: ${Instance.directory}`,
` Workspace root folder: ${Instance.worktree}`,
` Is directory a git repo: ${project.vcs === "git" ? "yes" : "no"}`,
` Platform: ${process.platform}`,
` Today's date: ${new Date().toDateString()}`,
`</env>`,
].join("\n"),
]
},
skills: Effect.fn("SystemPrompt.skills")(function* (agent: Agent.Info) {
if (Permission.disabled(["skill"], agent.permission).has("skill")) return
const list = yield* skill.available(agent)
return [
"Skills provide specialized instructions and workflows for specific tasks.",
"Use the skill tool to load a skill when a task matches its description.",
// the agents seem to ingest the information about skills a bit better if we present a more verbose
// version of them here and a less verbose version in tool description, rather than vice versa.
Skill.fmt(list, { verbose: true }),
].join("\n")
}),
})
}),
)
export const defaultLayer = layer.pipe(Layer.provide(Skill.defaultLayer))
}
export interface Interface {
readonly environment: (model: Provider.Model) => string[]
readonly skills: (agent: Agent.Info) => Effect.Effect<string | undefined>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SystemPrompt") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const skill = yield* Skill.Service
return Service.of({
environment(model) {
const project = Instance.project
return [
[
`You are powered by the model named ${model.api.id}. The exact model ID is ${model.providerID}/${model.api.id}`,
`Here is some useful information about the environment you are running in:`,
`<env>`,
` Working directory: ${Instance.directory}`,
` Workspace root folder: ${Instance.worktree}`,
` Is directory a git repo: ${project.vcs === "git" ? "yes" : "no"}`,
` Platform: ${process.platform}`,
` Today's date: ${new Date().toDateString()}`,
`</env>`,
].join("\n"),
]
},
skills: Effect.fn("SystemPrompt.skills")(function* (agent: Agent.Info) {
if (Permission.disabled(["skill"], agent.permission).has("skill")) return
const list = yield* skill.available(agent)
return [
"Skills provide specialized instructions and workflows for specific tasks.",
"Use the skill tool to load a skill when a task matches its description.",
// the agents seem to ingest the information about skills a bit better if we present a more verbose
// version of them here and a less verbose version in tool description, rather than vice versa.
Skill.fmt(list, { verbose: true }),
].join("\n")
}),
})
}),
)
export const defaultLayer = layer.pipe(Layer.provide(Skill.defaultLayer))
+74 -72
View File
@@ -6,78 +6,80 @@ import z from "zod"
import { Database, eq, asc } from "../storage/db"
import { TodoTable } from "./session.sql"
export const Info = z
.object({
content: z.string().describe("Brief description of the task"),
status: z.string().describe("Current status of the task: pending, in_progress, completed, cancelled"),
priority: z.string().describe("Priority level of the task: high, medium, low"),
})
.meta({ ref: "Todo" })
export type Info = z.infer<typeof Info>
export namespace Todo {
export const Info = z
.object({
content: z.string().describe("Brief description of the task"),
status: z.string().describe("Current status of the task: pending, in_progress, completed, cancelled"),
priority: z.string().describe("Priority level of the task: high, medium, low"),
})
.meta({ ref: "Todo" })
export type Info = z.infer<typeof Info>
export const Event = {
Updated: BusEvent.define(
"todo.updated",
z.object({
sessionID: SessionID.zod,
todos: z.array(Info),
export const Event = {
Updated: BusEvent.define(
"todo.updated",
z.object({
sessionID: SessionID.zod,
todos: z.array(Info),
}),
),
}
export interface Interface {
readonly update: (input: { sessionID: SessionID; todos: Info[] }) => Effect.Effect<void>
readonly get: (sessionID: SessionID) => Effect.Effect<Info[]>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionTodo") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const update = Effect.fn("Todo.update")(function* (input: { sessionID: SessionID; todos: Info[] }) {
yield* Effect.sync(() =>
Database.transaction((db) => {
db.delete(TodoTable).where(eq(TodoTable.session_id, input.sessionID)).run()
if (input.todos.length === 0) return
db.insert(TodoTable)
.values(
input.todos.map((todo, position) => ({
session_id: input.sessionID,
content: todo.content,
status: todo.status,
priority: todo.priority,
position,
})),
)
.run()
}),
)
yield* bus.publish(Event.Updated, input)
})
const get = Effect.fn("Todo.get")(function* (sessionID: SessionID) {
const rows = yield* Effect.sync(() =>
Database.use((db) =>
db
.select()
.from(TodoTable)
.where(eq(TodoTable.session_id, sessionID))
.orderBy(asc(TodoTable.position))
.all(),
),
)
return rows.map((row) => ({
content: row.content,
status: row.status,
priority: row.priority,
}))
})
return Service.of({ update, get })
}),
),
)
export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
}
export interface Interface {
readonly update: (input: { sessionID: SessionID; todos: Info[] }) => Effect.Effect<void>
readonly get: (sessionID: SessionID) => Effect.Effect<Info[]>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionTodo") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const update = Effect.fn("Todo.update")(function* (input: { sessionID: SessionID; todos: Info[] }) {
yield* Effect.sync(() =>
Database.transaction((db) => {
db.delete(TodoTable).where(eq(TodoTable.session_id, input.sessionID)).run()
if (input.todos.length === 0) return
db.insert(TodoTable)
.values(
input.todos.map((todo, position) => ({
session_id: input.sessionID,
content: todo.content,
status: todo.status,
priority: todo.priority,
position,
})),
)
.run()
}),
)
yield* bus.publish(Event.Updated, input)
})
const get = Effect.fn("Todo.get")(function* (sessionID: SessionID) {
const rows = yield* Effect.sync(() =>
Database.use((db) =>
db
.select()
.from(TodoTable)
.where(eq(TodoTable.session_id, sessionID))
.orderBy(asc(TodoTable.position))
.all(),
),
)
return rows.map((row) => ({
content: row.content,
status: row.status,
priority: row.priority,
}))
})
return Service.of({ update, get })
}),
)
export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
+2
View File
@@ -0,0 +1,2 @@
export * as ShareNext from "./share-next"
export * as SessionShare from "./session"
+49 -51
View File
@@ -4,56 +4,54 @@ import { SyncEvent } from "@/sync"
import { Effect, Layer, Scope, Context } from "effect"
import { Config } from "../config"
import { Flag } from "../flag/flag"
import { ShareNext } from "./share-next"
import { ShareNext } from "."
export namespace SessionShare {
export interface Interface {
readonly create: (input?: Session.CreateInput) => Effect.Effect<Session.Info>
readonly share: (sessionID: SessionID) => Effect.Effect<{ url: string }, unknown>
readonly unshare: (sessionID: SessionID) => Effect.Effect<void, unknown>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionShare") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const cfg = yield* Config.Service
const session = yield* Session.Service
const shareNext = yield* ShareNext.Service
const scope = yield* Scope.Scope
const share = Effect.fn("SessionShare.share")(function* (sessionID: SessionID) {
const conf = yield* cfg.get()
if (conf.share === "disabled") throw new Error("Sharing is disabled in configuration")
const result = yield* shareNext.create(sessionID)
yield* Effect.sync(() =>
SyncEvent.run(Session.Event.Updated, { sessionID, info: { share: { url: result.url } } }),
)
return result
})
const unshare = Effect.fn("SessionShare.unshare")(function* (sessionID: SessionID) {
yield* shareNext.remove(sessionID)
yield* Effect.sync(() => SyncEvent.run(Session.Event.Updated, { sessionID, info: { share: { url: null } } }))
})
const create = Effect.fn("SessionShare.create")(function* (input?: Session.CreateInput) {
const result = yield* session.create(input)
if (result.parentID) return result
const conf = yield* cfg.get()
if (!(Flag.OPENCODE_AUTO_SHARE || conf.share === "auto")) return result
yield* share(result.id).pipe(Effect.ignore, Effect.forkIn(scope))
return result
})
return Service.of({ create, share, unshare })
}),
)
export const defaultLayer = layer.pipe(
Layer.provide(ShareNext.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(Config.defaultLayer),
)
export interface Interface {
readonly create: (input?: Session.CreateInput) => Effect.Effect<Session.Info>
readonly share: (sessionID: SessionID) => Effect.Effect<{ url: string }, unknown>
readonly unshare: (sessionID: SessionID) => Effect.Effect<void, unknown>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionShare") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const cfg = yield* Config.Service
const session = yield* Session.Service
const shareNext = yield* ShareNext.Service
const scope = yield* Scope.Scope
const share = Effect.fn("SessionShare.share")(function* (sessionID: SessionID) {
const conf = yield* cfg.get()
if (conf.share === "disabled") throw new Error("Sharing is disabled in configuration")
const result = yield* shareNext.create(sessionID)
yield* Effect.sync(() =>
SyncEvent.run(Session.Event.Updated, { sessionID, info: { share: { url: result.url } } }),
)
return result
})
const unshare = Effect.fn("SessionShare.unshare")(function* (sessionID: SessionID) {
yield* shareNext.remove(sessionID)
yield* Effect.sync(() => SyncEvent.run(Session.Event.Updated, { sessionID, info: { share: { url: null } } }))
})
const create = Effect.fn("SessionShare.create")(function* (input?: Session.CreateInput) {
const result = yield* session.create(input)
if (result.parentID) return result
const conf = yield* cfg.get()
if (!(Flag.OPENCODE_AUTO_SHARE || conf.share === "auto")) return result
yield* share(result.id).pipe(Effect.ignore, Effect.forkIn(scope))
return result
})
return Service.of({ create, share, unshare })
}),
)
export const defaultLayer = layer.pipe(
Layer.provide(ShareNext.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(Config.defaultLayer),
)
+307 -309
View File
@@ -7,344 +7,342 @@ import { InstanceState } from "@/effect"
import { Provider } from "@/provider"
import { ModelID, ProviderID } from "@/provider/schema"
import { Session } from "@/session"
import { MessageV2 } from "@/session"
import { MessageV2 } from "@/session/message-v2"
import type { SessionID } from "@/session/schema"
import { Database, eq } from "@/storage/db"
import { Config } from "@/config"
import { Log } from "@/util/log"
import { SessionShareTable } from "./share.sql"
export namespace ShareNext {
const log = Log.create({ service: "share-next" })
const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1"
const log = Log.create({ service: "share-next" })
const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1"
export type Api = {
create: string
sync: (shareID: string) => string
remove: (shareID: string) => string
data: (shareID: string) => string
}
export type Api = {
create: string
sync: (shareID: string) => string
remove: (shareID: string) => string
data: (shareID: string) => string
}
export type Req = {
headers: Record<string, string>
api: Api
baseUrl: string
}
export type Req = {
headers: Record<string, string>
api: Api
baseUrl: string
}
const ShareSchema = Schema.Struct({
id: Schema.String,
url: Schema.String,
secret: Schema.String,
})
export type Share = typeof ShareSchema.Type
const ShareSchema = Schema.Struct({
id: Schema.String,
url: Schema.String,
secret: Schema.String,
})
export type Share = typeof ShareSchema.Type
type State = {
queue: Map<string, { data: Map<string, Data> }>
scope: Scope.Closeable
}
type State = {
queue: Map<string, { data: Map<string, Data> }>
scope: Scope.Closeable
}
type Data =
| {
type: "session"
data: SDK.Session
}
| {
type: "message"
data: SDK.Message
}
| {
type: "part"
data: SDK.Part
}
| {
type: "session_diff"
data: SDK.SnapshotFileDiff[]
}
| {
type: "model"
data: SDK.Model[]
}
export interface Interface {
readonly init: () => Effect.Effect<void, unknown>
readonly url: () => Effect.Effect<string, unknown>
readonly request: () => Effect.Effect<Req, unknown>
readonly create: (sessionID: SessionID) => Effect.Effect<Share, unknown>
readonly remove: (sessionID: SessionID) => Effect.Effect<void, unknown>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/ShareNext") {}
const db = <T>(fn: (d: Parameters<typeof Database.use>[0] extends (trx: infer D) => any ? D : never) => T) =>
Effect.sync(() => Database.use(fn))
function api(resource: string): Api {
return {
create: `/api/${resource}`,
sync: (shareID) => `/api/${resource}/${shareID}/sync`,
remove: (shareID) => `/api/${resource}/${shareID}`,
data: (shareID) => `/api/${resource}/${shareID}/data`,
type Data =
| {
type: "session"
data: SDK.Session
}
}
const legacyApi = api("share")
const consoleApi = api("shares")
function key(item: Data) {
switch (item.type) {
case "session":
return "session"
case "message":
return `message/${item.data.id}`
case "part":
return `part/${item.data.messageID}/${item.data.id}`
case "session_diff":
return "session_diff"
case "model":
return "model"
| {
type: "message"
data: SDK.Message
}
| {
type: "part"
data: SDK.Part
}
| {
type: "session_diff"
data: SDK.SnapshotFileDiff[]
}
| {
type: "model"
data: SDK.Model[]
}
export interface Interface {
readonly init: () => Effect.Effect<void, unknown>
readonly url: () => Effect.Effect<string, unknown>
readonly request: () => Effect.Effect<Req, unknown>
readonly create: (sessionID: SessionID) => Effect.Effect<Share, unknown>
readonly remove: (sessionID: SessionID) => Effect.Effect<void, unknown>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/ShareNext") {}
const db = <T>(fn: (d: Parameters<typeof Database.use>[0] extends (trx: infer D) => any ? D : never) => T) =>
Effect.sync(() => Database.use(fn))
function api(resource: string): Api {
return {
create: `/api/${resource}`,
sync: (shareID) => `/api/${resource}/${shareID}/sync`,
remove: (shareID) => `/api/${resource}/${shareID}`,
data: (shareID) => `/api/${resource}/${shareID}/data`,
}
}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const account = yield* Account.Service
const bus = yield* Bus.Service
const cfg = yield* Config.Service
const http = yield* HttpClient.HttpClient
const httpOk = HttpClient.filterStatusOk(http)
const provider = yield* Provider.Service
const session = yield* Session.Service
const legacyApi = api("share")
const consoleApi = api("shares")
function sync(sessionID: SessionID, data: Data[]): Effect.Effect<void> {
return Effect.gen(function* () {
if (disabled) return
const s = yield* InstanceState.get(state)
const existing = s.queue.get(sessionID)
if (existing) {
for (const item of data) {
existing.data.set(key(item), item)
}
return
function key(item: Data) {
switch (item.type) {
case "session":
return "session"
case "message":
return `message/${item.data.id}`
case "part":
return `part/${item.data.messageID}/${item.data.id}`
case "session_diff":
return "session_diff"
case "model":
return "model"
}
}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const account = yield* Account.Service
const bus = yield* Bus.Service
const cfg = yield* Config.Service
const http = yield* HttpClient.HttpClient
const httpOk = HttpClient.filterStatusOk(http)
const provider = yield* Provider.Service
const session = yield* Session.Service
function sync(sessionID: SessionID, data: Data[]): Effect.Effect<void> {
return Effect.gen(function* () {
if (disabled) return
const s = yield* InstanceState.get(state)
const existing = s.queue.get(sessionID)
if (existing) {
for (const item of data) {
existing.data.set(key(item), item)
}
const next = new Map(data.map((item) => [key(item), item]))
s.queue.set(sessionID, { data: next })
yield* flush(sessionID).pipe(
Effect.delay(1000),
Effect.catchCause((cause) =>
Effect.sync(() => {
log.error("share flush failed", { sessionID, cause })
}),
),
Effect.forkIn(s.scope),
)
})
}
const state: InstanceState.InstanceState<State> = yield* InstanceState.make<State>(
Effect.fn("ShareNext.state")(function* (_ctx) {
const cache: State = { queue: new Map(), scope: yield* Scope.make() }
yield* Effect.addFinalizer(() =>
Scope.close(cache.scope, Exit.void).pipe(
Effect.andThen(
Effect.sync(() => {
cache.queue.clear()
}),
),
),
)
if (disabled) return cache
const watch = <D extends { type: string }>(
def: D,
fn: (evt: { properties: any }) => Effect.Effect<void, unknown>,
) =>
bus.subscribe(def as never).pipe(
Stream.runForEach((evt) =>
fn(evt).pipe(
Effect.catchCause((cause) =>
Effect.sync(() => {
log.error("share subscriber failed", { type: def.type, cause })
}),
),
),
),
Effect.forkScoped,
)
yield* watch(Session.Event.Updated, (evt) =>
Effect.gen(function* () {
const info = yield* session.get(evt.properties.sessionID)
yield* sync(info.id, [{ type: "session", data: info }])
}),
)
yield* watch(MessageV2.Event.Updated, (evt) =>
Effect.gen(function* () {
const info = evt.properties.info
yield* sync(info.sessionID, [{ type: "message", data: info }])
if (info.role !== "user") return
const model = yield* provider.getModel(info.model.providerID, info.model.modelID)
yield* sync(info.sessionID, [{ type: "model", data: [model] }])
}),
)
yield* watch(MessageV2.Event.PartUpdated, (evt) =>
sync(evt.properties.part.sessionID, [{ type: "part", data: evt.properties.part }]),
)
yield* watch(Session.Event.Diff, (evt) =>
sync(evt.properties.sessionID, [{ type: "session_diff", data: evt.properties.diff }]),
)
yield* watch(Session.Event.Deleted, (evt) => remove(evt.properties.sessionID))
return cache
}),
)
const request = Effect.fn("ShareNext.request")(function* () {
const headers: Record<string, string> = {}
const active = yield* account.active()
if (Option.isNone(active) || !active.value.active_org_id) {
const baseUrl = (yield* cfg.get()).enterprise?.url ?? "https://opncd.ai"
return { headers, api: legacyApi, baseUrl } satisfies Req
return
}
const token = yield* account.token(active.value.id)
if (Option.isNone(token)) {
throw new Error("No active account token available for sharing")
}
headers.authorization = `Bearer ${token.value}`
headers["x-org-id"] = active.value.active_org_id
return { headers, api: consoleApi, baseUrl: active.value.url } satisfies Req
})
const get = Effect.fnUntraced(function* (sessionID: SessionID) {
const row = yield* db((db) =>
db.select().from(SessionShareTable).where(eq(SessionShareTable.session_id, sessionID)).get(),
)
if (!row) return
return { id: row.id, secret: row.secret, url: row.url } satisfies Share
})
const flush = Effect.fn("ShareNext.flush")(function* (sessionID: SessionID) {
if (disabled) return
const s = yield* InstanceState.get(state)
const queued = s.queue.get(sessionID)
if (!queued) return
s.queue.delete(sessionID)
const share = yield* get(sessionID)
if (!share) return
const req = yield* request()
const res = yield* HttpClientRequest.post(`${req.baseUrl}${req.api.sync(share.id)}`).pipe(
HttpClientRequest.setHeaders(req.headers),
HttpClientRequest.bodyJson({ secret: share.secret, data: Array.from(queued.data.values()) }),
Effect.flatMap((r) => http.execute(r)),
)
if (res.status >= 400) {
log.warn("failed to sync share", { sessionID, shareID: share.id, status: res.status })
}
})
const full = Effect.fn("ShareNext.full")(function* (sessionID: SessionID) {
log.info("full sync", { sessionID })
const info = yield* session.get(sessionID)
const diffs = yield* session.diff(sessionID)
const messages = yield* Effect.sync(() => Array.from(MessageV2.stream(sessionID)))
const models = yield* Effect.forEach(
Array.from(
new Map(
messages
.filter((msg) => msg.info.role === "user")
.map((msg) => (msg.info as SDK.UserMessage).model)
.map((item) => [`${item.providerID}/${item.modelID}`, item] as const),
).values(),
),
(item) => provider.getModel(ProviderID.make(item.providerID), ModelID.make(item.modelID)),
{ concurrency: 8 },
)
yield* sync(sessionID, [
{ type: "session", data: info },
...messages.map((item) => ({ type: "message" as const, data: item.info })),
...messages.flatMap((item) => item.parts.map((part) => ({ type: "part" as const, data: part }))),
{ type: "session_diff", data: diffs },
{ type: "model", data: models },
])
})
const init = Effect.fn("ShareNext.init")(function* () {
if (disabled) return
yield* InstanceState.get(state)
})
const url = Effect.fn("ShareNext.url")(function* () {
return (yield* request()).baseUrl
})
const create = Effect.fn("ShareNext.create")(function* (sessionID: SessionID) {
if (disabled) return { id: "", url: "", secret: "" }
log.info("creating share", { sessionID })
const req = yield* request()
const result = yield* HttpClientRequest.post(`${req.baseUrl}${req.api.create}`).pipe(
HttpClientRequest.setHeaders(req.headers),
HttpClientRequest.bodyJson({ sessionID }),
Effect.flatMap((r) => httpOk.execute(r)),
Effect.flatMap(HttpClientResponse.schemaBodyJson(ShareSchema)),
)
yield* db((db) =>
db
.insert(SessionShareTable)
.values({ session_id: sessionID, id: result.id, secret: result.secret, url: result.url })
.onConflictDoUpdate({
target: SessionShareTable.session_id,
set: { id: result.id, secret: result.secret, url: result.url },
})
.run(),
)
const s = yield* InstanceState.get(state)
yield* full(sessionID).pipe(
const next = new Map(data.map((item) => [key(item), item]))
s.queue.set(sessionID, { data: next })
yield* flush(sessionID).pipe(
Effect.delay(1000),
Effect.catchCause((cause) =>
Effect.sync(() => {
log.error("share full sync failed", { sessionID, cause })
log.error("share flush failed", { sessionID, cause })
}),
),
Effect.forkIn(s.scope),
)
return result
})
}
const remove = Effect.fn("ShareNext.remove")(function* (sessionID: SessionID) {
if (disabled) return
log.info("removing share", { sessionID })
const share = yield* get(sessionID)
if (!share) return
const state: InstanceState.InstanceState<State> = yield* InstanceState.make<State>(
Effect.fn("ShareNext.state")(function* (_ctx) {
const cache: State = { queue: new Map(), scope: yield* Scope.make() }
const req = yield* request()
yield* HttpClientRequest.delete(`${req.baseUrl}${req.api.remove(share.id)}`).pipe(
HttpClientRequest.setHeaders(req.headers),
HttpClientRequest.bodyJson({ secret: share.secret }),
Effect.flatMap((r) => httpOk.execute(r)),
yield* Effect.addFinalizer(() =>
Scope.close(cache.scope, Exit.void).pipe(
Effect.andThen(
Effect.sync(() => {
cache.queue.clear()
}),
),
),
)
yield* db((db) => db.delete(SessionShareTable).where(eq(SessionShareTable.session_id, sessionID)).run())
})
if (disabled) return cache
return Service.of({ init, url, request, create, remove })
}),
)
const watch = <D extends { type: string }>(
def: D,
fn: (evt: { properties: any }) => Effect.Effect<void, unknown>,
) =>
bus.subscribe(def as never).pipe(
Stream.runForEach((evt) =>
fn(evt).pipe(
Effect.catchCause((cause) =>
Effect.sync(() => {
log.error("share subscriber failed", { type: def.type, cause })
}),
),
),
),
Effect.forkScoped,
)
export const defaultLayer = layer.pipe(
Layer.provide(Bus.layer),
Layer.provide(Account.defaultLayer),
Layer.provide(Config.defaultLayer),
Layer.provide(FetchHttpClient.layer),
Layer.provide(Provider.defaultLayer),
Layer.provide(Session.defaultLayer),
)
}
yield* watch(Session.Event.Updated, (evt) =>
Effect.gen(function* () {
const info = yield* session.get(evt.properties.sessionID)
yield* sync(info.id, [{ type: "session", data: info }])
}),
)
yield* watch(MessageV2.Event.Updated, (evt) =>
Effect.gen(function* () {
const info = evt.properties.info
yield* sync(info.sessionID, [{ type: "message", data: info }])
if (info.role !== "user") return
const model = yield* provider.getModel(info.model.providerID, info.model.modelID)
yield* sync(info.sessionID, [{ type: "model", data: [model] }])
}),
)
yield* watch(MessageV2.Event.PartUpdated, (evt) =>
sync(evt.properties.part.sessionID, [{ type: "part", data: evt.properties.part }]),
)
yield* watch(Session.Event.Diff, (evt) =>
sync(evt.properties.sessionID, [{ type: "session_diff", data: evt.properties.diff }]),
)
yield* watch(Session.Event.Deleted, (evt) => remove(evt.properties.sessionID))
return cache
}),
)
const request = Effect.fn("ShareNext.request")(function* () {
const headers: Record<string, string> = {}
const active = yield* account.active()
if (Option.isNone(active) || !active.value.active_org_id) {
const baseUrl = (yield* cfg.get()).enterprise?.url ?? "https://opncd.ai"
return { headers, api: legacyApi, baseUrl } satisfies Req
}
const token = yield* account.token(active.value.id)
if (Option.isNone(token)) {
throw new Error("No active account token available for sharing")
}
headers.authorization = `Bearer ${token.value}`
headers["x-org-id"] = active.value.active_org_id
return { headers, api: consoleApi, baseUrl: active.value.url } satisfies Req
})
const get = Effect.fnUntraced(function* (sessionID: SessionID) {
const row = yield* db((db) =>
db.select().from(SessionShareTable).where(eq(SessionShareTable.session_id, sessionID)).get(),
)
if (!row) return
return { id: row.id, secret: row.secret, url: row.url } satisfies Share
})
const flush = Effect.fn("ShareNext.flush")(function* (sessionID: SessionID) {
if (disabled) return
const s = yield* InstanceState.get(state)
const queued = s.queue.get(sessionID)
if (!queued) return
s.queue.delete(sessionID)
const share = yield* get(sessionID)
if (!share) return
const req = yield* request()
const res = yield* HttpClientRequest.post(`${req.baseUrl}${req.api.sync(share.id)}`).pipe(
HttpClientRequest.setHeaders(req.headers),
HttpClientRequest.bodyJson({ secret: share.secret, data: Array.from(queued.data.values()) }),
Effect.flatMap((r) => http.execute(r)),
)
if (res.status >= 400) {
log.warn("failed to sync share", { sessionID, shareID: share.id, status: res.status })
}
})
const full = Effect.fn("ShareNext.full")(function* (sessionID: SessionID) {
log.info("full sync", { sessionID })
const info = yield* session.get(sessionID)
const diffs = yield* session.diff(sessionID)
const messages = yield* Effect.sync(() => Array.from(MessageV2.stream(sessionID)))
const models = yield* Effect.forEach(
Array.from(
new Map(
messages
.filter((msg) => msg.info.role === "user")
.map((msg) => (msg.info as SDK.UserMessage).model)
.map((item) => [`${item.providerID}/${item.modelID}`, item] as const),
).values(),
),
(item) => provider.getModel(ProviderID.make(item.providerID), ModelID.make(item.modelID)),
{ concurrency: 8 },
)
yield* sync(sessionID, [
{ type: "session", data: info },
...messages.map((item) => ({ type: "message" as const, data: item.info })),
...messages.flatMap((item) => item.parts.map((part) => ({ type: "part" as const, data: part }))),
{ type: "session_diff", data: diffs },
{ type: "model", data: models },
])
})
const init = Effect.fn("ShareNext.init")(function* () {
if (disabled) return
yield* InstanceState.get(state)
})
const url = Effect.fn("ShareNext.url")(function* () {
return (yield* request()).baseUrl
})
const create = Effect.fn("ShareNext.create")(function* (sessionID: SessionID) {
if (disabled) return { id: "", url: "", secret: "" }
log.info("creating share", { sessionID })
const req = yield* request()
const result = yield* HttpClientRequest.post(`${req.baseUrl}${req.api.create}`).pipe(
HttpClientRequest.setHeaders(req.headers),
HttpClientRequest.bodyJson({ sessionID }),
Effect.flatMap((r) => httpOk.execute(r)),
Effect.flatMap(HttpClientResponse.schemaBodyJson(ShareSchema)),
)
yield* db((db) =>
db
.insert(SessionShareTable)
.values({ session_id: sessionID, id: result.id, secret: result.secret, url: result.url })
.onConflictDoUpdate({
target: SessionShareTable.session_id,
set: { id: result.id, secret: result.secret, url: result.url },
})
.run(),
)
const s = yield* InstanceState.get(state)
yield* full(sessionID).pipe(
Effect.catchCause((cause) =>
Effect.sync(() => {
log.error("share full sync failed", { sessionID, cause })
}),
),
Effect.forkIn(s.scope),
)
return result
})
const remove = Effect.fn("ShareNext.remove")(function* (sessionID: SessionID) {
if (disabled) return
log.info("removing share", { sessionID })
const share = yield* get(sessionID)
if (!share) return
const req = yield* request()
yield* HttpClientRequest.delete(`${req.baseUrl}${req.api.remove(share.id)}`).pipe(
HttpClientRequest.setHeaders(req.headers),
HttpClientRequest.bodyJson({ secret: share.secret }),
Effect.flatMap((r) => httpOk.execute(r)),
)
yield* db((db) => db.delete(SessionShareTable).where(eq(SessionShareTable.session_id, sessionID)).run())
})
return Service.of({ init, url, request, create, remove })
}),
)
export const defaultLayer = layer.pipe(
Layer.provide(Bus.layer),
Layer.provide(Account.defaultLayer),
Layer.provide(Config.defaultLayer),
Layer.provide(FetchHttpClient.layer),
Layer.provide(Provider.defaultLayer),
Layer.provide(Session.defaultLayer),
)
+1 -1
View File
@@ -4,7 +4,7 @@ import { Effect } from "effect"
import { Tool } from "./tool"
import { Question } from "../question"
import { Session } from "../session"
import { MessageV2 } from "../session"
import { MessageV2 } from "../session/message-v2"
import { Provider } from "../provider"
import { Instance } from "../project/instance"
import { type SessionID, MessageID, PartID } from "../session/schema"
+1 -1
View File
@@ -11,7 +11,7 @@ import { FileTime } from "../file/time"
import DESCRIPTION from "./read.txt"
import { Instance } from "../project/instance"
import { assertExternalDirectoryEffect } from "./external-directory"
import { Instruction } from "../session"
import { Instruction } from "../session/instruction"
const DEFAULT_READ_LIMIT = 2000
const MAX_LINE_LENGTH = 2000
+2 -2
View File
@@ -37,10 +37,10 @@ import { Ripgrep } from "../file/ripgrep"
import { Format } from "../format"
import { InstanceState } from "@/effect"
import { Question } from "../question"
import { Todo } from "../session"
import { Todo } from "../session/todo"
import { LSP } from "../lsp"
import { FileTime } from "../file/time"
import { Instruction } from "../session"
import { Instruction } from "../session/instruction"
import { AppFileSystem } from "@opencode-ai/shared/filesystem"
import { Bus } from "../bus"
import { Agent } from "../agent/agent"
+2 -2
View File
@@ -3,9 +3,9 @@ import DESCRIPTION from "./task.txt"
import z from "zod"
import { Session } from "../session"
import { SessionID, MessageID } from "../session/schema"
import { MessageV2 } from "../session"
import { MessageV2 } from "../session/message-v2"
import { Agent } from "../agent/agent"
import type { SessionPrompt } from "../session"
import type { SessionPrompt } from "../session/prompt"
import { Config } from "../config"
import { Effect } from "effect"
+1 -1
View File
@@ -2,7 +2,7 @@ import z from "zod"
import { Effect } from "effect"
import { Tool } from "./tool"
import DESCRIPTION_WRITE from "./todowrite.txt"
import { Todo } from "../session"
import { Todo } from "../session/todo"
const parameters = z.object({
todos: z.array(z.object(Todo.Info.shape)).describe("The updated todo list"),
+1 -1
View File
@@ -1,6 +1,6 @@
import z from "zod"
import { Effect } from "effect"
import type { MessageV2 } from "../session"
import type { MessageV2 } from "../session/message-v2"
import type { Permission } from "../permission"
import type { SessionID, MessageID } from "../session/schema"
import { Truncate } from "./truncate"
@@ -1,6 +1,6 @@
import { test, expect, describe } from "bun:test"
import { extractResponseText, formatPromptTooLargeError } from "../../src/cli/cmd/github"
import type { MessageV2 } from "../../src/session"
import type { MessageV2 } from "../../src/session/message-v2"
import { SessionID, MessageID, PartID } from "../../src/session/schema"
// Helper to create minimal valid parts
@@ -3,7 +3,7 @@ import { Effect } from "effect"
import { Instance } from "../../src/project/instance"
import { Server } from "../../src/server/server"
import { Session as SessionNs } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import { MessageID, PartID, type SessionID } from "../../src/session/schema"
import { Log } from "../../src/util/log"
import { tmpdir } from "../fixture/fixture"
@@ -6,8 +6,8 @@ import z from "zod"
import { Bus } from "../../src/bus"
import { Config } from "../../src/config"
import { Agent } from "../../src/agent/agent"
import { LLM } from "../../src/session"
import { SessionCompaction } from "../../src/session"
import { LLM } from "../../src/session/llm"
import { SessionCompaction } from "../../src/session/compaction"
import { Token } from "../../src/util/token"
import { Instance } from "../../src/project/instance"
import { Log } from "../../src/util/log"
@@ -15,13 +15,13 @@ import { Permission } from "../../src/permission"
import { Plugin } from "../../src/plugin"
import { provideTmpdirInstance, tmpdir } from "../fixture/fixture"
import { Session as SessionNs } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import { MessageID, PartID, SessionID } from "../../src/session/schema"
import { SessionStatus } from "../../src/session"
import { SessionSummary } from "../../src/session"
import { SessionStatus } from "../../src/session/status"
import { SessionSummary } from "../../src/session/summary"
import { ModelID, ProviderID } from "../../src/provider/schema"
import type { Provider } from "../../src/provider"
import * as SessionProcessorModule from "../../src/session"
import * as SessionProcessorModule from "../../src/session/processor"
import { Snapshot } from "../../src/snapshot"
import { ProviderTest } from "../fake/provider"
import { testEffect } from "../lib/effect"
@@ -2,8 +2,8 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"
import path from "path"
import { Effect } from "effect"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { Instruction } from "../../src/session"
import type { MessageV2 } from "../../src/session"
import { Instruction } from "../../src/session/instruction"
import type { MessageV2 } from "../../src/session/message-v2"
import { Instance } from "../../src/project/instance"
import { MessageID, PartID, SessionID } from "../../src/session/schema"
import { Global } from "../../src/global"
+2 -2
View File
@@ -4,7 +4,7 @@ import { tool, type ModelMessage } from "ai"
import { Cause, Effect, Exit, Stream } from "effect"
import z from "zod"
import { makeRuntime } from "../../src/effect/run-service"
import { LLM } from "../../src/session"
import { LLM } from "../../src/session/llm"
import { Instance } from "../../src/project/instance"
import { Provider } from "../../src/provider"
import { ProviderTransform } from "../../src/provider/transform"
@@ -13,7 +13,7 @@ import { ProviderID, ModelID } from "../../src/provider/schema"
import { Filesystem } from "../../src/util/filesystem"
import { tmpdir } from "../fixture/fixture"
import type { Agent } from "../../src/agent/agent"
import { MessageV2 } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import { SessionID, MessageID } from "../../src/session/schema"
import { AppRuntime } from "../../src/effect/app-runtime"
@@ -1,6 +1,6 @@
import { describe, expect, test } from "bun:test"
import { APICallError } from "ai"
import { MessageV2 } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import type { Provider } from "../../src/provider"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { SessionID, MessageID, PartID } from "../../src/session/schema"
@@ -3,7 +3,7 @@ import { Effect } from "effect"
import path from "path"
import { Instance } from "../../src/project/instance"
import { Session as SessionNs } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import { MessageID, PartID, type SessionID } from "../../src/session/schema"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { Log } from "../../src/util/log"
@@ -11,12 +11,12 @@ import { Plugin } from "../../src/plugin"
import { Provider } from "../../src/provider"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { Session } from "../../src/session"
import { LLM } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { SessionProcessor } from "../../src/session"
import { LLM } from "../../src/session/llm"
import { MessageV2 } from "../../src/session/message-v2"
import { SessionProcessor } from "../../src/session/processor"
import { MessageID, PartID, SessionID } from "../../src/session/schema"
import { SessionStatus } from "../../src/session"
import { SessionSummary } from "../../src/session"
import { SessionStatus } from "../../src/session/status"
import { SessionSummary } from "../../src/session/summary"
import { Snapshot } from "../../src/snapshot"
import { Log } from "../../src/util/log"
import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner"
@@ -16,22 +16,22 @@ import { Provider as ProviderSvc } from "../../src/provider"
import { Env } from "../../src/env"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { Question } from "../../src/question"
import { Todo } from "../../src/session"
import { Todo } from "../../src/session/todo"
import { Session } from "../../src/session"
import { LLM } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { LLM } from "../../src/session/llm"
import { MessageV2 } from "../../src/session/message-v2"
import { AppFileSystem } from "@opencode-ai/shared/filesystem"
import { SessionCompaction } from "../../src/session"
import { SessionSummary } from "../../src/session"
import { Instruction } from "../../src/session"
import { SessionProcessor } from "../../src/session"
import { SessionPrompt } from "../../src/session"
import { SessionRevert } from "../../src/session"
import { SessionRunState } from "../../src/session"
import { SessionCompaction } from "../../src/session/compaction"
import { SessionSummary } from "../../src/session/summary"
import { Instruction } from "../../src/session/instruction"
import { SessionProcessor } from "../../src/session/processor"
import { SessionPrompt } from "../../src/session/prompt"
import { SessionRevert } from "../../src/session/revert"
import { SessionRunState } from "../../src/session/run-state"
import { MessageID, PartID, SessionID } from "../../src/session/schema"
import { SessionStatus } from "../../src/session"
import { SessionStatus } from "../../src/session/status"
import { Skill } from "../../src/skill"
import { SystemPrompt } from "../../src/session"
import { SystemPrompt } from "../../src/session/system"
import { Shell } from "../../src/shell/shell"
import { Snapshot } from "../../src/snapshot"
import { ToolRegistry } from "../../src/tool/registry"
@@ -6,8 +6,8 @@ import { Effect, Layer } from "effect"
import { Instance } from "../../src/project/instance"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { Session } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { SessionPrompt } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import { SessionPrompt } from "../../src/session/prompt"
import { Log } from "../../src/util/log"
import { tmpdir } from "../fixture/fixture"
+3 -3
View File
@@ -3,12 +3,12 @@ import type { NamedError } from "@opencode-ai/shared/util/error"
import { APICallError } from "ai"
import { setTimeout as sleep } from "node:timers/promises"
import { Effect, Schedule } from "effect"
import { SessionRetry } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { SessionRetry } from "../../src/session/retry"
import { MessageV2 } from "../../src/session/message-v2"
import { ProviderID } from "../../src/provider/schema"
import { AppRuntime } from "../../src/effect/app-runtime"
import { SessionID } from "../../src/session/schema"
import { SessionStatus } from "../../src/session"
import { SessionStatus } from "../../src/session/status"
import { Instance } from "../../src/project/instance"
import { tmpdir } from "../fixture/fixture"
@@ -4,8 +4,8 @@ import path from "path"
import { Effect, Layer } from "effect"
import { Session } from "../../src/session"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { SessionRevert } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { SessionRevert } from "../../src/session/revert"
import { MessageV2 } from "../../src/session/message-v2"
import { Snapshot } from "../../src/snapshot"
import { Log } from "../../src/util/log"
import { MessageID, PartID, SessionID } from "../../src/session/schema"
@@ -4,7 +4,7 @@ import { Session as SessionNs } from "../../src/session"
import { Bus } from "../../src/bus"
import { Log } from "../../src/util/log"
import { Instance } from "../../src/project/instance"
import { MessageV2 } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import { MessageID, PartID, type SessionID } from "../../src/session/schema"
import { AppRuntime } from "../../src/effect/app-runtime"
import { tmpdir } from "../fixture/fixture"
@@ -17,11 +17,11 @@ import { FetchHttpClient } from "effect/unstable/http"
import fs from "fs/promises"
import path from "path"
import { Session } from "../../src/session"
import { LLM } from "../../src/session"
import { SessionPrompt } from "../../src/session"
import { SessionRevert } from "../../src/session"
import { SessionSummary } from "../../src/session"
import { MessageV2 } from "../../src/session"
import { LLM } from "../../src/session/llm"
import { SessionPrompt } from "../../src/session/prompt"
import { SessionRevert } from "../../src/session/revert"
import { SessionSummary } from "../../src/session/summary"
import { MessageV2 } from "../../src/session/message-v2"
import { Log } from "../../src/util/log"
import { provideTmpdirServer } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
@@ -42,13 +42,13 @@ import { Provider as ProviderSvc } from "../../src/provider"
import { Env } from "../../src/env"
import { Question } from "../../src/question"
import { Skill } from "../../src/skill"
import { SystemPrompt } from "../../src/session"
import { Todo } from "../../src/session"
import { SessionCompaction } from "../../src/session"
import { Instruction } from "../../src/session"
import { SessionProcessor } from "../../src/session"
import { SessionRunState } from "../../src/session"
import { SessionStatus } from "../../src/session"
import { SystemPrompt } from "../../src/session/system"
import { Todo } from "../../src/session/todo"
import { SessionCompaction } from "../../src/session/compaction"
import { Instruction } from "../../src/session/instruction"
import { SessionProcessor } from "../../src/session/processor"
import { SessionRunState } from "../../src/session/run-state"
import { SessionStatus } from "../../src/session/status"
import { Snapshot } from "../../src/snapshot"
import { ToolRegistry } from "../../src/tool/registry"
import { Truncate } from "../../src/tool/truncate"
@@ -2,10 +2,10 @@ import { describe, expect, test } from "bun:test"
import path from "path"
import { Effect, Layer } from "effect"
import { Session } from "../../src/session"
import { SessionPrompt } from "../../src/session"
import { SessionPrompt } from "../../src/session/prompt"
import { Log } from "../../src/util/log"
import { Instance } from "../../src/project/instance"
import { MessageV2 } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
const projectRoot = path.join(__dirname, "../..")
Log.init({ print: false })
@@ -1,6 +1,6 @@
import { describe, expect, test } from "bun:test"
import { MessageV2 } from "../../src/session"
import { SessionPrompt } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import { SessionPrompt } from "../../src/session/prompt"
import { SessionID, MessageID } from "../../src/session/schema"
describe("structured-output.OutputFormat", () => {
@@ -3,7 +3,7 @@ import path from "path"
import { Effect } from "effect"
import { Agent } from "../../src/agent/agent"
import { Instance } from "../../src/project/instance"
import { SystemPrompt } from "../../src/session"
import { SystemPrompt } from "../../src/session/system"
import { provideInstance, tmpdir } from "../fixture/fixture"
function load<A>(dir: string, fn: (svc: Agent.Interface) => Effect.Effect<A>) {
@@ -12,7 +12,7 @@ import { Config } from "../../src/config"
import { Provider } from "../../src/provider"
import { Session } from "../../src/session"
import type { SessionID } from "../../src/session/schema"
import { ShareNext } from "../../src/share/share-next"
import { ShareNext } from "../../src/share"
import { SessionShareTable } from "../../src/share/share.sql"
import { Database, eq } from "../../src/storage/db"
import { provideTmpdirInstance } from "../fixture/fixture"
+1 -1
View File
@@ -9,7 +9,7 @@ import { LSP } from "../../src/lsp"
import { Permission } from "../../src/permission"
import { Instance } from "../../src/project/instance"
import { SessionID, MessageID } from "../../src/session/schema"
import { Instruction } from "../../src/session"
import { Instruction } from "../../src/session/instruction"
import { ReadTool } from "../../src/tool/read"
import { Truncate } from "../../src/tool/truncate"
import { Tool } from "../../src/tool/tool"
+2 -2
View File
@@ -5,8 +5,8 @@ import { Config } from "../../src/config"
import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner"
import { Instance } from "../../src/project/instance"
import { Session } from "../../src/session"
import { MessageV2 } from "../../src/session"
import type { SessionPrompt } from "../../src/session"
import { MessageV2 } from "../../src/session/message-v2"
import type { SessionPrompt } from "../../src/session/prompt"
import { MessageID, PartID } from "../../src/session/schema"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { TaskTool, type TaskPromptOps } from "../../src/tool/task"