refactor(core): move exec tools onto environment (#41095)

This commit is contained in:
Kit Langton
2026-08-07 11:59:59 -04:00
committed by GitHub
parent db31c42e39
commit 917d6449e3
9 changed files with 424 additions and 399 deletions
+16
View File
@@ -51,4 +51,20 @@ export interface FilesImpl {
export interface Files extends FilesImpl {}
/**
* Derives a follow-stat kind from the lstat-like Files contract. A dangling
* symlink fails with `NotFound`.
*/
export const typeFollowing = (files: Files, path: string) =>
files.stat(path).pipe(
Effect.flatMap((info) =>
info.type === "symlink"
? files.read(path, { offset: 0, length: 0 }).pipe(
Effect.map((result) => result.info.type),
Effect.catchTag("Environment.WrongKind", (error) => Effect.succeed(error.actual)),
)
: Effect.succeed(info.type),
),
)
export * as EnvironmentFiles from "./files"
+1
View File
@@ -9,6 +9,7 @@ export {
type FilesImpl,
type FileType,
NotFound,
typeFollowing,
WrongKind,
} from "./files"
export { execDefaults } from "./exec-defaults"
+7 -5
View File
@@ -3,8 +3,9 @@ export * as Ripgrep from "./ripgrep"
import { Context, Effect, Fiber, Layer, Schema, Stream } from "effect"
import { ChildProcess } from "effect/unstable/process"
import { Entry, Match } from "@opencode-ai/schema/filesystem"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { AppProcess, collectStream, waitForAbort } from "@opencode-ai/util/process"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { collectStream, waitForAbort } from "@opencode-ai/util/process"
import { Environment } from "./environment"
import { NonNegativeInt, PositiveInt, RelativePath } from "./schema"
import { RipgrepBinary } from "./ripgrep/binary"
@@ -93,7 +94,7 @@ const isInvalidPattern = (stderr: string) =>
const layer = Layer.effect(
Service,
Effect.gen(function* () {
const process = yield* AppProcess.Service
const environment = yield* Environment.Service
const binary = yield* RipgrepBinary.Service
const run = <A>(input: {
@@ -107,7 +108,8 @@ const layer = Layer.effect(
}) => {
const program = Effect.scoped(
Effect.gen(function* () {
const handle = yield* process.spawn(
// Hosted environments will resolve rg through their driver image; the spawner is the execution seam.
const handle = yield* environment.spawner.spawn(
ChildProcess.make(yield* binary.filepath, input.args, { cwd: input.cwd, extendEnv: true, stdin: "ignore" }),
)
const stderrFiber = yield* collectStream(handle.stderr, ERROR_BYTES).pipe(
@@ -275,4 +277,4 @@ const layer = Layer.effect(
}),
)
export const node = makeGlobalNode({ service: Service, layer: layer, deps: [RipgrepBinary.node, AppProcess.node] })
export const node = makeLocationNode({ service: Service, layer, deps: [Environment.node, RipgrepBinary.node] })
+257 -258
View File
@@ -6,9 +6,9 @@ import { ChildProcess } from "effect/unstable/process"
import { produce } from "immer"
import { Shell } from "@opencode-ai/schema/shell"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { AppProcess } from "@opencode-ai/util/process"
import { Config } from "./config"
import { Bus } from "./bus"
import { Environment } from "./environment"
import { Location } from "./location"
import { Global } from "@opencode-ai/util/global"
import { ShellSelect } from "./shell/select"
@@ -65,285 +65,284 @@ export interface Interface {
export class Service extends Context.Service<Service, Interface>()("@opencode/Shell") {}
export const layer = (options?: ShellSelect.Options) => Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const location = yield* Location.Service
const config = yield* Config.Service
const global = yield* Global.Service
const appProcess = yield* AppProcess.Service
const hooks = yield* PluginHooks.Service
const context = yield* Effect.context()
const runFork = Effect.runForkWith(context)
const sessions = new Map<string, Active>()
const exitOrder: string[] = []
export const layer = (options?: ShellSelect.Options) =>
Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const location = yield* Location.Service
const config = yield* Config.Service
const global = yield* Global.Service
const environment = yield* Environment.Service
const hooks = yield* PluginHooks.Service
const context = yield* Effect.context()
const runFork = Effect.runForkWith(context)
const sessions = new Map<string, Active>()
const exitOrder: string[] = []
const outputDir = path.join(global.data, "shell", location.project.id)
const { mkdir, unlink } = yield* Effect.promise(() => import("fs/promises"))
const { createWriteStream, createReadStream } = yield* Effect.promise(() => import("fs"))
yield* Effect.promise(() => mkdir(outputDir, { recursive: true }))
const outputDir = path.join(global.data, "shell", location.project.id)
const { mkdir, unlink } = yield* Effect.promise(() => import("fs/promises"))
const { createWriteStream, createReadStream } = yield* Effect.promise(() => import("fs"))
yield* Effect.promise(() => mkdir(outputDir, { recursive: true }))
yield* Effect.addFinalizer(() =>
Effect.gen(function* () {
for (const session of sessions.values()) {
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
// Unblock waiters still pending at teardown; succeed is a no-op once already resolved.
yield* Deferred.fail(session.done, new NotFoundError({ id: Shell.ID.make(session.info.id) }))
yield* Effect.addFinalizer(() =>
Effect.gen(function* () {
for (const session of sessions.values()) {
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
// Unblock waiters still pending at teardown; succeed is a no-op once already resolved.
yield* Deferred.fail(session.done, new NotFoundError({ id: Shell.ID.make(session.info.id) }))
}
sessions.clear()
exitOrder.length = 0
}),
)
const require = Effect.fn("Shell.require")(function* (id: Shell.ID) {
const session = sessions.get(id)
if (!session) return yield* new NotFoundError({ id })
return session
})
const removeSession = Effect.fnUntraced(function* (id: Shell.ID) {
const session = sessions.get(id)
if (!session) return
sessions.delete(id)
const index = exitOrder.indexOf(id)
if (index !== -1) exitOrder.splice(index, 1)
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
// Unblock any wait still pending when the command is removed before it terminated.
yield* Deferred.fail(session.done, new NotFoundError({ id }))
yield* Effect.promise(() => unlink(session.file).catch(() => {}))
yield* bus.publish(Shell.Event.Deleted, { id })
})
const remove = Effect.fn("Shell.remove")(function* (id: Shell.ID) {
yield* require(id)
yield* removeSession(id)
})
const list = Effect.fn("Shell.list")(function* () {
return Array.from(sessions.values())
.filter((session) => session.info.status === "running")
.map((session) => session.info)
})
const get = Effect.fn("Shell.get")(function* (id: Shell.ID) {
return (yield* require(id)).info
})
const wait = Effect.fn("Shell.wait")(function* (id: Shell.ID) {
return yield* Deferred.await((yield* require(id)).done)
})
const timeout = Effect.fn("Shell.timeout")(function* (id: Shell.ID, duration: number) {
const session = yield* require(id)
if (session.info.status !== "running" || !session.timeout) return session.info
yield* session.timeout(duration)
return session.info
})
const resolve = () =>
config.entries().pipe(Effect.map((entries) => ShellSelect.preferred(Config.latest(entries, "shell"), options)))
const name = () => resolve().pipe(Effect.map(ShellSelect.name))
const output = Effect.fn("Shell.output")(function* (id: Shell.ID, input?: Shell.OutputInput) {
const session = yield* require(id)
const cursor = input?.cursor ?? 0
const limit = input?.limit ?? 65536
if (cursor >= session.size) return { output: "", cursor: session.size, size: session.size, truncated: false }
const start = Math.max(0, cursor)
const length = Math.min(limit, session.size - start)
const buffer = Buffer.alloc(length)
const bytesRead = yield* Effect.promise(
() =>
new Promise<number>((resolve) => {
const stream = createReadStream(session.file, { start, end: start + length - 1 })
let offset = 0
stream.on("data", (chunk: string | Buffer) => {
const bytes = Buffer.from(chunk)
bytes.copy(buffer, offset)
offset += bytes.length
})
stream.on("end", () => resolve(offset))
stream.on("error", () => resolve(0))
}),
)
return {
output: buffer.subarray(0, bytesRead).toString("utf8"),
cursor: start + bytesRead,
size: session.size,
truncated: false,
}
sessions.clear()
exitOrder.length = 0
}),
)
})
const require = Effect.fn("Shell.require")(function* (id: Shell.ID) {
const session = sessions.get(id)
if (!session) return yield* new NotFoundError({ id })
return session
})
const create = Effect.fn("Shell.create")(function* <E = never, R = never>(
input: Shell.CreateInput,
before?: (input: ShellCreateBefore) => Effect.Effect<void, E, R>,
) {
const invocation: ShellCreateBefore = {
command: input.command,
cwd: input.cwd ?? location.directory,
timeout: input.timeout,
shell: yield* resolve(),
env: {
...process.env,
TERM: "xterm-256color",
OPENCODE_TERMINAL: "1",
},
}
yield* hooks.trigger("shell", "create.before", invocation)
if (before) yield* before(invocation)
const removeSession = Effect.fnUntraced(function* (id: Shell.ID) {
const session = sessions.get(id)
if (!session) return
sessions.delete(id)
const index = exitOrder.indexOf(id)
if (index !== -1) exitOrder.splice(index, 1)
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
// Unblock any wait still pending when the command is removed before it terminated.
yield* Deferred.fail(session.done, new NotFoundError({ id }))
yield* Effect.promise(() => unlink(session.file).catch(() => {}))
yield* bus.publish(Shell.Event.Deleted, { id })
})
const id = Shell.ID.ascending()
const args = ShellSelect.args(invocation.shell, invocation.command)
const file = path.join(outputDir, `${id}.out`)
const remove = Effect.fn("Shell.remove")(function* (id: Shell.ID) {
yield* require(id)
yield* removeSession(id)
})
const info: Info = {
id,
status: "running",
command: invocation.command,
cwd: invocation.cwd,
shell: invocation.shell,
file,
metadata: input.metadata ?? {},
time: { started: Date.now() },
}
const list = Effect.fn("Shell.list")(function* () {
return Array.from(sessions.values())
.filter((session) => session.info.status === "running")
.map((session) => session.info)
})
const get = Effect.fn("Shell.get")(function* (id: Shell.ID) {
return (yield* require(id)).info
})
const wait = Effect.fn("Shell.wait")(function* (id: Shell.ID) {
return yield* Deferred.await((yield* require(id)).done)
})
const timeout = Effect.fn("Shell.timeout")(function* (id: Shell.ID, duration: number) {
const session = yield* require(id)
if (session.info.status !== "running" || !session.timeout) return session.info
yield* session.timeout(duration)
return session.info
})
const resolve = () =>
config
.entries()
.pipe(Effect.map((entries) => ShellSelect.preferred(Config.latest(entries, "shell"), options)))
const name = () => resolve().pipe(Effect.map(ShellSelect.name))
const output = Effect.fn("Shell.output")(function* (id: Shell.ID, input?: Shell.OutputInput) {
const session = yield* require(id)
const cursor = input?.cursor ?? 0
const limit = input?.limit ?? 65536
if (cursor >= session.size) return { output: "", cursor: session.size, size: session.size, truncated: false }
const start = Math.max(0, cursor)
const length = Math.min(limit, session.size - start)
const buffer = Buffer.alloc(length)
const bytesRead = yield* Effect.promise(
() =>
new Promise<number>((resolve) => {
const stream = createReadStream(session.file, { start, end: start + length - 1 })
let offset = 0
stream.on("data", (chunk: string | Buffer) => {
const bytes = Buffer.from(chunk)
bytes.copy(buffer, offset)
offset += bytes.length
})
stream.on("end", () => resolve(offset))
stream.on("error", () => resolve(0))
}),
)
return {
output: buffer.subarray(0, bytesRead).toString("utf8"),
cursor: start + bytesRead,
size: session.size,
truncated: false,
}
})
const create = Effect.fn("Shell.create")(function* <E = never, R = never>(
input: Shell.CreateInput,
before?: (input: ShellCreateBefore) => Effect.Effect<void, E, R>,
) {
const invocation: ShellCreateBefore = {
command: input.command,
cwd: input.cwd ?? location.directory,
timeout: input.timeout,
shell: yield* resolve(),
env: {
...process.env,
TERM: "xterm-256color",
OPENCODE_TERMINAL: "1",
},
}
yield* hooks.trigger("shell", "create.before", invocation)
if (before) yield* before(invocation)
const id = Shell.ID.ascending()
const args = ShellSelect.args(invocation.shell, invocation.command)
const file = path.join(outputDir, `${id}.out`)
const info: Info = {
id,
status: "running",
command: invocation.command,
cwd: invocation.cwd,
shell: invocation.shell,
file,
metadata: input.metadata ?? {},
time: { started: Date.now() },
}
// Spawn via AppProcess and stream combined output to the file. The handle is scope-bound, so
// the managing fiber keeps its scope open until the command terminates (it awaits `done` at the
// end). `create` returns once `ready` resolves with the registered session.
const ready = Deferred.makeUnsafe<Active>()
runFork(
Effect.scoped(
Effect.gen(function* () {
const handle = yield* appProcess.spawn(
ChildProcess.make(invocation.shell, args, {
cwd: invocation.cwd,
env: invocation.env,
stdin: "ignore",
detached: process.platform !== "win32",
forceKillAfter: Duration.seconds(3),
}),
)
const session: Active = {
info: produce(info, (draft) => {
draft.pid = handle.pid
}),
file,
size: 0,
done: Deferred.makeUnsafe<Info, NotFoundError>(),
}
sessions.set(id, session)
const stream = createWriteStream(file)
const outputDone = Deferred.makeUnsafe<void>()
const pump = handle.all.pipe(
Stream.runForEach((chunk: Uint8Array) =>
Effect.sync(() => {
stream.write(chunk)
session.size += chunk.length
// Spawn through the Environment and stream combined output to the file. The handle is scope-bound, so
// the managing fiber keeps its scope open until the command terminates (it awaits `done` at the
// end). `create` returns once `ready` resolves with the registered session.
const ready = Deferred.makeUnsafe<Active>()
runFork(
Effect.scoped(
Effect.gen(function* () {
const handle = yield* environment.spawner.spawn(
ChildProcess.make(invocation.shell, args, {
cwd: invocation.cwd,
env: invocation.env,
stdin: "ignore",
detached: process.platform !== "win32",
forceKillAfter: Duration.seconds(3),
}),
),
)
runFork(
Effect.gen(function* () {
yield* pump.pipe(Effect.catch(() => Effect.void))
yield* Effect.promise(
() =>
new Promise<void>((resolve) => {
stream.end(() => resolve())
}),
)
yield* Deferred.succeed(outputDone, undefined)
}).pipe(Effect.catch(() => Deferred.succeed(outputDone, undefined))),
)
yield* Effect.promise(
() =>
new Promise<void>((resolve) => {
stream.once("open", () => resolve())
stream.once("error", () => resolve())
)
const session: Active = {
info: produce(info, (draft) => {
draft.pid = handle.pid
}),
)
file,
size: 0,
done: Deferred.makeUnsafe<Info, NotFoundError>(),
}
sessions.set(id, session)
const finish = (status: Info["status"], exit?: number, beforeWait = Effect.void) =>
Effect.gen(function* () {
if (session.info.status !== "running") return
session.info = produce(session.info, (draft) => {
draft.status = status
if (exit !== undefined) draft.exit = exit
draft.time.completed = Date.now()
})
yield* beforeWait
yield* Deferred.await(outputDone)
// Resolve waiters with the terminal Info before any retention eviction, so an evicted
// session still reports success rather than the removal NotFoundError. This runs before
// the timeout-fiber interrupt below, which on the timeout path would otherwise cancel
// this very fiber (finish is invoked by the timeout fiber) before waiters are resolved.
yield* Deferred.succeed(session.done, session.info)
yield* bus.publish(Shell.Event.Exited, {
id,
...(exit !== undefined ? { exit } : {}),
status,
})
exitOrder.push(id)
while (exitOrder.length > EXITED_LIMIT) {
const oldest = exitOrder[0]
if (!oldest) break
yield* removeSession(Shell.ID.make(oldest))
}
// Cancel a pending timeout once the command exits on its own. Interrupting last avoids
// aborting finish when finish itself runs on the timeout fiber.
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
})
const stream = createWriteStream(file)
const outputDone = Deferred.makeUnsafe<void>()
const pump = handle.all.pipe(
Stream.runForEach((chunk: Uint8Array) =>
Effect.sync(() => {
stream.write(chunk)
session.size += chunk.length
}),
),
)
runFork(
Effect.gen(function* () {
yield* pump.pipe(Effect.catch(() => Effect.void))
yield* Effect.promise(
() =>
new Promise<void>((resolve) => {
stream.end(() => resolve())
}),
)
yield* Deferred.succeed(outputDone, undefined)
}).pipe(Effect.catch(() => Deferred.succeed(outputDone, undefined))),
)
yield* Effect.promise(
() =>
new Promise<void>((resolve) => {
stream.once("open", () => resolve())
stream.once("error", () => resolve())
}),
)
session.timeout = (duration) =>
Effect.gen(function* () {
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
session.timeoutFiber = undefined
if (duration === 0 || session.info.status !== "running") return
session.timeoutFiber = runFork(
Effect.sleep(Duration.millis(duration)).pipe(
Effect.flatMap(() =>
finish("timeout", undefined, handle.kill().pipe(Effect.catch(() => Effect.void))),
const finish = (status: Info["status"], exit?: number, beforeWait = Effect.void) =>
Effect.gen(function* () {
if (session.info.status !== "running") return
session.info = produce(session.info, (draft) => {
draft.status = status
if (exit !== undefined) draft.exit = exit
draft.time.completed = Date.now()
})
yield* beforeWait
yield* Deferred.await(outputDone)
// Resolve waiters with the terminal Info before any retention eviction, so an evicted
// session still reports success rather than the removal NotFoundError. This runs before
// the timeout-fiber interrupt below, which on the timeout path would otherwise cancel
// this very fiber (finish is invoked by the timeout fiber) before waiters are resolved.
yield* Deferred.succeed(session.done, session.info)
yield* bus.publish(Shell.Event.Exited, {
id,
...(exit !== undefined ? { exit } : {}),
status,
})
exitOrder.push(id)
while (exitOrder.length > EXITED_LIMIT) {
const oldest = exitOrder[0]
if (!oldest) break
yield* removeSession(Shell.ID.make(oldest))
}
// Cancel a pending timeout once the command exits on its own. Interrupting last avoids
// aborting finish when finish itself runs on the timeout fiber.
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
})
session.timeout = (duration) =>
Effect.gen(function* () {
if (session.timeoutFiber) yield* Fiber.interrupt(session.timeoutFiber)
session.timeoutFiber = undefined
if (duration === 0 || session.info.status !== "running") return
session.timeoutFiber = runFork(
Effect.sleep(Duration.millis(duration)).pipe(
Effect.flatMap(() =>
finish("timeout", undefined, handle.kill().pipe(Effect.catch(() => Effect.void))),
),
Effect.catch(() => Effect.void),
),
Effect.catch(() => Effect.void),
),
)
})
)
})
yield* session.timeout(invocation.timeout)
yield* session.timeout(invocation.timeout)
runFork(
handle.exitCode.pipe(
Effect.flatMap((code) => finish("exited", code)),
Effect.catch(() => Effect.void),
),
)
runFork(
handle.exitCode.pipe(
Effect.flatMap((code) => finish("exited", code)),
Effect.catch(() => Effect.void),
),
)
yield* bus.publish(Shell.Event.Created, { info })
yield* Deferred.succeed(ready, session)
// Hold the handle's scope open until the command terminates; closing it earlier would
// release (kill) the process before its exit is observed.
yield* Deferred.await(session.done).pipe(Effect.catch(() => Effect.void))
}),
).pipe(Effect.catch(() => Effect.void)),
)
yield* bus.publish(Shell.Event.Created, { info })
yield* Deferred.succeed(ready, session)
// Hold the handle's scope open until the command terminates; closing it earlier would
// release (kill) the process before its exit is observed.
yield* Deferred.await(session.done).pipe(Effect.catch(() => Effect.void))
}),
).pipe(Effect.catch(() => Effect.void)),
)
const session = yield* Deferred.await(ready)
return session.info
})
const session = yield* Deferred.await(ready)
return session.info
})
return Service.of({ name, create, list, get, wait, timeout, output, remove })
}),
)
return Service.of({ name, create, list, get, wait, timeout, output, remove })
}),
)
export function configured(options?: ShellSelect.Options) {
return makeLocationNode({
service: Service,
layer: layer(options),
deps: [Bus.node, Location.node, Config.node, Global.node, AppProcess.node, PluginHooks.node],
deps: [Bus.node, Location.node, Config.node, Global.node, Environment.node, PluginHooks.node],
})
}
+10 -12
View File
@@ -4,8 +4,8 @@ import { ToolFailure } from "@opencode-ai/ai"
import type { Context as PluginContext } from "@opencode-ai/plugin/effect/plugin"
import { Effect, Schema } from "effect"
import path from "path"
import { Environment } from "../../environment"
import { FileSystem } from "../../filesystem"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Location } from "../../location"
import { LocationMutation } from "../../location-mutation"
import { Ripgrep } from "../../ripgrep"
@@ -42,7 +42,7 @@ export const toModelContent = (entries: EncodedOutput, truncated = false) => {
export const Plugin = {
id: "opencode.tool.glob",
effect: Effect.fn("GlobTool.Plugin")(function* (ctx: PluginContext) {
const fs = yield* FSUtil.Service
const environment = yield* Environment.Service
const ripgrep = yield* Ripgrep.Service
const location = yield* Location.Service
const mutation = yield* LocationMutation.Service
@@ -82,22 +82,20 @@ export const Plugin = {
agent: context.agent,
source,
})
const info = yield* fs
.stat(target.absolute)
.pipe(
Effect.catchReason("PlatformError", "NotFound", () =>
Effect.fail(new ToolFailure({ message: `Search path does not exist: ${searchPath ?? "."}` })),
),
)
if (info.type !== "Directory")
const type = yield* Environment.typeFollowing(environment.files, target.absolute).pipe(
Effect.catchTag("Environment.NotFound", () =>
Effect.fail(new ToolFailure({ message: `Search path does not exist: ${searchPath ?? "."}` })),
),
)
if (type !== "directory")
return yield* Effect.fail(
new ToolFailure({ message: `Search path is not a directory: ${searchPath ?? "."}` }),
)
const root = path.resolve(location.directory, searchPath ?? ".")
const root = target.absolute
const limit = input.limit ?? FileSystem.DEFAULT_SEARCH_LIMIT
const entries = yield* ripgrep
.glob({
cwd: target.absolute,
cwd: root,
pattern: input.pattern,
limit: limit + 1,
})
+90 -94
View File
@@ -4,8 +4,8 @@ import type { Context as PluginContext } from "@opencode-ai/plugin/effect/plugin
import { ToolFailure } from "@opencode-ai/ai"
import { Effect, Schema } from "effect"
import path from "path"
import { Environment } from "../../environment"
import { FileSystem } from "../../filesystem"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Location } from "../../location"
import { LocationMutation } from "../../location-mutation"
import { Permission } from "../../permission"
@@ -15,11 +15,11 @@ import { RelativePath } from "../../schema"
export const name = "grep"
export const Input = Schema.Struct({
pattern: FileSystem.GrepInput.fields.pattern.check(
Schema.isMinLength(1, { message: "Pattern must not be empty" }),
).annotate({
description: "Regular expression to search for in file contents (ripgrep syntax)",
}),
pattern: FileSystem.GrepInput.fields.pattern
.check(Schema.isMinLength(1, { message: "Pattern must not be empty" }))
.annotate({
description: "Regular expression to search for in file contents (ripgrep syntax)",
}),
path: Schema.optionalKey(RelativePath).annotate({
description: "File or directory to search. Defaults to the current working directory.",
}),
@@ -58,7 +58,7 @@ export const toModelContent = (matches: EncodedOutput, truncated = false) => {
export const Plugin = {
id: "opencode.tool.grep",
effect: Effect.fn("GrepTool.Plugin")(function* (ctx: PluginContext) {
const fs = yield* FSUtil.Service
const environment = yield* Environment.Service
const ripgrep = yield* Ripgrep.Service
const location = yield* Location.Service
const mutation = yield* LocationMutation.Service
@@ -66,104 +66,100 @@ export const Plugin = {
yield* ctx.tool
.transform((draft) =>
draft.add(
({
name,
options: { codemode: false },
description:
"Search file contents using regular expressions. Use it to locate specific code, symbols, or text patterns, and narrow searches with `path` or `include`. Returns matching file paths, line numbers, and line previews.",
input: Input,
output: Output,
execute: (input, context) =>
Effect.gen(function* () {
const source = { type: "tool" as const, messageID: context.messageID, id: context.id }
const target = yield* mutation.resolve({ path: input.path ?? "." })
if (target.externalDirectory)
yield* permission.assert({
...LocationMutation.externalDirectoryPermission(target.externalDirectory),
sessionID: context.sessionID,
agent: context.agent,
source,
})
draft.add({
name,
options: { codemode: false },
description:
"Search file contents using regular expressions. Use it to locate specific code, symbols, or text patterns, and narrow searches with `path` or `include`. Returns matching file paths, line numbers, and line previews.",
input: Input,
output: Output,
execute: (input, context) =>
Effect.gen(function* () {
const source = { type: "tool" as const, messageID: context.messageID, id: context.id }
const target = yield* mutation.resolve({ path: input.path ?? "." })
if (target.externalDirectory)
yield* permission.assert({
action: name,
resources: [input.pattern],
save: ["*"],
metadata: {
root: ".",
path: input.path,
include: input.include,
limit: input.limit,
},
...LocationMutation.externalDirectoryPermission(target.externalDirectory),
sessionID: context.sessionID,
agent: context.agent,
source,
})
const root = path.resolve(location.directory, input.path ?? ".")
const info = yield* fs
.stat(root)
.pipe(
Effect.catchReason("PlatformError", "NotFound", () =>
Effect.fail(new ToolFailure({ message: `Search path does not exist: ${input.path ?? "."}` })),
),
)
const cwd = info?.type === "Directory" ? root : path.dirname(root)
const limit = input.limit ?? FileSystem.DEFAULT_SEARCH_LIMIT
const matches = yield* ripgrep
.grep({
cwd,
pattern: input.pattern,
file: info?.type === "File" ? path.basename(root) : undefined,
include: input.include,
limit: limit + 1,
})
.pipe(
Effect.timeoutOrElse({
duration: FileSystem.DEFAULT_SEARCH_TIMEOUT_MS,
orElse: () =>
Effect.fail(
new ToolFailure({
message: `Search timed out after ${FileSystem.DEFAULT_SEARCH_TIMEOUT_MS / 1_000} seconds. Consider using a more specific path or pattern.`,
}),
),
}),
Effect.map((result) =>
result.map((match) =>
FileSystem.Match.make({
...match,
entry: FileSystem.Entry.make({
...match.entry,
path: RelativePath.make(
path.relative(location.directory, path.resolve(cwd, match.entry.path)),
),
}),
yield* permission.assert({
action: name,
resources: [input.pattern],
save: ["*"],
metadata: {
root: ".",
path: input.path,
include: input.include,
limit: input.limit,
},
sessionID: context.sessionID,
agent: context.agent,
source,
})
const root = target.absolute
const type = yield* Environment.typeFollowing(environment.files, root).pipe(
Effect.catchTag("Environment.NotFound", () =>
Effect.fail(new ToolFailure({ message: `Search path does not exist: ${input.path ?? "."}` })),
),
)
const cwd = type === "directory" ? root : path.dirname(root)
const limit = input.limit ?? FileSystem.DEFAULT_SEARCH_LIMIT
const matches = yield* ripgrep
.grep({
cwd,
pattern: input.pattern,
file: type === "file" ? path.basename(root) : undefined,
include: input.include,
limit: limit + 1,
})
.pipe(
Effect.timeoutOrElse({
duration: FileSystem.DEFAULT_SEARCH_TIMEOUT_MS,
orElse: () =>
Effect.fail(
new ToolFailure({
message: `Search timed out after ${FileSystem.DEFAULT_SEARCH_TIMEOUT_MS / 1_000} seconds. Consider using a more specific path or pattern.`,
}),
),
}),
Effect.map((result) =>
result.map((match) =>
FileSystem.Match.make({
...match,
entry: FileSystem.Entry.make({
...match.entry,
path: RelativePath.make(
path.relative(location.directory, path.resolve(cwd, match.entry.path)),
),
}),
}),
),
)
return { matches: matches.slice(0, limit), truncated: matches.length > limit }
}).pipe(
Effect.map((result) => ({
output: result.matches,
content: toModelContent(
result.matches.map((match) => ({
...match,
entry: { ...match.entry, path: path.resolve(location.directory, match.entry.path) },
})),
result.truncated,
),
metadata: { matches: result.matches.length, truncated: result.truncated },
})),
Effect.mapError((error) =>
error instanceof ToolFailure
? error
: error instanceof Ripgrep.InvalidPatternError
? new ToolFailure({ message: `Invalid regex pattern: ${error.message}` })
: new ToolFailure({ message: `Unable to grep for ${input.pattern}`, error }),
)
return { matches: matches.slice(0, limit), truncated: matches.length > limit }
}).pipe(
Effect.map((result) => ({
output: result.matches,
content: toModelContent(
result.matches.map((match) => ({
...match,
entry: { ...match.entry, path: path.resolve(location.directory, match.entry.path) },
})),
result.truncated,
),
metadata: { matches: result.matches.length, truncated: result.truncated },
})),
Effect.mapError((error) =>
error instanceof ToolFailure
? error
: error instanceof Ripgrep.InvalidPatternError
? new ToolFailure({ message: `Invalid regex pattern: ${error.message}` })
: new ToolFailure({ message: `Unable to grep for ${input.pattern}`, error }),
),
}),
),
),
}),
)
.pipe(Effect.orDie)
}),
+8 -10
View File
@@ -5,8 +5,8 @@ import { ToolFailure } from "@opencode-ai/ai"
import type { Content } from "@opencode-ai/schema/tool"
import type { Context as PluginContext } from "@opencode-ai/plugin/effect/plugin"
import { Deferred, Effect, Schema, Scope } from "effect"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Config } from "../../config"
import { Environment } from "../../environment"
import { LocationMutation } from "../../location-mutation"
import { Permission } from "../../permission"
import { PluginRuntime } from "../../plugin/runtime"
@@ -83,7 +83,7 @@ export const Plugin = {
effect: Effect.fn("ShellTool.Plugin")(function* (ctx: PluginContext) {
const runtime = yield* PluginRuntime.Service
const scope = yield* Scope.Scope
const fsUtil = yield* FSUtil.Service
const environment = yield* Environment.Service
const mutation = yield* LocationMutation.Service
const shell = yield* Shell.Service
const permission = yield* Permission.Service
@@ -179,14 +179,12 @@ export const Plugin = {
agent: context.agent,
source,
})
const workdir = yield* fsUtil
.stat(target.absolute)
.pipe(
Effect.catchReason("PlatformError", "NotFound", () =>
Effect.fail(new Error(`Working directory does not exist: ${target.absolute}`)),
),
)
if (workdir.type !== "Directory")
const workdir = yield* Environment.typeFollowing(environment.files, target.absolute).pipe(
Effect.catchTag("Environment.NotFound", () =>
Effect.fail(new Error(`Working directory does not exist: ${target.absolute}`)),
),
)
if (workdir !== "directory")
return yield* Effect.fail(new Error(`Working directory is not a directory: ${target.absolute}`))
}),
)
+29 -1
View File
@@ -1,11 +1,39 @@
import fs from "node:fs/promises"
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { ChildProcessSpawner } from "effect/unstable/process"
import { CrossSpawnSpawner } from "@opencode-ai/util/cross-spawn-spawner"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { execDefaults, Failed, makeFiles, makeLocalDriver, makeMemoryDriver } from "../src/environment/index"
import {
execDefaults,
Failed,
makeFiles,
makeLocalDriver,
makeMemoryDriver,
NotFound,
typeFollowing,
} from "../src/environment/index"
import { tmpdir } from "./fixture/tmpdir"
import { environmentConformance } from "./lib/environment-conformance"
import { it } from "./lib/effect"
describe("typeFollowing", () => {
it.effect("follows symlinks without changing stat semantics", () =>
Effect.gen(function* () {
const driver = makeMemoryDriver()
const files = makeFiles(driver)
yield* files.mkdir("/directory")
yield* files.write("/file", new Uint8Array())
yield* driver.symlink("/directory", "/directory-link")
yield* driver.symlink("/file", "/file-link")
yield* driver.symlink("/missing", "/dangling-link")
expect(yield* typeFollowing(files, "/directory-link")).toBe("directory")
expect(yield* typeFollowing(files, "/file-link")).toBe("file")
expect(yield* typeFollowing(files, "/dangling-link").pipe(Effect.flip)).toBeInstanceOf(NotFound)
}),
)
})
environmentConformance("memory environment", () =>
Effect.sync(() => {
+6 -19
View File
@@ -5,8 +5,8 @@ import { Effect, Layer } from "effect"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Environment } from "@opencode-ai/core/environment"
import { FileSystem } from "@opencode-ai/core/filesystem"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Location } from "@opencode-ai/core/location"
import { LocationMutation } from "@opencode-ai/core/location-mutation"
import { Permission } from "@opencode-ai/core/permission"
@@ -24,19 +24,12 @@ import { executeTool, registerToolPlugin, toolIdentity } from "./lib/tool"
const globToolNode = makeLocationNode({
name: "test/glob-tool-plugin",
layer: Layer.effectDiscard(registerToolPlugin(GlobTool.Plugin)),
deps: [
Tool.node,
FSUtil.node,
Ripgrep.node,
Location.node,
LocationMutation.node,
Permission.node,
],
deps: [Tool.node, Environment.node, Ripgrep.node, Location.node, LocationMutation.node, Permission.node],
})
const grepToolNode = makeLocationNode({
name: "test/grep-tool-plugin",
layer: Layer.effectDiscard(registerToolPlugin(GrepTool.Plugin)),
deps: [Tool.node, FSUtil.node, Ripgrep.node, Location.node, LocationMutation.node, Permission.node],
deps: [Tool.node, Environment.node, Ripgrep.node, Location.node, LocationMutation.node, Permission.node],
})
const sessionID = Session.ID.make("ses_search_tool_test")
@@ -186,9 +179,7 @@ describe("search tools", () => {
Effect.promise(() => tmpdir()),
(tmp) =>
Effect.promise(() => fs.writeFile(path.join(tmp.path, "file.txt"), "haystack\n")).pipe(
Effect.andThen(
withTools(tmp.path, (registry) => executeTool(registry, call("grep", { pattern: "needle" }))),
),
Effect.andThen(withTools(tmp.path, (registry) => executeTool(registry, call("grep", { pattern: "needle" })))),
Effect.tap((result) =>
Effect.sync(() => {
expect(result).toMatchObject({
@@ -297,9 +288,7 @@ describe("search tools", () => {
(tmp) =>
Effect.promise(() => fs.writeFile(path.join(tmp.path, "file.txt"), "content\n")).pipe(
Effect.andThen(
withTools(tmp.path, (registry) =>
executeTool(registry, call("glob", { path: "file.txt", pattern: "*" })),
),
withTools(tmp.path, (registry) => executeTool(registry, call("glob", { path: "file.txt", pattern: "*" }))),
),
Effect.tap((result) =>
Effect.sync(() => {
@@ -331,9 +320,7 @@ describe("search tools", () => {
Effect.sync(() => {
expect(result.status).toBe("completed")
expect(assertions.map((input) => input.action)).toEqual(["external_directory", "glob"])
expect(assertions[0]?.resources).toEqual([
path.join(outside.path, "*").replaceAll("\\", "/"),
])
expect(assertions[0]?.resources).toEqual([path.join(outside.path, "*").replaceAll("\\", "/")])
}),
),
)