Compare commits

..

4 Commits

Author SHA1 Message Date
Kit Langton
0aa7525da8 Merge branch 'dev' into kit/windows-session-restore 2026-03-25 19:49:36 -04:00
Kit Langton
638a025ab4 fix(app): resync session model on session change 2026-03-25 15:24:15 -04:00
Kit Langton
31e92f1454 fix(app): normalize e2e workspace path checks 2026-03-25 15:01:43 -04:00
Kit Langton
6aaaa4c807 fix(app): normalize workspace keys for session restore 2026-03-25 14:46:10 -04:00
65 changed files with 1339 additions and 2655 deletions

1
.github/VOUCHED.td vendored
View File

@@ -25,4 +25,3 @@ r44vc0rp
rekram1-node
-spider-yamet clawdbot/llm psychosis, spam pinging the team
thdxr
-OpenCodeEngineer bot that spams issues

View File

@@ -358,7 +358,7 @@
"drizzle-orm": "catalog:",
"effect": "catalog:",
"fuzzysort": "3.1.0",
"gitlab-ai-provider": "5.3.3",
"gitlab-ai-provider": "5.3.2",
"glob": "13.0.5",
"google-auth-library": "10.5.0",
"gray-matter": "4.0.3",
@@ -3037,7 +3037,7 @@
"github-slugger": ["github-slugger@2.0.0", "", {}, "sha512-IaOQ9puYtjrkq7Y0Ygl9KDZnrf/aiUJYUpVf89y8kyaxbRG7Y1SrX/jaumrv81vc61+kiMempujsM3Yw7w5qcw=="],
"gitlab-ai-provider": ["gitlab-ai-provider@5.3.3", "", { "dependencies": { "@anthropic-ai/sdk": "^0.71.0", "@anycable/core": "^0.9.2", "graphql-request": "^6.1.0", "isomorphic-ws": "^5.0.0", "openai": "^6.16.0", "socket.io-client": "^4.8.1", "vscode-jsonrpc": "^8.2.1", "zod": "^3.25.76" }, "peerDependencies": { "@ai-sdk/provider": ">=2.0.0", "@ai-sdk/provider-utils": ">=3.0.0" } }, "sha512-k0kRUoAhDvoRC28hQW4sPp+A3cfpT5c/oL9Ng10S0oBiF2Tci1AtsX1iclJM5Os8C1nIIAXBW8LMr0GY7rwcGA=="],
"gitlab-ai-provider": ["gitlab-ai-provider@5.3.2", "", { "dependencies": { "@anthropic-ai/sdk": "^0.71.0", "@anycable/core": "^0.9.2", "graphql-request": "^6.1.0", "isomorphic-ws": "^5.0.0", "openai": "^6.16.0", "socket.io-client": "^4.8.1", "vscode-jsonrpc": "^8.2.1", "zod": "^3.25.76" }, "peerDependencies": { "@ai-sdk/provider": ">=2.0.0", "@ai-sdk/provider-utils": ">=3.0.0" } }, "sha512-EiAipDMa4Ngsxp4MMaua5YHWsHhc9kGXKmBxulJg1Gueb+5IZmMwxaVtgWTGWZITxC3tzKEeRt/3U4McE2vTIA=="],
"glob": ["glob@13.0.5", "", { "dependencies": { "minimatch": "^10.2.1", "minipass": "^7.1.2", "path-scurry": "^2.0.0" } }, "sha512-BzXxZg24Ibra1pbQ/zE7Kys4Ua1ks7Bn6pKLkVPZ9FZe4JQS6/Q7ef3LG1H+k7lUf5l4T3PLSyYyYJVYUvfgTw=="],

View File

@@ -1,8 +1,8 @@
{
"nodeModules": {
"x86_64-linux": "sha256-0VwVhbOtK1r16cVSZcHaI/8fUPc6aYQiUnh7Q3bSHqs=",
"aarch64-linux": "sha256-z5b234MIS0QqDYLopyaT2hd9CAtEbcSo28y0eMfPsBs=",
"aarch64-darwin": "sha256-sn16mtZIhF9OSBrfAHpDCJO6Nt19mdoxvYAOnwWgwDk=",
"x86_64-darwin": "sha256-FaZpwGuWzfypA28ct86xAnW2RuFFUiXjPkr5wVTLN/o="
"x86_64-linux": "sha256-l3k/1fRIAQkj7zdVj2Ad3QZWeTOf1CuIM6vgMHRaK1s=",
"aarch64-linux": "sha256-iN3YtrKAUTK1GIwVMoVYkMXhtDZOiP7sSJ+Z8v4B5xw=",
"aarch64-darwin": "sha256-FFedoiPyfHGdzQnITz1SRV7xv2XoT9vzxIDp4EcVdkU=",
"x86_64-darwin": "sha256-0HiHkhJiN73UixUq5CC6YP6DkZzLar8lKnEL1aoiiHg="
}
}

View File

