Compare commits

...

21 Commits

Author SHA1 Message Date
Kit Langton 68861727f9 refactor(core): remove unused agent default method 2026-08-05 16:33:27 -04:00
Kit Langton 934935963d refactor(core): remove unused filesystem search methods (#40703) 2026-08-05 19:42:30 +00:00
Kit Langton f3912a2a8a refactor(app): remove unused UI helpers (#40686) 2026-08-05 15:15:15 -04:00
Kit Langton 45d58717a4 fix(core): bound protected home searches (#40518) 2026-08-05 15:15:02 -04:00
Kit Langton 74e3155ef0 refactor(session-ui): remove unused exported components (#40685) 2026-08-05 15:13:39 -04:00
Kit Langton 0af6c82563 refactor(core): remove orphaned message projector paths (#40705) 2026-08-05 15:13:15 -04:00
Kit Langton 686127f809 refactor(core): remove unreachable repository validation (#40700) 2026-08-05 14:30:29 -04:00
Kit Langton 5256655c4d refactor(core): remove obsolete project commit bridge (#40701) 2026-08-05 18:16:55 +00:00
Kit Langton d10b652637 refactor(core): remove unused formatter methods (#40684) 2026-08-05 13:45:23 -04:00
Kit Langton b03ca0d4e2 refactor(core): remove unused bus sequences (#40681) 2026-08-05 13:41:38 -04:00
Kit Langton 25aaea3d31 refactor(core): remove unused shell tree killer (#40680) 2026-08-05 13:38:59 -04:00
Kit Langton cae7a139bc refactor(server): remove obsolete auth header helpers (#40682) 2026-08-05 13:37:29 -04:00
Kit Langton 5ea62ab05f refactor(core): remove orphaned git change operations (#40683) 2026-08-05 13:32:03 -04:00
Kit Langton faadc05c88 refactor(core): remove unused snapshot operations (#40687) 2026-08-05 13:01:48 -04:00
Kit Langton 3e253c589e refactor(core): persist instruction updates as messages (#40679) 2026-08-05 12:55:29 -04:00
Kit Langton 0a0fc09533 refactor(core): remove unused file mutation methods (#40667) 2026-08-05 12:33:20 -04:00
Kit Langton 5aa0413fea refactor(core): remove dead fork pending copy (#40675) 2026-08-05 12:30:42 -04:00
Kit Langton 6f4c199629 fix(tui): wait for diff request in tests (#40670) 2026-08-05 16:02:22 +00:00
Kit Langton ed8e1f4654 fix(core): reconcile promoted prompt retries from messages (#40664) 2026-08-05 11:43:03 -04:00
Kit Langton 3b0195e045 fix(core): avoid eager directory snapshots (#40552) 2026-08-05 14:52:07 +00:00
Shoubhit Dash 143a776373 fix(acp): surface subagent activity (#40438) 2026-08-05 18:57:15 +05:30
62 changed files with 943 additions and 1726 deletions
-5
View File
@@ -248,11 +248,6 @@ export function formatKeybind(config: string, t?: (key: KeyLabel) => string): st
return IS_MAC ? parts.join("") : parts.join("+")
}
// KeybindV2 takes an array instead of a string
export function formatKeybindKeys(config: string, t?: (key: KeyLabel) => string): string[] {
return formatKeybindParts(config, t)
}
function isEditableTarget(target: EventTarget | null) {
if (!(target instanceof HTMLElement)) return false
if (target.isContentEditable) return true
-7
View File
@@ -286,13 +286,6 @@ export const { use: useFile, provider: FileProvider } = createSimpleContext({
children: tree.children,
expand: tree.expandDir,
collapse: tree.collapseDir,
toggle(input: string) {
if (tree.dirState(input)?.expanded) {
tree.collapseDir(input)
return
}
tree.expandDir(input)
},
},
get,
load,
@@ -153,18 +153,6 @@ export function normalizeProviderList(
}
}
export function sanitizeProject(project: Project) {
if (!project.icon?.url && !project.icon?.override) return project
return {
...project,
icon: {
...project.icon,
url: undefined,
override: undefined,
},
}
}
export function normalizeProjectInfo(project: Project | CurrentProject): Project {
return {
...project,
-30
View File
@@ -753,9 +753,6 @@ export const { use: useLayout, provider: LayoutProvider } = createSimpleContext(
},
mobileSidebar: {
opened: createMemo(() => store.mobileSidebar?.opened ?? false),
show() {
setStore("mobileSidebar", "opened", true)
},
hide() {
setStore("mobileSidebar", "opened", false)
},
@@ -961,33 +958,6 @@ export const { use: useLayout, provider: LayoutProvider } = createSimpleContext(
if (current.reviewOpen.includes(path)) return
setStore("sessionView", session, "reviewOpen", current.reviewOpen.length, path)
},
closePath(path: string) {
const session = key()
const current = store.sessionView[session]?.reviewOpen
if (!current) return
const index = current.indexOf(path)
if (index === -1) return
setStore(
"sessionView",
session,
"reviewOpen",
produce((draft) => {
if (!draft) return
draft.splice(index, 1)
}),
)
},
togglePath(path: string) {
const session = key()
const current = store.sessionView[session]?.reviewOpen
if (!current || !current.includes(path)) {
this.openPath(path)
return
}
this.closePath(path)
},
},
}
},
@@ -22,8 +22,6 @@ type TabsInput = {
fileBrowser?: Accessor<boolean>
}
export const getSessionKey = (dir: string | undefined, id: string | undefined) => `${dir ?? ""}${id ? `/${id}` : ""}`
export function shouldShowFileTree(input: { visible: boolean; opened: boolean }) {
return input.opened && input.visible
}
+193 -38
View File
@@ -1,4 +1,4 @@
import type { AgentSideConnection, PromptResponse } from "@agentclientprotocol/sdk"
import type { AgentSideConnection, PromptResponse, SessionUpdate } from "@agentclientprotocol/sdk"
import type {
EventSubscribeOutput,
OpenCodeClient,
@@ -37,6 +37,34 @@ export type TurnStart =
| { readonly type: "skill"; readonly id: string }
| { readonly type: "compaction"; readonly id: string }
export const ChildSessionUpdatesCapability = "opencode/child-session-updates"
export const ChildSessionUpdateMethod = "opencode/session/child_update"
type ChildSessionUpdateBase = {
readonly rootSessionId: string
readonly childSessionId: string
readonly parentSessionId: string
readonly depth: number
readonly title?: string
}
type ChildSessionEvent =
| { readonly type: "update"; readonly update: SessionUpdate }
| {
readonly type: "status"
readonly status: "created" | "running" | "completed" | "failed" | "interrupted"
readonly error?: { readonly type: string; readonly message: string }
}
export type ChildSessionUpdate = ChildSessionUpdateBase & ChildSessionEvent
type ChildSession = {
readonly id: string
readonly parentID: string
readonly depth: number
readonly title?: string
}
function emptyToolState(): ToolState {
return { name: "tool", input: {}, metadata: {}, content: [] }
}
@@ -50,8 +78,13 @@ export async function streamTurn(input: {
readonly writeTextFile: boolean
readonly submit: (signal: AbortSignal) => Promise<unknown>
readonly control: TurnControl
readonly childSessionUpdate?: (update: ChildSessionUpdate) => Promise<void>
readonly connectionSignal?: AbortSignal
readonly sessionSignal?: AbortSignal
}): Promise<PromptResponse> {
const streamController = new AbortController()
const connectionAbort = () => streamController.abort()
input.connectionSignal?.addEventListener("abort", connectionAbort, { once: true })
const stream = input.client.event.subscribe({ signal: streamController.signal })[Symbol.asyncIterator]()
const connected = await stream.next()
if (connected.done) throw new Error("event stream disconnected before prompt admission")
@@ -62,47 +95,101 @@ export async function streamTurn(input: {
let finish: SessionMessageAssistant["finish"]
let executionError: { readonly type: string; readonly message: string } | undefined
const tools = new Map<string, ToolState>()
const children = new Map<string, ChildSession>()
const openChildren = new Set<string>()
let handedOff = false
const update = (value: Parameters<Connection["sessionUpdate"]>[0]["update"]) =>
input.connection.sessionUpdate({ sessionId: input.sessionID, update: value })
const notifyChild = async (child: ChildSession, value: ChildSessionEvent) => {
if (!input.childSessionUpdate) return
await input
.childSessionUpdate({
rootSessionId: input.sessionID,
childSessionId: child.id,
parentSessionId: child.parentID,
depth: child.depth,
...(child.title ? { title: child.title } : {}),
...value,
})
.catch(() => {})
}
const consume = async () => {
const updateSession = async (value: SessionUpdate, child: ChildSession | undefined, mode: "turn" | "background") => {
const projected = child ? projectChildUpdate(value, child) : value
if (mode === "turn" && (!child || !input.childSessionUpdate)) {
await input.connection.sessionUpdate({ sessionId: input.sessionID, update: projected })
}
if (child) await notifyChild(child, { type: "update", update: projected })
}
const consume = async (mode: "turn" | "background") => {
while (!streamController.signal.aborted) {
const next = await stream.next()
if (next.done) throw new Error("event stream disconnected during prompt execution")
const event = next.value
if (event.type === "permission.asked" && event.data.sessionID === input.sessionID) {
const tool = event.data.source?.id ? tools.get(event.data.source.id) : undefined
if (event.type === "session.created") {
const parentID = event.data.info.parentID
if (!parentID) continue
const parent = parentID === input.sessionID ? undefined : children.get(parentID)
if ((mode === "turn" && parentID === input.sessionID) || parent) {
const child = {
id: event.data.sessionID,
parentID,
depth: parent ? parent.depth + 1 : 1,
title: event.data.info.title,
}
children.set(child.id, child)
openChildren.add(child.id)
await notifyChild(child, { type: "status", status: "created" })
}
continue
}
const eventSessionID = sessionIDFromEvent(event)
const child = eventSessionID ? children.get(eventSessionID) : undefined
const send = (update: SessionUpdate) => updateSession(update, child, mode)
if (mode === "background" && !child) continue
if (event.type === "permission.asked" && (event.data.sessionID === input.sessionID || child)) {
const tool = event.data.source?.id ? tools.get(toolKey(event.data.sessionID, event.data.source.id)) : undefined
await replyPermission({
client: input.client,
connection: input.connection,
event,
sessionID: input.sessionID,
sessionID: event.data.sessionID,
clientSessionID: input.sessionID,
cwd: input.cwd,
tool,
...(child ? { toolCallPrefix: child.id, titlePrefix: child.title } : {}),
})
continue
}
if (event.type === "form.created" && event.data.form.sessionID === input.sessionID) {
if (event.type === "form.created" && (event.data.form.sessionID === input.sessionID || child)) {
await input.client.form
.cancel({ sessionID: input.sessionID, formID: event.data.form.id })
.catch(() => input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {}))
.cancel({ sessionID: event.data.form.sessionID, formID: event.data.form.id })
.catch(() => input.client.session.interrupt({ sessionID: event.data.form.sessionID }).catch(() => {}))
continue
}
if (!("sessionID" in event.data) || event.data.sessionID !== input.sessionID) continue
if (!eventSessionID || (eventSessionID !== input.sessionID && !child)) continue
if (matchesStart(event, input.start)) {
started = true
continue
}
if (!started) continue
if (event.type === "session.execution.started") {
if (child) {
await notifyChild(child, { type: "status", status: "running" })
}
continue
}
if (event.type === "session.step.started") {
assistantMessageID = event.data.assistantMessageID
if (!child) assistantMessageID = event.data.assistantMessageID
continue
}
if (event.type === "session.text.delta") {
assistantMessageID = event.data.assistantMessageID
await update({
if (!child) assistantMessageID = event.data.assistantMessageID
await send({
sessionUpdate: "agent_message_chunk",
messageId: event.data.assistantMessageID,
content: { type: "text", text: event.data.delta },
@@ -110,8 +197,8 @@ export async function streamTurn(input: {
continue
}
if (event.type === "session.reasoning.delta") {
assistantMessageID = event.data.assistantMessageID
await update({
if (!child) assistantMessageID = event.data.assistantMessageID
await send({
sessionUpdate: "agent_thought_chunk",
messageId: event.data.assistantMessageID,
content: { type: "text", text: event.data.delta },
@@ -119,9 +206,14 @@ export async function streamTurn(input: {
continue
}
if (event.type === "session.tool.input.started") {
assistantMessageID = event.data.assistantMessageID
tools.set(event.data.id, { name: event.data.name, input: {}, metadata: {}, content: [] })
await update({
if (!child) assistantMessageID = event.data.assistantMessageID
tools.set(toolKey(event.data.sessionID, event.data.id), {
name: event.data.name,
input: {},
metadata: {},
content: [],
})
await send({
sessionUpdate: "tool_call",
...pendingToolCall({
toolCallId: event.data.id,
@@ -133,11 +225,12 @@ export async function streamTurn(input: {
continue
}
if (event.type === "session.tool.called") {
assistantMessageID = event.data.assistantMessageID
const current = tools.get(event.data.id) ?? emptyToolState()
if (!child) assistantMessageID = event.data.assistantMessageID
const key = toolKey(event.data.sessionID, event.data.id)
const current = tools.get(key) ?? emptyToolState()
current.input = event.data.input
tools.set(event.data.id, current)
await update({
tools.set(key, current)
await send({
sessionUpdate: "tool_call_update",
...runningToolUpdate({
toolCallId: event.data.id,
@@ -149,10 +242,10 @@ export async function streamTurn(input: {
continue
}
if (event.type === "session.tool.progress") {
const current = tools.get(event.data.id)
const current = tools.get(toolKey(event.data.sessionID, event.data.id))
if (!current) continue
current.metadata = event.data.metadata
await update({
await send({
sessionUpdate: "tool_call_update",
...runningToolUpdate({
toolCallId: event.data.id,
@@ -164,8 +257,9 @@ export async function streamTurn(input: {
continue
}
if (event.type === "session.tool.success") {
const current = tools.get(event.data.id) ?? emptyToolState()
tools.delete(event.data.id)
const key = toolKey(event.data.sessionID, event.data.id)
const current = tools.get(key) ?? emptyToolState()
tools.delete(key)
await syncEditedFiles({
connection: input.connection,
writeTextFile: input.writeTextFile,
@@ -175,7 +269,7 @@ export async function streamTurn(input: {
toolInput: current.input,
metadata: event.data.metadata ?? {},
}).catch(() => {})
await update({
await send({
sessionUpdate: "tool_call_update",
...completedToolUpdate({
toolCallId: event.data.id,
@@ -188,9 +282,10 @@ export async function streamTurn(input: {
continue
}
if (event.type === "session.tool.failed") {
const current = tools.get(event.data.id) ?? emptyToolState()
tools.delete(event.data.id)
await update({
const key = toolKey(event.data.sessionID, event.data.id)
const current = tools.get(key) ?? emptyToolState()
tools.delete(key)
await send({
sessionUpdate: "tool_call_update",
...errorToolUpdate({
toolCallId: event.data.id,
@@ -205,13 +300,33 @@ export async function streamTurn(input: {
continue
}
if (event.type === "session.step.ended") {
assistantMessageID = event.data.assistantMessageID
finish = event.data.finish
if (!child) {
assistantMessageID = event.data.assistantMessageID
finish = event.data.finish
}
continue
}
if (event.type === "session.execution.succeeded") {
if (!child) return "succeeded" as const
openChildren.delete(child.id)
await notifyChild(child, { type: "status", status: "completed" })
if (mode === "background" && openChildren.size === 0) return "succeeded" as const
continue
}
if (event.type === "session.execution.interrupted") {
if (!child) return "interrupted" as const
openChildren.delete(child.id)
await notifyChild(child, { type: "status", status: "interrupted" })
if (mode === "background" && openChildren.size === 0) return "interrupted" as const
continue
}
if (event.type === "session.execution.succeeded") return "succeeded" as const
if (event.type === "session.execution.interrupted") return "interrupted" as const
if (event.type === "session.execution.failed") {
if (child) {
openChildren.delete(child.id)
await notifyChild(child, { type: "status", status: "failed", error: event.data.error })
if (mode === "background" && openChildren.size === 0) return "failed" as const
continue
}
executionError = event.data.error
return "failed" as const
}
@@ -219,7 +334,13 @@ export async function streamTurn(input: {
return "interrupted" as const
}
const completed = consume()
const completed = consume("turn")
const closeStream = async () => {
streamController.abort()
input.connectionSignal?.removeEventListener("abort", connectionAbort)
input.sessionSignal?.removeEventListener("abort", connectionAbort)
await stream.return?.(undefined).catch(() => {})
}
try {
await input.submit(control.admission.signal).catch((error) => {
if (!control.cancelled) throw error
@@ -233,6 +354,13 @@ export async function streamTurn(input: {
}
}
const terminal = await completed
if (input.childSessionUpdate && openChildren.size > 0 && !input.sessionSignal?.aborted) {
handedOff = true
input.sessionSignal?.addEventListener("abort", connectionAbort, { once: true })
void consume("background")
.catch(() => {})
.finally(closeStream)
}
const assistant = assistantMessageID
? await input.client.session
.message({ sessionID: input.sessionID, messageID: assistantMessageID })
@@ -250,11 +378,38 @@ export async function streamTurn(input: {
await completed.catch(() => {})
throw error
} finally {
streamController.abort()
await stream.return?.(undefined).catch(() => {})
if (!handedOff) await closeStream()
}
}
function sessionIDFromEvent(event: EventSubscribeOutput) {
if ("sessionID" in event.data && typeof event.data.sessionID === "string") return event.data.sessionID
if (event.type === "form.created") return event.data.form.sessionID
return undefined
}
function toolKey(sessionID: string, id: string) {
return `${sessionID}:${id}`
}
function projectChildUpdate(update: SessionUpdate, child: ChildSession) {
const projected = { ...update }
projected._meta = {
...projected._meta,
"opencode/child-session": {
id: child.id,
parentID: child.parentID,
depth: child.depth,
...(child.title ? { title: child.title } : {}),
},
}
if (projected.sessionUpdate === "tool_call" || projected.sessionUpdate === "tool_call_update") {
projected.toolCallId = `${child.id}:${projected.toolCallId}`
if (projected.title && child.title) projected.title = `${child.title}: ${projected.title}`
}
return projected
}
export async function replayMessages(
connection: Pick<AgentSideConnection, "sessionUpdate">,
sessionID: string,
+17 -3
View File
@@ -20,20 +20,28 @@ export async function replyPermission(input: {
readonly connection: Connection
readonly event: PermissionEvent
readonly sessionID: string
readonly clientSessionID?: string
readonly cwd: string
readonly tool?: Tool
readonly toolCallPrefix?: string
readonly titlePrefix?: string
}) {
const toolName = input.tool?.name ?? input.event.data.action
const toolInput = { ...input.event.data.metadata, ...input.tool?.input }
const previews = await permissionPreviews(toolName, toolInput, input.cwd)
const toolCallID = input.event.data.source?.id ?? input.event.data.id
const title = permissionTitle(toolName, toolInput, previews)
const result = await input.connection
.requestPermission({
sessionId: input.sessionID,
sessionId: input.clientSessionID ?? input.sessionID,
toolCall: {
...pendingToolCall({
toolCallId: input.event.data.source?.id ?? input.event.data.id,
toolCallId: input.toolCallPrefix ? `${input.toolCallPrefix}:${toolCallID}` : toolCallID,
toolName,
state: { input: toolInput, title: permissionTitle(toolName, toolInput, previews) },
state: {
input: toolInput,
title: prefixedTitle(input.titlePrefix, title),
},
cwd: input.cwd,
}),
locations: permissionLocations(toolName, toolInput, input.event.data.resources, input.cwd, previews),
@@ -51,6 +59,12 @@ export async function replyPermission(input: {
})
}
function prefixedTitle(prefix: string | undefined, title: string | undefined) {
if (!prefix) return title
if (!title) return prefix
return `${prefix}: ${title}`
}
export async function syncEditedFiles(input: {
readonly connection: Partial<Pick<AgentSideConnection, "writeTextFile">>
readonly writeTextFile: boolean
+32 -7
View File
@@ -43,13 +43,21 @@ import { OPENCODE_VERSION } from "../version"
import { SessionMessage } from "@opencode-ai/schema/session-message"
import { buildConfigOptions, parseModelSelection, type ConfigOptionProvider } from "./config-option"
import { promptContentToParts } from "./content"
import { replayMessages, streamTurn, type TurnControl, type TurnStart } from "./event"
import {
ChildSessionUpdateMethod,
ChildSessionUpdatesCapability,
replayMessages,
streamTurn,
type ChildSessionUpdate,
type TurnControl,
type TurnStart,
} from "./event"
import { ACPError } from "./error"
export const AuthMethodID = "opencode-login"
type Connection = Pick<AgentSideConnection, "sessionUpdate" | "requestPermission"> &
Partial<Pick<AgentSideConnection, "writeTextFile">>
Partial<Pick<AgentSideConnection, "writeTextFile" | "extNotification" | "signal">>
type Catalog = {
readonly providers: ConfigOptionProvider[]
@@ -64,6 +72,7 @@ type Catalog = {
type Attached = {
readonly id: string
readonly cwd: string
readonly abort: AbortController
catalog: Catalog
model: ModelRef
modeID: string
@@ -100,7 +109,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
const catalogs = new Map<string, Promise<Catalog>>()
const registeredMcp = new Map<string, Set<string>>()
const active = new Map<string, TurnControl>()
const capabilities = { writeTextFile: false }
const capabilities = { writeTextFile: false, childSessionUpdates: false }
const catalog = (cwd: string) => {
const cached = catalogs.get(cwd)
@@ -119,11 +128,19 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
throw new ACPError.SessionNotFoundError({ sessionId: sessionID })
}
const detach = (sessionID: string) => {
sessions.get(sessionID)?.abort.abort()
sessions.delete(sessionID)
registeredMcp.delete(sessionID)
}
const attach = async (session: SessionInfo, cwd: string, mcpServers: readonly McpServer[]) => {
const currentCatalog = await catalog(cwd)
sessions.get(session.id)?.abort.abort()
const state: Attached = {
id: session.id,
cwd,
abort: new AbortController(),
catalog: currentCatalog,
model: session.model ?? currentCatalog.defaultModel,
modeID: session.agent ?? currentCatalog.defaultModeID,
@@ -161,6 +178,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
return {
initialize: async (params) => {
capabilities.writeTextFile = params.clientCapabilities?.fs?.writeTextFile === true
capabilities.childSessionUpdates = params.clientCapabilities?._meta?.[ChildSessionUpdatesCapability] === true
const authMethod: AuthMethod = {
description: "Run `opencode auth login` in the terminal",
name: "Login with opencode",
@@ -178,6 +196,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
mcpCapabilities: { http: true, sse: false },
promptCapabilities: { embeddedContext: true, image: true },
sessionCapabilities: { close: {}, delete: {}, fork: {}, list: {}, resume: {} },
_meta: { [ChildSessionUpdatesCapability]: true },
},
authMethods: [authMethod],
agentInfo: { name: "OpenCode", version: OPENCODE_VERSION },
@@ -224,8 +243,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
await input.client.session.remove({ sessionID: params.sessionId }).catch((error) => {
if (!isSessionNotFoundError(error)) throw error
})
sessions.delete(params.sessionId)
registeredMcp.delete(params.sessionId)
detach(params.sessionId)
return {}
},
resumeSession: async (params) => {
@@ -234,8 +252,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
return { configOptions: configOptions(state) }
},
closeSession: async (params) => {
sessions.delete(params.sessionId)
registeredMcp.delete(params.sessionId)
detach(params.sessionId)
const turn = active.get(params.sessionId)
if (turn) {
turn.cancelled = true
@@ -296,6 +313,11 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
const messageID = SessionMessage.ID.create()
const prepared = preparePrompt(state.catalog, params.prompt, messageID)
const control: TurnControl = { cancelled: false, admission: new AbortController() }
const extNotification = input.connection.extNotification
const childSessionUpdate =
capabilities.childSessionUpdates && extNotification
? (update: ChildSessionUpdate) => extNotification(ChildSessionUpdateMethod, update).then(() => {})
: undefined
active.set(state.id, control)
const response = await streamTurn({
client: input.client,
@@ -305,7 +327,10 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
start: prepared.start,
writeTextFile: capabilities.writeTextFile,
control,
connectionSignal: input.connection.signal,
sessionSignal: state.abort.signal,
submit: (signal) => submitPrompt(input.client, state, prepared, signal),
...(childSessionUpdate ? { childSessionUpdate } : {}),
}).finally(() => {
if (active.get(state.id) === control) active.delete(state.id)
})
+191 -1
View File
@@ -2,7 +2,7 @@ import { describe, expect, test } from "bun:test"
import type { AgentSideConnection } from "@agentclientprotocol/sdk"
import type { SessionMessageInfo } from "@opencode-ai/client/promise"
import { resolve } from "node:path"
import { replayMessages, streamTurn, type TurnControl } from "../../src/acp/event"
import { replayMessages, streamTurn, type ChildSessionUpdate, type TurnControl } from "../../src/acp/event"
import { createSseFixture, durableEvent, ephemeralEvent, withTimeout } from "./sse-fixture"
type SessionUpdateParams = Parameters<AgentSideConnection["sessionUpdate"]>[0]
@@ -191,6 +191,181 @@ describe("acp event behavior", () => {
}
})
test("projects foreground child session updates onto the parent turn", async () => {
const updates: SessionUpdateParams[] = []
const fixture = createSseFixture({
onPrompt({ id, send }) {
send(durableEvent("session.input.promoted", { sessionID: "ses_parent", inputID: id }))
send(
durableEvent("session.created", {
sessionID: "ses_child",
info: childSession("ses_child", "ses_parent", "Explore code"),
}),
)
send(durableEvent("session.execution.started", { sessionID: "ses_child" }))
send(
durableEvent("session.tool.input.started", {
sessionID: "ses_child",
assistantMessageID: "msg_child",
id: "call_read",
name: "read",
}),
)
send(
durableEvent("session.tool.called", {
sessionID: "ses_child",
assistantMessageID: "msg_child",
id: "call_read",
input: { path: "/workspace/src/index.ts" },
executed: false,
}),
)
send(
durableEvent("session.tool.success", {
sessionID: "ses_child",
assistantMessageID: "msg_child",
id: "call_read",
metadata: {},
content: [{ type: "text", text: "source" }],
executed: true,
}),
)
send(durableEvent("session.execution.succeeded", { sessionID: "ses_child" }))
send(durableEvent("session.execution.succeeded", { sessionID: "ses_parent" }))
},
})
try {
const response = await turn({
fixture,
connection: recordingConnection(updates),
sessionID: "ses_parent",
inputID: "input_parent",
})
expect(updates.map((item) => [item.sessionId, item.update.sessionUpdate])).toEqual([
["ses_parent", "tool_call"],
["ses_parent", "tool_call_update"],
["ses_parent", "tool_call_update"],
])
expect(updates.map((item) => ("toolCallId" in item.update ? item.update.toolCallId : undefined))).toEqual([
"ses_child:call_read",
"ses_child:call_read",
"ses_child:call_read",
])
expect(updates[0]?.update).toMatchObject({
title: "Explore code: read",
_meta: {
"opencode/child-session": {
id: "ses_child",
parentID: "ses_parent",
depth: 1,
title: "Explore code",
},
},
})
expect(response.stopReason).toBe("end_turn")
} finally {
await fixture.stop()
}
})
test("continues child extension updates after the parent turn ends", async () => {
const updates: SessionUpdateParams[] = []
const childUpdates: ChildSessionUpdate[] = []
const completed = Promise.withResolvers<void>()
const fixture = createSseFixture({
onPrompt({ id, send }) {
send(durableEvent("session.input.promoted", { sessionID: "ses_parent", inputID: id }))
send(
durableEvent("session.created", {
sessionID: "ses_background",
info: childSession("ses_background", "ses_parent", "Background research"),
}),
)
send(durableEvent("session.execution.succeeded", { sessionID: "ses_parent" }))
},
})
try {
const response = await turn({
fixture,
connection: recordingConnection(updates),
sessionID: "ses_parent",
inputID: "input_parent",
childSessionUpdate: async (update) => {
childUpdates.push(update)
if (update.type === "status" && update.status === "completed") completed.resolve()
},
})
expect(response.stopReason).toBe("end_turn")
fixture.send(
durableEvent("session.created", {
sessionID: "ses_future",
info: childSession("ses_future", "ses_parent", "Later turn child"),
}),
)
fixture.send(durableEvent("session.execution.started", { sessionID: "ses_future" }))
fixture.send(durableEvent("session.execution.started", { sessionID: "ses_background" }))
fixture.send(
durableEvent("session.tool.input.started", {
sessionID: "ses_background",
assistantMessageID: "msg_background",
id: "call_shell",
name: "shell",
}),
)
fixture.send(
durableEvent("session.tool.called", {
sessionID: "ses_background",
assistantMessageID: "msg_background",
id: "call_shell",
input: { command: "pwd" },
executed: false,
}),
)
fixture.send(
durableEvent("session.tool.success", {
sessionID: "ses_background",
assistantMessageID: "msg_background",
id: "call_shell",
metadata: { exit: 0 },
content: [{ type: "text", text: "/workspace" }],
executed: true,
}),
)
fixture.send(durableEvent("session.execution.succeeded", { sessionID: "ses_background" }))
await withTimeout(completed.promise, "background child completion was not delivered")
expect(updates).toEqual([])
expect(
childUpdates.map((update) =>
update.type === "status" ? [update.type, update.status] : [update.type, update.update.sessionUpdate],
),
).toEqual([
["status", "created"],
["status", "running"],
["update", "tool_call"],
["update", "tool_call_update"],
["update", "tool_call_update"],
["status", "completed"],
])
expect(childUpdates[2]).toMatchObject({
rootSessionId: "ses_parent",
childSessionId: "ses_background",
parentSessionId: "ses_parent",
depth: 1,
title: "Background research",
type: "update",
update: { toolCallId: "ses_background:call_shell" },
})
expect(childUpdates.some((update) => update.childSessionId === "ses_future")).toBe(false)
} finally {
await fixture.stop()
}
})
test("streams tool pending, progress, success, and failure updates", async () => {
const updates: SessionUpdateParams[] = []
const fixture = createSseFixture({
@@ -556,6 +731,7 @@ function turn(input: {
readonly connection: Connection
readonly sessionID: string
readonly inputID: string
readonly childSessionUpdate?: (update: ChildSessionUpdate) => Promise<void>
}) {
return streamTurn({
client: input.fixture.client,
@@ -565,11 +741,25 @@ function turn(input: {
start: { type: "input", id: input.inputID },
writeTextFile: false,
control: { cancelled: false, admission: new AbortController() },
childSessionUpdate: input.childSessionUpdate,
submit: (signal) =>
input.fixture.client.session.prompt({ sessionID: input.sessionID, id: input.inputID, text: "hello" }, { signal }),
})
}
function childSession(id: string, parentID: string, title: string) {
return {
id,
slug: id,
projectID: "project",
directory: "/workspace",
parentID,
title,
version: "test",
time: { created: 1, updated: 1 },
}
}
function tokens() {
return { input: 1, output: 1, reasoning: 0, cache: { read: 0, write: 0 } }
}
@@ -153,6 +153,68 @@ describe("acp permission behavior", () => {
}
})
test("routes foreground child permissions through the parent ACP session", async () => {
const permissionRequests: RequestPermissionRequest[] = []
const fixture = createSseFixture({
onPrompt({ id, send }) {
send(durableEvent("session.input.promoted", { sessionID: "ses_parent", inputID: id }))
send(
durableEvent("session.created", {
sessionID: "ses_child",
info: {
id: "ses_child",
slug: "ses_child",
projectID: "project",
directory: "/workspace",
parentID: "ses_parent",
title: "Review code",
version: "test",
time: { created: 1, updated: 1 },
},
}),
)
send(durableEvent("session.execution.started", { sessionID: "ses_child" }))
send(
permissionAsked("ses_child", "perm_child", {
action: "read",
metadata: { path: "/workspace/child.ts" },
source: { type: "tool", messageID: "msg_child", id: "call_child" },
}),
)
send(durableEvent("session.execution.succeeded", { sessionID: "ses_child" }))
send(durableEvent("session.execution.succeeded", { sessionID: "ses_parent" }))
},
})
const connection = {
sessionUpdate: async () => {},
requestPermission: async (request) => {
permissionRequests.push(request)
return { outcome: { outcome: "selected", optionId: "once" } } as const
},
} satisfies Connection
try {
await startTurn(fixture, connection, "ses_parent", "input_parent")
expect(permissionRequests).toHaveLength(1)
expect(permissionRequests[0]).toMatchObject({
sessionId: "ses_parent",
toolCall: {
toolCallId: "ses_child:call_child",
title: "Review code: /workspace/child.ts",
},
})
expect(fixture.requests).toContainEqual(
expect.objectContaining({
method: "POST",
path: "/api/session/ses_child/permission/perm_child/reply",
}),
)
} finally {
await fixture.stop()
}
})
test("previews edits during approval and syncs the completed file", async () => {
const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-acp-permission-"))
const file = path.join(cwd, "file.ts")
+7
View File
@@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"
import type { AgentSideConnection } from "@agentclientprotocol/sdk"
import { OpenCode } from "@opencode-ai/client/promise"
import { ACPService } from "../../src/acp/service"
import { ChildSessionUpdatesCapability } from "../../src/acp/event"
describe("acp service", () => {
test("creates a v2 session, registers mcp, and publishes commands", async () => {
@@ -39,11 +40,17 @@ describe("acp service", () => {
})
try {
const initialized = await service.initialize({
protocolVersion: 1,
clientCapabilities: { _meta: { [ChildSessionUpdatesCapability]: true } },
clientInfo: { name: "test", version: "1" },
})
const result = await service.newSession({
cwd: "/workspace",
mcpServers: [{ name: "docs", command: "bun", args: ["docs.ts"], env: [{ name: "TOKEN", value: "x" }] }],
})
expect(result.sessionId).toBe("ses_acp")
expect(initialized.agentCapabilities?._meta).toEqual({ [ChildSessionUpdatesCapability]: true })
expect(result.configOptions?.map((option) => option.id)).toEqual(["model", "effort", "mode"])
expect(requests).toContainEqual({
method: "PUT",
+1
View File
@@ -420,6 +420,7 @@ export type Endpoint5_26Output =
readonly data: {
readonly sessionID: Session.ID
readonly delta: { readonly [x: string]: (string & Brand.Brand<"Instruction.Hash">) | "removed" }
readonly text?: string | undefined
}
}
| {
@@ -676,7 +676,7 @@ export type SessionInstructionsUpdated = {
type: "session.instructions.updated"
durable: { aggregateID: string; seq: number; version: 2 }
location?: LocationRef
data: { sessionID: string; delta: { [x: string]: string | "removed" } }
data: { sessionID: string; delta: { [x: string]: string | "removed" }; text?: string }
}
export type SessionSynthetic = {
-4
View File
@@ -44,7 +44,6 @@ export type Draft = {
export interface Interface extends State.Transformable<Draft> {
readonly get: (id: ID) => Effect.Effect<Info | undefined>
readonly default: () => Effect.Effect<Info | undefined>
readonly resolve: (id?: ID | string) => Effect.Effect<Info | undefined>
readonly select: (id?: ID | string) => Effect.Effect<Selection>
readonly list: () => Effect.Effect<Info[]>
@@ -110,9 +109,6 @@ const layer = Layer.effect(
get: Effect.fn("Agent.get")(function* (id) {
return state.get().agents.get(id)
}),
default: Effect.fn("Agent.default")(function* () {
return selectedDefault()
}),
resolve: Effect.fn("Agent.resolve")(function* (id) {
if (id !== undefined) return state.get().agents.get(ID.make(id))
return selectedDefault()
+1 -17
View File
@@ -3,7 +3,7 @@ export * as Bus from "./bus"
import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
import { Event } from "@opencode-ai/schema/event"
import type { EventLog } from "@opencode-ai/schema/event-log"
import { and, asc, eq, gt, inArray, lte, sql } from "drizzle-orm"
import { and, asc, eq, gt, lte, sql } from "drizzle-orm"
import { Database } from "./database/database"
import { EventSequenceTable, EventTable } from "./event/sql"
import { Location } from "./location"
@@ -134,8 +134,6 @@ export interface Interface {
readonly after?: number
readonly follow?: boolean
}) => Stream.Stream<LogItem>
/** Latest committed seq per aggregate. Aggregates without events are absent. */
readonly sequences: (aggregateIDs: ReadonlyArray<string>) => Effect.Effect<ReadonlyMap<string, Event.Seq>>
/** @deprecated Use `subscribe()` and consume the returned stream. */
readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
readonly project: <D extends Event.Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
@@ -657,19 +655,6 @@ export const layerWith = (options?: LayerOptions) =>
}),
)
const sequences = (aggregateIDs: ReadonlyArray<string>): Effect.Effect<ReadonlyMap<string, Event.Seq>> => {
if (aggregateIDs.length === 0) return Effect.succeed(new Map())
return db
.select({ aggregateID: EventSequenceTable.aggregate_id, seq: EventSequenceTable.seq })
.from(EventSequenceTable)
.where(inArray(EventSequenceTable.aggregate_id, Array.from(aggregateIDs)))
.all()
.pipe(
Effect.orDie,
Effect.map((rows) => new Map(rows.map((row) => [row.aggregateID, Event.Seq.make(row.seq)]))),
)
}
const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
Effect.sync(() => {
listeners.push(listener)
@@ -691,7 +676,6 @@ export const layerWith = (options?: LayerOptions) =>
publish,
subscribe,
log,
sequences,
listen,
project,
replay,
+2 -92
View File
@@ -1,8 +1,7 @@
export * as FileMutation from "./file-mutation"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Context, Effect, Layer, Schema } from "effect"
import { dirname } from "path"
import { Context, Effect, Layer } from "effect"
import { KeyedMutex } from "./effect/keyed-mutex"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Bom } from "@opencode-ai/util/bom"
@@ -22,22 +21,6 @@ export interface TextWriteInput {
readonly content: string
}
export interface ConditionalWriteInput extends WriteInput {
readonly expected: Uint8Array
}
export interface RemoveInput {
readonly target: Target
}
export class StaleContentError extends Schema.TaggedErrorClass<StaleContentError>()("FileMutation.StaleContentError", {
path: Schema.String,
}) {}
export class TargetExistsError extends Schema.TaggedErrorClass<TargetExistsError>()("FileMutation.TargetExistsError", {
path: Schema.String,
}) {}
export interface WriteResult {
readonly operation: "write"
readonly target: string
@@ -45,24 +28,10 @@ export interface WriteResult {
readonly existed: boolean
}
export interface RemoveResult {
readonly operation: "remove"
readonly target: string
readonly resource: string
readonly existed: boolean
}
export interface Interface {
/** Create without replacing an existing target. */
readonly create: (input: WriteInput) => Effect.Effect<WriteResult, TargetExistsError | FSUtil.Error>
readonly write: (input: WriteInput) => Effect.Effect<WriteResult, FSUtil.Error>
/** Write text while retaining an existing UTF-8 BOM and emitting at most one BOM. */
readonly writeTextPreservingBom: (input: TextWriteInput) => Effect.Effect<WriteResult, FSUtil.Error>
/** Commit only if an existing target still has the expected bytes. */
readonly writeIfUnchanged: (
input: ConditionalWriteInput,
) => Effect.Effect<WriteResult, StaleContentError | FSUtil.Error>
readonly remove: (input: RemoveInput) => Effect.Effect<RemoveResult, FSUtil.Error>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/FileMutation") {}
@@ -89,13 +58,6 @@ const layer = Layer.effect(
existed,
})
const removeResult = (target: Target, existed: boolean): RemoveResult => ({
operation: "remove",
target: target.canonical,
resource: target.resource,
existed,
})
const write = Effect.fn("FileMutation.write")((input: WriteInput) =>
withTargetLock(input.target)(
Effect.gen(function* () {
@@ -122,62 +84,10 @@ const layer = Layer.effect(
),
)
const create = Effect.fn("FileMutation.create")((input: WriteInput) =>
withTargetLock(input.target)(
Effect.gen(function* () {
const write =
typeof input.content === "string"
? fs.writeFileString(input.target.canonical, input.content, { flag: "wx" })
: fs.writeFile(input.target.canonical, input.content, { flag: "wx" })
yield* write.pipe(
Effect.catchReason("PlatformError", "NotFound", () =>
fs.ensureDir(dirname(input.target.canonical)).pipe(Effect.andThen(write)),
),
Effect.catchReason("PlatformError", "AlreadyExists", () =>
Effect.fail(new TargetExistsError({ path: input.target.canonical })),
),
)
return writeResult(input.target, false)
}),
),
)
const writeIfUnchanged = Effect.fn("FileMutation.writeIfUnchanged")((input: ConditionalWriteInput) =>
withTargetLock(input.target)(
Effect.gen(function* () {
const current = yield* fs.readFile(input.target.canonical)
if (!sameBytes(current, input.expected)) {
return yield* new StaleContentError({ path: input.target.canonical })
}
yield* typeof input.content === "string"
? fs.writeFileString(input.target.canonical, input.content)
: fs.writeFile(input.target.canonical, input.content)
return writeResult(input.target, true)
}),
),
)
const remove = Effect.fn("FileMutation.remove")((input: RemoveInput) =>
withTargetLock(input.target)(
Effect.gen(function* () {
const existed = yield* fs.remove(input.target.canonical).pipe(
Effect.as(true),
Effect.catchReason("PlatformError", "NotFound", () => Effect.succeed(false)),
)
return removeResult(input.target, existed)
}),
),
)
return Service.of({ create, write, writeTextPreservingBom, writeIfUnchanged, remove })
return Service.of({ write, writeTextPreservingBom })
}),
)
function sameBytes(left: Uint8Array, right: Uint8Array) {
if (left.length !== right.length) return false
return left.every((byte, index) => byte === right[index])
}
export const node = makeLocationNode({ service: Service, layer, deps: [FSUtil.node] })
/**
+1 -5
View File
@@ -7,7 +7,7 @@ import { FSUtil } from "@opencode-ai/util/fs-util"
import { Location } from "./location"
import { PositiveInt, RelativePath } from "./schema"
import { FileSystemSearch } from "./filesystem/search"
import { Entry, FileSystem, FindInput, Match } from "@opencode-ai/schema/filesystem"
import { Entry, FileSystem, FindInput } from "@opencode-ai/schema/filesystem"
export { Entry, Match, Submatch } from "@opencode-ai/schema/filesystem"
export const ReadInput = Schema.Struct({
@@ -53,8 +53,6 @@ export interface Interface {
readonly read: (input: ReadInput) => Effect.Effect<{ readonly content: Uint8Array; readonly mime: string }>
readonly list: (input?: ListInput) => Effect.Effect<Entry[]>
readonly find: (input: FindInput) => Effect.Effect<Entry[]>
readonly glob: (input: GlobInput) => Effect.Effect<readonly Entry[]>
readonly grep: (input: GrepInput) => Effect.Effect<readonly Match[]>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/FileSystem") {}
@@ -76,8 +74,6 @@ const baseLayer = Layer.effect(
})
return Service.of({
find: search.find,
glob: search.glob,
grep: search.grep,
read: Effect.fn("FileSystem.read")(function* (input) {
const target = yield* resolve(input.path)
const info = yield* fs.stat(target.real).pipe(Effect.orDie)
-38
View File
@@ -3,9 +3,6 @@ import {
type DirItem,
type DirSearchResult,
type FileItem,
type GrepCursor,
type GrepMatch,
type GrepResult,
type InitOptions,
type MixedItem,
type MixedSearchResult,
@@ -45,19 +42,6 @@ export interface MixedSearch {
export type File = FileItem
export type Directory = DirItem
export type Mixed = MixedItem
export type Cursor = GrepCursor | null
export type Hit = GrepMatch
export interface Grep {
items: GrepResult["items"]
totalMatched: number
totalFilesSearched: number
totalFiles: number
filteredFileCount: number
nextCursor: Cursor
regexFallbackError?: string
}
export interface Picker {
destroy(): void
isScanning(): boolean
@@ -71,14 +55,6 @@ export interface Picker {
pageSize?: number
},
): Result<Search>
glob(
pattern: string,
opts?: {
currentFile?: string
pageIndex?: number
pageSize?: number
},
): Result<Search>
directorySearch(
query: string,
opts?: {
@@ -95,18 +71,6 @@ export interface Picker {
pageSize?: number
},
): Result<MixedSearch>
grep(
query: string,
opts?: {
mode?: "plain" | "regex" | "fuzzy"
maxMatchesPerFile?: number
timeBudgetMs?: number
beforeContext?: number
afterContext?: number
cursor?: Cursor
pageSize?: number
},
): Result<Grep>
trackQuery(query: string, file: string): Result<boolean>
getHistoricalQuery(offset: number): Result<string | null>
}
@@ -127,10 +91,8 @@ export function create(opts: Init): Result<Picker> {
waitForScan: (timeoutMs) => pick.waitForScan(timeoutMs),
refreshGitStatus: () => pick.refreshGitStatus(),
fileSearch: (query, next) => pick.fileSearch(query, next),
glob: (pattern, next) => pick.glob(pattern, next),
directorySearch: (query, next) => pick.directorySearch(query, next),
mixedSearch: (query, next) => pick.mixedSearch(query, next),
grep: (query, next) => pick.grep(query, next),
trackQuery: (query, file) => pick.trackQuery(query, file),
getHistoricalQuery: (offset) => pick.getHistoricalQuery(offset),
},
-38
View File
@@ -2,9 +2,6 @@ import type {
DirItem,
DirSearchResult,
FileItem,
GrepCursor,
GrepMatch,
GrepResult,
InitOptions,
MixedItem,
MixedSearchResult,
@@ -42,19 +39,6 @@ export interface MixedSearch {
export type File = FileItem
export type Directory = DirItem
export type Mixed = MixedItem
export type Cursor = GrepCursor | null
export type Hit = GrepMatch
export interface Grep {
items: GrepResult["items"]
totalMatched: number
totalFilesSearched: number
totalFiles: number
filteredFileCount: number
nextCursor: Cursor
regexFallbackError?: string
}
export interface Picker {
destroy(): void
isScanning(): boolean
@@ -68,14 +52,6 @@ export interface Picker {
pageSize?: number
},
): Result<Search>
glob(
pattern: string,
opts?: {
currentFile?: string
pageIndex?: number
pageSize?: number
},
): Result<Search>
directorySearch(
query: string,
opts?: {
@@ -92,18 +68,6 @@ export interface Picker {
pageSize?: number
},
): Result<MixedSearch>
grep(
query: string,
opts?: {
mode?: "plain" | "regex" | "fuzzy"
maxMatchesPerFile?: number
timeBudgetMs?: number
beforeContext?: number
afterContext?: number
cursor?: Cursor
pageSize?: number
},
): Result<Grep>
trackQuery(query: string, file: string): Result<boolean>
getHistoricalQuery(offset: number): Result<string | null>
}
@@ -125,10 +89,8 @@ export function create(opts: Init): Result<Picker> {
waitForScan: (timeoutMs) => pick.waitForScan(timeoutMs),
refreshGitStatus: () => pick.refreshGitStatus(),
fileSearch: (query, next) => pick.fileSearch(query, next),
glob: (pattern, next) => pick.glob(pattern, next),
directorySearch: (query, next) => pick.directorySearch(query, next),
mixedSearch: (query, next) => pick.mixedSearch(query, next),
grep: (query, next) => pick.grep(query, next),
trackQuery: (query, file) => pick.trackQuery(query, file),
getHistoricalQuery: (offset) => pick.getHistoricalQuery(offset),
},
@@ -3,7 +3,6 @@ export * as LocationWatcher from "./location-watcher"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Context, Effect, Layer, Stream } from "effect"
import { FileSystem } from "@opencode-ai/schema/filesystem"
import os from "os"
import path from "path"
import { Config } from "../config"
import { Bus } from "../bus"
@@ -44,7 +43,7 @@ const layer = Layer.effect(
const config = (yield* configService.entries())
.filter((entry): entry is Config.Document => entry.type === "document")
.flatMap((item) => item.info.watcher?.ignore ?? [])
const home = path.resolve(location.directory) === path.resolve(os.homedir())
const home = Protected.isHome(location.directory)
if (!home && location.vcs) {
const updates = yield* watcher.subscribe({
@@ -3,6 +3,10 @@ import path from "path"
const home = os.homedir()
export function isHome(directory: string) {
return path.resolve(directory) === path.resolve(home)
}
const DARWIN_HOME = [
"Music",
"Pictures",
+20 -120
View File
@@ -6,15 +6,13 @@ import { Context, Effect, Layer, Schema, Scope } from "effect"
import { Fff } from "#fff"
import fuzzysort from "fuzzysort"
import { FileSystem } from "../filesystem"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Location } from "../location"
import { Ripgrep } from "../ripgrep"
import { RelativePath } from "../schema"
import { Protected } from "./protected"
export interface Interface {
readonly find: (input: FileSystem.FindInput) => Effect.Effect<FileSystem.Entry[]>
readonly glob: (input: FileSystem.GlobInput) => Effect.Effect<readonly FileSystem.Entry[]>
readonly grep: (input: FileSystem.GrepInput) => Effect.Effect<readonly FileSystem.Match[]>
}
export const Options = Schema.Struct({
@@ -27,89 +25,35 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Fi
export const ripgrepLayer = Layer.effect(
Service,
Effect.gen(function* () {
const fs = yield* FSUtil.Service
const location = yield* Location.Service
const ripgrep = yield* Ripgrep.Service
const scope = yield* Scope.Scope
const state = {
files: [] as string[],
directories: [] as string[],
}
const files: string[] = []
const directories = new Set<string>()
const home = Protected.isHome(location.directory)
yield* ripgrep
.find({
cwd: location.directory,
pattern: "*",
limit: location.vcs ? Number.MAX_SAFE_INTEGER : 100_000,
limit: location.vcs && !home ? Number.MAX_SAFE_INTEGER : 100_000,
exclude: home ? [...Protected.names()].map((name) => `${name}/**`) : undefined,
onEntry: (entry) =>
Effect.sync(() => {
state.files.push(entry.path)
files.push(entry.path)
const parts = entry.path.split("/")
parts.slice(0, -1).forEach((_, index) => directories.add(parts.slice(0, index + 1).join("/") + path.sep))
state.directories = Array.from(directories)
}),
})
.pipe(Effect.orDie, Effect.asVoid, Effect.forkIn(scope))
return Service.of({
glob: (input) =>
Effect.gen(function* () {
const target = path.resolve(location.directory, input.path ?? ".")
const info = yield* fs.stat(target).pipe(Effect.orDie)
const cwd = info.type === "File" ? path.dirname(target) : target
return yield* ripgrep
.glob({
cwd,
pattern: input.pattern,
limit: input.limit ?? FileSystem.DEFAULT_SEARCH_LIMIT,
})
.pipe(
Effect.map((result) =>
result.map((entry) =>
FileSystem.Entry.make({
...entry,
path: RelativePath.make(path.relative(location.directory, path.resolve(cwd, entry.path))),
}),
),
),
Effect.orDie,
)
}),
grep: (input) =>
Effect.gen(function* () {
const target = path.resolve(location.directory, input.path ?? ".")
const info = yield* fs.stat(target).pipe(Effect.orDie)
const cwd = info.type === "File" ? path.dirname(target) : target
return yield* ripgrep
.grep({
cwd,
pattern: input.pattern,
file: info.type === "File" ? path.basename(target) : undefined,
include: input.include,
limit: input.limit ?? FileSystem.DEFAULT_SEARCH_LIMIT,
})
.pipe(
Effect.map((result) =>
result.map((match) =>
FileSystem.Match.make({
...match,
entry: FileSystem.Entry.make({
...match.entry,
path: RelativePath.make(path.relative(location.directory, path.resolve(cwd, match.entry.path))),
}),
}),
),
),
Effect.orDie,
)
}),
find: (input) =>
Effect.gen(function* () {
const items =
input.type === "file"
? state.files
? files
: input.type === "directory"
? state.directories
: [...state.files, ...state.directories]
? Array.from(directories)
: [...files, ...directories]
return fuzzysort.go(input.query, items, { limit: input.limit ?? 50 }).map((item) => {
const relative = item.target
const type = relative.endsWith(path.sep) ? ("directory" as const) : ("file" as const)
@@ -143,55 +87,10 @@ export const fffLayer = Layer.effect(
if (result) yield* Effect.logWarning("failed to initialize fff", { error: result.error })
return Service.of({
find: () => Effect.succeed([]),
glob: () => Effect.succeed([]),
grep: () => Effect.succeed([]),
})
}
yield* Effect.addFinalizer(() => Effect.sync(() => result.value.destroy()).pipe(Effect.ignore))
return Service.of({
glob: (input) =>
Effect.sync(() => {
const prefix = input.path?.replaceAll("\\", "/").replace(/\/$/, "")
const found = result.value.glob(prefix ? `${prefix}/${input.pattern}` : input.pattern, {
pageIndex: 0,
pageSize: input.limit ?? FileSystem.DEFAULT_SEARCH_LIMIT,
})
if (!found.ok) throw found.error
return found.value.items.map((item) =>
FileSystem.Entry.make({
path: RelativePath.make(item.relativePath.replaceAll("\\", "/")),
type: "file",
}),
)
}),
grep: (input) =>
Effect.sync(() => {
const prefix = input.path?.replaceAll("\\", "/").replace(/\/$/, "")
const found = result.value.grep(
[prefix ? `${prefix}/**` : undefined, input.include, input.pattern]
.filter((value) => value !== undefined)
.join(" "),
{ mode: "regex", pageSize: input.limit ?? FileSystem.DEFAULT_SEARCH_LIMIT, timeBudgetMs: 1_500 },
)
if (!found.ok) throw found.error
return found.value.items.map((match) => {
const bytes = Buffer.from(match.lineContent)
return FileSystem.Match.make({
entry: FileSystem.Entry.make({
path: RelativePath.make(match.relativePath.replaceAll("\\", "/")),
type: "file",
}),
line: match.lineNumber,
offset: match.byteOffset,
text: match.lineContent.length > 2_000 ? match.lineContent.slice(0, 2_000) + "..." : match.lineContent,
submatches: match.matchRanges.map(([start, end]) => ({
text: bytes.subarray(start, end).toString("utf8"),
start,
end,
})),
})
})
}),
find: (input) =>
Effect.sync(() => {
const options = { pageIndex: 0, pageSize: input.limit ?? 50 }
@@ -236,18 +135,19 @@ export const fffLayer = Layer.effect(
}),
)
export const layer = (options?: Options) => Layer.unwrap(
Effect.gen(function* () {
if (options?.fff === false || (options?.fff === undefined && process.platform === "win32") || !Fff.available())
return ripgrepLayer
const location = yield* Location.Service
// Non-VCS locations can contain many repositories, so avoid eagerly content-indexing the entire aggregate tree.
return location.vcs ? fffLayer : ripgrepLayer
}),
)
export const layer = (options?: Options) =>
Layer.unwrap(
Effect.gen(function* () {
if (options?.fff === false || (options?.fff === undefined && process.platform === "win32") || !Fff.available())
return ripgrepLayer
const location = yield* Location.Service
// Non-VCS locations can contain many repositories, so avoid eagerly content-indexing the entire aggregate tree.
return location.vcs && !Protected.isHome(location.directory) ? fffLayer : ripgrepLayer
}),
)
export function configured(options?: Options) {
return makeLocationNode({ service: Service, layer: layer(options), deps: [FSUtil.node, Location.node, Ripgrep.node] })
return makeLocationNode({ service: Service, layer: layer(options), deps: [Location.node, Ripgrep.node] })
}
export const node = configured()
+2 -28
View File
@@ -1,6 +1,6 @@
export * as Formatter from "./formatter"
import { Context, Effect, Layer, Schema } from "effect"
import { Context, Effect, Layer } from "effect"
import { ChildProcess } from "effect/unstable/process"
import path from "path"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
@@ -11,16 +11,7 @@ import { Config } from "./config"
import { Location } from "./location"
import { make, type Info } from "./formatter/builtins"
export const Status = Schema.Struct({
name: Schema.String,
extensions: Schema.Array(Schema.String),
enabled: Schema.Boolean,
}).annotate({ identifier: "FormatterStatus" })
export type Status = typeof Status.Type
export interface Interface {
readonly init: () => Effect.Effect<void>
readonly status: () => Effect.Effect<Status[]>
readonly file: (filepath: string) => Effect.Effect<boolean>
}
@@ -84,23 +75,6 @@ const layer = Layer.effect(
return result
})
const init = Effect.fn("Formatter.init")(function* () {
yield* load
})
const status = Effect.fn("Formatter.status")(function* () {
yield* load
return yield* Effect.forEach(formatters, (formatter) =>
command(formatter).pipe(
Effect.map((enabled) => ({
name: formatter.name,
extensions: [...formatter.extensions],
enabled: enabled !== false,
})),
),
)
})
const file = Effect.fn("Formatter.file")(function* (filepath: string) {
yield* load
const matching = formatters.filter((formatter) =>
@@ -143,7 +117,7 @@ const layer = Layer.effect(
return false
})
return Service.of({ init, status, file })
return Service.of({ file })
}),
)
+1 -224
View File
@@ -1,8 +1,7 @@
export * as Git from "./git"
import path from "path"
import { randomUUID } from "crypto"
import { Context, Effect, Layer, Schema, Stream } from "effect"
import { Context, Effect, Layer, Schema } from "effect"
import { ChildProcess } from "effect/unstable/process"
import { AbsolutePath, RelativePath } from "./schema"
import { FSUtil } from "@opencode-ai/util/fs-util"
@@ -36,9 +35,6 @@ const snapshotConfig = `[core]
threads = true
`
export const ChangeSet = Schema.String.pipe(Schema.brand("Git.ChangeSet"))
export type ChangeSet = typeof ChangeSet.Type
export const TreeID = Schema.String.pipe(Schema.brand("Git.TreeID"))
export type TreeID = typeof TreeID.Type
@@ -73,13 +69,6 @@ export class WorktreeError extends Schema.TaggedErrorClass<WorktreeError>()("Git
cause: Schema.optional(Schema.Defect()),
}) {}
export class PatchError extends Schema.TaggedErrorClass<PatchError>()("Git.PatchError", {
operation: Schema.Literals(["capture", "apply", "reset"]),
directory: AbsolutePath,
message: Schema.String,
cause: Schema.optional(Schema.Defect()),
}) {}
export interface Interface {
readonly repo: {
readonly discover: (input: AbsolutePath) => Effect.Effect<Repository | undefined>
@@ -116,20 +105,6 @@ export interface Interface {
) => Effect.Effect<void, OperationError>
readonly resetHard: (repository: Repository, revision: string) => Effect.Effect<void, OperationError>
}
readonly change: {
readonly capture: (input: { repository: Repository; path: AbsolutePath }) => Effect.Effect<ChangeSet, PatchError>
readonly apply: (input: {
repository: Repository
path: AbsolutePath
changes: ChangeSet
}) => Effect.Effect<void, PatchError>
readonly discard: (input: {
repository: Repository
path: AbsolutePath
index: "preserve" | "reset"
untracked: "preserve" | "remove"
}) => Effect.Effect<void, PatchError>
}
readonly worktree: {
readonly create: (input: {
repository: Repository
@@ -175,17 +150,10 @@ export interface Interface {
context?: number
paths?: readonly RelativePath[]
}) => Effect.Effect<readonly File.Diff[], OperationError>
readonly preview: (input: {
repository: Repository
current: TreeID
files: ReadonlyMap<RelativePath, TreeID>
context?: number
}) => Effect.Effect<readonly File.Diff[], OperationError>
readonly restore: (input: {
repository: Repository
files: ReadonlyMap<RelativePath, TreeID>
}) => Effect.Effect<void, OperationError>
readonly checkout: (input: { repository: Repository; tree: TreeID }) => Effect.Effect<void, OperationError>
}
}
@@ -657,58 +625,6 @@ const layer = Layer.effect(
return { mode: match[1], object: match[2] }
})
const preview = Effect.fn("Git.tree.preview")(
(input: {
repository: Repository
current: TreeID
files: ReadonlyMap<RelativePath, TreeID>
context?: number
}) =>
locked(
input.repository,
Effect.gen(function* () {
const index = path.join(input.repository.gitDirectory, `preview-${randomUUID()}.index`)
const env = { GIT_INDEX_FILE: index }
return yield* Effect.gen(function* () {
yield* repositoryOperation("diff", input.repository, ["read-tree", input.current], { env })
yield* Effect.forEach(
input.files,
([file, tree]) =>
Effect.gen(function* () {
const source = yield* entry(input.repository, tree, file)
if (!source) {
yield* repositoryOperation(
"diff",
input.repository,
["update-index", "--force-remove", "--", file],
{ env },
)
return
}
yield* repositoryOperation(
"diff",
input.repository,
["update-index", "--add", "--cacheinfo", source.mode, source.object, file],
{ env },
)
}),
{ discard: true },
)
const target = TreeID.make(
(yield* repositoryOperation("diff", input.repository, ["write-tree"], { env })).text.trim(),
)
return yield* treeDiff({
repository: input.repository,
from: input.current,
to: target,
context: input.context,
paths: Array.from(input.files.keys()),
})
}).pipe(Effect.ensuring(fs.remove(index).pipe(Effect.catch(() => Effect.void))))
}),
),
)
const restore = Effect.fn("Git.tree.restore")(
(input: { repository: Repository; files: ReadonlyMap<RelativePath, TreeID> }) =>
locked(
@@ -738,142 +654,6 @@ const layer = Layer.effect(
),
)
const checkoutTree = Effect.fn("Git.tree.checkout")((input: { repository: Repository; tree: TreeID }) =>
locked(
input.repository,
Effect.gen(function* () {
yield* repositoryOperation("restore", input.repository, ["read-tree", input.tree])
yield* repositoryOperation("restore", input.repository, ["checkout-index", "--all", "--force"])
}),
),
)
const capture = Effect.fn("Git.change.capture")(function* (input: { repository: Repository; path: AbsolutePath }) {
const scope = path.relative(input.repository.worktree, input.path).replaceAll("\\", "/") || "."
const tracked = yield* execute(
input.repository.worktree,
proc,
)(["diff", "--binary", "HEAD", "--", scope]).pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "capture", directory: input.path, message: cause.message, cause }),
),
)
if (tracked.exitCode !== 0) {
return yield* new PatchError({
operation: "capture",
directory: input.path,
message: tracked.stderr.trim() || tracked.text.trim() || "Failed to capture tracked changes",
})
}
const untracked = yield* execute(
input.repository.worktree,
proc,
)(["ls-files", "--others", "--exclude-standard", "-z", "--", scope]).pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "capture", directory: input.path, message: cause.message, cause }),
),
)
if (untracked.exitCode !== 0) {
return yield* new PatchError({
operation: "capture",
directory: input.path,
message: untracked.stderr.trim() || untracked.text.trim() || "Failed to list untracked changes",
})
}
const created = yield* Effect.forEach(untracked.text.split("\0").filter(Boolean), (file) =>
execute(
input.repository.worktree,
proc,
)(["diff", "--binary", "--no-index", "--", "/dev/null", file]).pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "capture", directory: input.path, message: cause.message, cause }),
),
Effect.flatMap((result) =>
// git diff --no-index returns 1 when differences were found.
result.exitCode === 0 || result.exitCode === 1
? Effect.succeed(result.text)
: Effect.fail(
new PatchError({
operation: "capture",
directory: input.path,
message:
result.stderr.trim() || result.text.trim() || `Failed to capture untracked change: ${file}`,
}),
),
),
),
)
return ChangeSet.make([tracked.text, ...created].filter(Boolean).join("\n"))
})
const apply = Effect.fn("Git.change.apply")(function* (input: {
repository: Repository
path: AbsolutePath
changes: ChangeSet
}) {
const result = yield* proc
.run(
ChildProcess.make("git", ["apply", "-"], {
cwd: input.path,
extendEnv: true,
stdin: Stream.make(new TextEncoder().encode(input.changes)),
}),
)
.pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "apply", directory: input.path, message: cause.message, cause }),
),
)
if (result.exitCode === 0) return
return yield* new PatchError({
operation: "apply",
directory: input.path,
message:
result.stderr.toString("utf8").trim() || result.stdout.toString("utf8").trim() || "Failed to apply changes",
})
})
const discard = Effect.fn("Git.change.discard")(function* (input: {
repository: Repository
path: AbsolutePath
index: "preserve" | "reset"
untracked: "preserve" | "remove"
}) {
const scope = path.relative(input.repository.worktree, input.path).replaceAll("\\", "/") || "."
const restore = yield* execute(
input.repository.worktree,
proc,
)(input.index === "reset" ? ["checkout", "HEAD", "--", scope] : ["checkout", "--", scope]).pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "reset", directory: input.path, message: cause.message, cause }),
),
)
if (restore.exitCode !== 0) {
return yield* new PatchError({
operation: "reset",
directory: input.path,
message: restore.stderr.trim() || restore.text.trim() || "Failed to restore tracked changes",
})
}
if (input.untracked === "preserve") return
const clean = yield* execute(
input.repository.worktree,
proc,
)(["clean", "-fd", "--", scope]).pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "reset", directory: input.path, message: cause.message, cause }),
),
)
if (clean.exitCode === 0) return
return yield* new PatchError({
operation: "reset",
directory: input.path,
message: clean.stderr.trim() || clean.text.trim() || "Failed to clean untracked changes",
})
})
const worktreeRun = Effect.fnUntraced(function* (
operation: "create" | "remove" | "list",
repository: Repository,
@@ -949,7 +729,6 @@ const layer = Layer.effect(
remote: { get: remote },
history: { head, branch, defaultRemoteBranch: remoteHead, rootCommits: roots },
sync: { fetchRemotes: fetch, fetchBranch, checkoutRemoteBranch: checkout, resetHard: reset },
change: { capture, apply, discard },
worktree: { create: worktreeCreate, remove: worktreeRemove, list: worktreeList },
index: { refresh, ignored },
tree: {
@@ -957,9 +736,7 @@ const layer = Layer.effect(
write: writeTree,
files: treeFiles,
diff: treeDiff,
preview,
restore,
checkout: checkoutTree,
},
})
}),
+1 -15
View File
@@ -59,16 +59,6 @@ export interface Interface {
readonly list: () => Effect.Effect<ReadonlyArray<Info>>
readonly directories: (input: DirectoriesInput) => Effect.Effect<Directories>
readonly resolve: (input: AbsolutePath) => Effect.Effect<Resolved>
/**
* Temporary bridge method for writing the resolved project ID to the repo-local cache.
*
* This exists while the old opencode project service and this core project
* service work together: core resolves the ID, while the old service still owns
* database migration and persistence. The old service should call this after it
* finishes migrating from `resolve().previous` to `resolve().id`; once project
* persistence moves into core, this separate bridge method can go away.
*/
readonly commit: (input: { store: AbsolutePath; id: ID }) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/Project") {}
@@ -268,11 +258,7 @@ const layer = Layer.effect(
return yield* persist({ id: ID.global, directory, canonical: directory, vcs: undefined })
})
const commit = Effect.fn("Project.commit")(function* (input: { store: AbsolutePath; id: ID }) {
yield* fs.writeFileString(path.join(input.store, "opencode"), input.id).pipe(Effect.ignore)
})
return Service.of({ list, directories, resolve, commit })
return Service.of({ list, directories, resolve })
}),
)
-17
View File
@@ -31,14 +31,6 @@ export type EnsureInput = {
readonly branch?: string
}
export class InvalidRepositoryError extends Schema.TaggedErrorClass<InvalidRepositoryError>()(
"RepositoryCacheInvalidRepositoryError",
{
repository: Schema.String,
message: Schema.String,
},
) {}
export class InvalidBranchError extends Schema.TaggedErrorClass<InvalidBranchError>()(
"RepositoryCacheInvalidBranchError",
{
@@ -86,7 +78,6 @@ export class CacheOperationError extends Schema.TaggedErrorClass<CacheOperationE
) {}
export type Error =
| InvalidRepositoryError
| InvalidBranchError
| CloneFailedError
| FetchFailedError
@@ -103,7 +94,6 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Re
export function isError(error: unknown): error is Error {
return (
error instanceof InvalidRepositoryError ||
error instanceof InvalidBranchError ||
error instanceof CloneFailedError ||
error instanceof FetchFailedError ||
@@ -114,13 +104,6 @@ export function isError(error: unknown): error is Error {
)
}
export const parseRemote = Effect.fn("RepositoryCache.parseRemote")(function* (repository: string) {
return yield* Effect.try({
try: () => Repository.parseRemote(repository),
catch: (error) => new InvalidRepositoryError({ repository, message: errorMessage(error) }),
})
})
export const validateBranch = Effect.fn("RepositoryCache.validateBranch")(function* (branch: string) {
return yield* Effect.try({
try: () => Repository.validateBranch(branch),
-10
View File
@@ -44,16 +44,6 @@ export class InvalidBranchError extends Schema.TaggedErrorClass<InvalidBranchErr
message: Schema.String,
}) {}
export type Error = InvalidReferenceError | UnsupportedLocalRepositoryError | InvalidBranchError
export function isError(error: unknown): error is Error {
return (
error instanceof InvalidReferenceError ||
error instanceof UnsupportedLocalRepositoryError ||
error instanceof InvalidBranchError
)
}
export function parse(input: string): Reference | undefined {
const cleaned = normalizeInput(input)
if (!cleaned) return
+2
View File
@@ -52,6 +52,7 @@ export interface FindInput {
readonly cwd: string
readonly pattern: string
readonly limit: number
readonly exclude?: readonly string[]
readonly hidden?: boolean
readonly follow?: boolean
readonly signal?: AbortSignal
@@ -195,6 +196,7 @@ const layer = Layer.effect(
...(input.hidden ? ["--hidden"] : []),
...(input.follow ? ["--follow"] : []),
...(input.pattern === "*" ? [] : [`--glob=${input.pattern}`]),
...(input.exclude ?? []).map((pattern) => `--glob=!${pattern}`),
"--glob=!**/.git/**",
".",
],
+5 -4
View File
@@ -410,7 +410,7 @@ const layer = Layer.effect(
fork: Effect.fn("Session.fork")(function* (input) {
const parent = yield* result.get(input.sessionID)
const boundary = yield* db
.select({ id: SessionMessageTable.id, seq: SessionMessageTable.seq })
.select({ id: SessionMessageTable.id })
.from(SessionMessageTable)
.where(
and(
@@ -429,13 +429,14 @@ const layer = Layer.effect(
})
if (!boundary) return yield* new ForkEmptyError({ sessionID: input.sessionID })
const sessionID = SessionSchema.ID.create()
const instructionThrough =
input.boundary.type === "before" ? boundary.seq - 1 : yield* Bus.latestSequence(db, parent.id)
// The fork adopts the parent's newest instruction values rather than the
// values in effect at the boundary; copied history may contain frozen
// instruction-update text the initial baseline already reflects.
yield* bus.publish(SessionEvent.Forked, {
sessionID,
parentID: parent.id,
boundary: { ...input.boundary, messageID: boundary.id },
instructions: yield* InstructionState.valuesAt(db, parent.id, instructionThrough),
instructions: yield* InstructionState.current(db, parent.id),
})
return yield* result.get(sessionID).pipe(Effect.orDie)
}),
+3 -5
View File
@@ -80,10 +80,9 @@ export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(fun
.transaction(() =>
Effect.gen(function* () {
const messages = yield* messageEntries(db, sessionID)
const assembled = yield* InstructionState.assemble(db, sessionID, instructions)
return {
initial: assembled.initial,
entries: [...messages, ...assembled.updates].toSorted((a, b) => a.seq - b.seq),
initial: yield* InstructionState.initial(db, sessionID, instructions),
entries: messages,
}
}),
)
@@ -106,10 +105,9 @@ export const preview = Effect.fn("SessionHistory.preview")(function* (
)
const settled = unsettled === -1 ? messages : messages.slice(0, unsettled)
const assembled = yield* InstructionState.preview(db, sessionID, instructions, observed)
const entries = [...settled, ...assembled.updates].toSorted((a, b) => a.seq - b.seq)
return {
initial: assembled.initial,
messages: entries.map((entry) => entry.message),
messages: settled.map((entry) => entry.message),
instructionUpdate: assembled.update,
}
}),
+56 -230
View File
@@ -1,25 +1,20 @@
export * as InstructionState from "./instruction-state"
import { and, asc, desc, eq, gt, inArray, lte, sql } from "drizzle-orm"
import { DateTime, Effect, Option, Schema } from "effect"
import { eq, inArray, sql } from "drizzle-orm"
import { Effect, Option, Schema } from "effect"
import type { Database } from "../database/database"
import { Bus } from "../bus"
import { EventTable } from "../event/sql"
import type { Bus } from "../bus"
import { Instructions } from "../instructions/index"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
import { Event } from "@opencode-ai/schema/event"
import { SessionSchema } from "./schema"
import { InstructionBlobTable, InstructionStateTable } from "./sql"
type DatabaseService = Database.Interface["db"]
const decodeInstructionsUpdated = Schema.decodeUnknownSync(SessionEvent.InstructionsUpdated.data)
const decodeForked = Schema.decodeUnknownSync(SessionEvent.Forked.data)
export interface Observation extends Instructions.Admission {
readonly sessionID: SessionSchema.ID
readonly initial: boolean
readonly previous: Instructions.Values
readonly current: Instructions.Values
}
@@ -28,13 +23,14 @@ export const observe = Effect.fn("InstructionState.observe")(function* (
instructions: Instructions.Instructions,
sessionID: SessionSchema.ID,
): Effect.fn.Return<Observation, Instructions.InitializationBlocked> {
const [observed, stored] = yield* Effect.all([Instructions.read(instructions), ensure(db, sessionID)], {
const [observed, stored] = yield* Effect.all([Instructions.read(instructions), find(db, sessionID)], {
concurrency: "unbounded",
})
const result = yield* observeAgainst(observed, stored?.current_values)
return {
sessionID,
initial: !stored,
previous: stored?.current_values ?? {},
...result,
}
})
@@ -42,12 +38,20 @@ export const observe = Effect.fn("InstructionState.observe")(function* (
export const commit = Effect.fn("InstructionState.commit")(function* (
db: DatabaseService,
bus: Bus.Interface,
instructions: Instructions.Instructions,
observation: Observation,
) {
if (!observation.initial && Object.keys(observation.delta).length === 0) return
// The rendered text is frozen into the durable event: replaying it later would
// require the Location-scoped registry that produced it.
const text = observation.initial ? "" : yield* renderUpdateText(db, instructions, observation)
yield* bus.publish(
SessionEvent.InstructionsUpdated,
{ sessionID: observation.sessionID, delta: observation.delta },
{
sessionID: observation.sessionID,
delta: observation.delta,
...(text.length > 0 ? { text } : {}),
},
{
// Initial sync establishes the baseline; unlike later deltas it is not chronological history.
...(observation.initial ? { metadata: { instructions: { initial: true } } } : {}),
@@ -56,13 +60,27 @@ export const commit = Effect.fn("InstructionState.commit")(function* (
)
})
const renderUpdateText = Effect.fnUntraced(function* (
db: DatabaseService,
instructions: Instructions.Instructions,
observation: Observation,
) {
const replaced = Object.entries(observation.previous).filter(([key]) => Object.hasOwn(observation.delta, key))
const blobs = yield* loadBlobs(db, replaced.map(([, hash]) => hash))
const previous = Object.fromEntries(replaced.map(([key, hash]) => [key, requireBlob(blobs, hash)]))
const admitted = new Map(
Object.entries(observation.blobs).map(([hash, value]) => [Instructions.Hash.make(hash), value]),
)
return Instructions.renderUpdate(instructions, previous, dereferenceDelta(observation.delta, admitted))
})
export const prepare = Effect.fn("InstructionState.prepare")(function* (
db: DatabaseService,
bus: Bus.Interface,
instructions: Instructions.Instructions,
sessionID: SessionSchema.ID,
) {
yield* commit(db, bus, yield* observe(db, instructions, sessionID))
yield* commit(db, bus, instructions, yield* observe(db, instructions, sessionID))
})
export const apply = Effect.fn("InstructionState.apply")(function* (
@@ -140,79 +158,24 @@ export const reset = Effect.fn("InstructionState.reset")(function* (db: Database
.pipe(Effect.orDie)
})
export const rebuild = Effect.fn("InstructionState.rebuild")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
const state = yield* stateFromEvents(db, sessionID)
if (!state) {
yield* reset(db, sessionID)
return undefined
}
yield* db
.insert(InstructionStateTable)
.values(state)
.onConflictDoUpdate({
target: InstructionStateTable.session_id,
set: {
epoch_start: state.epoch_start,
through_seq: state.through_seq,
initial_values: state.initial_values,
current_values: state.current_values,
},
})
.run()
.pipe(Effect.orDie)
return state
})
const assembleState = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
instructions: Instructions.Instructions,
state: typeof InstructionStateTable.$inferSelect,
) {
const rows = yield* instructionUpdatesAfter(db, sessionID, state.epoch_start)
const updates = rows.map((row) => ({
row,
delta: decodeInstructionsUpdated(row.data).delta,
}))
const blobs = yield* loadBlobs(db, [
...Object.values(state.initial_values),
...updates.flatMap((update) =>
Object.values(update.delta).filter((hash): hash is Instructions.Hash => hash !== "removed"),
),
])
const valuesAtStart = dereference(state.initial_values, blobs)
let values = valuesAtStart
const result: Array<{ readonly seq: number; readonly message: SessionMessage.System }> = []
for (const update of updates) {
const delta = dereferenceDelta(update.delta, blobs)
const text = Instructions.renderUpdate(instructions, values, delta)
if (text.length > 0)
result.push({
seq: update.row.seq,
message: SessionMessage.System.make({
id: SessionMessage.ID.fromEvent(Event.ID.make(update.row.id)),
type: "system",
text,
time: { created: DateTime.makeUnsafe(update.row.created) },
}),
})
values = Instructions.applyDelta(values, delta)
}
return { initial: Instructions.renderInitial(instructions, valuesAtStart), updates: result, current: values }
})
export const assemble = Effect.fn("InstructionState.assemble")(function* (
/** Renders the epoch baseline shown at the start of every model request. */
export const initial = Effect.fn("InstructionState.initial")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
instructions: Instructions.Instructions,
) {
const state = yield* find(db, sessionID)
if (!state) return yield* Effect.die(new Error(`Instruction state not found during assembly: ${sessionID}`))
const assembled = yield* assembleState(db, sessionID, instructions, state)
return { initial: assembled.initial, updates: assembled.updates }
const blobs = yield* loadBlobs(db, Object.values(state.initial_values))
return Instructions.renderInitial(instructions, dereference(state.initial_values, blobs))
})
/** The current instruction values, used to seed a fork's baseline. */
export const current = Effect.fn("InstructionState.current")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
return (yield* find(db, sessionID))?.current_values
})
export const preview = Effect.fn("InstructionState.preview")(function* (
@@ -221,20 +184,26 @@ export const preview = Effect.fn("InstructionState.preview")(function* (
instructions: Instructions.Instructions,
observed: Instructions.ReadResult,
) {
const state = yield* readState(db, sessionID)
const state = yield* find(db, sessionID)
const result = yield* observeAgainst(observed, state?.current_values)
const blobs = new Map<Instructions.Hash, Schema.Json>(
const observedBlobs = new Map<Instructions.Hash, Schema.Json>(
Object.entries(result.blobs).map(([hash, value]) => [Instructions.Hash.make(hash), value]),
)
if (!state) {
const values = dereference(result.current, blobs)
return { initial: Instructions.renderInitial(instructions, values), updates: [], update: "" }
const values = dereference(result.current, observedBlobs)
return { initial: Instructions.renderInitial(instructions, values), update: "" }
}
const assembled = yield* assembleState(db, sessionID, instructions, state)
const stored = yield* loadBlobs(db, [
...Object.values(state.initial_values),
...Object.values(state.current_values),
])
return {
initial: assembled.initial,
updates: assembled.updates,
update: Instructions.renderUpdate(instructions, assembled.current, dereferenceDelta(result.delta, blobs)),
initial: Instructions.renderInitial(instructions, dereference(state.initial_values, stored)),
update: Instructions.renderUpdate(
instructions,
dereference(state.current_values, stored),
dereferenceDelta(result.delta, new Map([...stored, ...observedBlobs])),
),
}
})
@@ -255,46 +224,6 @@ const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: Sessio
.pipe(Effect.orDie)
})
const ensure = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const stored = yield* find(db, sessionID)
if (!stored) return yield* rebuild(db, sessionID)
const latest = yield* latestRelevantSequence(db, sessionID)
if (!latest || latest.seq <= stored.through_seq) return stored
return yield* rebuild(db, sessionID)
})
const readState = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const stored = yield* find(db, sessionID)
if (!stored) return yield* stateFromEvents(db, sessionID)
const latest = yield* latestRelevantSequence(db, sessionID)
if (!latest || latest.seq <= stored.through_seq) return stored
return yield* stateFromEvents(db, sessionID)
})
const stateFromEvents = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const folded = fold(yield* instructionEvents(db, sessionID))
return folded ? foldedState(sessionID, folded) : undefined
})
export const valuesAt = Effect.fn("InstructionState.valuesAt")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
through: number,
) {
return fold(yield* instructionEvents(db, sessionID, through))?.current
})
const latestRelevantSequence = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select({ seq: EventTable.seq })
.from(EventTable)
.where(and(eq(EventTable.aggregate_id, sessionID), inArray(EventTable.type, relevantEventTypes)))
.orderBy(desc(EventTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
})
const insertBlobs = Effect.fnUntraced(function* (db: DatabaseService, blobs: Readonly<Record<string, Schema.Json>>) {
const rows = Object.entries(blobs).map(([hash, value]) => ({ hash: Instructions.Hash.make(hash), value }))
if (rows.length === 0) return
@@ -339,106 +268,3 @@ function requireBlob(blobs: ReadonlyMap<Instructions.Hash, Schema.Json>, hash: I
if (value === undefined) throw new Error(`Instruction blob not found: ${hash}`)
return value
}
const instructionEventType = Bus.versionedType(
SessionEvent.InstructionsUpdated.type,
SessionEvent.InstructionsUpdated.durable.version,
)
const compactionEventType = Bus.versionedType(
SessionEvent.Compaction.Ended.type,
SessionEvent.Compaction.Ended.durable.version,
)
const movedEventType = Bus.versionedType(SessionEvent.Moved.type, SessionEvent.Moved.durable.version)
const revertedEventType = Bus.versionedType(
SessionEvent.RevertEvent.Committed.type,
SessionEvent.RevertEvent.Committed.durable.version,
)
const forkedEventType = Bus.versionedType(SessionEvent.Forked.type, SessionEvent.Forked.durable.version)
const relevantEventTypes = [
forkedEventType,
instructionEventType,
compactionEventType,
movedEventType,
revertedEventType,
]
type InstructionEventRow = typeof EventTable.$inferSelect
const instructionEvents = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
through?: number,
): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
return yield* eventRows(db, sessionID, relevantEventTypes, undefined, through)
})
const instructionUpdatesAfter = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
after: number,
) {
return yield* eventRows(db, sessionID, [instructionEventType], after)
})
const eventRows = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
types: ReadonlyArray<string>,
after?: number,
through?: number,
): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
return yield* db
.select()
.from(EventTable)
.where(
and(
eq(EventTable.aggregate_id, sessionID),
inArray(EventTable.type, types),
after === undefined ? undefined : gt(EventTable.seq, after),
through === undefined ? undefined : lte(EventTable.seq, through),
),
)
.orderBy(asc(EventTable.seq))
.all()
.pipe(Effect.orDie)
})
function fold(rows: ReadonlyArray<InstructionEventRow>) {
return rows.reduce<
| {
readonly epochStart: number
readonly throughSeq: number
readonly initial: Instructions.Values
readonly current: Instructions.Values
}
| undefined
>((state, row) => {
if (row.type === forkedEventType) {
const instructions = decodeForked(row.data).instructions
return instructions
? { epochStart: row.seq, throughSeq: row.seq, initial: instructions, current: instructions }
: undefined
}
if (row.type === movedEventType || row.type === revertedEventType) return undefined
if (row.type === compactionEventType)
return state
? { epochStart: row.seq, throughSeq: row.seq, initial: state.current, current: state.current }
: undefined
if (row.type !== instructionEventType) return state
const delta = decodeInstructionsUpdated(row.data).delta
const current = Instructions.applyHashDelta(state?.current ?? {}, delta)
return state
? { ...state, throughSeq: row.seq, current }
: { epochStart: row.seq, throughSeq: row.seq, initial: current, current }
}, undefined)
}
function foldedState(sessionID: SessionSchema.ID, folded: NonNullable<ReturnType<typeof fold>>) {
return {
session_id: sessionID,
epoch_start: folded.epochStart,
through_seq: folded.throughSeq,
initial_values: folded.initial,
current_values: folded.current,
}
}
+20 -121
View File
@@ -1,12 +1,8 @@
import { castDraft, produce, type WritableDraft } from "immer"
import { DateTime, Effect } from "effect"
import { DateTime, Effect, Match, pipe } from "effect"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
export type MemoryState = {
messages: SessionMessage.Info[]
}
export interface Adapter {
readonly getModel: () => Effect.Effect<SessionMessage.ModelSelected["model"] | undefined, never, never>
readonly getCurrentAssistant: () => Effect.Effect<SessionMessage.Assistant | undefined, never, never>
@@ -23,89 +19,7 @@ export interface Adapter {
readonly appendMessage: (message: SessionMessage.Info) => Effect.Effect<void, never, never>
}
export function memory(state: MemoryState): Adapter {
const assistantIndex = (messageID: SessionMessage.ID) =>
state.messages.findLastIndex((message) => message.id === messageID)
const shellIndex = (messageID: SessionMessage.ID) =>
state.messages.findLastIndex((message) => message.id === messageID)
const compactionIndex = () =>
state.messages.findLastIndex((message) => message.type === "compaction" && message.status === "running")
// A newer step supersedes stale incomplete rows; never resume an older assistant projection.
const latestAssistantIndex = () => state.messages.findLastIndex((message) => message.type === "assistant")
return {
getModel() {
return Effect.sync(
() =>
state.messages.findLast(
(message): message is SessionMessage.ModelSelected | SessionMessage.Assistant =>
message.type === "model-switched" || message.type === "assistant",
)?.model,
)
},
getCurrentAssistant() {
return Effect.sync(() => {
const index = latestAssistantIndex()
if (index < 0) return
const assistant = state.messages[index]
return assistant?.type === "assistant" && !assistant.time.completed ? assistant : undefined
})
},
getAssistant(messageID) {
return Effect.sync(() => {
const index = assistantIndex(messageID)
if (index < 0) return
const assistant = state.messages[index]
return assistant?.type === "assistant" ? assistant : undefined
})
},
getShell(shellID) {
return Effect.sync(() => {
return state.messages.find((message): message is SessionMessage.Shell => {
return message.type === "shell" && message.shellID === shellID
})
})
},
getCompaction() {
return Effect.sync(() => {
const index = compactionIndex()
const message = state.messages[index]
return message?.type === "compaction" ? message : undefined
})
},
updateAssistant(assistant) {
return Effect.sync(() => {
const index = assistantIndex(assistant.id)
if (index < 0) return
const current = state.messages[index]
if (current?.type !== "assistant") return
state.messages[index] = assistant
})
},
updateShell(shell) {
return Effect.sync(() => {
const index = shellIndex(shell.id)
if (index < 0) return
const current = state.messages[index]
if (current?.type !== "shell") return
state.messages[index] = shell
})
},
updateCompaction(compaction) {
return Effect.sync(() => {
const index = state.messages.findLastIndex((message) => message.id === compaction.id)
if (index >= 0) state.messages[index] = compaction
})
},
appendMessage(message) {
return Effect.sync(() => {
state.messages.push(message)
})
},
}
}
export function update(adapter: Adapter, event: SessionEvent.Event) {
export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
type DraftAssistant = WritableDraft<SessionMessage.Assistant>
type DraftTool = WritableDraft<SessionMessage.AssistantTool>
type DraftText = WritableDraft<SessionMessage.AssistantText>
@@ -139,9 +53,9 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
}
})
return Effect.gen(function* () {
yield* SessionEvent.All.match(event, {
"session.usage.updated": () => Effect.void,
const project = pipe(
Match.type<SessionEvent.DurableEvent>(),
Match.discriminatorsExhaustive("type")({
"session.usage.recorded": () => Effect.void,
"session.agent.selected": (event) => {
return adapter.appendMessage(
@@ -179,7 +93,18 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.execution.succeeded": () => clearCurrentRetry,
"session.execution.failed": () => clearCurrentRetry,
"session.execution.interrupted": () => clearCurrentRetry,
"session.instructions.updated": () => Effect.void,
"session.instructions.updated": (event) => {
if (event.data.text === undefined) return Effect.void
return adapter.appendMessage(
SessionMessage.System.make({
id: SessionMessage.ID.fromEvent(event.id),
type: "system",
text: event.data.text,
metadata: event.metadata,
time: { created: event.created },
}),
)
},
"session.synthetic": (event) => {
return adapter.appendMessage(
SessionMessage.Synthetic.make({
@@ -310,12 +235,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
draft.content.push(castDraft(SessionMessage.AssistantText.make({ type: "text", text: "" })))
})
},
"session.text.delta": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestText(draft)
if (match) match.text += event.data.delta
})
},
"session.text.ended": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestText(draft)
@@ -340,7 +259,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
)
})
},
"session.tool.input.delta": () => Effect.void,
"session.tool.input.ended": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestTool(draft, event.data.id)
@@ -364,14 +282,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
}
})
},
"session.tool.progress": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestTool(draft, event.data.id)
if (match && match.state.status === "running") {
match.state.metadata = event.data.metadata
}
})
},
// Terminal tool events are self-contained; projection is a direct copy and
// never reaches into ephemeral progress history.
"session.tool.success": (event) => {
@@ -425,12 +335,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
)
})
},
"session.reasoning.delta": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestReasoning(draft)
if (match) match.text += event.data.delta
})
},
"session.reasoning.ended": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestReasoning(draft)
@@ -464,12 +368,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
time: { created: event.created },
}),
),
"session.compaction.delta": (event) =>
Effect.gen(function* () {
const current = yield* adapter.getCompaction()
if (current?.status !== "running") return
yield* adapter.updateCompaction({ ...current, summary: current.summary + event.data.text })
}),
"session.compaction.ended": (event) => {
return Effect.gen(function* () {
const current = yield* adapter.getCompaction()
@@ -515,8 +413,9 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.revert.staged": () => Effect.void,
"session.revert.cleared": () => Effect.void,
"session.revert.committed": () => Effect.void,
})
})
}),
)
return project(event)
}
export * as SessionMessageUpdater from "./message-updater"
+22 -39
View File
@@ -12,10 +12,8 @@ import {
User,
UserData,
} from "@opencode-ai/schema/session-pending"
import { Event } from "@opencode-ai/schema/event"
import type { Database } from "../database/database"
import { Bus } from "../bus"
import { EventTable } from "../event/sql"
import { KeyedMutex } from "../effect/keyed-mutex"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
@@ -37,11 +35,7 @@ const decodeUser = Schema.decodeUnknownSync(UserData)
const encodeUser = Schema.encodeSync(UserData)
const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData)
const encodeSynthetic = Schema.encodeSync(SyntheticData)
const decodeAdmittedEvent = Schema.decodeUnknownOption(SessionEvent.InputAdmitted.data)
const admittedEventType = Bus.versionedType(
SessionEvent.InputAdmitted.type,
SessionEvent.InputAdmitted.durable.version,
)
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
const inboxLocks = KeyedMutex.makeUnsafe<SessionSchema.ID>()
export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()(
@@ -103,46 +97,35 @@ export const compaction = Effect.fn("SessionPending.compaction")(function* (
return entry.type === "compaction" ? entry : undefined
})
/**
* Reconstruct the admitted record for a pending row that was already consumed
* by promotion. The projected `session_message` row proves promotion happened;
* the durable `session.input.admitted` event retains the exact admitted
* message, including delivery.
*/
const promotedFromHistory = Effect.fn("SessionPending.promotedFromHistory")(function* (
const promotedFromMessage = Effect.fn("SessionPending.promotedFromMessage")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
id: SessionMessage.ID,
delivery: Delivery,
) {
const message = yield* db
const row = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, id))
.get()
.pipe(Effect.orDie)
if (message === undefined) return undefined
if (message.session_id !== sessionID || (message.type !== "user" && message.type !== "synthetic"))
if (row === undefined) return undefined
if (row.session_id !== sessionID || (row.type !== "user" && row.type !== "synthetic"))
return yield* Effect.die(new LifecycleConflict({ id }))
const rows = yield* db
.select()
.from(EventTable)
.where(and(eq(EventTable.aggregate_id, sessionID), eq(EventTable.type, admittedEventType)))
.all()
.pipe(Effect.orDie)
for (const row of rows) {
const decoded = decodeAdmittedEvent(row.data)
if (decoded._tag !== "Some" || decoded.value.inputID !== id) continue
const base = {
id,
sessionID,
timeCreated: DateTime.makeUnsafe(row.created),
}
return decoded.value.input.type === "user"
? User.make({ ...base, ...decoded.value.input })
: Synthetic.make({ ...base, ...decoded.value.input })
}
// A projected message without an admitted event in this aggregate (for
// example fork-copied history) is not a retryable admission.
const message = decodeMessage({ ...row.data, id: row.id, type: row.type })
const base = { id, sessionID, timeCreated: message.time.created, delivery }
if (message.type === "user")
return User.make({
...base,
type: "user",
data: decodeUser(message),
})
if (message.type === "synthetic")
return Synthetic.make({
...base,
type: "synthetic",
data: decodeSynthetic(message),
})
return yield* Effect.die(new LifecycleConflict({ id }))
})
@@ -160,7 +143,7 @@ export const admit = Effect.fn("SessionPending.admit")(function* (
if (existing.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: request.id }))
return existing
}
const promoted = yield* promotedFromHistory(db, request.sessionID, request.id)
const promoted = yield* promotedFromMessage(db, request.sessionID, request.id, request.input.delivery)
if (promoted !== undefined) return promoted
return yield* bus
.publish(SessionEvent.InputAdmitted, {
@@ -426,7 +409,7 @@ const publish = Effect.fn("SessionPending.publish")(function* (
.pipe(
Effect.catchDefect((defect) =>
defect instanceof LifecycleConflict
? promotedFromHistory(db, sessionID, entry.id).pipe(
? promotedFromMessage(db, sessionID, entry.id, entry.delivery).pipe(
Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))),
)
: Effect.die(defect),
+15 -59
View File
@@ -1,6 +1,6 @@
export * as SessionProjector from "./projector"
import { and, asc, desc, eq, gt, gte, inArray, lt, lte, sql } from "drizzle-orm"
import { and, asc, desc, eq, gt, gte, lt, lte, sql } from "drizzle-orm"
import { DateTime, Effect, Layer, Schema, Stream } from "effect"
import { Database } from "../database/database"
import { Bus } from "../bus"
@@ -21,10 +21,7 @@ import { Money } from "@opencode-ai/schema/money"
type DatabaseService = Database.Interface["db"]
type CurrentDurableEvent = Extract<SessionEvent.Event, { readonly durable: object }>
type MessageEvent = Exclude<
CurrentDurableEvent,
typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type | typeof SessionEvent.InstructionsUpdated.Type
>
type MessageEvent = Exclude<CurrentDurableEvent, typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type>
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
const encodeMessage = Schema.encodeSync(SessionMessage.Info)
@@ -255,66 +252,22 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.pipe(Effect.orDie)
if (rows.length === 0) break
const idMap = new Map(rows.map((row) => [row.id, SessionMessage.ID.create()]))
yield* db
.insert(SessionMessageTable)
.values(
rows.map((row) => {
const id = idMap.get(row.id)
if (!id) throw new Error(`Fork message ID mapping missing: ${row.id}`)
return {
id,
session_id: event.data.sessionID,
type: row.type,
seq: row.seq,
time_created: row.time_created,
time_updated: row.time_updated,
data: row.data,
}
}),
rows.map((row) => ({
id: SessionMessage.ID.create(),
session_id: event.data.sessionID,
type: row.type,
seq: row.seq,
time_created: row.time_created,
time_updated: row.time_updated,
data: row.data,
})),
)
.run()
.pipe(Effect.orDie)
const pendingRows = yield* db
.select()
.from(SessionPendingTable)
.where(
and(
eq(SessionPendingTable.session_id, event.data.parentID),
inArray(
SessionPendingTable.id,
rows.map((row) => row.id),
),
),
)
.all()
.pipe(Effect.orDie)
if (pendingRows.length > 0) {
yield* db
.insert(SessionPendingTable)
.values(
pendingRows.flatMap((row) => {
const id = idMap.get(row.id)
return id && row.type !== "compaction"
? [
{
id,
session_id: event.data.sessionID,
type: row.type,
data: row.data,
delivery: row.delivery,
admitted_seq: row.admitted_seq,
time_created: row.time_created,
},
]
: []
}),
)
.run()
.pipe(Effect.orDie)
}
cursor = rows.at(-1)!.seq
}
if (copiedSeq !== undefined) yield* Bus.reserveSequence(db, event.data.sessionID, copiedSeq)
@@ -682,7 +635,10 @@ const layer = Layer.effectDiscard(
yield* bus.project(SessionEvent.Execution.Failed, (event) => run(db, event))
yield* bus.project(SessionEvent.Execution.Interrupted, (event) => run(db, event))
yield* bus.project(SessionEvent.InstructionsUpdated, (event) =>
InstructionState.apply(db, event.data.sessionID, event.durable.seq, event.data.delta),
Effect.gen(function* () {
yield* run(db, event)
yield* InstructionState.apply(db, event.data.sessionID, event.durable.seq, event.data.delta)
}),
)
yield* bus.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* bus.project(SessionEvent.Skill.Activated, (event) => run(db, event))
-34
View File
@@ -1,15 +1,12 @@
export * as ShellSelect from "./select"
import path from "path"
import { spawn, type ChildProcess } from "child_process"
import { readFile } from "fs/promises"
import { statSync } from "fs"
import { setTimeout } from "node:timers/promises"
import { Schema } from "effect"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { which } from "../util/which"
const SIGKILL_TIMEOUT_MS = 200
const META: Record<string, { deny?: boolean; login?: boolean; posix?: boolean; ps?: boolean }> = {
bash: { login: true, posix: true },
dash: { login: true, posix: true },
@@ -33,37 +30,6 @@ export const Options = Schema.Struct({
})
export type Options = typeof Options.Type
export async function killTree(proc: ChildProcess, opts?: { exited?: () => boolean }): Promise<void> {
const pid = proc.pid
if (!pid || opts?.exited?.()) return
if (process.platform === "win32") {
await new Promise<void>((resolve) => {
const killer = spawn("taskkill", ["/pid", String(pid), "/f", "/t"], {
stdio: "ignore",
windowsHide: true,
})
killer.once("exit", () => resolve())
killer.once("error", () => resolve())
})
return
}
try {
process.kill(-pid, "SIGTERM")
await setTimeout(SIGKILL_TIMEOUT_MS)
if (!opts?.exited?.()) {
process.kill(-pid, "SIGKILL")
}
} catch {
proc.kill("SIGTERM")
await setTimeout(SIGKILL_TIMEOUT_MS)
if (!opts?.exited?.()) {
proc.kill("SIGKILL")
}
}
}
function stat(file: string) {
return statSync(file, { throwIfNoEntry: false }) ?? undefined
}
+6 -59
View File
@@ -16,7 +16,7 @@ import { Hash } from "@opencode-ai/util/hash"
export { ID }
export class Error extends Schema.TaggedErrorClass<Error>()("Snapshot.Error", {
operation: Schema.Literals(["capture", "files", "diff", "preview", "restore"]),
operation: Schema.Literals(["capture", "files", "diff", "restore"]),
message: Schema.String,
cause: Schema.optional(Schema.Defect()),
}) {}
@@ -36,10 +36,6 @@ export interface RestoreInput {
readonly files: ReadonlyMap<RelativePath, ID>
}
export interface PreviewInput extends RestoreInput {
readonly context?: number
}
export interface Interface {
/**
* Capture the current Location-scoped filesystem state as a content-addressed
@@ -60,25 +56,11 @@ export interface Interface {
*/
readonly diff: (input: DiffInput) => Effect.Effect<readonly File.Diff[], Error>
/**
* Preview the filesystem result of a selective restore without modifying the
* worktree. Each project-relative path maps to the tree it would be restored
* from.
*/
readonly preview: (input: PreviewInput) => Effect.Effect<readonly File.Diff[], Error>
/**
* Restore selected project-relative paths from their associated trees. A path
* absent from its selected tree is removed; paths outside the map are untouched.
*/
*/
readonly restore: (input: RestoreInput) => Effect.Effect<void, Error>
/**
* Replace the snapshot index with a captured tree and check out all its entries.
* Files absent from the tree remain untouched. Prefer selective `restore` when
* only known paths should change.
*/
readonly checkout: (snapshot: ID) => Effect.Effect<void, Error>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/Snapshot") {}
@@ -176,59 +158,26 @@ const layer = Layer.effect(
.pipe(Effect.mapError((cause) => failure("diff", cause)))
})
const plan = Effect.fnUntraced(function* (
operation: "preview" | "restore",
worktree: AbsolutePath,
input: RestoreInput,
) {
const plan = Effect.fnUntraced(function* (worktree: AbsolutePath, input: RestoreInput) {
const files = new Map<RelativePath, Git.TreeID>()
for (const [file, snapshot] of input.files) {
const absolute = path.resolve(worktree, file)
if (!FSUtil.contains(worktree, absolute))
return yield* new Error({ operation, message: `Path escapes the project: ${file}` })
return yield* new Error({ operation: "restore", message: `Path escapes the project: ${file}` })
files.set(file, Git.TreeID.make(snapshot))
}
return files
})
const preview = Effect.fn("Snapshot.preview")(function* (input: PreviewInput) {
if (!(yield* enabled())) return yield* new Error({ operation: "preview", message: "Snapshots are disabled" })
const repo = yield* repository.pipe(Effect.mapError((cause) => failure("preview", cause)))
const files = yield* plan("preview", repo.worktree, input)
const current = yield* git.tree
.capture({
repository: repo.snapshotRepository,
scopes: Array.from(files.keys()),
ignores: repo.source,
maximumUntrackedFileBytes: 2 * 1024 * 1024,
})
.pipe(Effect.mapError((cause) => failure("preview", cause)))
return yield* git.tree
.preview({
repository: repo.snapshotRepository,
current,
files,
context: input.context,
})
.pipe(Effect.mapError((cause) => failure("preview", cause)))
})
const restore = Effect.fn("Snapshot.restore")(function* (input: RestoreInput) {
if (!(yield* enabled())) return yield* new Error({ operation: "restore", message: "Snapshots are disabled" })
const repo = yield* repository.pipe(Effect.mapError((cause) => failure("restore", cause)))
yield* git.tree
.restore({ repository: repo.snapshotRepository, files: yield* plan("restore", repo.worktree, input) })
.restore({ repository: repo.snapshotRepository, files: yield* plan(repo.worktree, input) })
.pipe(Effect.mapError((cause) => failure("restore", cause)))
})
const checkout = Effect.fn("Snapshot.checkout")(function* (snapshot: ID) {
const repo = yield* repository.pipe(Effect.mapError((cause) => failure("restore", cause)))
yield* git.tree
.checkout({ repository: repo.snapshotRepository, tree: Git.TreeID.make(snapshot) })
.pipe(Effect.mapError((cause) => failure("restore", cause)))
})
return Service.of({ capture, files, diff, preview, restore, checkout })
return Service.of({ capture, files, diff, restore })
}).pipe(Effect.withSpan("Snapshot.boot")),
)
@@ -244,9 +193,7 @@ export const noopLayer = Layer.succeed(
capture: () => Effect.succeed(undefined),
files: () => Effect.succeed([]),
diff: () => Effect.succeed([]),
preview: () => Effect.succeed([]),
restore: () => Effect.void,
checkout: () => Effect.void,
}),
)
-20
View File
@@ -1298,24 +1298,4 @@ describe("Bus", () => {
}),
)
it.effect("sequences returns the latest committed seq per aggregate and omits unknown aggregates", () =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const first = Session.ID.create()
const second = Session.ID.create()
yield* bus.publish(DurableMessage, durableData(first, "zero"))
yield* bus.publish(DurableMessage, durableData(first, "one"))
yield* bus.publish(DurableMessage, durableData(second, "zero"))
const sequences = yield* bus.sequences([first, second, Session.ID.create()])
expect(sequences).toEqual(
new Map([
[first, Event.Seq.make(1)],
[second, Event.Seq.make(0)],
]),
)
expect(yield* bus.sequences([])).toEqual(new Map())
}),
)
})
@@ -80,7 +80,6 @@ describe("node build", () => {
list: () => Effect.succeed([]),
directories: () => Effect.succeed([]),
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
commit: () => Effect.void,
})
}),
)
-162
View File
@@ -89,68 +89,6 @@ describe("FileMutation", () => {
),
)
it.live("rejects create when a prospective target appears after resolution", () =>
withTmp((directory) =>
Effect.gen(function* () {
const targetPath = path.join(directory, "appeared.txt")
const target = yield* (yield* LocationMutation.Service).resolve({ path: "appeared.txt" })
yield* Effect.promise(() => fs.writeFile(targetPath, "winner"))
expect(
yield* (yield* FileMutation.Service).create({ target, content: "replacement" }).pipe(Effect.flip),
).toMatchObject({
_tag: "FileMutation.TargetExistsError",
})
expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("winner")
}).pipe(provide(directory)),
),
)
it.live("creates when an existing target disappears after resolution", () =>
withTmp((directory) =>
Effect.gen(function* () {
const targetPath = path.join(directory, "removed.txt")
yield* Effect.promise(() => fs.writeFile(targetPath, "before"))
const target = yield* (yield* LocationMutation.Service).resolve({ path: "removed.txt" })
yield* Effect.promise(() => fs.rm(targetPath))
expect(yield* (yield* FileMutation.Service).create({ target, content: "after" })).toEqual({
operation: "write",
target: target.canonical,
resource: "removed.txt",
existed: false,
})
expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("after")
}).pipe(provide(directory)),
),
)
it.live("removes an existing internal file", () =>
withTmp((directory) =>
Effect.gen(function* () {
const targetPath = path.join(directory, "remove.txt")
yield* Effect.promise(() => fs.writeFile(targetPath, "remove"))
const target = yield* (yield* LocationMutation.Service).resolve({ path: "remove.txt" })
const result = yield* (yield* FileMutation.Service).remove({ target })
expect(result).toEqual({
operation: "remove",
target: target.canonical,
resource: "remove.txt",
existed: true,
})
expect(
yield* Effect.promise(() =>
fs.stat(targetPath).then(
() => true,
() => false,
),
),
).toBe(false)
}).pipe(provide(directory)),
),
)
it.live("writes an explicitly resolved external target", () =>
withTmp((directory) =>
withTmp((outside) =>
@@ -171,49 +109,6 @@ describe("FileMutation", () => {
),
)
it.live("removes an explicitly resolved external target", () =>
withTmp((directory) =>
withTmp((outside) =>
Effect.gen(function* () {
const targetPath = path.join(outside, "external.txt")
yield* Effect.promise(() => fs.writeFile(targetPath, "external"))
const target = yield* (yield* LocationMutation.Service).resolve({ path: targetPath })
const result = yield* (yield* FileMutation.Service).remove({ target })
expect(result).toEqual({
operation: "remove",
target: target.canonical,
resource: target.resource,
existed: true,
})
expect(
yield* Effect.promise(() =>
fs.stat(targetPath).then(
() => true,
() => false,
),
),
).toBe(false)
}).pipe(provide(directory)),
),
),
)
it.live("reports a missing target as not removed without checking existence first", () =>
withTmp((directory) =>
Effect.gen(function* () {
const target = yield* (yield* LocationMutation.Service).resolve({ path: "missing.txt" })
expect(yield* (yield* FileMutation.Service).remove({ target })).toEqual({
operation: "remove",
target: target.canonical,
resource: "missing.txt",
existed: false,
})
}).pipe(provide(directory)),
),
)
it.live("serializes concurrent writes to the same canonical target", () =>
withTmp((directory) =>
Effect.gen(function* () {
@@ -257,63 +152,6 @@ describe("FileMutation", () => {
),
)
it.live("allows only one concurrent conditional write based on the same bytes", () =>
withTmp((directory) =>
Effect.gen(function* () {
const targetPath = path.join(directory, "shared.txt")
yield* Effect.promise(() => fs.writeFile(targetPath, "initial"))
const firstStarted = yield* Deferred.make<void>()
const releaseFirst = yield* Deferred.make<void>()
let writes = 0
const filesystem = instrumentWrites((write) =>
Effect.gen(function* () {
writes++
if (writes === 1) {
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(releaseFirst)
}
yield* write
}),
)
yield* Effect.gen(function* () {
const mutation = yield* LocationMutation.Service
const files = yield* FileMutation.Service
const target = yield* mutation.resolve({ path: "shared.txt" })
const expected = new TextEncoder().encode("initial")
const first = yield* files.writeIfUnchanged({ target, expected, content: "first" }).pipe(Effect.forkChild)
yield* Deferred.await(firstStarted)
const second = yield* files
.writeIfUnchanged({ target, expected, content: "second" })
.pipe(Effect.flip, Effect.forkChild)
yield* Deferred.succeed(releaseFirst, undefined)
yield* Fiber.join(first)
expect(yield* Fiber.join(second)).toMatchObject({ _tag: "FileMutation.StaleContentError" })
expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("first")
expect(writes).toBe(1)
}).pipe(provide(directory, filesystem))
}),
),
)
it.live("rejects a conditional write when target content is already stale", () =>
withTmp((directory) =>
Effect.gen(function* () {
const targetPath = path.join(directory, "stale.txt")
yield* Effect.promise(() => fs.writeFile(targetPath, "current"))
const target = yield* (yield* LocationMutation.Service).resolve({ path: "stale.txt" })
expect(
yield* (yield* FileMutation.Service)
.writeIfUnchanged({ target, expected: new TextEncoder().encode("older"), content: "replacement" })
.pipe(Effect.flip),
).toMatchObject({ _tag: "FileMutation.StaleContentError", path: target.canonical })
expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("current")
}).pipe(provide(directory)),
),
)
it.live("allows distinct canonical targets to proceed independently", () =>
withTmp((directory) =>
Effect.gen(function* () {
+52 -37
View File
@@ -1,44 +1,59 @@
import { describe, expect } from "bun:test"
import fs from "fs/promises"
import { describe, expect, test } from "bun:test"
import os from "os"
import path from "path"
import { Effect } from "effect"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Effect, Layer } from "effect"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { FileSystem } from "@opencode-ai/core/filesystem"
import { Protected } from "@opencode-ai/core/filesystem/protected"
import { FileSystemSearch } from "@opencode-ai/core/filesystem/search"
import { Location } from "@opencode-ai/core/location"
import { Ripgrep } from "@opencode-ai/core/ripgrep"
import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema"
import { tmpdir } from "../fixture/tmpdir"
import { testEffect } from "../lib/effect"
import { location } from "../fixture/location"
const it = testEffect(LayerNode.compile(Ripgrep.node))
describe("FileSystemSearch", () => {
test("bounds a home scan even when home is detected as a repository", async () => {
let observed: Ripgrep.FindInput | undefined
const home = AbsolutePath.make(os.homedir())
const layer = AppNodeBuilder.build(FileSystemSearch.node, [
[
Location.node,
Layer.succeed(
Location.Service,
Location.Service.of(
location({ directory: home }, { vcs: { type: "git", store: AbsolutePath.make(path.join(home, ".git")) } }),
),
),
],
[
Ripgrep.node,
Layer.succeed(
Ripgrep.Service,
Ripgrep.Service.of({
find: (input) =>
Effect.gen(function* () {
observed = input
if (input.onEntry)
yield* input.onEntry(FileSystem.Entry.make({ path: RelativePath.make("src/index.ts"), type: "file" }))
return []
}),
glob: () => Effect.succeed([]),
grep: () => Effect.succeed([]),
}),
),
],
])
const withTmp = <A, E, R>(f: (directory: AbsolutePath) => Effect.Effect<A, E, R>) =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
).pipe(Effect.flatMap((tmp) => f(AbsolutePath.make(tmp.path))))
describe("Ripgrep", () => {
it.live("globs files as an array", () =>
withTmp((cwd) =>
await Effect.runPromise(
Effect.gen(function* () {
yield* Effect.promise(() => fs.mkdir(path.join(cwd, "src")))
yield* Effect.promise(() => fs.writeFile(path.join(cwd, "src", "match.ts"), "needle\n"))
const result = yield* (yield* Ripgrep.Service).glob({ cwd, pattern: "**/*.ts", limit: 10 })
expect(result.map((item) => item.path)).toEqual([RelativePath.make("src/match.ts")])
}),
),
)
it.live("greps files with include filtering", () =>
withTmp((cwd) =>
Effect.gen(function* () {
yield* Effect.promise(() => fs.mkdir(path.join(cwd, "src")))
yield* Effect.promise(() => fs.writeFile(path.join(cwd, "src", "match.ts"), "needle\n"))
yield* Effect.promise(() => fs.writeFile(path.join(cwd, "src", "skip.txt"), "needle\n"))
const result = yield* (yield* Ripgrep.Service).grep({ cwd, pattern: "needle", include: "*.ts", limit: 10 })
expect(result).toHaveLength(1)
expect(result[0]?.entry.path).toBe(RelativePath.make("src/match.ts"))
expect(result[0]?.submatches[0]?.text).toBe("needle")
}),
),
)
const search = yield* FileSystemSearch.Service
yield* Effect.sleep("10 millis")
expect(observed?.limit).toBe(100_000)
expect(observed?.exclude).toEqual([...Protected.names()].map((name) => `${name}/**`))
expect((yield* search.find({ query: "src", type: "directory" }))[0]?.path).toBe(
RelativePath.make(`src${path.sep}`),
)
}).pipe(Effect.provide(layer), Effect.scoped),
)
})
})
+34 -57
View File
@@ -56,52 +56,22 @@ function withTemp<A, E, R>(body: (directory: string) => Effect.Effect<A, E, R>)
}
describe("Formatter", () => {
it.live("status() returns empty list when no formatters are configured", () =>
it.live("does not run formatters marked as disabled in config", () =>
withTemp((directory) =>
Formatter.Service.use((formatter) => formatter.status()).pipe(Effect.provide(formatterLayer(directory))),
),
)
it.live("status() returns built-in formatters when formatter is true", () =>
withTemp((directory) =>
Formatter.Service.use((formatter) =>
Effect.gen(function* () {
const statuses = yield* formatter.status()
const gofmt = statuses.find((item) => item.name === "gofmt")
expect(gofmt).toBeDefined()
expect(gofmt?.extensions).toContain(".go")
}),
).pipe(Effect.provide(formatterLayer(directory, true))),
),
)
it.live("status() keeps built-in formatters when config object is provided", () =>
withTemp((directory) =>
Formatter.Service.use((formatter) =>
Effect.gen(function* () {
const statuses = yield* formatter.status()
expect(statuses.find((item) => item.name === "gofmt")?.extensions).toContain(".go")
expect(statuses.find((item) => item.name === "mix")).toBeDefined()
}),
).pipe(Effect.provide(formatterLayer(directory, { gofmt: {} }))),
),
)
it.live("status() excludes formatters marked as disabled in config", () =>
withTemp((directory) =>
Formatter.Service.use((formatter) =>
Effect.gen(function* () {
const statuses = yield* formatter.status()
expect(statuses.find((item) => item.name === "gofmt")).toBeUndefined()
expect(statuses.find((item) => item.name === "mix")).toBeDefined()
}),
).pipe(Effect.provide(formatterLayer(directory, { gofmt: { disabled: true } }))),
),
)
it.live("service initializes without error", () =>
withTemp((directory) =>
Formatter.Service.use((formatter) => formatter.init()).pipe(Effect.provide(formatterLayer(directory))),
Effect.gen(function* () {
const file = path.join(directory, "test.disabled")
expect(yield* Formatter.Service.use((formatter) => formatter.file(file))).toBe(false)
}).pipe(
Effect.provide(
formatterLayer(directory, {
disabled: {
disabled: true,
command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
extensions: [".disabled"],
},
}),
),
),
),
)
@@ -115,22 +85,29 @@ describe("Formatter", () => {
),
)
it.live("status() initializes formatter state per directory", () =>
Effect.acquireUseRelease(
Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
([off, on]) =>
it.live("loads formatter state per directory", () =>
withTemp((off) =>
withTemp((on) =>
Effect.gen(function* () {
const disabled = yield* Formatter.Service.use((formatter) => formatter.status()).pipe(
Effect.provide(formatterLayer(off.path, false)),
const offFile = path.join(off, "test.isolated")
const onFile = path.join(on, "test.isolated")
const disabled = yield* Formatter.Service.use((formatter) => formatter.file(offFile)).pipe(
Effect.provide(formatterLayer(off, false)),
)
const enabled = yield* Formatter.Service.use((formatter) => formatter.status()).pipe(
Effect.provide(formatterLayer(on.path, true)),
const enabled = yield* Formatter.Service.use((formatter) => formatter.file(onFile)).pipe(
Effect.provide(
formatterLayer(on, {
isolated: {
command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
extensions: [".isolated"],
},
}),
),
)
expect(disabled).toEqual([])
expect(enabled.find((item) => item.name === "gofmt")).toBeDefined()
expect(disabled).toBe(false)
expect(enabled).toBe(true)
}),
(directories) =>
Effect.promise(() => Promise.all(directories.map((tmp) => tmp[Symbol.asyncDispose]())).then(() => undefined)),
),
),
)
-3
View File
@@ -185,9 +185,6 @@ describe("Git trees", () => {
])
const files = new Map([[RelativePath.make("scope/tracked.txt"), before]])
const preview = yield* git.tree.preview({ repository, current: after, files, context: 1 })
expect(preview).toHaveLength(1)
expect(preview[0]?.file).toBe(RelativePath.make("scope/tracked.txt"))
yield* git.tree.restore({ repository, files })
expect(yield* read(path.join(root.path, "scope", "tracked.txt"))).toBe("one\n")
expect(yield* read(path.join(root.path, "scope", "added.txt"))).toBe("added\n")
+55 -11
View File
@@ -14,7 +14,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
import { InstructionState } from "@opencode-ai/core/session/instruction-state"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionSchema } from "@opencode-ai/core/session/schema"
import { InstructionBlobTable, InstructionStateTable, SessionTable } from "@opencode-ai/core/session/sql"
import { InstructionBlobTable, InstructionStateTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
import { testEffect } from "./lib/effect"
const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node])))
@@ -105,6 +105,7 @@ describe("InstructionState", () => {
expect(observation).toEqual({
sessionID,
initial: true,
previous: {},
current: {
"test/first": Instructions.hash("first"),
"test/second": Instructions.hash("second"),
@@ -156,7 +157,7 @@ describe("InstructionState", () => {
const initial = yield* InstructionState.observe(db, instructions, sessionID)
expect(reads).toBe(2)
yield* InstructionState.commit(db, events, initial)
yield* InstructionState.commit(db, events, instructions, initial)
expect(reads).toBe(2)
current = "changed"
@@ -166,6 +167,10 @@ describe("InstructionState", () => {
expect(changed).toMatchObject({
sessionID,
initial: false,
previous: {
"test/current": Instructions.hash("initial"),
"test/retired": Instructions.hash("retired"),
},
current: { "test/current": Instructions.hash("changed") },
delta: {
"test/current": Instructions.hash("changed"),
@@ -173,7 +178,7 @@ describe("InstructionState", () => {
},
blobs: { [Instructions.hash("changed")]: "changed" },
})
yield* InstructionState.commit(db, events, changed)
yield* InstructionState.commit(db, events, instructions, changed)
expect(reads).toBe(4)
yield* unsubscribe
@@ -190,6 +195,11 @@ describe("InstructionState", () => {
"test/retired": "removed",
},
])
// The chronological update text is frozen into the event; the baseline has none.
expect((yield* instructionEvents(db, sessionID)).map((event) => event.data.text)).toEqual([
undefined,
"changed\n\nRemoved retired",
])
expect(yield* db.select().from(InstructionStateTable).get().pipe(Effect.orDie)).toMatchObject({
initial_values: {
"test/current": Instructions.hash("initial"),
@@ -222,18 +232,19 @@ describe("InstructionState", () => {
expect(observation).toEqual({
sessionID,
initial: false,
previous: { "test/context": Instructions.hash("unchanged") },
current: { "test/context": Instructions.hash("unchanged") },
delta: {},
blobs: {},
})
yield* InstructionState.commit(db, events, observation)
yield* InstructionState.commit(db, events, instructions, observation)
expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs)
}),
)
it.effect("assembles a fresh private update without repairing a missing cache", () =>
it.effect("treats a missing state row as a fresh baseline without repairing it", () =>
Effect.gen(function* () {
const sessionID = SessionSchema.ID.make("ses_instruction_generate")
const { db, events } = yield* setup(sessionID)
@@ -254,7 +265,7 @@ describe("InstructionState", () => {
const assembled = yield* preview(db, sessionID, instructions)
expect(assembled).toEqual({ initial: "Initial context", updates: [], update: "Changed context" })
expect(assembled).toEqual({ initial: "Changed context", update: "" })
expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs)
expect(
@@ -268,7 +279,7 @@ describe("InstructionState", () => {
}),
)
it.effect("reads through a stale cache without repairing it", () =>
it.effect("trusts the projected state without consulting durable events", () =>
Effect.gen(function* () {
const sessionID = SessionSchema.ID.make("ses_instruction_generate_stale")
const { db, events } = yield* setup(sessionID)
@@ -280,6 +291,7 @@ describe("InstructionState", () => {
yield* InstructionState.prepare(db, events, instructions, sessionID)
value = "Committed update"
yield* InstructionState.prepare(db, events, instructions, sessionID)
// Tamper with the projected state; the authoritative row wins over event history.
yield* db
.update(InstructionStateTable)
.set({ through_seq: 0, current_values: { "test/context": Instructions.hash("Initial context") } })
@@ -294,7 +306,6 @@ describe("InstructionState", () => {
const assembled = yield* preview(db, sessionID, instructions)
expect(assembled.initial).toBe("Initial context")
expect(assembled.updates.map((entry) => entry.message.text)).toEqual(["Committed update"])
expect(assembled.update).toBe("Private update")
expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs)
@@ -302,6 +313,41 @@ describe("InstructionState", () => {
}),
)
it.effect("persists chronological updates as system messages", () =>
Effect.gen(function* () {
const sessionID = SessionSchema.ID.make("ses_instruction_messages")
const { db, events } = yield* setup(sessionID)
let value = "Initial context"
const instructions = source(
"test/context",
Effect.sync(() => value),
)
const messages = () =>
db
.select()
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "system")))
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
// The initial baseline is not chronological history and produces no message.
yield* InstructionState.prepare(db, events, instructions, sessionID)
expect(yield* messages()).toEqual([])
value = "Changed context"
yield* InstructionState.prepare(db, events, instructions, sessionID)
const rows = yield* messages()
expect(rows).toHaveLength(1)
expect(rows[0]?.data).toMatchObject({ text: "Changed context" })
expect(rows.map((row) => row.seq)).toEqual([(yield* instructionEvents(db, sessionID)).at(-1)!.seq])
// A no-op observation adds nothing.
yield* InstructionState.prepare(db, events, instructions, sessionID)
expect(yield* messages()).toHaveLength(1)
}),
)
it.effect("assembles initial instructions without persisting a baseline", () =>
Effect.gen(function* () {
const sessionID = SessionSchema.ID.make("ses_instruction_generate_initial")
@@ -310,7 +356,6 @@ describe("InstructionState", () => {
expect(yield* preview(db, sessionID, instructions)).toEqual({
initial: "Initial context",
updates: [],
update: "",
})
expect(yield* instructionEvents(db, sessionID)).toEqual([])
@@ -336,7 +381,6 @@ describe("InstructionState", () => {
expect(yield* preview(db, sessionID, instructions)).toEqual({
initial: "Committed context",
updates: [],
update: "",
})
expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents)
@@ -388,7 +432,7 @@ describe("InstructionState", () => {
for (const next of ["initial", "changed", "changed", Instructions.removed] as const) {
value = next
yield* InstructionState.observe(db, observedInstructions, observedSessionID).pipe(
Effect.flatMap((observation) => InstructionState.commit(db, events, observation)),
Effect.flatMap((observation) => InstructionState.commit(db, events, observedInstructions, observation)),
)
yield* InstructionState.prepare(db, events, preparedInstructions, preparedSessionID)
}
-1
View File
@@ -21,7 +21,6 @@ const projectLayer = Layer.succeed(
canonical: AbsolutePath.make("/main/repo"),
vcs: { type: "git", store: AbsolutePath.make("/repo/.git") },
}),
commit: () => Effect.void,
}),
)
const it = testEffect(AppNodeBuilder.build(Location.boundNode(ref), [[Project.node, projectLayer]]))
+1 -4
View File
@@ -101,13 +101,10 @@ describe("RepositoryCache", () => {
),
)
it.live("returns typed validation and clone failures", () =>
it.live("returns typed branch validation and clone failures", () =>
withRemote((fixture) =>
Effect.gen(function* () {
const cache = yield* RepositoryCache.Service
const invalidRepository = yield* Effect.flip(RepositoryCache.parseRemote("not-a-repo"))
expect(invalidRepository).toBeInstanceOf(RepositoryCache.InvalidRepositoryError)
const invalidBranch = yield* Effect.flip(cache.ensure({ reference: fixture.reference, branch: "../unsafe" }))
expect(invalidBranch).toBeInstanceOf(RepositoryCache.InvalidBranchError)
+61
View File
@@ -11,6 +11,44 @@ import { testEffect } from "./lib/effect"
const it = testEffect(LayerNode.compile(Ripgrep.node))
describe("Ripgrep", () => {
it.live("globs files as an array", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
(tmp) =>
Effect.gen(function* () {
yield* Effect.promise(() => fs.mkdir(path.join(tmp.path, "src")))
yield* Effect.promise(() => fs.writeFile(path.join(tmp.path, "src", "match.ts"), "needle\n"))
const result = yield* (yield* Ripgrep.Service).glob({ cwd: tmp.path, pattern: "**/*.ts", limit: 10 })
expect(result.map((item) => item.path)).toEqual([RelativePath.make("src/match.ts")])
}),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
it.live("greps files with include filtering", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
(tmp) =>
Effect.gen(function* () {
yield* Effect.promise(() => fs.mkdir(path.join(tmp.path, "src")))
yield* Effect.promise(() => fs.writeFile(path.join(tmp.path, "src", "match.ts"), "needle\n"))
yield* Effect.promise(() => fs.writeFile(path.join(tmp.path, "src", "skip.txt"), "needle\n"))
const result = yield* (yield* Ripgrep.Service).grep({
cwd: tmp.path,
pattern: "needle",
include: "*.ts",
limit: 10,
})
expect(result).toHaveLength(1)
expect(result[0]?.entry.path).toBe(RelativePath.make("src/match.ts"))
expect(result[0]?.submatches[0]?.text).toBe("needle")
}),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
it.live("keeps ignored files out of catch-all find results", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
@@ -63,6 +101,29 @@ describe("Ripgrep", () => {
),
)
it.live("excludes protected directory trees from catch-all find results", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
(tmp) =>
Effect.gen(function* () {
yield* Effect.promise(() => fs.mkdir(path.join(tmp.path, "Pictures")))
yield* Effect.promise(() => fs.writeFile(path.join(tmp.path, "Pictures", "private.jpg"), "private\n"))
yield* Effect.promise(() => fs.writeFile(path.join(tmp.path, "visible.txt"), "visible\n"))
const files = yield* (yield* Ripgrep.Service).find({
cwd: tmp.path,
pattern: "*",
limit: 10,
exclude: ["Pictures/**"],
})
expect(files.map((item) => item.path)).toContain(RelativePath.make("visible.txt"))
expect(files.map((item) => item.path)).not.toContain(RelativePath.make("Pictures/private.jpg"))
}),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
it.live("returns a bounded preview for matches on oversized lines", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
@@ -36,7 +36,6 @@ const projects = Layer.succeed(
list: () => Effect.succeed([]),
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
directories: () => Effect.succeed([]),
commit: () => Effect.void,
}),
)
let requests: LLMRequest[] = []
+2 -7
View File
@@ -34,7 +34,6 @@ const projects = Layer.succeed(
list: () => Effect.succeed([]),
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
directories: () => Effect.succeed([]),
commit: () => Effect.void,
}),
)
const it = testEffect(
@@ -284,13 +283,9 @@ describe("Session.create", () => {
})
expect(yield* SessionPending.find(db, forkContext[0].id)).toBeUndefined()
expect(yield* SessionPending.find(db, forkContext[1].id)).toBeUndefined()
// Fork-copied messages have no admitted event in the fork aggregate, so
// reusing their IDs as prompt IDs is conflicting reuse, not a retry.
expect(
yield* session
.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false })
.pipe(Effect.flip),
).toMatchObject({ _tag: "Session.PromptConflictError", messageID: forkContext[0].id })
yield* session.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }),
).toMatchObject({ id: forkContext[0].id, type: "user", data: { text: "First" } })
yield* session.prompt({
sessionID: parent.id,
@@ -56,7 +56,6 @@ const projects = Layer.succeed(
list: () => Effect.succeed([]),
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
directories: () => Effect.succeed([]),
commit: () => Effect.void,
}),
)
const permission = Layer.succeed(
+1 -4
View File
@@ -23,7 +23,6 @@ const projects = Layer.succeed(
list: () => Effect.succeed([]),
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
directories: () => Effect.succeed([]),
commit: () => Effect.void,
}),
)
const it = testEffect(
@@ -41,17 +40,15 @@ describe("Session.log", () => {
it.effect("replays public session events and marks synced at the aggregate watermark", () =>
Effect.gen(function* () {
const session = yield* Session.Service
const bus = yield* Bus.Service
const created = yield* session.create({ location })
yield* session.rename({ sessionID: created.id, title: "session.renamed" })
const items = Array.from(yield* Stream.runCollect(session.log({ sessionID: created.id })))
const watermark = (yield* bus.sequences([created.id])).get(created.id)
// Session creation commits a non-public durable event, so the marker's
// seq covers more of the aggregate than the public events emitted.
expect(items.map((item) => item.type)).toEqual(["session.renamed", "log.synced"])
expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: watermark })
expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: Event.Seq.make(1) })
}),
)
@@ -17,7 +17,6 @@ import { Session } from "@opencode-ai/core/session"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { Money } from "@opencode-ai/schema/money"
import { SessionMessageUpdater } from "@opencode-ai/core/session/message-updater"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { fromRow } from "@opencode-ai/core/session/info"
@@ -126,34 +125,6 @@ describe("SessionProjector", () => {
}),
)
it.effect("folds live compaction deltas into running memory state", () =>
Effect.gen(function* () {
const state = {
messages: [
SessionMessage.CompactionRunning.make({
id: SessionMessage.ID.make("msg_compaction"),
type: "compaction",
status: "running",
reason: "manual",
summary: "partial ",
recent: "recent",
time: { created },
}),
],
}
yield* SessionMessageUpdater.update(
SessionMessageUpdater.memory(state),
SessionEvent.Compaction.Delta.make({
id: Event.ID.make("evt_delta"),
type: "session.compaction.delta",
created,
data: { sessionID, text: "summary" },
}),
)
expect(state.messages[0]).toMatchObject({ status: "running", summary: "partial summary", recent: "recent" })
}),
)
it.effect("projects staged, cleared, and committed reverts", () =>
Effect.gen(function* () {
const db = (yield* Database.Service).db
@@ -550,31 +521,6 @@ describe("SessionProjector", () => {
}),
)
it.effect("does not revive a stale incomplete in-memory assistant projection", () =>
Effect.gen(function* () {
const stale = SessionMessage.Assistant.make({
id: SessionMessage.ID.make("msg_assistant_stale"),
type: "assistant",
agent: build,
model,
content: [],
time: { created },
})
const completed = SessionMessage.Assistant.make({
id: SessionMessage.ID.make("msg_assistant_completed"),
type: "assistant",
agent: build,
model,
content: [],
time: { created: DateTime.makeUnsafe(1), completed: DateTime.makeUnsafe(2) },
})
expect(
yield* SessionMessageUpdater.memory({ messages: [stale, completed] }).getCurrentAssistant(),
).toBeUndefined()
}),
)
it.effect("projects retry state and clears it at the next step or execution terminal", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
+41
View File
@@ -553,6 +553,47 @@ describe("Session.prompt", () => {
}),
)
it.effect("reconciles an exact retry from the promoted message without admission history", () =>
Effect.gen(function* () {
yield* setup
const session = yield* Session.Service
const bus = yield* Bus.Service
const { db } = yield* Database.Service
const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
const first = yield* session.prompt(input)
yield* SessionPending.promote(db, bus, sessionID, "steer")
yield* db
.delete(EventTable)
.where(eq(EventTable.aggregate_id, sessionID))
.run()
.pipe(Effect.orDie)
const retried = yield* session.prompt(input)
expect(retried).toMatchObject({ id: first.id, type: "user", data: { text: first.data.text } })
expect(yield* session.messages({ sessionID })).toMatchObject([
{ id: messageID, type: "user", text: "Fix the failing tests" },
])
}),
)
it.effect("ignores delivery when retrying a promoted message", () =>
Effect.gen(function* () {
yield* setup
const session = yield* Session.Service
const bus = yield* Bus.Service
const { db } = yield* Database.Service
const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
yield* session.prompt(input)
yield* SessionPending.promote(db, bus, sessionID, "steer")
const retried = yield* session.prompt({ ...input, delivery: "queue" })
expect(retried).toMatchObject({ id: messageID, type: "user", data: { text: input.text } })
expect(yield* admitted(messageID)).toBeUndefined()
}),
)
it.effect("wakes execution when an exact prompt retry recovers a committed message", () =>
Effect.gen(function* () {
yield* setup
@@ -19,7 +19,6 @@ const projects = Layer.succeed(
list: () => Effect.succeed([]),
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
directories: () => Effect.succeed([]),
commit: () => Effect.void,
}),
)
const it = testEffect(
+22 -12
View File
@@ -1180,7 +1180,7 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("forks instruction values at the selected message instead of the parent's latest state", () =>
it.effect("seeds a fork with the parent's newest instruction values", () =>
Effect.gen(function* () {
const session = yield* setup
yield* runPrompt(session, "First")
@@ -1197,14 +1197,16 @@ describe("SessionRunnerLLM", () => {
.where(eq(InstructionStateTable.session_id, forked.id))
.get(),
).toMatchObject({
initial_values: { "test/context": Instructions.hash("Changed context") },
current_values: { "test/context": Instructions.hash("Changed context") },
initial_values: { "test/context": Instructions.hash("Latest context") },
current_values: { "test/context": Instructions.hash("Latest context") },
})
yield* session.prompt({ sessionID: forked.id, text: "Forked", resume: false })
yield* session.resume(forked.id)
expect(requests.at(-1)?.system.map((part) => part.text)).toEqual([defaultSystem, "Changed context"])
expect(systemTexts(requests.at(-1)!)).toContain("Latest context")
expect(requests.at(-1)?.system.map((part) => part.text)).toEqual([defaultSystem, "Latest context"])
// Copied history keeps the frozen chronological update; no new update is emitted.
expect(systemTexts(requests.at(-1)!)).toContain("Changed context")
expect(systemTexts(requests.at(-1)!)).not.toContain("Latest context")
const { db } = yield* Database.Service
const bus = yield* Bus.Service
@@ -1263,7 +1265,7 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("rebuilds a missing instruction cache without admitting another delta", () =>
it.effect("re-establishes a fresh baseline when instruction state is missing", () =>
Effect.gen(function* () {
const session = yield* setup
const { db } = yield* Database.Service
@@ -1277,13 +1279,15 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(1)
expect(requests[0]?.system.map((part) => part.text)).toEqual([defaultSystem, "Initial context"])
expect(messageRoles(requests[0])).toEqual(["user", "user"])
// The projected row is authoritative: a missing row admits a fresh baseline
// instead of rebuilding from durable events.
expect(
yield* db
.select({ id: EventTable.id })
.select({ data: EventTable.data })
.from(EventTable)
.where(eq(EventTable.type, "session.instructions.updated.2"))
.all(),
).toHaveLength(1)
).toHaveLength(2)
expect(yield* db.select().from(InstructionStateTable).get()).toMatchObject({
initial_values: { "test/context": Instructions.hash("Initial context") },
current_values: { "test/context": Instructions.hash("Initial context") },
@@ -1310,7 +1314,10 @@ describe("SessionRunnerLLM", () => {
])
expect(messageRoles(requests[1])).toEqual(["user", "system", "user"])
expect(requests[1]?.messages.at(1)?.content).toEqual([{ type: "text", text: "Changed context" }])
expect(yield* session.messages({ sessionID })).toHaveLength(2)
// The chronological update is a durable client-visible system message.
const messages = yield* session.messages({ sessionID })
expect(messages).toHaveLength(3)
expect(messages[1]).toMatchObject({ type: "system", text: "Changed context" })
const { db } = yield* Database.Service
const updates = yield* db
.select({ data: EventTable.data })
@@ -1327,9 +1334,10 @@ describe("SessionRunnerLLM", () => {
expect(updates[1]?.data).toEqual({
sessionID,
delta: { "test/context": Instructions.hash("Changed context") },
text: "Changed context",
})
yield* replaySessionProjection(sessionID)
expect(yield* session.messages({ sessionID })).toHaveLength(2)
expect(yield* session.messages({ sessionID })).toHaveLength(3)
}),
)
@@ -1596,7 +1604,7 @@ describe("SessionRunnerLLM", () => {
expect(requests[1]?.messages.at(1)?.content).toEqual([
{ type: "text", text: "System context source removed: test/context" },
])
expect(yield* session.messages({ sessionID })).toHaveLength(2)
expect(yield* session.messages({ sessionID })).toHaveLength(3)
}),
)
@@ -1708,12 +1716,14 @@ describe("SessionRunnerLLM", () => {
expect(requests[2]?.messages.filter((message) => message.role === "system")).toHaveLength(2)
expect((yield* session.context(sessionID)).map((message) => message.type)).toEqual([
"user",
"system",
"user",
"model-switched",
"system",
"user",
])
yield* replaySessionProjection(sessionID)
expect(yield* session.messages({ sessionID })).toHaveLength(4)
expect(yield* session.messages({ sessionID })).toHaveLength(6)
yield* runPrompt(session, "Fourth")
}),
)
-33
View File
@@ -117,9 +117,6 @@ describe("Snapshot", () => {
RelativePath.make("scope/tracked.txt"),
])
const plan = new Map([[RelativePath.make("scope/tracked.txt"), before]])
const preview = yield* snapshot.preview({ files: plan, context: 1 })
expect(preview).toHaveLength(1)
expect(preview[0]?.file).toBe(RelativePath.make("scope/tracked.txt"))
yield* snapshot.restore({ files: plan })
expect(yield* read(path.join(location, "tracked.txt"))).toBe("one\n")
expect(yield* read(path.join(location, "added.txt"))).toBe("added\n")
@@ -185,36 +182,6 @@ describe("Snapshot", () => {
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
testEffect(Layer.empty).live("checks out a legacy revert snapshot without removing unrelated files", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
(tmp) =>
Effect.gen(function* () {
const project = path.join(tmp.path, "project")
yield* Effect.promise(async () => {
await fs.mkdir(project)
await fs.writeFile(path.join(project, "tracked.txt"), "one\n")
await initGit(project)
})
yield* Effect.gen(function* () {
const snapshot = yield* Snapshot.Service
const before = yield* snapshot.capture()
expect(before).toBeDefined()
if (!before) return
yield* Effect.promise(async () => {
await fs.writeFile(path.join(project, "tracked.txt"), "two\n")
await fs.writeFile(path.join(project, "unrelated.txt"), "keep\n")
})
yield* snapshot.checkout(before)
expect(yield* read(path.join(project, "tracked.txt"))).toBe("one\n")
expect(yield* read(path.join(project, "unrelated.txt"))).toBe("keep\n")
}).pipe(Effect.provide(snapshotLayer(tmp.path, project)))
}),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
})
function snapshotLayer(data: string, directory: string) {
+5
View File
@@ -186,6 +186,11 @@ export const InstructionsUpdated = Event.durable({
schema: {
...Base,
delta: Instruction.Delta,
/**
* The rendered chronological update shown to the model, frozen at emit time.
* Absent for the initial baseline observation and for deltas that render empty.
*/
text: Schema.String.pipe(optional),
},
})
export type InstructionsUpdated = typeof InstructionsUpdated.Type
-17
View File
@@ -2,10 +2,6 @@ export * as ServerAuth from "./auth"
import { Context, Layer, Option, Redacted } from "effect"
export type Credentials = {
password?: string
}
export type DecodedCredentials = {
readonly username: string
readonly password: Redacted.Redacted
@@ -37,16 +33,3 @@ export function authorized(credentials: DecodedCredentials, config: Info) {
Redacted.value(credentials.password) === config.password.value
)
}
export function header(credentials?: Credentials) {
const password = credentials?.password
if (!password) return undefined
return `Basic ${Buffer.from(`opencode:${password}`).toString("base64")}`
}
export function headers(credentials?: Credentials) {
const authorization = header(credentials)
if (!authorization) return undefined
return { Authorization: authorization }
}
-4
View File
@@ -7,7 +7,3 @@ test("accepts only the fixed opencode username", () => {
expect(ServerAuth.authorized({ username: "opencode", password: Redacted.make("secret") }, config)).toBe(true)
expect(ServerAuth.authorized({ username: "custom", password: Redacted.make("secret") }, config)).toBe(false)
})
test("encodes the fixed opencode username", () => {
expect(ServerAuth.header({ password: "secret" })).toBe(`Basic ${Buffer.from("opencode:secret").toString("base64")}`)
})
@@ -36,10 +36,6 @@ export const lineCommentStyles = `
border: none;
}
[data-component="line-comment"][data-variant="add"] [data-slot="line-comment-button"] {
background: var(--syntax-diff-add);
}
[data-component="line-comment"] [data-component="icon"] {
color: var(--white);
}
@@ -9,7 +9,7 @@ import { useI18n } from "@opencode-ai/ui/context/i18n"
installLineCommentStyles()
export type LineCommentVariant = "default" | "editor" | "add"
export type LineCommentVariant = "default" | "editor"
function InlineGlyph(props: { icon: "comment" | "plus" }) {
return (
@@ -156,25 +156,6 @@ export const LineComment = (props: LineCommentProps) => {
)
}
export type LineCommentAddProps = Omit<LineCommentAnchorProps, "children" | "variant" | "open" | "icon"> & {
label?: string
}
export const LineCommentAdd = (props: LineCommentAddProps) => {
const [split, rest] = splitProps(props, ["label"])
const i18n = useI18n()
return (
<LineCommentAnchor
{...rest}
open={false}
variant="add"
icon="plus"
buttonLabel={split.label ?? i18n.t("ui.lineComment.submit")}
/>
)
}
export type LineCommentEditorProps = Omit<LineCommentAnchorProps, "children" | "open" | "variant" | "onClick"> & {
value: string
selection: JSX.Element
@@ -948,10 +948,6 @@ function ExaOutput(props: { output?: string }) {
)
}
export function registerPartComponent(type: string, component: PartComponent) {
PART_MAPPING[type] = component
}
export function Message(props: MessageProps) {
return (
<Switch>
@@ -147,12 +147,12 @@ async function renderDiffViewer(vcsDiff: unknown[], height = 20, initialRoute?:
const config = createTuiResolvedConfig()
const transport = createFetch((url) => {
if (url.pathname !== "/api/vcs/diff") return
if (fail) return json({ message: "boom" }, { status: 500 })
vcsDiffInput = {
location: { directory: url.searchParams.get("location[directory]") },
mode: url.searchParams.get("mode"),
context: url.searchParams.get("context"),
}
if (fail) return json({ message: "boom" }, { status: 500 })
return json({
location: { directory: "/repo/session", project: { id: "project-1", directory: "/repo/session" } },
data: vcsDiff,
@@ -238,6 +238,7 @@ async function renderDiffViewer(vcsDiff: unknown[], height = 20, initialRoute?:
const app = await testRender(() => <Harness />, { width: 80, height })
await waitForCommand(app, commands, "diff.close")
await app.waitFor(() => vcsDiffInput !== undefined)
return {
app,
commands,