@@ -5,7 +5,7 @@ import os from "node:os"
import path from "node:path"
import { execSync } from "node:child_process"
import { terminalAttr, type E2EWindow } from "../src/testing/terminal"
import { createSdk, modKey, resolveDirectory, serverUrl } from "./utils"
import { createSdk, modKey, resolveDirectory, serverUrl, workspaceKey } from "./utils"
import {
dropdownMenuTriggerSelector,
dropdownMenuContentSelector,
@@ -438,7 +438,7 @@ export async function resolveSlug(slug: string) {
}
export async function waitDir(page: Page, directory: string) {
const target = await resolveDirectory(directory)
const target = workspaceKey(await resolveDirectory(directory))
await expect
.poll(
async () => {
@@ -446,7 +446,7 @@ export async function waitDir(page: Page, directory: string) {
const slug = slugFromUrl(page.url())
if (!slug) return ""
return resolveSlug(slug)
.then((item) => item.directory)
.then((item) => workspaceKey(item.directory))
.catch(() => "")
},
{ timeout: 45_000 },
@@ -456,7 +456,7 @@ export async function waitDir(page: Page, directory: string) {
}
export async function waitSession(page: Page, input: { directory: string; sessionID?: string }) {
const target = await resolveDirectory(input.directory)
const target = workspaceKey(await resolveDirectory(input.directory))
await expect
.poll(
async () => {
@@ -464,14 +464,14 @@ export async function waitSession(page: Page, input: { directory: string; sessio
const slug = slugFromUrl(page.url())
if (!slug) return false
const resolved = await resolveSlug(slug).catch(() => undefined)
if (!resolved || resolved.directory !== target) return false
if (!resolved || workspaceKey(resolved.directory) !== target) return false
if (input.sessionID && sessionIDFromUrl(page.url()) !== input.sessionID) return false
const state = await probeSession(page)
if (input.sessionID && (!state || state.sessionID !== input.sessionID)) return false
if (state?.dir) {
const dir = await resolveDirectory(state.dir).catch(() => state.dir ?? "")
if (dir !== target) return false
if (workspaceKey(dir) !== target) return false
}
return page
@@ -488,7 +488,7 @@ export async function waitSession(page: Page, input: { directory: string; sessio
export async function waitSessionSaved(directory: string, sessionID: string, timeout = 30_000) {
const sdk = createSdk(directory)
const target = await resolveDirectory(directory)
const target = workspaceKey(await resolveDirectory(directory))
await expect
.poll(
@@ -498,7 +498,9 @@ export async function waitSessionSaved(directory: string, sessionID: string, tim
.then((x) => x.data)
.catch(() => undefined)
if (!data?.directory) return ""
return resolveDirectory(data.directory).catch(() => data.directory)
return resolveDirectory(data.directory)
.then(workspaceKey)
.catch(() => workspaceKey(data.directory))
},
{ timeout },
)

View File

@@ -36,6 +36,14 @@ export async function resolveDirectory(directory: string) {
.then((x) => x.data?.directory ?? directory)
}
export function workspaceKey(dir: string) {
const value = dir.replaceAll("\\", "/")
const drive = value.match(/^([A-Za-z]:)\/+$/)
if (drive) return `${drive[1]}/`
if (/^\/+$/i.test(value)) return "/"
return value.replace(/\/+$/, "")
}
export async function getWorktree() {
const sdk = createSdk()
const result = await sdk.path.get()
@@ -57,7 +65,8 @@ export function sessionPath(directory: string, sessionID?: string) {
}
export function workspacePersistKey(directory: string, key: string) {
const head = (directory.slice(0, 12) || "workspace").replace(/[^a-zA-Z0-9._-]/g, "-")
const sum = checksum(directory) ?? "0"
const dir = workspaceKey(directory)
const head = (dir.slice(0, 12) || "workspace").replace(/[^a-zA-Z0-9._-]/g, "-")
const sum = checksum(dir) ?? "0"
return `opencode.workspace.${head}.${sum}.dat:workspace:${key}`
}

View File

@@ -6,7 +6,8 @@ const serverHost = process.env.PLAYWRIGHT_SERVER_HOST ?? "127.0.0.1"
const serverPort = process.env.PLAYWRIGHT_SERVER_PORT ?? "4096"
const command = `bun run dev -- --host 0.0.0.0 --port ${port}`
const reuse = !process.env.CI
const workers = Number(process.env.PLAYWRIGHT_WORKERS ?? (process.env.CI ? 5 : 0)) || undefined
const workers =
Number(process.env.PLAYWRIGHT_WORKERS ?? (process.env.CI ? (process.platform === "win32" ? 2 : 5) : 0)) || undefined
export default defineConfig({
testDir: "./e2e",

View File

@@ -7,6 +7,7 @@ import { useModels } from "@/context/models"
import { useProviders } from "@/hooks/use-providers"
import { modelEnabled, modelProbe } from "@/testing/model-selection"
import { Persist, persisted } from "@/utils/persist"
import { workspaceKey } from "@/utils/workspace"
import { cycleModelVariant, getConfiguredAgentVariant, resolveModelVariant } from "./model-variant"
import { useSDK } from "./sdk"
import { useSync } from "./sync"
@@ -26,7 +27,7 @@ type Saved = {
const WORKSPACE_KEY = "__workspace__"
const handoff = new Map<string, State>()
const handoffKey = (dir: string, id: string) => `${dir}\n${id}`
const handoffKey = (dir: string, id: string) => `${workspaceKey(dir)}\n${id}`
const migrate = (value: unknown) => {
if (!value || typeof value !== "object") return { session: {} }
@@ -364,7 +365,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const next = clone(snapshot())
if (!next) return
if (dir === sdk.directory) {
if (workspaceKey(dir) === workspaceKey(sdk.directory)) {
setSaved("session", session, next)
setStore("draft", undefined)
return

View File

@@ -1,19 +1,13 @@
import { getFilename } from "@opencode-ai/util/path"
import { type Session } from "@opencode-ai/sdk/v2/client"
export { workspaceKey } from "@/utils/workspace"
import { workspaceKey } from "@/utils/workspace"
type SessionStore = {
session?: Session[]
path: { directory: string }
}
export const workspaceKey = (directory: string) => {
const value = directory.replaceAll("\\", "/")
const drive = value.match(/^([A-Za-z]:)\/+$/)
if (drive) return `${drive[1]}/`
if (/^\/+$/i.test(value)) return "/"
return value.replace(/\/+$/, "")
}
function sortSessions(now: number) {
const oneMinuteAgo = now - 60 * 1000
return (a: Session, b: Session) => {

View File

@@ -494,7 +494,7 @@ export default function Page() {
createEffect(
on(
() => lastUserMessage()?.id,
() => [params.id, lastUserMessage()?.id] as const,
() => {
const msg = lastUserMessage()
if (!msg) return

View File

@@ -112,4 +112,11 @@ describe("persist localStorage resilience", () => {
expect(result.endsWith(".dat")).toBeTrue()
expect(/[:\\/]/.test(result)).toBeFalse()
})
test("workspace storage treats slash variants as the same workspace", () => {
const a = persistTesting.workspaceStorage("C:\\Users\\foo\\bar\\")
const b = persistTesting.workspaceStorage("C:/Users/foo/bar")
expect(a).toBe(b)
})
})

View File

@@ -1,4 +1,5 @@
import { Platform, usePlatform } from "@/context/platform"
import { workspaceKey } from "@/utils/workspace"
import { makePersisted, type AsyncStorage, type SyncStorage } from "@solid-primitives/storage"
import { checksum } from "@opencode-ai/util/encode"
import { createResource, type Accessor } from "solid-js"
@@ -209,8 +210,9 @@ function normalize(defaults: unknown, raw: string, migrate?: (value: unknown) =>
}
function workspaceStorage(dir: string) {
const head = (dir.slice(0, 12) || "workspace").replace(/[^a-zA-Z0-9._-]/g, "-")
const sum = checksum(dir) ?? "0"
const key = workspaceKey(dir)
const head = (key.slice(0, 12) || "workspace").replace(/[^a-zA-Z0-9._-]/g, "-")
const sum = checksum(key) ?? "0"
return `opencode.workspace.${head}.${sum}.dat`
}

View File

@@ -0,0 +1,7 @@
export const workspaceKey = (dir: string) => {
const value = dir.replaceAll("\\", "/")
const drive = value.match(/^([A-Za-z]:)\/+$/)
if (drive) return `${drive[1]}/`
if (/^\/+$/i.test(value)) return "/"
return value.replace(/\/+$/, "")
}

View File

@@ -132,7 +132,7 @@ export async function handler(
retry,
stickyProvider,
)
validateModelSettings(billingSource, authInfo)
validateModelSettings(authInfo)
updateProviderKey(authInfo, providerInfo)
logger.metric({ provider: providerInfo.id })
@@ -768,10 +768,9 @@ export async function handler(
return "balance"
}
function validateModelSettings(billingSource: BillingSource, authInfo: AuthInfo) {
if (billingSource === "lite") return
if (billingSource === "anonymous") return
if (authInfo!.isDisabled) throw new ModelError(t("zen.api.error.modelDisabled"))
function validateModelSettings(authInfo: AuthInfo) {
if (!authInfo) return
if (authInfo.isDisabled) throw new ModelError(t("zen.api.error.modelDisabled"))
}
function updateProviderKey(authInfo: AuthInfo, providerInfo: ProviderInfo) {

View File

@@ -14,7 +14,6 @@ import { KeyTable } from "../src/schema/key.sql.js"
import { BlackData } from "../src/black.js"
import { centsToMicroCents } from "../src/util/price.js"
import { getWeekBounds } from "../src/util/date.js"
import { ModelTable } from "../src/schema/model.sql.js"
// get input from command line
const identifier = process.argv[2]
@@ -179,8 +178,9 @@ async function printWorkspace(workspaceID: string) {
balance: `$${(row.balance / 100000000).toFixed(2)}`,
reload: row.reload ? "yes" : "no",
customerID: row.customerID,
GO: row.liteSubscriptionID,
Black: row.blackSubscriptionID
liteSubscriptionID: row.liteSubscriptionID,
blackSubscriptionID: row.blackSubscriptionID,
blackSubscription: row.blackSubscriptionID
? [
`Black ${row.blackSubscription.enrichment!.plan}`,
row.blackSubscription.enrichment!.seats > 1
@@ -223,50 +223,6 @@ async function printWorkspace(workspaceID: string) {
),
)
await printTable("28-Day Usage", (tx) =>
tx
.select({
date: sql<string>`DATE(${UsageTable.timeCreated})`.as("date"),
requests: sql<number>`COUNT(*)`.as("requests"),
inputTokens: sql<number>`SUM(${UsageTable.inputTokens})`.as("input_tokens"),
outputTokens: sql<number>`SUM(${UsageTable.outputTokens})`.as("output_tokens"),
reasoningTokens: sql<number>`SUM(${UsageTable.reasoningTokens})`.as("reasoning_tokens"),
cacheReadTokens: sql<number>`SUM(${UsageTable.cacheReadTokens})`.as("cache_read_tokens"),
cacheWrite5mTokens: sql<number>`SUM(${UsageTable.cacheWrite5mTokens})`.as("cache_write_5m_tokens"),
cacheWrite1hTokens: sql<number>`SUM(${UsageTable.cacheWrite1hTokens})`.as("cache_write_1h_tokens"),
cost: sql<number>`SUM(${UsageTable.cost})`.as("cost"),
})
.from(UsageTable)
.where(
and(
eq(UsageTable.workspaceID, workspace.id),
sql`${UsageTable.timeCreated} >= DATE_SUB(NOW(), INTERVAL 28 DAY)`,
),
)
.groupBy(sql`DATE(${UsageTable.timeCreated})`)
.orderBy(sql`DATE(${UsageTable.timeCreated}) DESC`)
.then((rows) => {
const totalCost = rows.reduce((sum, r) => sum + Number(r.cost), 0)
const mapped = rows.map((row) => ({
...row,
cost: `$${(Number(row.cost) / 100000000).toFixed(2)}`,
}))
if (mapped.length > 0) {
mapped.push({
date: "TOTAL",
requests: null as any,
inputTokens: null as any,
outputTokens: null as any,
reasoningTokens: null as any,
cacheReadTokens: null as any,
cacheWrite5mTokens: null as any,
cacheWrite1hTokens: null as any,
cost: `$${(totalCost / 100000000).toFixed(2)}`,
})
}
return mapped
}),
)
/*
await printTable("Usage", (tx) =>
tx
@@ -292,22 +248,6 @@ async function printWorkspace(workspaceID: string) {
cost: `$${(row.cost / 100000000).toFixed(2)}`,
})),
),
)
await printTable("Disabled Models", (tx) =>
tx
.select({
model: ModelTable.model,
timeCreated: ModelTable.timeCreated,
})
.from(ModelTable)
.where(eq(ModelTable.workspaceID, workspace.id))
.orderBy(sql`${ModelTable.timeCreated} DESC`)
.then((rows) =>
rows.map((row) => ({
model: row.model,
timeCreated: formatDate(row.timeCreated),
})),
),
)
*/
}

View File

@@ -31,14 +31,12 @@ See `specs/effect-migration.md` for the compact pattern reference and examples.
- Use `Schema.Defect` instead of `unknown` for defect-like causes.
- In `Effect.gen` / `Effect.fn`, prefer `yield* new MyError(...)` over `yield* Effect.fail(new MyError(...))` for direct early-failure branches.
## Runtime vs InstanceState
## Runtime vs Instances
- Use `makeRuntime` (from `src/effect/run-service.ts`) for all services. It returns `{ runPromise, runFork, runCallback }` backed by a shared `memoMap` that deduplicates layers.
- Use `InstanceState` (from `src/effect/instance-state.ts`) for per-directory or per-project state that needs per-instance cleanup. It uses `ScopedCache` keyed by directory — each open project gets its own state, automatically cleaned up on disposal.
- If two open directories should not share one copy of the service, it needs `InstanceState`.
- Do the work directly in the `InstanceState.make` closure — `ScopedCache` handles run-once semantics. Don't add fibers, `ensure()` callbacks, or `started` flags on top.
- Use `Effect.addFinalizer` or `Effect.acquireRelease` inside the `InstanceState.make` closure for cleanup (subscriptions, process teardown, etc.).
- Use `Effect.forkScoped` inside the closure for background stream consumers — the fiber is interrupted when the instance is disposed.
- Use the shared runtime for process-wide services with one lifecycle for the whole app.
- Use `src/effect/instances.ts` for per-directory or per-project services that need `InstanceContext`, per-instance state, or per-instance cleanup.
- If two open directories should not share one copy of the service, it belongs in `Instances`.
- Instance-scoped services should read context from `InstanceContext`, not `Instance.*` globals.
## Preferred Effect services
@@ -53,7 +51,7 @@ See `specs/effect-migration.md` for the compact pattern reference and examples.
`Instance.bind(fn)` captures the current Instance AsyncLocalStorage context and restores it synchronously when called.
Use it for native addon callbacks (`@parcel/watcher`, `node-pty`, native `fs.watch`, etc.) that need to call `Bus.publish` or anything that reads `Instance.directory`.
Use it for native addon callbacks (`@parcel/watcher`, `node-pty`, native `fs.watch`, etc.) that need to call `Bus.publish`, `Instance.state()`, or anything that reads `Instance.directory`.
You do not need it for `setTimeout`, `Promise.then`, `EventEmitter.on`, or Effect fibers.

View File

@@ -121,7 +121,7 @@
"drizzle-orm": "catalog:",
"effect": "catalog:",
"fuzzysort": "3.1.0",
"gitlab-ai-provider": "5.3.3",
"gitlab-ai-provider": "5.3.2",
"glob": "13.0.5",
"google-auth-library": "10.5.0",
"gray-matter": "4.0.3",

View File

@@ -6,7 +6,7 @@ Practical reference for new and migrated Effect code in `packages/opencode`.
Use `InstanceState` (from `src/effect/instance-state.ts`) for services that need per-directory state, per-instance cleanup, or project-bound background work. InstanceState uses a `ScopedCache` keyed by directory, so each open project gets its own copy of the state that is automatically cleaned up on disposal.
Use `makeRuntime` (from `src/effect/run-service.ts`) to create a per-service `ManagedRuntime` that lazily initializes and shares layers via a global `memoMap`. Returns `{ runPromise, runFork, runCallback }`.
Use `makeRunPromise` (from `src/effect/run-service.ts`) to create a per-service `ManagedRuntime` that lazily initializes and shares layers via a global `memoMap`.
- Global services (no per-directory state): Account, Auth, Installation, Truncate
- Instance-scoped (per-directory state via InstanceState): File, FileTime, FileWatcher, Format, Permission, Question, Skill, Snapshot, Vcs, ProviderAuth
@@ -46,7 +46,7 @@ export namespace Foo {
export const defaultLayer = layer.pipe(Layer.provide(FooDep.layer))
// Per-service runtime (inside the namespace)
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
// Async facade functions
export async function get(id: FooID) {
@@ -79,24 +79,22 @@ See `Auth.ZodInfo` for the canonical example.
The `InstanceState.make` init callback receives a `Scope`, so you can use `Effect.acquireRelease`, `Effect.addFinalizer`, and `Effect.forkScoped` inside it. Resources acquired this way are automatically cleaned up when the instance is disposed or invalidated by `ScopedCache`. This makes it the right place for:
- **Subscriptions**: Yield `Bus.Service` at the layer level, then use `Stream` + `forkScoped` inside the init closure. The fiber is automatically interrupted when the instance scope closes:
- **Subscriptions**: Use `Effect.acquireRelease` to subscribe and auto-unsubscribe:
```ts
const bus = yield * Bus.Service
const cache =
yield *
InstanceState.make<State>(
Effect.fn("Foo.state")(function* (ctx) {
// ... load state ...
yield* bus.subscribeAll().pipe(
Stream.runForEach((event) =>
Effect.sync(() => {
yield* Effect.acquireRelease(
Effect.sync(() =>
Bus.subscribeAll((event) => {
/* handle */
}),
),
Effect.forkScoped,
(unsub) => Effect.sync(unsub),
)
return {
@@ -106,16 +104,6 @@ const cache =
)
```
- **Resource cleanup**: Use `Effect.acquireRelease` or `Effect.addFinalizer` for resources that need teardown (native watchers, process handles, etc.):
```ts
yield *
Effect.acquireRelease(
Effect.sync(() => nativeAddon.watch(dir)),
(watcher) => Effect.sync(() => watcher.close()),
)
```
- **Background fibers**: Use `Effect.forkScoped` — the fiber is interrupted on disposal.
- **Side effects at init**: Config notification, event wiring, etc. all belong in the init closure. Callers just do `InstanceState.get(cache)` to trigger everything, and `ScopedCache` deduplicates automatically.
@@ -177,7 +165,7 @@ Still open and likely worth migrating:
- [x] `ToolRegistry`
- [ ] `Pty`
- [x] `Worktree`
- [x] `Bus`
- [ ] `Bus`
- [x] `Command`
- [ ] `Config`
- [ ] `Session`
@@ -187,4 +175,4 @@ Still open and likely worth migrating:
- [ ] `Provider`
- [x] `Project`
- [ ] `LSP`
- [x] `MCP`
- [ ] `MCP`

View File

@@ -1,7 +1,7 @@
import { Clock, Duration, Effect, Layer, Option, Schema, SchemaGetter, ServiceMap } from "effect"
import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { withTransientReadRetry } from "@/util/effect-http-client"
import { AccountRepo, type AccountRow } from "./repo"
import {
@@ -379,7 +379,7 @@ export namespace Account {
export const defaultLayer = layer.pipe(Layer.provide(AccountRepo.layer), Layer.provide(FetchHttpClient.layer))
export const { runPromise } = makeRuntime(Service, defaultLayer)
export const runPromise = makeRunPromise(Service, defaultLayer)
export async function active(): Promise<Info | undefined> {
return Option.getOrUndefined(await runPromise((service) => service.active()))

View File

@@ -21,7 +21,7 @@ import { Plugin } from "@/plugin"
import { Skill } from "../skill"
import { Effect, ServiceMap, Layer } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
export namespace Agent {
export const Info = z
@@ -393,7 +393,7 @@ export namespace Agent {
export const defaultLayer = layer.pipe(Layer.provide(Auth.layer))
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export async function get(agent: string) {
return runPromise((svc) => svc.get(agent))

View File

@@ -1,6 +1,6 @@
import path from "path"
import { Effect, Layer, Record, Result, Schema, ServiceMap } from "effect"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { zod } from "@/util/effect-zod"
import { Global } from "../global"
import { Filesystem } from "../util/filesystem"
@@ -95,7 +95,7 @@ export namespace Auth {
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export async function get(providerID: string) {
return runPromise((service) => service.get(providerID))

View File

@@ -1,14 +1,12 @@
import z from "zod"
import { Effect, Exit, Layer, PubSub, Scope, ServiceMap, Stream } from "effect"
import { Log } from "../util/log"
import { Instance } from "../project/instance"
import { BusEvent } from "./bus-event"
import { GlobalBus } from "./global"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
export namespace Bus {
const log = Log.create({ service: "bus" })
type Subscription = (event: any) => void
export const InstanceDisposed = BusEvent.define(
"server.instance.disposed",
@@ -17,168 +15,91 @@ export namespace Bus {
}),
)
type Payload<D extends BusEvent.Definition = BusEvent.Definition> = {
type: D["type"]
properties: z.infer<D["properties"]>
}
const state = Instance.state(
() => {
const subscriptions = new Map<any, Subscription[]>()
type State = {
wildcard: PubSub.PubSub<Payload>
typed: Map<string, PubSub.PubSub<Payload>>
}
export interface Interface {
readonly publish: <D extends BusEvent.Definition>(
def: D,
properties: z.output<D["properties"]>,
) => Effect.Effect<void>
readonly subscribe: <D extends BusEvent.Definition>(def: D) => Stream.Stream<Payload<D>>
readonly subscribeAll: () => Stream.Stream<Payload>
readonly subscribeCallback: <D extends BusEvent.Definition>(
def: D,
callback: (event: Payload<D>) => unknown,
) => Effect.Effect<() => void>
readonly subscribeAllCallback: (callback: (event: any) => unknown) => Effect.Effect<() => void>
}
export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Bus") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const cache = yield* InstanceState.make<State>(
Effect.fn("Bus.state")(function* (ctx) {
const wildcard = yield* PubSub.unbounded<Payload>()
const typed = new Map<string, PubSub.PubSub<Payload>>()
yield* Effect.addFinalizer(() =>
Effect.gen(function* () {
// Publish InstanceDisposed before shutting down so subscribers see it
yield* PubSub.publish(wildcard, {
type: InstanceDisposed.type,
properties: { directory: ctx.directory },
})
yield* PubSub.shutdown(wildcard)
for (const ps of typed.values()) {
yield* PubSub.shutdown(ps)
}
}),
)
return { wildcard, typed }
}),
)
function getOrCreate<D extends BusEvent.Definition>(state: State, def: D) {
return Effect.gen(function* () {
let ps = state.typed.get(def.type)
if (!ps) {
ps = yield* PubSub.unbounded<Payload>()
state.typed.set(def.type, ps)
}
return ps as unknown as PubSub.PubSub<Payload<D>>
})
return {
subscriptions,
}
function publish<D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) {
return Effect.gen(function* () {
const state = yield* InstanceState.get(cache)
const payload: Payload = { type: def.type, properties }
log.info("publishing", { type: def.type })
const ps = state.typed.get(def.type)
if (ps) yield* PubSub.publish(ps, payload)
yield* PubSub.publish(state.wildcard, payload)
GlobalBus.emit("event", {
directory: Instance.directory,
payload,
})
})
},
async (entry) => {
const wildcard = entry.subscriptions.get("*")
if (!wildcard) return
const event = {
type: InstanceDisposed.type,
properties: {
directory: Instance.directory,
},
}
function subscribe<D extends BusEvent.Definition>(def: D): Stream.Stream<Payload<D>> {
log.info("subscribing", { type: def.type })
return Stream.unwrap(
Effect.gen(function* () {
const state = yield* InstanceState.get(cache)
const ps = yield* getOrCreate(state, def)
return Stream.fromPubSub(ps)
}),
).pipe(Stream.ensuring(Effect.sync(() => log.info("unsubscribing", { type: def.type }))))
for (const sub of [...wildcard]) {
sub(event)
}
function subscribeAll(): Stream.Stream<Payload> {
log.info("subscribing", { type: "*" })
return Stream.unwrap(
Effect.gen(function* () {
const state = yield* InstanceState.get(cache)
return Stream.fromPubSub(state.wildcard)
}),
).pipe(Stream.ensuring(Effect.sync(() => log.info("unsubscribing", { type: "*" }))))
}
function on<T>(pubsub: PubSub.PubSub<T>, type: string, callback: (event: T) => unknown) {
return Effect.gen(function* () {
log.info("subscribing", { type })
const scope = yield* Scope.make()
const subscription = yield* Scope.provide(scope)(PubSub.subscribe(pubsub))
yield* Scope.provide(scope)(
Stream.fromSubscription(subscription).pipe(
Stream.runForEach((msg) =>
Effect.tryPromise({
try: () => Promise.resolve().then(() => callback(msg)),
catch: (cause) => {
log.error("subscriber failed", { type, cause })
},
}).pipe(Effect.ignore),
),
Effect.forkScoped,
),
)
return () => {
log.info("unsubscribing", { type })
Effect.runFork(Scope.close(scope, Exit.void))
}
})
}
const subscribeCallback = Effect.fn("Bus.subscribeCallback")(function* <D extends BusEvent.Definition>(
def: D,
callback: (event: Payload<D>) => unknown,
) {
const state = yield* InstanceState.get(cache)
const ps = yield* getOrCreate(state, def)
return yield* on(ps, def.type, callback)
})
const subscribeAllCallback = Effect.fn("Bus.subscribeAllCallback")(function* (callback: (event: any) => unknown) {
const state = yield* InstanceState.get(cache)
return yield* on(state.wildcard, "*", callback)
})
return Service.of({ publish, subscribe, subscribeAll, subscribeCallback, subscribeAllCallback })
}),
},
)
const { runPromise, runSync } = makeRuntime(Service, layer)
// runSync is safe here because the subscribe chain (InstanceState.get, PubSub.subscribe,
// Scope.make, Effect.forkScoped) is entirely synchronous. If any step becomes async, this will throw.
export async function publish<D extends BusEvent.Definition>(def: D, properties: z.output<D["properties"]>) {
return runPromise((svc) => svc.publish(def, properties))
}
export function subscribe<D extends BusEvent.Definition>(
def: D,
callback: (event: { type: D["type"]; properties: z.infer<D["properties"]> }) => unknown,
export async function publish<Definition extends BusEvent.Definition>(
def: Definition,
properties: z.output<Definition["properties"]>,
) {
return runSync((svc) => svc.subscribeCallback(def, callback))
const payload = {
type: def.type,
properties,
}
log.info("publishing", {
type: def.type,
})
const pending = []
for (const key of [def.type, "*"]) {
const match = [...(state().subscriptions.get(key) ?? [])]
for (const sub of match) {
pending.push(sub(payload))
}
}
GlobalBus.emit("event", {
directory: Instance.directory,
payload,
})
return Promise.all(pending)
}
export function subscribeAll(callback: (event: any) => unknown) {
return runSync((svc) => svc.subscribeAllCallback(callback))
export function subscribe<Definition extends BusEvent.Definition>(
def: Definition,
callback: (event: { type: Definition["type"]; properties: z.infer<Definition["properties"]> }) => void,
) {
return raw(def.type, callback)
}
export function once<Definition extends BusEvent.Definition>(
def: Definition,
callback: (event: {
type: Definition["type"]
properties: z.infer<Definition["properties"]>
}) => "done" | undefined,
) {
const unsub = subscribe(def, (event) => {
if (callback(event)) unsub()
})
}
export function subscribeAll(callback: (event: any) => void) {
return raw("*", callback)
}
function raw(type: string, callback: (event: any) => void) {
log.info("subscribing", { type })
const subscriptions = state().subscriptions
let match = subscriptions.get(type) ?? []
match.push(callback)
subscriptions.set(type, match)
return () => {
log.info("unsubscribing", { type })
const match = subscriptions.get(type)
if (!match) return
const index = match.indexOf(callback)
if (index === -1) return
match.splice(index, 1)
}
}
}

View File

@@ -186,7 +186,7 @@ export function tui(input: {
targetFps: 60,
gatherStats: false,
exitOnCtrlC: false,
useKittyKeyboard: { events: process.platform === "win32" },
useKittyKeyboard: {},
autoFocus: false,
openConsoleOnError: false,
consoleOptions: {

View File

@@ -18,7 +18,7 @@ import { usePromptStash } from "./stash"
import { DialogStash } from "../dialog-stash"
import { type AutocompleteRef, Autocomplete } from "./autocomplete"
import { useCommandDialog } from "../dialog-command"
import { useKeyboard, useRenderer } from "@opentui/solid"
import { useRenderer } from "@opentui/solid"
import { Editor } from "@tui/util/editor"
import { useExit } from "../../context/exit"
import { Clipboard } from "../../util/clipboard"
@@ -356,20 +356,6 @@ export function Prompt(props: PromptProps) {
]
})
// Windows Terminal 1.25+ handles Ctrl+V on keydown when kitty events are
// enabled, but still reports the kitty key-release event. Probe on release.
if (process.platform === "win32") {
useKeyboard(
(evt) => {
if (!input.focused) return
if (evt.name === "v" && evt.ctrl && evt.eventType === "release") {
command.trigger("prompt.paste")
}
},
{ release: true },
)
}
const ref: PromptRef = {
get focused() {
return input.focused
@@ -864,9 +850,10 @@ export function Prompt(props: PromptProps) {
e.preventDefault()
return
}
// Check clipboard for images before terminal-handled paste runs.
// This helps terminals that forward Ctrl+V to the app; Windows
// Terminal 1.25+ usually handles Ctrl+V before this path.
// Handle clipboard paste (Ctrl+V) - check for images first on Windows
// This is needed because Windows terminal doesn't properly send image data
// through bracketed paste, so we need to intercept the keypress and
// directly read from clipboard before the terminal handles it
if (keybind.match("input_paste", e)) {
const content = await Clipboard.read()
if (content?.mime.startsWith("image/")) {
@@ -949,9 +936,6 @@ export function Prompt(props: PromptProps) {
// Replace CRLF first, then any remaining CR
const normalizedText = decodePasteBytes(event.bytes).replace(/\r\n/g, "\n").replace(/\r/g, "\n")
const pastedContent = normalizedText.trim()
// Windows Terminal <1.25 can surface image-only clipboard as an
// empty bracketed paste. Windows Terminal 1.25+ does not.
if (!pastedContent) {
command.trigger("prompt.paste")
return

View File

@@ -28,14 +28,6 @@ export namespace Clipboard {
mime: string
}
// Checks clipboard for images first, then falls back to text.
//
// On Windows prompt/ can call this from multiple paste signals because
// terminals surface image paste differently:
// 1. A forwarded Ctrl+V keypress
// 2. An empty bracketed-paste hint for image-only clipboard in Windows
// Terminal <1.25
// 3. A kitty Ctrl+V key-release fallback for Windows Terminal 1.25+
export async function read(): Promise<Content | undefined> {
const os = platform()
@@ -66,8 +58,6 @@ export namespace Clipboard {
}
}
// Windows/WSL: probe clipboard for images via PowerShell.
// Bracketed paste can't carry image data so we read it directly.
if (os === "win32" || release().includes("WSL")) {
const script =
"Add-Type -AssemblyName System.Windows.Forms; $img = [System.Windows.Forms.Clipboard]::GetImage(); if ($img) { $ms = New-Object System.IO.MemoryStream; $img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png); [System.Convert]::ToBase64String($ms.ToArray()) }"

View File

@@ -1,6 +1,6 @@
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { SessionID, MessageID } from "@/session/schema"
import { Effect, Layer, ServiceMap } from "effect"
import z from "zod"
@@ -173,7 +173,7 @@ export namespace Command {
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export async function get(name: string) {
return runPromise((svc) => svc.get(name))

View File

@@ -0,0 +1,14 @@
import { ServiceMap } from "effect"
import type { Project } from "@/project/project"
export declare namespace InstanceContext {
export interface Shape {
readonly directory: string
readonly worktree: string
readonly project: Project.Info
}
}
export class InstanceContext extends ServiceMap.Service<InstanceContext, InstanceContext.Shape>()(
"opencode/InstanceContext",
) {}

View File

@@ -3,15 +3,11 @@ import * as ServiceMap from "effect/ServiceMap"
export const memoMap = Layer.makeMemoMapUnsafe()
export function makeRuntime<I, S, E>(service: ServiceMap.Service<I, S>, layer: Layer.Layer<I, E>) {
export function makeRunPromise<I, S, E>(service: ServiceMap.Service<I, S>, layer: Layer.Layer<I, E>) {
let rt: ManagedRuntime.ManagedRuntime<I, E> | undefined
const getRuntime = () => (rt ??= ManagedRuntime.make(layer, { memoMap }))
return {
runSync: <A, Err>(fn: (svc: S) => Effect.Effect<A, Err, I>) => getRuntime().runSync(service.use(fn)),
runPromise: <A, Err>(fn: (svc: S) => Effect.Effect<A, Err, I>, options?: Effect.RunOptions) =>
getRuntime().runPromise(service.use(fn), options),
runFork: <A, Err>(fn: (svc: S) => Effect.Effect<A, Err, I>) => getRuntime().runFork(service.use(fn)),
runCallback: <A, Err>(fn: (svc: S) => Effect.Effect<A, Err, I>) => getRuntime().runCallback(service.use(fn)),
return <A, Err>(fn: (svc: S) => Effect.Effect<A, Err, I>, options?: Effect.RunOptions) => {
rt ??= ManagedRuntime.make(layer, { memoMap })
return rt.runPromise(service.use(fn), options)
}
}

View File

@@ -1,8 +1,8 @@
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { Git } from "@/git"
import { Effect, Layer, ServiceMap } from "effect"
import { Effect, Fiber, Layer, Scope, ServiceMap } from "effect"
import { formatPatch, structuredPatch } from "diff"
import fs from "fs"
import fuzzysort from "fuzzysort"
@@ -323,6 +323,7 @@ export namespace File {
interface State {
cache: Entry
fiber: Fiber.Fiber<void> | undefined
}
export interface Interface {
@@ -347,6 +348,7 @@ export namespace File {
Effect.fn("File.state")(() =>
Effect.succeed({
cache: { files: [], dirs: [] } as Entry,
fiber: undefined as Fiber.Fiber<void> | undefined,
}),
),
)
@@ -404,11 +406,21 @@ export namespace File {
s.cache = next
})
let cachedScan = yield* Effect.cached(scan().pipe(Effect.catchCause(() => Effect.void)))
const scope = yield* Scope.Scope
const ensure = Effect.fn("File.ensure")(function* () {
yield* cachedScan
cachedScan = yield* Effect.cached(scan().pipe(Effect.catchCause(() => Effect.void)))
const s = yield* InstanceState.get(state)
if (!s.fiber)
s.fiber = yield* scan().pipe(
Effect.catchCause(() => Effect.void),
Effect.ensuring(
Effect.sync(() => {
s.fiber = undefined
}),
),
Effect.forkIn(scope),
)
yield* Fiber.join(s.fiber)
})
const init = Effect.fn("File.init")(function* () {
@@ -676,7 +688,7 @@ export namespace File {
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export function init() {
return runPromise((svc) => svc.init())

View File

@@ -1,6 +1,6 @@
import { DateTime, Effect, Layer, Semaphore, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { Flag } from "@/flag/flag"
import type { SessionID } from "@/session/schema"
import { Filesystem } from "../util/filesystem"
@@ -108,7 +108,7 @@ export namespace FileTime {
}),
).pipe(Layer.orDie)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export function read(sessionID: SessionID, file: string) {
return runPromise((s) => s.read(sessionID, file))

View File

@@ -8,7 +8,7 @@ import z from "zod"
import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { Flag } from "@/flag/flag"
import { Git } from "@/git"
import { Instance } from "@/project/instance"
@@ -159,7 +159,7 @@ export namespace FileWatcher {
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export function init() {
return runPromise((svc) => svc.init())

View File

@@ -1,10 +1,12 @@
import { Effect, Layer, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import path from "path"
import { mergeDeep } from "remeda"
import z from "zod"
import { Bus } from "../bus"
import { Config } from "../config/config"
import { File } from "../file"
import { Instance } from "../project/instance"
import { Process } from "../util/process"
import { Log } from "../util/log"
@@ -27,7 +29,6 @@ export namespace Format {
export interface Interface {
readonly init: () => Effect.Effect<void>
readonly status: () => Effect.Effect<Status[]>
readonly file: (filepath: string) => Effect.Effect<void>
}
export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Format") {}
@@ -96,46 +97,53 @@ export namespace Format {
return checks.filter((x) => x.enabled).map((x) => x.item)
}
async function formatFile(filepath: string) {
log.info("formatting", { file: filepath })
const ext = path.extname(filepath)
for (const item of await getFormatter(ext)) {
log.info("running", { command: item.command })
try {
const proc = Process.spawn(
item.command.map((x) => x.replace("$FILE", filepath)),
{
cwd: Instance.directory,
env: { ...process.env, ...item.environment },
stdout: "ignore",
stderr: "ignore",
},
)
const exit = await proc.exited
if (exit !== 0) {
log.error("failed", {
command: item.command,
...item.environment,
})
}
} catch (error) {
log.error("failed to format file", {
error,
command: item.command,
...item.environment,
file: filepath,
})
}
}
}
yield* Effect.acquireRelease(
Effect.sync(() =>
Bus.subscribe(
File.Event.Edited,
Instance.bind(async (payload) => {
const file = payload.properties.file
log.info("formatting", { file })
const ext = path.extname(file)
for (const item of await getFormatter(ext)) {
log.info("running", { command: item.command })
try {
const proc = Process.spawn(
item.command.map((x) => x.replace("$FILE", file)),
{
cwd: Instance.directory,
env: { ...process.env, ...item.environment },
stdout: "ignore",
stderr: "ignore",
},
)
const exit = await proc.exited
if (exit !== 0) {
log.error("failed", {
command: item.command,
...item.environment,
})
}
} catch (error) {
log.error("failed to format file", {
error,
command: item.command,
...item.environment,
file,
})
}
}
}),
),
),
(unsubscribe) => Effect.sync(unsubscribe),
)
log.info("init")
return {
formatters,
isEnabled,
formatFile,
}
}),
)
@@ -158,16 +166,11 @@ export namespace Format {
return result
})
const file = Effect.fn("Format.file")(function* (filepath: string) {
const { formatFile } = yield* InstanceState.get(state)
yield* Effect.promise(() => formatFile(filepath))
})
return Service.of({ init, status, file })
return Service.of({ init, status })
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export async function init() {
return runPromise((s) => s.init())
@@ -176,8 +179,4 @@ export namespace Format {
export async function status() {
return runPromise((s) => s.status())
}
export async function file(filepath: string) {
return runPromise((s) => s.file(filepath))
}
}

View File

@@ -2,7 +2,7 @@ import { NodeFileSystem, NodePath } from "@effect/platform-node"
import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
import { Effect, Layer, ServiceMap, Stream } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
export namespace Git {
const cfg = [
@@ -264,7 +264,7 @@ export namespace Git {
Layer.provide(NodePath.layer),
)
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export function run(args: string[], opts: Options) {
return runPromise((git) => git.run(args, opts))

View File

@@ -2,7 +2,7 @@ import { NodeFileSystem, NodePath } from "@effect/platform-node"
import { Effect, Layer, Schema, ServiceMap, Stream } from "effect"
import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { withTransientReadRetry } from "@/util/effect-http-client"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import path from "path"
@@ -346,7 +346,7 @@ export namespace Installation {
Layer.provide(NodePath.layer),
)
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export async function info(): Promise<Info> {
return runPromise((svc) => svc.info())

View File

@@ -1,9 +1,7 @@
import path from "path"
import z from "zod"
import { Global } from "../global"
import { Effect, Layer, ServiceMap } from "effect"
import { AppFileSystem } from "@/filesystem"
import { makeRuntime } from "@/effect/run-service"
import { Filesystem } from "../util/filesystem"
export namespace McpAuth {
export const Tokens = z.object({
@@ -27,155 +25,106 @@ export namespace McpAuth {
clientInfo: ClientInfo.optional(),
codeVerifier: z.string().optional(),
oauthState: z.string().optional(),
serverUrl: z.string().optional(),
serverUrl: z.string().optional(), // Track the URL these credentials are for
})
export type Entry = z.infer<typeof Entry>
const filepath = path.join(Global.Path.data, "mcp-auth.json")
export interface Interface {
readonly all: () => Effect.Effect<Record<string, Entry>>
readonly get: (mcpName: string) => Effect.Effect<Entry | undefined>
readonly getForUrl: (mcpName: string, serverUrl: string) => Effect.Effect<Entry | undefined>
readonly set: (mcpName: string, entry: Entry, serverUrl?: string) => Effect.Effect<void>
readonly remove: (mcpName: string) => Effect.Effect<void>
readonly updateTokens: (mcpName: string, tokens: Tokens, serverUrl?: string) => Effect.Effect<void>
readonly updateClientInfo: (mcpName: string, clientInfo: ClientInfo, serverUrl?: string) => Effect.Effect<void>
readonly updateCodeVerifier: (mcpName: string, codeVerifier: string) => Effect.Effect<void>
readonly clearCodeVerifier: (mcpName: string) => Effect.Effect<void>
readonly updateOAuthState: (mcpName: string, oauthState: string) => Effect.Effect<void>
readonly getOAuthState: (mcpName: string) => Effect.Effect<string | undefined>
readonly clearOAuthState: (mcpName: string) => Effect.Effect<void>
readonly isTokenExpired: (mcpName: string) => Effect.Effect<boolean | null>
export async function get(mcpName: string): Promise<Entry | undefined> {
const data = await all()
return data[mcpName]
}
export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/McpAuth") {}
/**
* Get auth entry and validate it's for the correct URL.
* Returns undefined if URL has changed (credentials are invalid).
*/
export async function getForUrl(mcpName: string, serverUrl: string): Promise<Entry | undefined> {
const entry = await get(mcpName)
if (!entry) return undefined
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
// If no serverUrl is stored, this is from an old version - consider it invalid
if (!entry.serverUrl) return undefined
const all = Effect.fn("McpAuth.all")(function* () {
return yield* fs.readJson(filepath).pipe(
Effect.map((data) => data as Record<string, Entry>),
Effect.catch(() => Effect.succeed({} as Record<string, Entry>)),
)
})
// If URL has changed, credentials are invalid
if (entry.serverUrl !== serverUrl) return undefined
const get = Effect.fn("McpAuth.get")(function* (mcpName: string) {
const data = yield* all()
return data[mcpName]
})
return entry
}
const getForUrl = Effect.fn("McpAuth.getForUrl")(function* (mcpName: string, serverUrl: string) {
const entry = yield* get(mcpName)
if (!entry) return undefined
if (!entry.serverUrl) return undefined
if (entry.serverUrl !== serverUrl) return undefined
return entry
})
export async function all(): Promise<Record<string, Entry>> {
return Filesystem.readJson<Record<string, Entry>>(filepath).catch(() => ({}))
}
const set = Effect.fn("McpAuth.set")(function* (mcpName: string, entry: Entry, serverUrl?: string) {
const data = yield* all()
if (serverUrl) entry.serverUrl = serverUrl
yield* fs.writeJson(filepath, { ...data, [mcpName]: entry }, 0o600).pipe(Effect.orDie)
})
export async function set(mcpName: string, entry: Entry, serverUrl?: string): Promise<void> {
const data = await all()
// Always update serverUrl if provided
if (serverUrl) {
entry.serverUrl = serverUrl
}
await Filesystem.writeJson(filepath, { ...data, [mcpName]: entry }, 0o600)
}
const remove = Effect.fn("McpAuth.remove")(function* (mcpName: string) {
const data = yield* all()
delete data[mcpName]
yield* fs.writeJson(filepath, data, 0o600).pipe(Effect.orDie)
})
export async function remove(mcpName: string): Promise<void> {
const data = await all()
delete data[mcpName]
await Filesystem.writeJson(filepath, data, 0o600)
}
const updateField = <K extends keyof Entry>(field: K, spanName: string) =>
Effect.fn(`McpAuth.${spanName}`)(function* (mcpName: string, value: NonNullable<Entry[K]>, serverUrl?: string) {
const entry = (yield* get(mcpName)) ?? {}
entry[field] = value
yield* set(mcpName, entry, serverUrl)
})
export async function updateTokens(mcpName: string, tokens: Tokens, serverUrl?: string): Promise<void> {
const entry = (await get(mcpName)) ?? {}
entry.tokens = tokens
await set(mcpName, entry, serverUrl)
}
const clearField = <K extends keyof Entry>(field: K, spanName: string) =>
Effect.fn(`McpAuth.${spanName}`)(function* (mcpName: string) {
const entry = yield* get(mcpName)
if (entry) {
delete entry[field]
yield* set(mcpName, entry)
}
})
export async function updateClientInfo(mcpName: string, clientInfo: ClientInfo, serverUrl?: string): Promise<void> {
const entry = (await get(mcpName)) ?? {}
entry.clientInfo = clientInfo
await set(mcpName, entry, serverUrl)
}
const updateTokens = updateField("tokens", "updateTokens")
const updateClientInfo = updateField("clientInfo", "updateClientInfo")
const updateCodeVerifier = updateField("codeVerifier", "updateCodeVerifier")
const updateOAuthState = updateField("oauthState", "updateOAuthState")
const clearCodeVerifier = clearField("codeVerifier", "clearCodeVerifier")
const clearOAuthState = clearField("oauthState", "clearOAuthState")
export async function updateCodeVerifier(mcpName: string, codeVerifier: string): Promise<void> {
const entry = (await get(mcpName)) ?? {}
entry.codeVerifier = codeVerifier
await set(mcpName, entry)
}
const getOAuthState = Effect.fn("McpAuth.getOAuthState")(function* (mcpName: string) {
const entry = yield* get(mcpName)
return entry?.oauthState
})
export async function clearCodeVerifier(mcpName: string): Promise<void> {
const entry = await get(mcpName)
if (entry) {
delete entry.codeVerifier
await set(mcpName, entry)
}
}
const isTokenExpired = Effect.fn("McpAuth.isTokenExpired")(function* (mcpName: string) {
const entry = yield* get(mcpName)
if (!entry?.tokens) return null
if (!entry.tokens.expiresAt) return false
return entry.tokens.expiresAt < Date.now() / 1000
})
export async function updateOAuthState(mcpName: string, oauthState: string): Promise<void> {
const entry = (await get(mcpName)) ?? {}
entry.oauthState = oauthState
await set(mcpName, entry)
}
return Service.of({
all,
get,
getForUrl,
set,
remove,
updateTokens,
updateClientInfo,
updateCodeVerifier,
clearCodeVerifier,
updateOAuthState,
getOAuthState,
clearOAuthState,
isTokenExpired,
})
}),
)
export async function getOAuthState(mcpName: string): Promise<string | undefined> {
const entry = await get(mcpName)
return entry?.oauthState
}
const defaultLayer = layer.pipe(Layer.provide(AppFileSystem.defaultLayer))
export async function clearOAuthState(mcpName: string): Promise<void> {
const entry = await get(mcpName)
if (entry) {
delete entry.oauthState
await set(mcpName, entry)
}
}
const { runPromise } = makeRuntime(Service, defaultLayer)
// Async facades for backward compat (used by McpOAuthProvider, CLI)
export const get = async (mcpName: string) => runPromise((svc) => svc.get(mcpName))
export const getForUrl = async (mcpName: string, serverUrl: string) =>
runPromise((svc) => svc.getForUrl(mcpName, serverUrl))
export const all = async () => runPromise((svc) => svc.all())
export const set = async (mcpName: string, entry: Entry, serverUrl?: string) =>
runPromise((svc) => svc.set(mcpName, entry, serverUrl))
export const remove = async (mcpName: string) => runPromise((svc) => svc.remove(mcpName))
export const updateTokens = async (mcpName: string, tokens: Tokens, serverUrl?: string) =>
runPromise((svc) => svc.updateTokens(mcpName, tokens, serverUrl))
export const updateClientInfo = async (mcpName: string, clientInfo: ClientInfo, serverUrl?: string) =>
runPromise((svc) => svc.updateClientInfo(mcpName, clientInfo, serverUrl))
export const updateCodeVerifier = async (mcpName: string, codeVerifier: string) =>
runPromise((svc) => svc.updateCodeVerifier(mcpName, codeVerifier))
export const clearCodeVerifier = async (mcpName: string) => runPromise((svc) => svc.clearCodeVerifier(mcpName))
export const updateOAuthState = async (mcpName: string, oauthState: string) =>
runPromise((svc) => svc.updateOAuthState(mcpName, oauthState))
export const getOAuthState = async (mcpName: string) => runPromise((svc) => svc.getOAuthState(mcpName))
export const clearOAuthState = async (mcpName: string) => runPromise((svc) => svc.clearOAuthState(mcpName))
export const isTokenExpired = async (mcpName: string) => runPromise((svc) => svc.isTokenExpired(mcpName))
/**
* Check if stored tokens are expired.
* Returns null if no tokens exist, false if no expiry or not expired, true if expired.
*/
export async function isTokenExpired(mcpName: string): Promise<boolean | null> {
const entry = await get(mcpName)
if (!entry?.tokens) return null
if (!entry.tokens.expiresAt) return false
return entry.tokens.expiresAt < Date.now() / 1000
}
}

File diff suppressed because it is too large Load Diff

View File

@@ -54,9 +54,6 @@ interface PendingAuth {
export namespace McpOAuthCallback {
let server: ReturnType<typeof Bun.serve> | undefined
const pendingAuths = new Map<string, PendingAuth>()
// Reverse index: mcpName → oauthState, so cancelPending(mcpName) can
// find the right entry in pendingAuths (which is keyed by oauthState).
const mcpNameToState = new Map<string, string>()
const CALLBACK_TIMEOUT_MS = 5 * 60 * 1000 // 5 minutes
@@ -101,12 +98,6 @@ export namespace McpOAuthCallback {
const pending = pendingAuths.get(state)!
clearTimeout(pending.timeout)
pendingAuths.delete(state)
for (const [name, s] of mcpNameToState) {
if (s === state) {
mcpNameToState.delete(name)
break
}
}
pending.reject(new Error(errorMsg))
}
return new Response(HTML_ERROR(errorMsg), {
@@ -135,13 +126,6 @@ export namespace McpOAuthCallback {
clearTimeout(pending.timeout)
pendingAuths.delete(state)
// Clean up reverse index
for (const [name, s] of mcpNameToState) {
if (s === state) {
mcpNameToState.delete(name)
break
}
}
pending.resolve(code)
return new Response(HTML_SUCCESS, {
@@ -153,13 +137,11 @@ export namespace McpOAuthCallback {
log.info("oauth callback server started", { port: OAUTH_CALLBACK_PORT })
}
export function waitForCallback(oauthState: string, mcpName?: string): Promise<string> {
if (mcpName) mcpNameToState.set(mcpName, oauthState)
export function waitForCallback(oauthState: string): Promise<string> {
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
if (pendingAuths.has(oauthState)) {
pendingAuths.delete(oauthState)
if (mcpName) mcpNameToState.delete(mcpName)
reject(new Error("OAuth callback timeout - authorization took too long"))
}
}, CALLBACK_TIMEOUT_MS)
@@ -169,14 +151,10 @@ export namespace McpOAuthCallback {
}
export function cancelPending(mcpName: string): void {
// Look up the oauthState for this mcpName via the reverse index
const oauthState = mcpNameToState.get(mcpName)
const key = oauthState ?? mcpName
const pending = pendingAuths.get(key)
const pending = pendingAuths.get(mcpName)
if (pending) {
clearTimeout(pending.timeout)
pendingAuths.delete(key)
mcpNameToState.delete(mcpName)
pendingAuths.delete(mcpName)
pending.reject(new Error("Authorization cancelled"))
}
}
@@ -206,7 +184,6 @@ export namespace McpOAuthCallback {
pending.reject(new Error("OAuth callback server stopped"))
}
pendingAuths.clear()
mcpNameToState.clear()
}
export function isRunning(): boolean {

View File

@@ -2,7 +2,7 @@ import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { Config } from "@/config/config"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { ProjectID } from "@/project/schema"
import { Instance } from "@/project/instance"
import { MessageID, SessionID } from "@/session/schema"
@@ -306,7 +306,7 @@ export namespace Permission {
return result
}
export const { runPromise } = makeRuntime(Service, layer)
export const runPromise = makeRunPromise(Service, layer)
export async function ask(input: z.infer<typeof AskInput>) {
return runPromise((s) => s.ask(input))

View File

@@ -11,9 +11,9 @@ import { NamedError } from "@opencode-ai/util/error"
import { CopilotAuthPlugin } from "./copilot"
import { gitlabAuthPlugin as GitlabAuthPlugin } from "opencode-gitlab-auth"
import { PoeAuthPlugin } from "opencode-poe-auth"
import { Effect, Layer, ServiceMap, Stream } from "effect"
import { Effect, Layer, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
export namespace Plugin {
const log = Log.create({ service: "plugin" })
@@ -52,8 +52,6 @@ export namespace Plugin {
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const cache = yield* InstanceState.make<State>(
Effect.fn("Plugin.state")(function* (ctx) {
const hooks: Hooks[] = []
@@ -148,16 +146,16 @@ export namespace Plugin {
}
})
// Subscribe to bus events, fiber interrupted when scope closes
yield* bus.subscribeAll().pipe(
Stream.runForEach((input) =>
Effect.sync(() => {
// Subscribe to bus events, clean up when scope is closed
yield* Effect.acquireRelease(
Effect.sync(() =>
Bus.subscribeAll(async (input) => {
for (const hook of hooks) {
hook["event"]?.({ event: input as any })
hook["event"]?.({ event: input })
}
}),
),
Effect.forkScoped,
(unsub) => Effect.sync(unsub),
)
return { hooks }
@@ -194,8 +192,7 @@ export namespace Plugin {
}),
)
const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, layer)
export async function trigger<
Name extends TriggerName,

View File

@@ -11,7 +11,7 @@ import { ProjectID } from "./schema"
import { Effect, Layer, Path, Scope, ServiceMap, Stream } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { NodeFileSystem, NodePath } from "@effect/platform-node"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { AppFileSystem } from "@/filesystem"
import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
@@ -462,7 +462,7 @@ export namespace Project {
Layer.provide(NodeFileSystem.layer),
Layer.provide(NodePath.layer),
)
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
// ---------------------------------------------------------------------------
// Promise-based API (delegates to Effect service via runPromise)

View File

@@ -1,9 +1,9 @@
import { Effect, Layer, ServiceMap, Stream } from "effect"
import { Effect, Layer, ServiceMap } from "effect"
import path from "path"
import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { AppFileSystem } from "@/filesystem"
import { FileWatcher } from "@/file/watcher"
import { Git } from "@/git"
@@ -139,12 +139,11 @@ export namespace Vcs {
export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Vcs") {}
export const layer: Layer.Layer<Service, never, AppFileSystem.Service | Git.Service | Bus.Service> = Layer.effect(
export const layer: Layer.Layer<Service, never, AppFileSystem.Service | Git.Service> = Layer.effect(
Service,
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const git = yield* Git.Service
const bus = yield* Bus.Service
const state = yield* InstanceState.make<State>(
Effect.fn("Vcs.state")((ctx) =>
Effect.gen(function* () {
@@ -159,19 +158,21 @@ export namespace Vcs {
const value = { current, root }
log.info("initialized", { branch: value.current, default_branch: value.root?.name })
yield* bus.subscribe(FileWatcher.Event.Updated).pipe(
Stream.filter((evt) => evt.properties.file.endsWith("HEAD")),
Stream.runForEach((_evt) =>
Effect.gen(function* () {
const next = yield* Effect.promise(() => get())
if (next !== value.current) {
yield* Effect.acquireRelease(
Effect.sync(() =>
Bus.subscribe(
FileWatcher.Event.Updated,
Instance.bind(async (evt) => {
if (!evt.properties.file.endsWith("HEAD")) return
const next = await get()
if (next === value.current) return
log.info("branch changed", { from: value.current, to: next })
value.current = next
yield* bus.publish(Event.BranchUpdated, { branch: next })
}
}),
Bus.publish(Event.BranchUpdated, { branch: next })
}),
),
),
Effect.forkScoped,
(unsubscribe) => Effect.sync(unsubscribe),
)
return value
@@ -211,13 +212,9 @@ export namespace Vcs {
}),
)
export const defaultLayer = layer.pipe(
Layer.provide(Git.defaultLayer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(Bus.layer),
)
export const defaultLayer = layer.pipe(Layer.provide(Git.defaultLayer), Layer.provide(AppFileSystem.defaultLayer))
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export function init() {
return runPromise((svc) => svc.init())

View File

@@ -2,7 +2,7 @@ import type { AuthOuathResult, Hooks } from "@opencode-ai/plugin"
import { NamedError } from "@opencode-ai/util/error"
import { Auth } from "@/auth"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { Plugin } from "../plugin"
import { ProviderID } from "./schema"
import { Array as Arr, Effect, Layer, Record, Result, ServiceMap } from "effect"
@@ -215,13 +215,12 @@ export namespace ProviderAuth {
}
if ("refresh" in result) {
const { type: _, provider: __, refresh, access, expires, ...extra } = result
yield* auth.set(input.providerID, {
type: "oauth",
access,
refresh,
expires,
...extra,
access: result.access,
refresh: result.refresh,
expires: result.expires,
...(result.accountId ? { accountId: result.accountId } : {}),
})
}
})
@@ -232,7 +231,7 @@ export namespace ProviderAuth {
export const defaultLayer = layer.pipe(Layer.provide(Auth.layer))
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export async function methods() {
return runPromise((svc) => svc.methods())

View File

@@ -1,7 +1,7 @@
import { BusEvent } from "@/bus/bus-event"
import { Bus } from "@/bus"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { Instance } from "@/project/instance"
import { type IPty } from "bun-pty"
import z from "zod"
@@ -361,7 +361,7 @@ export namespace Pty {
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export async function list() {
return runPromise((svc) => svc.list())

View File

@@ -2,7 +2,7 @@ import { Deferred, Effect, Layer, Schema, ServiceMap } from "effect"
import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { SessionID, MessageID } from "@/session/schema"
import { Log } from "@/util/log"
import z from "zod"
@@ -197,7 +197,7 @@ export namespace Question {
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export async function ask(input: {
sessionID: SessionID

View File

@@ -1,7 +1,7 @@
import { BusEvent } from "@/bus/bus-event"
import { Bus } from "@/bus"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { SessionID } from "./schema"
import { Effect, Layer, ServiceMap } from "effect"
import z from "zod"
@@ -55,8 +55,6 @@ export namespace SessionStatus {
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const state = yield* InstanceState.make(
Effect.fn("SessionStatus.state")(() => Effect.succeed(new Map<SessionID, Info>())),
)
@@ -72,9 +70,9 @@ export namespace SessionStatus {
const set = Effect.fn("SessionStatus.set")(function* (sessionID: SessionID, status: Info) {
const data = yield* InstanceState.get(state)
yield* bus.publish(Event.Status, { sessionID, status })
yield* Effect.promise(() => Bus.publish(Event.Status, { sessionID, status }))
if (status.type === "idle") {
yield* bus.publish(Event.Idle, { sessionID })
yield* Effect.promise(() => Bus.publish(Event.Idle, { sessionID }))
data.delete(sessionID)
return
}
@@ -85,8 +83,7 @@ export namespace SessionStatus {
}),
)
const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, layer)
export async function get(sessionID: SessionID) {
return runPromise((svc) => svc.get(sessionID))

View File

@@ -7,7 +7,7 @@ import { NamedError } from "@opencode-ai/util/error"
import type { Agent } from "@/agent/agent"
import { Bus } from "@/bus"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { Flag } from "@/flag/flag"
import { Global } from "@/global"
import { Permission } from "@/permission"
@@ -54,6 +54,11 @@ export namespace Skill {
type State = {
skills: Record<string, Info>
dirs: Set<string>
task?: Promise<void>
}
type Cache = State & {
ensure: () => Promise<void>
}
export interface Interface {
@@ -111,47 +116,66 @@ export namespace Skill {
})
}
async function loadSkills(state: State, discovery: Discovery.Interface, directory: string, worktree: string) {
if (!Flag.OPENCODE_DISABLE_EXTERNAL_SKILLS) {
for (const dir of EXTERNAL_DIRS) {
const root = path.join(Global.Path.home, dir)
if (!(await Filesystem.isDir(root))) continue
await scan(state, root, EXTERNAL_SKILL_PATTERN, { dot: true, scope: "global" })
}
for await (const root of Filesystem.up({
targets: EXTERNAL_DIRS,
start: directory,
stop: worktree,
})) {
await scan(state, root, EXTERNAL_SKILL_PATTERN, { dot: true, scope: "project" })
}
// TODO: Migrate to Effect
const create = (discovery: Discovery.Interface, directory: string, worktree: string): Cache => {
const state: State = {
skills: {},
dirs: new Set<string>(),
}
for (const dir of await Config.directories()) {
await scan(state, dir, OPENCODE_SKILL_PATTERN)
}
const load = async () => {
if (!Flag.OPENCODE_DISABLE_EXTERNAL_SKILLS) {
for (const dir of EXTERNAL_DIRS) {
const root = path.join(Global.Path.home, dir)
if (!(await Filesystem.isDir(root))) continue
await scan(state, root, EXTERNAL_SKILL_PATTERN, { dot: true, scope: "global" })
}
const cfg = await Config.get()
for (const item of cfg.skills?.paths ?? []) {
const expanded = item.startsWith("~/") ? path.join(os.homedir(), item.slice(2)) : item
const dir = path.isAbsolute(expanded) ? expanded : path.join(directory, expanded)
if (!(await Filesystem.isDir(dir))) {
log.warn("skill path not found", { path: dir })
continue
for await (const root of Filesystem.up({
targets: EXTERNAL_DIRS,
start: directory,
stop: worktree,
})) {
await scan(state, root, EXTERNAL_SKILL_PATTERN, { dot: true, scope: "project" })
}
}
await scan(state, dir, SKILL_PATTERN)
}
for (const dir of await Config.directories()) {
await scan(state, dir, OPENCODE_SKILL_PATTERN)
}
const cfg = await Config.get()
for (const item of cfg.skills?.paths ?? []) {
const expanded = item.startsWith("~/") ? path.join(os.homedir(), item.slice(2)) : item
const dir = path.isAbsolute(expanded) ? expanded : path.join(directory, expanded)
if (!(await Filesystem.isDir(dir))) {
log.warn("skill path not found", { path: dir })
continue
}
for (const url of cfg.skills?.urls ?? []) {
for (const dir of await Effect.runPromise(discovery.pull(url))) {
state.dirs.add(dir)
await scan(state, dir, SKILL_PATTERN)
}
for (const url of cfg.skills?.urls ?? []) {
for (const dir of await Effect.runPromise(discovery.pull(url))) {
state.dirs.add(dir)
await scan(state, dir, SKILL_PATTERN)
}
}
log.info("init", { count: Object.keys(state.skills).length })
}
log.info("init", { count: Object.keys(state.skills).length })
const ensure = () => {
if (state.task) return state.task
state.task = load().catch((err) => {
state.task = undefined
throw err
})
return state.task
}
return { ...state, ensure }
}
export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Skill") {}
@@ -161,33 +185,33 @@ export namespace Skill {
Effect.gen(function* () {
const discovery = yield* Discovery.Service
const state = yield* InstanceState.make(
Effect.fn("Skill.state")((ctx) =>
Effect.gen(function* () {
const s: State = { skills: {}, dirs: new Set() }
yield* Effect.promise(() => loadSkills(s, discovery, ctx.directory, ctx.worktree))
return s
}),
),
Effect.fn("Skill.state")((ctx) => Effect.sync(() => create(discovery, ctx.directory, ctx.worktree))),
)
const ensure = Effect.fn("Skill.ensure")(function* () {
const cache = yield* InstanceState.get(state)
yield* Effect.promise(() => cache.ensure())
return cache
})
const get = Effect.fn("Skill.get")(function* (name: string) {
const s = yield* InstanceState.get(state)
return s.skills[name]
const cache = yield* ensure()
return cache.skills[name]
})
const all = Effect.fn("Skill.all")(function* () {
const s = yield* InstanceState.get(state)
return Object.values(s.skills)
const cache = yield* ensure()
return Object.values(cache.skills)
})
const dirs = Effect.fn("Skill.dirs")(function* () {
const s = yield* InstanceState.get(state)
return Array.from(s.dirs)
const cache = yield* ensure()
return Array.from(cache.dirs)
})
const available = Effect.fn("Skill.available")(function* (agent?: Agent.Info) {
const s = yield* InstanceState.get(state)
const list = Object.values(s.skills).toSorted((a, b) => a.name.localeCompare(b.name))
const cache = yield* ensure()
const list = Object.values(cache.skills).toSorted((a, b) => a.name.localeCompare(b.name))
if (!agent) return list
return list.filter((skill) => Permission.evaluate("skill", skill.name, agent.permission).action !== "deny")
})
@@ -218,7 +242,7 @@ export namespace Skill {
return ["## Available Skills", ...list.map((skill) => `- **${skill.name}**: ${skill.description}`)].join("\n")
}
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export async function get(name: string) {
return runPromise((skill) => skill.get(name))

View File

@@ -5,7 +5,7 @@ import path from "path"
import z from "zod"
import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { AppFileSystem } from "@/filesystem"
import { Hash } from "@/util/hash"
import { Config } from "../config/config"
@@ -459,7 +459,7 @@ export namespace Snapshot {
Layer.provide(NodePath.layer),
)
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export async function init() {
return runPromise((svc) => svc.init())

View File

@@ -13,7 +13,6 @@ import { LSP } from "../lsp"
import { Filesystem } from "../util/filesystem"
import DESCRIPTION from "./apply_patch.txt"
import { File } from "../file"
import { Format } from "../format"
const PatchParams = z.object({
patchText: z.string().describe("The full patch text that describes all changes to be made"),
@@ -221,8 +220,9 @@ export const ApplyPatchTool = Tool.define("apply_patch", {
}
if (edited) {
await Format.file(edited)
Bus.publish(File.Event.Edited, { file: edited })
await Bus.publish(File.Event.Edited, {
file: edited,
})
}
}

View File

@@ -12,7 +12,6 @@ import DESCRIPTION from "./edit.txt"
import { File } from "../file"
import { FileWatcher } from "../file/watcher"
import { Bus } from "../bus"
import { Format } from "../format"
import { FileTime } from "../file/time"
import { Filesystem } from "../util/filesystem"
import { Instance } from "../project/instance"
@@ -72,8 +71,9 @@ export const EditTool = Tool.define("edit", {
},
})
await Filesystem.write(filePath, params.newString)
await Format.file(filePath)
Bus.publish(File.Event.Edited, { file: filePath })
await Bus.publish(File.Event.Edited, {
file: filePath,
})
await Bus.publish(FileWatcher.Event.Updated, {
file: filePath,
event: existed ? "change" : "add",
@@ -108,8 +108,9 @@ export const EditTool = Tool.define("edit", {
})
await Filesystem.write(filePath, contentNew)
await Format.file(filePath)
Bus.publish(File.Event.Edited, { file: filePath })
await Bus.publish(File.Event.Edited, {
file: filePath,
})
await Bus.publish(FileWatcher.Event.Updated, {
file: filePath,
event: "change",

View File

@@ -31,7 +31,7 @@ import { Glob } from "../util/glob"
import { pathToFileURL } from "url"
import { Effect, Layer, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
export namespace ToolRegistry {
const log = Log.create({ service: "tool.registry" })
@@ -198,7 +198,7 @@ export namespace ToolRegistry {
}),
)
const { runPromise } = makeRuntime(Service, layer)
const runPromise = makeRunPromise(Service, layer)
export async function register(tool: Tool.Info) {
return runPromise((svc) => svc.register(tool))

View File

@@ -2,7 +2,7 @@ import { NodePath } from "@effect/platform-node"
import { Cause, Duration, Effect, Layer, Schedule, ServiceMap } from "effect"
import path from "path"
import type { Agent } from "../agent/agent"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import { AppFileSystem } from "@/filesystem"
import { evaluate } from "@/permission/evaluate"
import { Identifier } from "../id/id"
@@ -136,7 +136,7 @@ export namespace Truncate {
export const defaultLayer = layer.pipe(Layer.provide(AppFileSystem.defaultLayer), Layer.provide(NodePath.layer))
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export async function output(text: string, options: Options = {}, agent?: Agent.Info): Promise<Result> {
return runPromise((s) => s.output(text, options, agent))

View File

@@ -7,7 +7,6 @@ import DESCRIPTION from "./write.txt"
import { Bus } from "../bus"
import { File } from "../file"
import { FileWatcher } from "../file/watcher"
import { Format } from "../format"
import { FileTime } from "../file/time"
import { Filesystem } from "../util/filesystem"
import { Instance } from "../project/instance"
@@ -43,8 +42,9 @@ export const WriteTool = Tool.define("write", {
})
await Filesystem.write(filepath, params.content)
await Format.file(filepath)
Bus.publish(File.Event.Edited, { file: filepath })
await Bus.publish(File.Event.Edited, {
file: filepath,
})
await Bus.publish(FileWatcher.Event.Updated, {
file: filepath,
event: exists ? "change" : "add",

View File

@@ -15,7 +15,7 @@ import { Git } from "@/git"
import { Effect, FileSystem, Layer, Path, Scope, ServiceMap, Stream } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { NodeFileSystem, NodePath } from "@effect/platform-node"
import { makeRuntime } from "@/effect/run-service"
import { makeRunPromise } from "@/effect/run-service"
import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
export namespace Worktree {
@@ -576,7 +576,7 @@ export namespace Worktree {
Layer.provide(NodeFileSystem.layer),
Layer.provide(NodePath.layer),
)
const { runPromise } = makeRuntime(Service, defaultLayer)
const runPromise = makeRunPromise(Service, defaultLayer)
export async function makeWorktreeInfo(name?: string) {
return runPromise((svc) => svc.makeWorktreeInfo(name))

View File

@@ -1,164 +0,0 @@
import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"
import { describe, expect } from "bun:test"
import { Deferred, Effect, Layer, Stream } from "effect"
import z from "zod"
import { Bus } from "../../src/bus"
import { BusEvent } from "../../src/bus/bus-event"
import { Instance } from "../../src/project/instance"
import { provideInstance, provideTmpdirInstance, tmpdirScoped } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
const TestEvent = {
Ping: BusEvent.define("test.effect.ping", z.object({ value: z.number() })),
Pong: BusEvent.define("test.effect.pong", z.object({ message: z.string() })),
}
const node = NodeChildProcessSpawner.layer.pipe(
Layer.provideMerge(Layer.mergeAll(NodeFileSystem.layer, NodePath.layer)),
)
const live = Layer.mergeAll(Bus.layer, node)
const it = testEffect(live)
describe("Bus (Effect-native)", () => {
it.effect("publish + subscribe stream delivers events", () =>
provideTmpdirInstance(() =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const received: number[] = []
const done = yield* Deferred.make<void>()
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
Effect.sync(() => {
received.push(evt.properties.value)
if (received.length === 2) Deferred.doneUnsafe(done, Effect.void)
}),
).pipe(Effect.forkScoped)
yield* Effect.sleep("10 millis")
yield* bus.publish(TestEvent.Ping, { value: 1 })
yield* bus.publish(TestEvent.Ping, { value: 2 })
yield* Deferred.await(done)
expect(received).toEqual([1, 2])
}),
),
)
it.effect("subscribe filters by event type", () =>
provideTmpdirInstance(() =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const pings: number[] = []
const done = yield* Deferred.make<void>()
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
Effect.sync(() => {
pings.push(evt.properties.value)
Deferred.doneUnsafe(done, Effect.void)
}),
).pipe(Effect.forkScoped)
yield* Effect.sleep("10 millis")
yield* bus.publish(TestEvent.Pong, { message: "ignored" })
yield* bus.publish(TestEvent.Ping, { value: 42 })
yield* Deferred.await(done)
expect(pings).toEqual([42])
}),
),
)
it.effect("subscribeAll receives all types", () =>
provideTmpdirInstance(() =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const types: string[] = []
const done = yield* Deferred.make<void>()
yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
Effect.sync(() => {
types.push(evt.type)
if (types.length === 2) Deferred.doneUnsafe(done, Effect.void)
}),
).pipe(Effect.forkScoped)
yield* Effect.sleep("10 millis")
yield* bus.publish(TestEvent.Ping, { value: 1 })
yield* bus.publish(TestEvent.Pong, { message: "hi" })
yield* Deferred.await(done)
expect(types).toContain("test.effect.ping")
expect(types).toContain("test.effect.pong")
}),
),
)
it.effect("multiple subscribers each receive the event", () =>
provideTmpdirInstance(() =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const a: number[] = []
const b: number[] = []
const doneA = yield* Deferred.make<void>()
const doneB = yield* Deferred.make<void>()
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
Effect.sync(() => {
a.push(evt.properties.value)
Deferred.doneUnsafe(doneA, Effect.void)
}),
).pipe(Effect.forkScoped)
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
Effect.sync(() => {
b.push(evt.properties.value)
Deferred.doneUnsafe(doneB, Effect.void)
}),
).pipe(Effect.forkScoped)
yield* Effect.sleep("10 millis")
yield* bus.publish(TestEvent.Ping, { value: 99 })
yield* Deferred.await(doneA)
yield* Deferred.await(doneB)
expect(a).toEqual([99])
expect(b).toEqual([99])
}),
),
)
it.effect("subscribeAll stream sees InstanceDisposed on disposal", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped()
const types: string[] = []
const seen = yield* Deferred.make<void>()
const disposed = yield* Deferred.make<void>()
// Set up subscriber inside the instance
yield* Effect.gen(function* () {
const bus = yield* Bus.Service
yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
Effect.sync(() => {
types.push(evt.type)
if (evt.type === TestEvent.Ping.type) Deferred.doneUnsafe(seen, Effect.void)
if (evt.type === Bus.InstanceDisposed.type) Deferred.doneUnsafe(disposed, Effect.void)
}),
).pipe(Effect.forkScoped)
yield* Effect.sleep("10 millis")
yield* bus.publish(TestEvent.Ping, { value: 1 })
yield* Deferred.await(seen)
}).pipe(provideInstance(dir))
// Dispose from OUTSIDE the instance scope
yield* Effect.promise(() => Instance.disposeAll())
yield* Deferred.await(disposed).pipe(Effect.timeout("2 seconds"))
expect(types).toContain("test.effect.ping")
expect(types).toContain(Bus.InstanceDisposed.type)
}),
)
})

View File

@@ -1,87 +0,0 @@
import { afterEach, describe, expect, test } from "bun:test"
import z from "zod"
import { Bus } from "../../src/bus"
import { BusEvent } from "../../src/bus/bus-event"
import { Instance } from "../../src/project/instance"
import { tmpdir } from "../fixture/fixture"
const TestEvent = BusEvent.define("test.integration", z.object({ value: z.number() }))
function withInstance(directory: string, fn: () => Promise<void>) {
return Instance.provide({ directory, fn })
}
describe("Bus integration: acquireRelease subscriber pattern", () => {
afterEach(() => Instance.disposeAll())
test("subscriber via callback facade receives events and cleans up on unsub", async () => {
await using tmp = await tmpdir()
const received: number[] = []
await withInstance(tmp.path, async () => {
const unsub = Bus.subscribe(TestEvent, (evt) => {
received.push(evt.properties.value)
})
await Bun.sleep(10)
await Bus.publish(TestEvent, { value: 1 })
await Bus.publish(TestEvent, { value: 2 })
await Bun.sleep(10)
expect(received).toEqual([1, 2])
unsub()
await Bun.sleep(10)
await Bus.publish(TestEvent, { value: 3 })
await Bun.sleep(10)
expect(received).toEqual([1, 2])
})
})
test("subscribeAll receives events from multiple types", async () => {
await using tmp = await tmpdir()
const received: Array<{ type: string; value?: number }> = []
const OtherEvent = BusEvent.define("test.other", z.object({ value: z.number() }))
await withInstance(tmp.path, async () => {
Bus.subscribeAll((evt) => {
received.push({ type: evt.type, value: evt.properties.value })
})
await Bun.sleep(10)
await Bus.publish(TestEvent, { value: 10 })
await Bus.publish(OtherEvent, { value: 20 })
await Bun.sleep(10)
})
expect(received).toEqual([
{ type: "test.integration", value: 10 },
{ type: "test.other", value: 20 },
])
})
test("subscriber cleanup on instance disposal interrupts the stream", async () => {
await using tmp = await tmpdir()
const received: number[] = []
let disposed = false
await withInstance(tmp.path, async () => {
Bus.subscribeAll((evt) => {
if (evt.type === Bus.InstanceDisposed.type) {
disposed = true
return
}
received.push(evt.properties.value)
})
await Bun.sleep(10)
await Bus.publish(TestEvent, { value: 1 })
await Bun.sleep(10)
})
await Instance.disposeAll()
await Bun.sleep(50)
expect(received).toEqual([1])
expect(disposed).toBe(true)
})
})

View File

@@ -1,219 +0,0 @@
import { afterEach, describe, expect, test } from "bun:test"
import z from "zod"
import { Bus } from "../../src/bus"
import { BusEvent } from "../../src/bus/bus-event"
import { Instance } from "../../src/project/instance"
import { tmpdir } from "../fixture/fixture"
const TestEvent = {
Ping: BusEvent.define("test.ping", z.object({ value: z.number() })),
Pong: BusEvent.define("test.pong", z.object({ message: z.string() })),
}
function withInstance(directory: string, fn: () => Promise<void>) {
return Instance.provide({ directory, fn })
}
describe("Bus", () => {
afterEach(() => Instance.disposeAll())
describe("publish + subscribe", () => {
test("subscriber is live immediately after subscribe returns", async () => {
await using tmp = await tmpdir()
const received: number[] = []
await withInstance(tmp.path, async () => {
Bus.subscribe(TestEvent.Ping, (evt) => {
received.push(evt.properties.value)
})
await Bus.publish(TestEvent.Ping, { value: 42 })
await Bun.sleep(10)
})
expect(received).toEqual([42])
})
test("subscriber receives matching events", async () => {
await using tmp = await tmpdir()
const received: number[] = []
await withInstance(tmp.path, async () => {
Bus.subscribe(TestEvent.Ping, (evt) => {
received.push(evt.properties.value)
})
// Give the subscriber fiber time to start consuming
await Bun.sleep(10)
await Bus.publish(TestEvent.Ping, { value: 42 })
await Bus.publish(TestEvent.Ping, { value: 99 })
// Give subscriber time to process
await Bun.sleep(10)
})
expect(received).toEqual([42, 99])
})
test("subscriber does not receive events of other types", async () => {
await using tmp = await tmpdir()
const pings: number[] = []
await withInstance(tmp.path, async () => {
Bus.subscribe(TestEvent.Ping, (evt) => {
pings.push(evt.properties.value)
})
await Bun.sleep(10)
await Bus.publish(TestEvent.Pong, { message: "hello" })
await Bus.publish(TestEvent.Ping, { value: 1 })
await Bun.sleep(10)
})
expect(pings).toEqual([1])
})
test("publish with no subscribers does not throw", async () => {
await using tmp = await tmpdir()
await withInstance(tmp.path, async () => {
await Bus.publish(TestEvent.Ping, { value: 1 })
})
})
})
describe("unsubscribe", () => {
test("unsubscribe stops delivery", async () => {
await using tmp = await tmpdir()
const received: number[] = []
await withInstance(tmp.path, async () => {
const unsub = Bus.subscribe(TestEvent.Ping, (evt) => {
received.push(evt.properties.value)
})
await Bun.sleep(10)
await Bus.publish(TestEvent.Ping, { value: 1 })
await Bun.sleep(10)
unsub()
await Bun.sleep(10)
await Bus.publish(TestEvent.Ping, { value: 2 })
await Bun.sleep(10)
})
expect(received).toEqual([1])
})
})
describe("subscribeAll", () => {
test("subscribeAll is live immediately after subscribe returns", async () => {
await using tmp = await tmpdir()
const received: string[] = []
await withInstance(tmp.path, async () => {
Bus.subscribeAll((evt) => {
received.push(evt.type)
})
await Bus.publish(TestEvent.Ping, { value: 1 })
await Bun.sleep(10)
})
expect(received).toEqual(["test.ping"])
})
test("receives all event types", async () => {
await using tmp = await tmpdir()
const received: string[] = []
await withInstance(tmp.path, async () => {
Bus.subscribeAll((evt) => {
received.push(evt.type)
})
await Bun.sleep(10)
await Bus.publish(TestEvent.Ping, { value: 1 })
await Bus.publish(TestEvent.Pong, { message: "hi" })
await Bun.sleep(10)
})
expect(received).toContain("test.ping")
expect(received).toContain("test.pong")
})
})
describe("multiple subscribers", () => {
test("all subscribers for same event type are called", async () => {
await using tmp = await tmpdir()
const a: number[] = []
const b: number[] = []
await withInstance(tmp.path, async () => {
Bus.subscribe(TestEvent.Ping, (evt) => {
a.push(evt.properties.value)
})
Bus.subscribe(TestEvent.Ping, (evt) => {
b.push(evt.properties.value)
})
await Bun.sleep(10)
await Bus.publish(TestEvent.Ping, { value: 7 })
await Bun.sleep(10)
})
expect(a).toEqual([7])
expect(b).toEqual([7])
})
})
describe("instance isolation", () => {
test("events in one directory do not reach subscribers in another", async () => {
await using tmpA = await tmpdir()
await using tmpB = await tmpdir()
const receivedA: number[] = []
const receivedB: number[] = []
await withInstance(tmpA.path, async () => {
Bus.subscribe(TestEvent.Ping, (evt) => {
receivedA.push(evt.properties.value)
})
await Bun.sleep(10)
})
await withInstance(tmpB.path, async () => {
Bus.subscribe(TestEvent.Ping, (evt) => {
receivedB.push(evt.properties.value)
})
await Bun.sleep(10)
})
await withInstance(tmpA.path, async () => {
await Bus.publish(TestEvent.Ping, { value: 1 })
await Bun.sleep(10)
})
await withInstance(tmpB.path, async () => {
await Bus.publish(TestEvent.Ping, { value: 2 })
await Bun.sleep(10)
})
expect(receivedA).toEqual([1])
expect(receivedB).toEqual([2])
})
})
describe("instance disposal", () => {
test("InstanceDisposed is delivered to wildcard subscribers before stream ends", async () => {
await using tmp = await tmpdir()
const received: string[] = []
await withInstance(tmp.path, async () => {
Bus.subscribeAll((evt) => {
received.push(evt.type)
})
await Bun.sleep(10)
await Bus.publish(TestEvent.Ping, { value: 1 })
await Bun.sleep(10)
})
// Instance.disposeAll triggers the finalizer which publishes InstanceDisposed
await Instance.disposeAll()
await Bun.sleep(50)
expect(received).toContain("test.ping")
expect(received).toContain(Bus.InstanceDisposed.type)
})
})
})

View File

@@ -1,10 +1,10 @@
import { expect, test } from "bun:test"
import { Effect, Layer, ServiceMap } from "effect"
import { makeRuntime } from "../../src/effect/run-service"
import { makeRunPromise } from "../../src/effect/run-service"
class Shared extends ServiceMap.Service<Shared, { readonly id: number }>()("@test/Shared") {}
test("makeRuntime shares dependent layers through the shared memo map", async () => {
test("makeRunPromise shares dependent layers through the shared memo map", async () => {
let n = 0
const shared = Layer.effect(
@@ -37,8 +37,8 @@ test("makeRuntime shares dependent layers through the shared memo map", async ()
}),
).pipe(Layer.provide(shared))
const { runPromise: runOne } = makeRuntime(One, one)
const { runPromise: runTwo } = makeRuntime(Two, two)
const runOne = makeRunPromise(One, one)
const runTwo = makeRunPromise(Two, two)
expect(await runOne((svc) => svc.get())).toBe(1)
expect(await runTwo((svc) => svc.get())).toBe(1)

View File

@@ -2,8 +2,9 @@ import { $ } from "bun"
import { afterEach, describe, expect, test } from "bun:test"
import fs from "fs/promises"
import path from "path"
import { ConfigProvider, Deferred, Effect, Layer, ManagedRuntime, Option } from "effect"
import { Deferred, Effect, Option } from "effect"
import { tmpdir } from "../fixture/fixture"
import { watcherConfigLayer, withServices } from "../fixture/instance"
import { Bus } from "../../src/bus"
import { FileWatcher } from "../../src/file/watcher"
import { Instance } from "../../src/project/instance"
@@ -15,33 +16,20 @@ const describeWatcher = FileWatcher.hasNativeBinding() && !process.env.CI ? desc
// Helpers
// ---------------------------------------------------------------------------
const watcherConfigLayer = ConfigProvider.layer(
ConfigProvider.fromUnknown({
OPENCODE_EXPERIMENTAL_FILEWATCHER: "true",
OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "false",
}),
)
type WatcherEvent = { file: string; event: "add" | "change" | "unlink" }
/** Run `body` with a live FileWatcher service. */
function withWatcher<E>(directory: string, body: Effect.Effect<void, E>) {
return Instance.provide({
return withServices(
directory,
fn: async () => {
const layer: Layer.Layer<FileWatcher.Service, never, never> = FileWatcher.layer.pipe(
Layer.provide(watcherConfigLayer),
)
const rt = ManagedRuntime.make(layer)
try {
await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
await Effect.runPromise(ready(directory))
await Effect.runPromise(body)
} finally {
await rt.dispose()
}
FileWatcher.layer,
async (rt) => {
await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
await Effect.runPromise(ready(directory))
await Effect.runPromise(body)
},
})
{ provide: [watcherConfigLayer] },
)
}
function listen(directory: string, check: (evt: WatcherEvent) => boolean, hit: (evt: WatcherEvent) => void) {

View File

@@ -2,10 +2,7 @@ import { $ } from "bun"
import * as fs from "fs/promises"
import os from "os"
import path from "path"
import { Effect, FileSystem, ServiceMap } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import type { Config } from "../../src/config/config"
import { Instance } from "../../src/project/instance"
// Strip null bytes from paths (defensive fix for CI environment issues)
function sanitizePath(p: string): string {
@@ -74,68 +71,3 @@ export async function tmpdir<T>(options?: TmpDirOptions<T>) {
}
return result
}
/** Effectful scoped tmpdir. Cleaned up when the scope closes. Make sure these stay in sync */
export function tmpdirScoped(options?: { git?: boolean; config?: Partial<Config.Info> }) {
return Effect.gen(function* () {
const fs = yield* FileSystem.FileSystem
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
const dir = yield* fs.makeTempDirectoryScoped({ prefix: "opencode-test-" })
const git = (...args: string[]) =>
spawner.spawn(ChildProcess.make("git", args, { cwd: dir })).pipe(Effect.flatMap((handle) => handle.exitCode))
if (options?.git) {
yield* git("init")
yield* git("config", "core.fsmonitor", "false")
yield* git("config", "user.email", "test@opencode.test")
yield* git("config", "user.name", "Test")
yield* git("commit", "--allow-empty", "-m", "root commit")
}
if (options?.config) {
yield* fs.writeFileString(
path.join(dir, "opencode.json"),
JSON.stringify({ $schema: "https://opencode.ai/config.json", ...options.config }),
)
}
return dir
})
}
export const provideInstance =
(directory: string) =>
<A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
Effect.servicesWith((services: ServiceMap.ServiceMap<R>) =>
Effect.promise<A>(async () =>
Instance.provide({
directory,
fn: () => Effect.runPromiseWith(services)(self),
}),
),
)
export function provideTmpdirInstance<A, E, R>(
self: (path: string) => Effect.Effect<A, E, R>,
options?: { git?: boolean; config?: Partial<Config.Info> },
) {
return Effect.gen(function* () {
const path = yield* tmpdirScoped(options)
let provided = false
yield* Effect.addFinalizer(() =>
provided
? Effect.promise(() =>
Instance.provide({
directory: path,
fn: () => Instance.dispose(),
}),
).pipe(Effect.ignore)
: Effect.void,
)
provided = true
return yield* self(path).pipe(provideInstance(path))
})
}

View File

@@ -0,0 +1,51 @@
import { ConfigProvider, Layer, ManagedRuntime } from "effect"
import { InstanceContext } from "../../src/effect/instance-context"
import { Instance } from "../../src/project/instance"
/** ConfigProvider that enables the experimental file watcher. */
export const watcherConfigLayer = ConfigProvider.layer(
ConfigProvider.fromUnknown({
OPENCODE_EXPERIMENTAL_FILEWATCHER: "true",
OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "false",
}),
)
/**
* Boot an Instance with the given service layers and run `body` with
* the ManagedRuntime. Cleanup is automatic — the runtime is disposed
* and Instance context is torn down when `body` completes.
*
* Layers may depend on InstanceContext (provided automatically).
* Pass extra layers via `options.provide` (e.g. ConfigProvider.layer).
*/
export function withServices<S>(
directory: string,
layer: Layer.Layer<S, any, InstanceContext>,
body: (rt: ManagedRuntime.ManagedRuntime<S, never>) => Promise<void>,
options?: { provide?: Layer.Layer<never>[] },
) {
return Instance.provide({
directory,
fn: async () => {
const ctx = Layer.sync(InstanceContext, () =>
InstanceContext.of({
directory: Instance.directory,
worktree: Instance.worktree,
project: Instance.project,
}),
)
let resolved: Layer.Layer<S> = layer.pipe(Layer.provide(ctx)) as any
if (options?.provide) {
for (const l of options.provide) {
resolved = resolved.pipe(Layer.provide(l)) as any
}
}
const rt = ManagedRuntime.make(resolved)
try {
await body(rt)
} finally {
await rt.dispose()
}
},
})
}

View File

@@ -1,182 +1,172 @@
import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"
import { describe, expect } from "bun:test"
import { Effect, Layer } from "effect"
import { provideTmpdirInstance } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
import { Effect } from "effect"
import { afterEach, describe, expect, test } from "bun:test"
import { tmpdir } from "../fixture/fixture"
import { withServices } from "../fixture/instance"
import { Bus } from "../../src/bus"
import { File } from "../../src/file"
import { Format } from "../../src/format"
import * as Formatter from "../../src/format/formatter"
const node = NodeChildProcessSpawner.layer.pipe(
Layer.provideMerge(Layer.mergeAll(NodeFileSystem.layer, NodePath.layer)),
)
const it = testEffect(Layer.mergeAll(Format.layer, node))
import { Instance } from "../../src/project/instance"
describe("Format", () => {
it.effect("status() returns built-in formatters when no config overrides", () =>
provideTmpdirInstance(() =>
Format.Service.use((fmt) =>
Effect.gen(function* () {
const statuses = yield* fmt.status()
expect(Array.isArray(statuses)).toBe(true)
expect(statuses.length).toBeGreaterThan(0)
afterEach(async () => {
await Instance.disposeAll()
})
for (const item of statuses) {
expect(typeof item.name).toBe("string")
expect(Array.isArray(item.extensions)).toBe(true)
expect(typeof item.enabled).toBe("boolean")
}
test("status() returns built-in formatters when no config overrides", async () => {
await using tmp = await tmpdir()
const gofmt = statuses.find((item) => item.name === "gofmt")
expect(gofmt).toBeDefined()
expect(gofmt!.extensions).toContain(".go")
}),
),
),
)
await withServices(tmp.path, Format.layer, async (rt) => {
const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
expect(Array.isArray(statuses)).toBe(true)
expect(statuses.length).toBeGreaterThan(0)
it.effect("status() returns empty list when formatter is disabled", () =>
provideTmpdirInstance(
() =>
Format.Service.use((fmt) =>
Effect.gen(function* () {
expect(yield* fmt.status()).toEqual([])
}),
),
{ config: { formatter: false } },
),
)
for (const s of statuses) {
expect(typeof s.name).toBe("string")
expect(Array.isArray(s.extensions)).toBe(true)
expect(typeof s.enabled).toBe("boolean")
}
it.effect("status() excludes formatters marked as disabled in config", () =>
provideTmpdirInstance(
() =>
Format.Service.use((fmt) =>
Effect.gen(function* () {
const statuses = yield* fmt.status()
const gofmt = statuses.find((item) => item.name === "gofmt")
expect(gofmt).toBeUndefined()
}),
),
{
config: {
formatter: {
gofmt: { disabled: true },
},
const gofmt = statuses.find((s) => s.name === "gofmt")
expect(gofmt).toBeDefined()
expect(gofmt!.extensions).toContain(".go")
})
})
test("status() returns empty list when formatter is disabled", async () => {
await using tmp = await tmpdir({
config: { formatter: false },
})
await withServices(tmp.path, Format.layer, async (rt) => {
const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
expect(statuses).toEqual([])
})
})
test("status() excludes formatters marked as disabled in config", async () => {
await using tmp = await tmpdir({
config: {
formatter: {
gofmt: { disabled: true },
},
},
),
)
})
it.effect("service initializes without error", () =>
provideTmpdirInstance(() => Format.Service.use(() => Effect.void)),
)
await withServices(tmp.path, Format.layer, async (rt) => {
const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
const gofmt = statuses.find((s) => s.name === "gofmt")
expect(gofmt).toBeUndefined()
})
})
it.effect("status() initializes formatter state per directory", () =>
Effect.gen(function* () {
const a = yield* provideTmpdirInstance(() => Format.Service.use((fmt) => fmt.status()), {
config: { formatter: false },
test("service initializes without error", async () => {
await using tmp = await tmpdir()
await withServices(tmp.path, Format.layer, async (rt) => {
await rt.runPromise(Format.Service.use(() => Effect.void))
})
})
test("status() initializes formatter state per directory", async () => {
await using off = await tmpdir({
config: { formatter: false },
})
await using on = await tmpdir()
const a = await Instance.provide({
directory: off.path,
fn: () => Format.status(),
})
const b = await Instance.provide({
directory: on.path,
fn: () => Format.status(),
})
expect(a).toEqual([])
expect(b.length).toBeGreaterThan(0)
})
test("runs enabled checks for matching formatters in parallel", async () => {
await using tmp = await tmpdir()
const file = `${tmp.path}/test.parallel`
await Bun.write(file, "x")
const one = {
extensions: Formatter.gofmt.extensions,
enabled: Formatter.gofmt.enabled,
command: Formatter.gofmt.command,
}
const two = {
extensions: Formatter.mix.extensions,
enabled: Formatter.mix.enabled,
command: Formatter.mix.command,
}
let active = 0
let max = 0
Formatter.gofmt.extensions = [".parallel"]
Formatter.mix.extensions = [".parallel"]
Formatter.gofmt.command = ["sh", "-c", "true"]
Formatter.mix.command = ["sh", "-c", "true"]
Formatter.gofmt.enabled = async () => {
active++
max = Math.max(max, active)
await Bun.sleep(20)
active--
return true
}
Formatter.mix.enabled = async () => {
active++
max = Math.max(max, active)
await Bun.sleep(20)
active--
return true
}
try {
await withServices(tmp.path, Format.layer, async (rt) => {
await rt.runPromise(Format.Service.use((s) => s.init()))
await Bus.publish(File.Event.Edited, { file })
})
const b = yield* provideTmpdirInstance(() => Format.Service.use((fmt) => fmt.status()))
} finally {
Formatter.gofmt.extensions = one.extensions
Formatter.gofmt.enabled = one.enabled
Formatter.gofmt.command = one.command
Formatter.mix.extensions = two.extensions
Formatter.mix.enabled = two.enabled
Formatter.mix.command = two.command
}
expect(a).toEqual([])
expect(b.length).toBeGreaterThan(0)
}),
)
expect(max).toBe(2)
})
it.effect("runs enabled checks for matching formatters in parallel", () =>
provideTmpdirInstance((path) =>
Effect.gen(function* () {
const file = `${path}/test.parallel`
yield* Effect.promise(() => Bun.write(file, "x"))
const one = {
extensions: Formatter.gofmt.extensions,
enabled: Formatter.gofmt.enabled,
command: Formatter.gofmt.command,
}
const two = {
extensions: Formatter.mix.extensions,
enabled: Formatter.mix.enabled,
command: Formatter.mix.command,
}
let active = 0
let max = 0
yield* Effect.acquireUseRelease(
Effect.sync(() => {
Formatter.gofmt.extensions = [".parallel"]
Formatter.mix.extensions = [".parallel"]
Formatter.gofmt.command = ["sh", "-c", "true"]
Formatter.mix.command = ["sh", "-c", "true"]
Formatter.gofmt.enabled = async () => {
active++
max = Math.max(max, active)
await Bun.sleep(20)
active--
return true
}
Formatter.mix.enabled = async () => {
active++
max = Math.max(max, active)
await Bun.sleep(20)
active--
return true
}
}),
() =>
Format.Service.use((fmt) =>
Effect.gen(function* () {
yield* fmt.init()
yield* fmt.file(file)
}),
),
() =>
Effect.sync(() => {
Formatter.gofmt.extensions = one.extensions
Formatter.gofmt.enabled = one.enabled
Formatter.gofmt.command = one.command
Formatter.mix.extensions = two.extensions
Formatter.mix.enabled = two.enabled
Formatter.mix.command = two.command
}),
)
expect(max).toBe(2)
}),
),
)
it.effect("runs matching formatters sequentially for the same file", () =>
provideTmpdirInstance(
(path) =>
Effect.gen(function* () {
const file = `${path}/test.seq`
yield* Effect.promise(() => Bun.write(file, "x"))
yield* Format.Service.use((fmt) =>
Effect.gen(function* () {
yield* fmt.init()
yield* fmt.file(file)
}),
)
expect(yield* Effect.promise(() => Bun.file(file).text())).toBe("xAB")
}),
{
config: {
formatter: {
first: {
command: ["sh", "-c", 'sleep 0.05; v=$(cat "$1"); printf \'%sA\' "$v" > "$1"', "sh", "$FILE"],
extensions: [".seq"],
},
second: {
command: ["sh", "-c", 'v=$(cat "$1"); printf \'%sB\' "$v" > "$1"', "sh", "$FILE"],
extensions: [".seq"],
},
test("runs matching formatters sequentially for the same file", async () => {
await using tmp = await tmpdir({
config: {
formatter: {
first: {
command: ["sh", "-c", 'sleep 0.05; v=$(cat "$1"); printf \'%sA\' "$v" > "$1"', "sh", "$FILE"],
extensions: [".seq"],
},
second: {
command: ["sh", "-c", 'v=$(cat "$1"); printf \'%sB\' "$v" > "$1"', "sh", "$FILE"],
extensions: [".seq"],
},
},
},
),
)
})
const file = `${tmp.path}/test.seq`
await Bun.write(file, "x")
await withServices(tmp.path, Format.layer, async (rt) => {
await rt.runPromise(Format.Service.use((s) => s.init()))
await Bus.publish(File.Event.Edited, { file })
})
expect(await Bun.file(file).text()).toBe("xAB")
})
})

View File

@@ -1,660 +0,0 @@
import { test, expect, mock, beforeEach } from "bun:test"
// --- Mock infrastructure ---
// Per-client state for controlling mock behavior
interface MockClientState {
tools: Array<{ name: string; description?: string; inputSchema: object }>
listToolsCalls: number
listToolsShouldFail: boolean
listToolsError: string
listPromptsShouldFail: boolean
listResourcesShouldFail: boolean
prompts: Array<{ name: string; description?: string }>
resources: Array<{ name: string; uri: string; description?: string }>
closed: boolean
notificationHandlers: Map<unknown, (...args: any[]) => any>
}
const clientStates = new Map<string, MockClientState>()
let lastCreatedClientName: string | undefined
let connectShouldFail = false
let connectError = "Mock transport cannot connect"
// Tracks how many Client instances were created (detects leaks)
let clientCreateCount = 0
function getOrCreateClientState(name?: string): MockClientState {
const key = name ?? "default"
let state = clientStates.get(key)
if (!state) {
state = {
tools: [{ name: "test_tool", description: "A test tool", inputSchema: { type: "object", properties: {} } }],
listToolsCalls: 0,
listToolsShouldFail: false,
listToolsError: "listTools failed",
listPromptsShouldFail: false,
listResourcesShouldFail: false,
prompts: [],
resources: [],
closed: false,
notificationHandlers: new Map(),
}
clientStates.set(key, state)
}
return state
}
// Mock transport that succeeds or fails based on connectShouldFail
class MockStdioTransport {
stderr: null = null
pid = 12345
constructor(_opts: any) {}
async start() {
if (connectShouldFail) throw new Error(connectError)
}
async close() {}
}
class MockStreamableHTTP {
constructor(_url: URL, _opts?: any) {}
async start() {
if (connectShouldFail) throw new Error(connectError)
}
async close() {}
async finishAuth() {}
}
class MockSSE {
constructor(_url: URL, _opts?: any) {}
async start() {
throw new Error("SSE fallback - not used in these tests")
}
async close() {}
}
mock.module("@modelcontextprotocol/sdk/client/stdio.js", () => ({
StdioClientTransport: MockStdioTransport,
}))
mock.module("@modelcontextprotocol/sdk/client/streamableHttp.js", () => ({
StreamableHTTPClientTransport: MockStreamableHTTP,
}))
mock.module("@modelcontextprotocol/sdk/client/sse.js", () => ({
SSEClientTransport: MockSSE,
}))
mock.module("@modelcontextprotocol/sdk/client/auth.js", () => ({
UnauthorizedError: class extends Error {
constructor() {
super("Unauthorized")
}
},
}))
// Mock Client that delegates to per-name MockClientState
mock.module("@modelcontextprotocol/sdk/client/index.js", () => ({
Client: class MockClient {
_state!: MockClientState
transport: any
constructor(_opts: any) {
clientCreateCount++
}
async connect(transport: { start: () => Promise<void> }) {
this.transport = transport
await transport.start()
// After successful connect, bind to the last-created client name
this._state = getOrCreateClientState(lastCreatedClientName)
}
setNotificationHandler(schema: unknown, handler: (...args: any[]) => any) {
this._state?.notificationHandlers.set(schema, handler)
}
async listTools() {
if (this._state) this._state.listToolsCalls++
if (this._state?.listToolsShouldFail) {
throw new Error(this._state.listToolsError)
}
return { tools: this._state?.tools ?? [] }
}
async listPrompts() {
if (this._state?.listPromptsShouldFail) {
throw new Error("listPrompts failed")
}
return { prompts: this._state?.prompts ?? [] }
}
async listResources() {
if (this._state?.listResourcesShouldFail) {
throw new Error("listResources failed")
}
return { resources: this._state?.resources ?? [] }
}
async close() {
if (this._state) this._state.closed = true
}
},
}))
beforeEach(() => {
clientStates.clear()
lastCreatedClientName = undefined
connectShouldFail = false
connectError = "Mock transport cannot connect"
clientCreateCount = 0
})
// Import after mocks
const { MCP } = await import("../../src/mcp/index")
const { Instance } = await import("../../src/project/instance")
const { tmpdir } = await import("../fixture/fixture")
// --- Helper ---
function withInstance(config: Record<string, any>, fn: () => Promise<void>) {
return async () => {
await using tmp = await tmpdir({
init: async (dir) => {
await Bun.write(
`${dir}/opencode.json`,
JSON.stringify({
$schema: "https://opencode.ai/config.json",
mcp: config,
}),
)
},
})
await Instance.provide({
directory: tmp.path,
fn: async () => {
await fn()
// dispose instance to clean up state between tests
await Instance.dispose()
},
})
}
}
// ========================================================================
// Test: tools() are cached after connect
// ========================================================================
test(
"tools() reuses cached tool definitions after connect",
withInstance({}, async () => {
lastCreatedClientName = "my-server"
const serverState = getOrCreateClientState("my-server")
serverState.tools = [
{ name: "do_thing", description: "does a thing", inputSchema: { type: "object", properties: {} } },
]
// First: add the server successfully
const addResult = await MCP.add("my-server", {
type: "local",
command: ["echo", "test"],
})
expect((addResult.status as any)["my-server"]?.status ?? (addResult.status as any).status).toBe("connected")
expect(serverState.listToolsCalls).toBe(1)
const toolsA = await MCP.tools()
const toolsB = await MCP.tools()
expect(Object.keys(toolsA).length).toBeGreaterThan(0)
expect(Object.keys(toolsB).length).toBeGreaterThan(0)
expect(serverState.listToolsCalls).toBe(1)
}),
)
// ========================================================================
// Test: tool change notifications refresh the cache
// ========================================================================
test(
"tool change notifications refresh cached tool definitions",
withInstance({}, async () => {
lastCreatedClientName = "status-server"
const serverState = getOrCreateClientState("status-server")
await MCP.add("status-server", {
type: "local",
command: ["echo", "test"],
})
const before = await MCP.tools()
expect(Object.keys(before).some((key) => key.includes("test_tool"))).toBe(true)
expect(serverState.listToolsCalls).toBe(1)
serverState.tools = [{ name: "next_tool", description: "next", inputSchema: { type: "object", properties: {} } }]
const handler = Array.from(serverState.notificationHandlers.values())[0]
expect(handler).toBeDefined()
await handler?.()
const after = await MCP.tools()
expect(Object.keys(after).some((key) => key.includes("next_tool"))).toBe(true)
expect(Object.keys(after).some((key) => key.includes("test_tool"))).toBe(false)
expect(serverState.listToolsCalls).toBe(2)
}),
)
// ========================================================================
// Test: connect() / disconnect() lifecycle
// ========================================================================
test(
"disconnect sets status to disabled and removes client",
withInstance(
{
"disc-server": {
type: "local",
command: ["echo", "test"],
},
},
async () => {
lastCreatedClientName = "disc-server"
getOrCreateClientState("disc-server")
await MCP.add("disc-server", {
type: "local",
command: ["echo", "test"],
})
const statusBefore = await MCP.status()
expect(statusBefore["disc-server"]?.status).toBe("connected")
await MCP.disconnect("disc-server")
const statusAfter = await MCP.status()
expect(statusAfter["disc-server"]?.status).toBe("disabled")
// Tools should be empty after disconnect
const tools = await MCP.tools()
const serverTools = Object.keys(tools).filter((k) => k.startsWith("disc-server"))
expect(serverTools.length).toBe(0)
},
),
)
test(
"connect() after disconnect() re-establishes the server",
withInstance(
{
"reconn-server": {
type: "local",
command: ["echo", "test"],
},
},
async () => {
lastCreatedClientName = "reconn-server"
const serverState = getOrCreateClientState("reconn-server")
serverState.tools = [{ name: "my_tool", description: "a tool", inputSchema: { type: "object", properties: {} } }]
await MCP.add("reconn-server", {
type: "local",
command: ["echo", "test"],
})
await MCP.disconnect("reconn-server")
expect((await MCP.status())["reconn-server"]?.status).toBe("disabled")
// Reconnect
await MCP.connect("reconn-server")
expect((await MCP.status())["reconn-server"]?.status).toBe("connected")
const tools = await MCP.tools()
expect(Object.keys(tools).some((k) => k.includes("my_tool"))).toBe(true)
},
),
)
// ========================================================================
// Test: add() closes existing client before replacing
// ========================================================================
test(
"add() closes the old client when replacing a server",
// Don't put the server in config — add it dynamically so we control
// exactly which client instance is "first" vs "second".
withInstance({}, async () => {
lastCreatedClientName = "replace-server"
const firstState = getOrCreateClientState("replace-server")
await MCP.add("replace-server", {
type: "local",
command: ["echo", "test"],
})
expect(firstState.closed).toBe(false)
// Create new state for second client
clientStates.delete("replace-server")
const secondState = getOrCreateClientState("replace-server")
// Re-add should close the first client
await MCP.add("replace-server", {
type: "local",
command: ["echo", "test"],
})
expect(firstState.closed).toBe(true)
expect(secondState.closed).toBe(false)
}),
)
// ========================================================================
// Test: state init with mixed success/failure
// ========================================================================
test(
"init connects available servers even when one fails",
withInstance(
{
"good-server": {
type: "local",
command: ["echo", "good"],
},
"bad-server": {
type: "local",
command: ["echo", "bad"],
},
},
async () => {
// Set up good server
const goodState = getOrCreateClientState("good-server")
goodState.tools = [{ name: "good_tool", description: "works", inputSchema: { type: "object", properties: {} } }]
// Set up bad server - will fail on listTools during create()
const badState = getOrCreateClientState("bad-server")
badState.listToolsShouldFail = true
// Add good server first
lastCreatedClientName = "good-server"
await MCP.add("good-server", {
type: "local",
command: ["echo", "good"],
})
// Add bad server - should fail but not affect good server
lastCreatedClientName = "bad-server"
await MCP.add("bad-server", {
type: "local",
command: ["echo", "bad"],
})
const status = await MCP.status()
expect(status["good-server"]?.status).toBe("connected")
expect(status["bad-server"]?.status).toBe("failed")
// Good server's tools should still be available
const tools = await MCP.tools()
expect(Object.keys(tools).some((k) => k.includes("good_tool"))).toBe(true)
},
),
)
// ========================================================================
// Test: disabled server via config
// ========================================================================
test(
"disabled server is marked as disabled without attempting connection",
withInstance(
{
"disabled-server": {
type: "local",
command: ["echo", "test"],
enabled: false,
},
},
async () => {
const countBefore = clientCreateCount
await MCP.add("disabled-server", {
type: "local",
command: ["echo", "test"],
enabled: false,
} as any)
// No client should have been created
expect(clientCreateCount).toBe(countBefore)
const status = await MCP.status()
expect(status["disabled-server"]?.status).toBe("disabled")
},
),
)
// ========================================================================
// Test: prompts() and resources()
// ========================================================================
test(
"prompts() returns prompts from connected servers",
withInstance(
{
"prompt-server": {
type: "local",
command: ["echo", "test"],
},
},
async () => {
lastCreatedClientName = "prompt-server"
const serverState = getOrCreateClientState("prompt-server")
serverState.prompts = [{ name: "my-prompt", description: "A test prompt" }]
await MCP.add("prompt-server", {
type: "local",
command: ["echo", "test"],
})
const prompts = await MCP.prompts()
expect(Object.keys(prompts).length).toBe(1)
const key = Object.keys(prompts)[0]
expect(key).toContain("prompt-server")
expect(key).toContain("my-prompt")
},
),
)
test(
"resources() returns resources from connected servers",
withInstance(
{
"resource-server": {
type: "local",
command: ["echo", "test"],
},
},
async () => {
lastCreatedClientName = "resource-server"
const serverState = getOrCreateClientState("resource-server")
serverState.resources = [{ name: "my-resource", uri: "file:///test.txt", description: "A test resource" }]
await MCP.add("resource-server", {
type: "local",
command: ["echo", "test"],
})
const resources = await MCP.resources()
expect(Object.keys(resources).length).toBe(1)
const key = Object.keys(resources)[0]
expect(key).toContain("resource-server")
expect(key).toContain("my-resource")
},
),
)
test(
"prompts() skips disconnected servers",
withInstance(
{
"prompt-disc-server": {
type: "local",
command: ["echo", "test"],
},
},
async () => {
lastCreatedClientName = "prompt-disc-server"
const serverState = getOrCreateClientState("prompt-disc-server")
serverState.prompts = [{ name: "hidden-prompt", description: "Should not appear" }]
await MCP.add("prompt-disc-server", {
type: "local",
command: ["echo", "test"],
})
await MCP.disconnect("prompt-disc-server")
const prompts = await MCP.prompts()
expect(Object.keys(prompts).length).toBe(0)
},
),
)
// ========================================================================
// Test: connect() on nonexistent server
// ========================================================================
test(
"connect() on nonexistent server does not throw",
withInstance({}, async () => {
// Should not throw
await MCP.connect("nonexistent")
const status = await MCP.status()
expect(status["nonexistent"]).toBeUndefined()
}),
)
// ========================================================================
// Test: disconnect() on nonexistent server
// ========================================================================
test(
"disconnect() on nonexistent server does not throw",
withInstance({}, async () => {
await MCP.disconnect("nonexistent")
// Should complete without error
}),
)
// ========================================================================
// Test: tools() with no MCP servers configured
// ========================================================================
test(
"tools() returns empty when no MCP servers are configured",
withInstance({}, async () => {
const tools = await MCP.tools()
expect(Object.keys(tools).length).toBe(0)
}),
)
// ========================================================================
// Test: connect failure during create()
// ========================================================================
test(
"server that fails to connect is marked as failed",
withInstance(
{
"fail-connect": {
type: "local",
command: ["echo", "test"],
},
},
async () => {
lastCreatedClientName = "fail-connect"
getOrCreateClientState("fail-connect")
connectShouldFail = true
connectError = "Connection refused"
await MCP.add("fail-connect", {
type: "local",
command: ["echo", "test"],
})
const status = await MCP.status()
expect(status["fail-connect"]?.status).toBe("failed")
if (status["fail-connect"]?.status === "failed") {
expect(status["fail-connect"].error).toContain("Connection refused")
}
// No tools should be available
const tools = await MCP.tools()
expect(Object.keys(tools).length).toBe(0)
},
),
)
// ========================================================================
// Bug #5: McpOAuthCallback.cancelPending uses wrong key
// ========================================================================
test("McpOAuthCallback.cancelPending is keyed by mcpName but pendingAuths uses oauthState", async () => {
const { McpOAuthCallback } = await import("../../src/mcp/oauth-callback")
// Register a pending auth with an oauthState key, associated to an mcpName
const oauthState = "abc123hexstate"
const callbackPromise = McpOAuthCallback.waitForCallback(oauthState, "my-mcp-server")
// cancelPending is called with mcpName — should find the entry via reverse index
McpOAuthCallback.cancelPending("my-mcp-server")
// The callback should still be pending because cancelPending looked up
// "my-mcp-server" in a map keyed by "abc123hexstate"
let resolved = false
let rejected = false
callbackPromise.then(() => (resolved = true)).catch(() => (rejected = true))
// Give it a tick
await new Promise((r) => setTimeout(r, 50))
// cancelPending("my-mcp-server") should have rejected the pending callback
expect(rejected).toBe(true)
await McpOAuthCallback.stop()
})
// ========================================================================
// Test: multiple tools from same server get correct name prefixes
// ========================================================================
test(
"tools() prefixes tool names with sanitized server name",
withInstance(
{
"my.special-server": {
type: "local",
command: ["echo", "test"],
},
},
async () => {
lastCreatedClientName = "my.special-server"
const serverState = getOrCreateClientState("my.special-server")
serverState.tools = [
{ name: "tool-a", description: "Tool A", inputSchema: { type: "object", properties: {} } },
{ name: "tool.b", description: "Tool B", inputSchema: { type: "object", properties: {} } },
]
await MCP.add("my.special-server", {
type: "local",
command: ["echo", "test"],
})
const tools = await MCP.tools()
const keys = Object.keys(tools)
// Server name dots should be replaced with underscores
expect(keys.some((k) => k.startsWith("my_special-server_"))).toBe(true)
// Tool name dots should be replaced with underscores
expect(keys.some((k) => k.endsWith("tool_b"))).toBe(true)
expect(keys.length).toBe(2)
},
),
)

View File

@@ -2,7 +2,9 @@ import { $ } from "bun"
import { afterEach, describe, expect, test } from "bun:test"
import fs from "fs/promises"
import path from "path"
import { Effect, Layer, ManagedRuntime } from "effect"
import { tmpdir } from "../fixture/fixture"
import { watcherConfigLayer, withServices } from "../fixture/instance"
import { FileWatcher } from "../../src/file/watcher"
import { Instance } from "../../src/project/instance"
import { GlobalBus } from "../../src/bus/global"
@@ -15,26 +17,28 @@ const describeVcs = FileWatcher.hasNativeBinding() && !process.env.CI ? describe
// Helpers
// ---------------------------------------------------------------------------
async function withVcs(directory: string, body: () => Promise<void>) {
return Instance.provide({
function withVcs(
directory: string,
body: (rt: ManagedRuntime.ManagedRuntime<FileWatcher.Service | Vcs.Service, never>) => Promise<void>,
) {
return withServices(
directory,
fn: async () => {
FileWatcher.init()
Vcs.init()
Layer.merge(FileWatcher.layer, Vcs.defaultLayer),
async (rt) => {
await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
await rt.runPromise(Vcs.Service.use((s) => s.init()))
await Bun.sleep(500)
await body()
await body(rt)
},
})
{ provide: [watcherConfigLayer] },
)
}
function withVcsOnly(directory: string, body: () => Promise<void>) {
return Instance.provide({
directory,
fn: async () => {
Vcs.init()
await body()
},
})
function withVcsOnly(
directory: string,
body: (rt: ManagedRuntime.ManagedRuntime<Vcs.Service, never>) => Promise<void>,
) {
return withServices(directory, Vcs.defaultLayer, body)
}
type BranchEvent = { directory?: string; payload: { type: string; properties: { branch?: string } } }
@@ -78,8 +82,8 @@ describeVcs("Vcs", () => {
test("branch() returns current branch name", async () => {
await using tmp = await tmpdir({ git: true })
await withVcs(tmp.path, async () => {
const branch = await Vcs.branch()
await withVcs(tmp.path, async (rt) => {
const branch = await rt.runPromise(Vcs.Service.use((s) => s.branch()))
expect(branch).toBeDefined()
expect(typeof branch).toBe("string")
})
@@ -88,8 +92,8 @@ describeVcs("Vcs", () => {
test("branch() returns undefined for non-git directories", async () => {
await using tmp = await tmpdir()
await withVcs(tmp.path, async () => {
const branch = await Vcs.branch()
await withVcs(tmp.path, async (rt) => {
const branch = await rt.runPromise(Vcs.Service.use((s) => s.branch()))
expect(branch).toBeUndefined()
})
})
@@ -115,14 +119,14 @@ describeVcs("Vcs", () => {
const branch = `test-${Math.random().toString(36).slice(2)}`
await $`git branch ${branch}`.cwd(tmp.path).quiet()
await withVcs(tmp.path, async () => {
await withVcs(tmp.path, async (rt) => {
const pending = nextBranchUpdate(tmp.path)
const head = path.join(tmp.path, ".git", "HEAD")
await fs.writeFile(head, `ref: refs/heads/${branch}\n`)
await pending
const current = await Vcs.branch()
const current = await rt.runPromise(Vcs.Service.use((s) => s.branch()))
expect(current).toBe(branch)
})
})
@@ -137,8 +141,8 @@ describe("Vcs diff", () => {
await using tmp = await tmpdir({ git: true })
await $`git branch -M main`.cwd(tmp.path).quiet()
await withVcsOnly(tmp.path, async () => {
const branch = await Vcs.defaultBranch()
await withVcsOnly(tmp.path, async (rt) => {
const branch = await rt.runPromise(Vcs.Service.use((s) => s.defaultBranch()))
expect(branch).toBe("main")
})
})
@@ -148,8 +152,8 @@ describe("Vcs diff", () => {
await $`git branch -M trunk`.cwd(tmp.path).quiet()
await $`git config init.defaultBranch trunk`.cwd(tmp.path).quiet()
await withVcsOnly(tmp.path, async () => {
const branch = await Vcs.defaultBranch()
await withVcsOnly(tmp.path, async (rt) => {
const branch = await rt.runPromise(Vcs.Service.use((s) => s.defaultBranch()))
expect(branch).toBe("trunk")
})
})
@@ -161,8 +165,11 @@ describe("Vcs diff", () => {
const dir = path.join(wt.path, "feature")
await $`git worktree add -b feature/test ${dir} HEAD`.cwd(tmp.path).quiet()
await withVcsOnly(dir, async () => {
const [branch, base] = await Promise.all([Vcs.branch(), Vcs.defaultBranch()])
await withVcsOnly(dir, async (rt) => {
const [branch, base] = await Promise.all([
rt.runPromise(Vcs.Service.use((s) => s.branch())),
rt.runPromise(Vcs.Service.use((s) => s.defaultBranch())),
])
expect(branch).toBe("feature/test")
expect(base).toBe("main")
})
@@ -175,8 +182,8 @@ describe("Vcs diff", () => {
await $`git commit --no-gpg-sign -m "add file"`.cwd(tmp.path).quiet()
await fs.writeFile(path.join(tmp.path, "file.txt"), "changed\n", "utf-8")
await withVcsOnly(tmp.path, async () => {
const diff = await Vcs.diff("git")
await withVcsOnly(tmp.path, async (rt) => {
const diff = await rt.runPromise(Vcs.Service.use((s) => s.diff("git")))
expect(diff).toEqual(
expect.arrayContaining([
expect.objectContaining({
@@ -192,8 +199,8 @@ describe("Vcs diff", () => {
await using tmp = await tmpdir({ git: true })
await fs.writeFile(path.join(tmp.path, weird), "hello\n", "utf-8")
await withVcsOnly(tmp.path, async () => {
const diff = await Vcs.diff("git")
await withVcsOnly(tmp.path, async (rt) => {
const diff = await rt.runPromise(Vcs.Service.use((s) => s.diff("git")))
expect(diff).toEqual(
expect.arrayContaining([
expect.objectContaining({
@@ -213,8 +220,8 @@ describe("Vcs diff", () => {
await $`git add .`.cwd(tmp.path).quiet()
await $`git commit --no-gpg-sign -m "branch file"`.cwd(tmp.path).quiet()
await withVcsOnly(tmp.path, async () => {
const diff = await Vcs.diff("branch")
await withVcsOnly(tmp.path, async (rt) => {
const diff = await rt.runPromise(Vcs.Service.use((s) => s.diff("branch")))
expect(diff).toEqual(
expect.arrayContaining([
expect.objectContaining({

View File

@@ -110,16 +110,10 @@ describe("SyncEvent", () => {
type: string
properties: { id: string; name: string }
}> = []
const received = new Promise<void>((resolve) => {
Bus.subscribeAll((event) => {
events.push(event)
resolve()
})
})
const unsub = Bus.subscribeAll((event) => events.push(event))
SyncEvent.run(Created, { id: "evt_1", name: "test" })
await received
expect(events).toHaveLength(1)
expect(events[0]).toEqual({
type: "item.created",
@@ -128,6 +122,8 @@ describe("SyncEvent", () => {
name: "test",
},
})
unsub()
}),
)
})

View File

@@ -89,6 +89,7 @@ describe("tool.edit", () => {
const { FileWatcher } = await import("../../src/file/watcher")
const events: string[] = []
const unsubEdited = Bus.subscribe(File.Event.Edited, () => events.push("edited"))
const unsubUpdated = Bus.subscribe(FileWatcher.Event.Updated, () => events.push("updated"))
const edit = await EditTool.init()
@@ -101,7 +102,9 @@ describe("tool.edit", () => {
ctx,
)
expect(events).toContain("edited")
expect(events).toContain("updated")
unsubEdited()
unsubUpdated()
},
})
@@ -302,9 +305,11 @@ describe("tool.edit", () => {
await FileTime.read(ctx.sessionID, filepath)
const { Bus } = await import("../../src/bus")
const { File } = await import("../../src/file")
const { FileWatcher } = await import("../../src/file/watcher")
const events: string[] = []
const unsubEdited = Bus.subscribe(File.Event.Edited, () => events.push("edited"))
const unsubUpdated = Bus.subscribe(FileWatcher.Event.Updated, () => events.push("updated"))
const edit = await EditTool.init()
@@ -317,7 +322,9 @@ describe("tool.edit", () => {
ctx,
)
expect(events).toContain("edited")
expect(events).toContain("updated")
unsubEdited()
unsubUpdated()
},
})

View File

@@ -129,7 +129,6 @@ export type AuthOuathResult = { url: string; instructions: string } & (
access: string
expires: number
accountId?: string
enterpriseUrl?: string
}
| { key: string }
))
@@ -150,7 +149,6 @@ export type AuthOuathResult = { url: string; instructions: string } & (
access: string
expires: number
accountId?: string
enterpriseUrl?: string
}
| { key: string }
))