Compare commits

..

2 Commits

Author SHA1 Message Date
Kit Langton bc3617bbe6 test(core): simplify mutation lock coverage 2026-08-07 23:44:17 -04:00
Kit Langton b850cee756 fix(core): unify file mutation transaction locks 2026-08-07 23:39:06 -04:00
5 changed files with 160 additions and 166 deletions
+30 -41
View File
@@ -48,15 +48,13 @@ export const readText = Effect.fn("FileMutation.readText")(function* (files: Fil
return Bom.decodeBytes((yield* files.read(target)).bytes)
})
export const syncTextBom = Effect.fn("FileMutation.syncTextBom")(function* (
files: Files,
target: string,
bom: boolean,
) {
const synced = Bom.syncBytes((yield* files.read(target)).bytes, bom)
if (synced.bytes) yield* files.write(target, synced.bytes)
return synced.text
})
export const syncTextBom = Effect.fn("FileMutation.syncTextBom")((files: Files, target: string, bom: boolean) =>
Effect.gen(function* () {
const synced = Bom.syncBytes((yield* files.read(target)).bytes, bom)
if (synced.bytes) yield* files.write(target, synced.bytes)
return synced.text
}).pipe(Effect.uninterruptible),
)
/** Share transaction locks across Location graphs that address the same file. */
const transactionLocks = KeyedMutex.makeUnsafe<string>()
@@ -70,15 +68,10 @@ const layer = Layer.effect(
Service,
Effect.gen(function* () {
const environment = yield* Environment.Service
const locks = KeyedMutex.makeUnsafe<string>()
const withLock: Interface["withLock"] = (targets) => (effect) =>
[...new Set(targets.map(FSUtil.resolve))]
.sort()
.reduceRight((result, target) => transactionLocks.withLock(target)(result), effect)
const withTargetLock =
(target: Target) =>
<A, E, R>(effect: Effect.Effect<A, E, R>) =>
locks.withLock(target.absolute)(Effect.uninterruptible(effect))
const writeResult = (target: Target, existed: boolean): WriteResult => ({
operation: "write",
@@ -88,36 +81,32 @@ const layer = Layer.effect(
})
const write = Effect.fn("FileMutation.write")((input: WriteInput) =>
withTargetLock(input.target)(
Effect.gen(function* () {
const existed = yield* environment.files.stat(input.target.absolute).pipe(
Effect.as(true),
Effect.catchTag("Environment.NotFound", () => Effect.succeed(false)),
)
yield* environment.files.write(
input.target.absolute,
typeof input.content === "string" ? new TextEncoder().encode(input.content) : input.content,
)
return writeResult(input.target, existed)
}),
),
Effect.gen(function* () {
const existed = yield* environment.files.stat(input.target.absolute).pipe(
Effect.as(true),
Effect.catchTag("Environment.NotFound", () => Effect.succeed(false)),
)
yield* environment.files.write(
input.target.absolute,
typeof input.content === "string" ? new TextEncoder().encode(input.content) : input.content,
)
return writeResult(input.target, existed)
}).pipe(Effect.uninterruptible),
)
const writeTextPreservingBom = Effect.fn("FileMutation.writeTextPreservingBom")((input: TextWriteInput) =>
withTargetLock(input.target)(
Effect.gen(function* () {
const next = Bom.split(input.content)
const current = yield* environment.files.read(input.target.absolute, { offset: 0, length: 3 }).pipe(
Effect.map((result) => result.bytes),
Effect.catchTag("Environment.NotFound", () => Effect.succeed(undefined)),
)
yield* environment.files.write(
input.target.absolute,
new TextEncoder().encode(Bom.join(next.text, Boolean(current && Bom.has(current)) || next.bom)),
)
return writeResult(input.target, current !== undefined)
}),
),
Effect.gen(function* () {
const next = Bom.split(input.content)
const current = yield* environment.files.read(input.target.absolute, { offset: 0, length: 3 }).pipe(
Effect.map((result) => result.bytes),
Effect.catchTag("Environment.NotFound", () => Effect.succeed(undefined)),
)
yield* environment.files.write(
input.target.absolute,
new TextEncoder().encode(Bom.join(next.text, Boolean(current && Bom.has(current)) || next.bom)),
)
return writeResult(input.target, current !== undefined)
}).pipe(Effect.uninterruptible),
)
return Service.of({ withLock, write, writeTextPreservingBom })
+2
View File
@@ -1,5 +1,6 @@
export * as PluginHooks from "./hooks"
import type { AISDKHooks } from "@opencode-ai/plugin/effect/aisdk"
import type { SessionHooks } from "@opencode-ai/plugin/effect/session"
import type { ShellHooks } from "@opencode-ai/plugin/effect/shell"
import type { ToolHooks } from "@opencode-ai/plugin/effect/tool"
@@ -8,6 +9,7 @@ import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { State } from "../state"
export interface Domains {
readonly aisdk: AISDKHooks
readonly session: SessionHooks
readonly shell: ShellHooks
readonly tool: ToolHooks
+36 -27
View File
@@ -69,34 +69,43 @@ export const Plugin = {
id: context.id,
}
const target = yield* mutation.resolve({ path: input.path, kind: "file" })
const external = target.externalDirectory
if (external)
yield* permission.assert({
...LocationMutation.externalDirectoryPermission(external),
sessionID: context.sessionID,
agent: context.agent,
source,
})
const current = yield* FileMutation.readText(environment.files, target.absolute).pipe(
Effect.catchTag("Environment.NotFound", () => Effect.succeed(undefined)),
return yield* fileMutation.withLock([target.absolute])(
Effect.gen(function* () {
const external = target.externalDirectory
if (external)
yield* permission.assert({
...LocationMutation.externalDirectoryPermission(external),
sessionID: context.sessionID,
agent: context.agent,
source,
})
const current = yield* FileMutation.readText(environment.files, target.absolute).pipe(
Effect.catchTag("Environment.NotFound", () => Effect.succeed(undefined)),
)
const next = Bom.split(input.content)
const preview = fileDiff(
target.resource,
current?.text ?? "",
next.text,
current ? "modified" : "added",
)
yield* permission.assert({
action: "edit",
resources: [target.resource],
save: ["*"],
metadata: { files: [preview] },
sessionID: context.sessionID,
agent: context.agent,
source,
})
const result = yield* fileMutation.writeTextPreservingBom({ target, content: input.content })
const bom = (yield* FileMutation.readText(environment.files, target.absolute)).bom
if (yield* formatter.file(target.absolute)) {
yield* FileMutation.syncTextBom(environment.files, target.absolute, bom)
}
return result
}),
)
const next = Bom.split(input.content)
const preview = fileDiff(target.resource, current?.text ?? "", next.text, current ? "modified" : "added")
yield* permission.assert({
action: "edit",
resources: [target.resource],
save: ["*"],
metadata: { files: [preview] },
sessionID: context.sessionID,
agent: context.agent,
source,
})
const result = yield* fileMutation.writeTextPreservingBom({ target, content: input.content })
const bom = (yield* FileMutation.readText(environment.files, target.absolute)).bom
if (yield* formatter.file(target.absolute)) {
yield* FileMutation.syncTextBom(environment.files, target.absolute, bom)
}
return result
}).pipe(
Effect.map((output) => ({ output, content: toModelOutput(output) })),
Effect.mapError((error) => new ToolFailure({ message: `Unable to write ${input.path}`, error })),
-95
View File
@@ -109,49 +109,6 @@ describe("FileMutation", () => {
),
)
it.live("serializes concurrent writes to the same absolute target", () =>
withTmp((directory) =>
Effect.gen(function* () {
const targetPath = path.join(directory, "shared.txt")
yield* Effect.promise(() => fs.writeFile(targetPath, "initial"))
const firstStarted = yield* Deferred.make<void>()
const releaseFirst = yield* Deferred.make<void>()
const secondStarted = yield* Deferred.make<void>()
let writes = 0
const filesystem = instrumentWrites((write) =>
Effect.gen(function* () {
writes++
if (writes === 1) {
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(releaseFirst)
} else {
yield* Deferred.succeed(secondStarted, undefined)
}
yield* write
}),
)
yield* Effect.gen(function* () {
const mutation = yield* LocationMutation.Service
const files = yield* FileMutation.Service
const firstPlan = yield* mutation.resolve({ path: "shared.txt" })
const secondPlan = yield* mutation.resolve({ path: "shared.txt" })
const first = yield* files.write({ target: firstPlan, content: "first" }).pipe(Effect.forkChild)
yield* Deferred.await(firstStarted)
const second = yield* files.write({ target: secondPlan, content: "second" }).pipe(Effect.forkChild)
yield* Effect.yieldNow
expect(yield* Deferred.isDone(secondStarted)).toBe(false)
yield* Deferred.succeed(releaseFirst, undefined)
yield* Deferred.await(secondStarted)
yield* Fiber.join(first)
yield* Fiber.join(second)
expect(yield* Effect.promise(() => fs.readFile(targetPath, "utf8"))).toBe("second")
}).pipe(provide(directory, filesystem))
}),
),
)
it.live("shares transaction locks across Location service instances", () =>
withTmp((directory) =>
Effect.gen(function* () {
@@ -202,56 +159,4 @@ describe("FileMutation", () => {
}).pipe(provide(directory)),
),
)
it.live("allows distinct absolute targets to proceed independently", () =>
withTmp((directory) =>
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const releaseFirst = yield* Deferred.make<void>()
const secondFinished = yield* Deferred.make<void>()
const secondPath = path.join(directory, "second.txt")
let writes = 0
const filesystem = instrumentWrites((write) =>
++writes === 1
? Deferred.succeed(firstStarted, undefined).pipe(
Effect.andThen(Deferred.await(releaseFirst)),
Effect.andThen(write),
)
: write.pipe(Effect.andThen(Deferred.succeed(secondFinished, undefined))),
)
yield* Effect.gen(function* () {
const mutation = yield* LocationMutation.Service
const files = yield* FileMutation.Service
const firstPlan = yield* mutation.resolve({ path: "first.txt" })
const secondPlan = yield* mutation.resolve({ path: "second.txt" })
const first = yield* files.write({ target: firstPlan, content: "first" }).pipe(Effect.forkChild)
yield* Deferred.await(firstStarted)
const second = yield* files.write({ target: secondPlan, content: "second" }).pipe(Effect.forkChild)
yield* Deferred.await(secondFinished)
expect(yield* Effect.promise(() => fs.readFile(secondPath, "utf8"))).toBe("second")
yield* Deferred.succeed(releaseFirst, undefined)
yield* Fiber.join(first)
yield* Fiber.join(second)
}).pipe(provide(directory, filesystem))
}),
),
)
})
function instrumentWrites(run: <E>(write: Effect.Effect<void, E>, target: string) => Effect.Effect<void, E>) {
return Layer.effect(
Environment.Service,
Effect.gen(function* () {
const environment = yield* Environment.Service
return Environment.Service.of({
...environment,
files: {
...environment.files,
write: (target, content) => run(environment.files.write(target, content), target),
},
})
}),
).pipe(Layer.provide(LayerNode.compile(Environment.node)))
}
+92 -3
View File
@@ -1,7 +1,7 @@
import fs from "fs/promises"
import path from "path"
import { describe, expect } from "bun:test"
import { Effect, Layer } from "effect"
import { Deferred, Effect, Fiber, Layer } from "effect"
import { FileMutation } from "@opencode-ai/core/file-mutation"
import { Formatter } from "@opencode-ai/core/formatter"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
@@ -13,6 +13,7 @@ import { Permission } from "@opencode-ai/core/permission"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { Session } from "@opencode-ai/core/session"
import { Tool } from "@opencode-ai/core/tool"
import { EditTool } from "@opencode-ai/core/tool/plugin/edit"
import { WriteTool } from "@opencode-ai/core/tool/plugin/write"
import { location } from "./fixture/location"
import { tmpdir } from "./fixture/tmpdir"
@@ -26,10 +27,25 @@ const writeToolNode = makeLocationNode({
deps: [Tool.node, LocationMutation.node, FileMutation.node, Environment.node, Formatter.node, Permission.node],
})
const editToolNode = makeLocationNode({
name: "test/edit-tool-plugin",
layer: Layer.effectDiscard(registerToolPlugin(EditTool.Plugin)),
deps: [
Tool.node,
LocationMutation.node,
FileMutation.node,
Environment.node,
Formatter.node,
Location.node,
Permission.node,
],
})
const sessionID = Session.ID.make("ses_write_tool_test")
const assertions: Permission.AssertInput[] = []
const writes: string[] = []
let formatFile = (_target: string): Effect.Effect<boolean> => Effect.succeed(false)
let afterPermission = (_input: Permission.AssertInput): Effect.Effect<void> => Effect.void
let denyAction: string | undefined
const permission = Layer.succeed(
@@ -37,6 +53,7 @@ const permission = Layer.succeed(
Permission.Service.of({
assert: (input) =>
Effect.sync(() => assertions.push(input)).pipe(
Effect.andThen(Effect.suspend(() => afterPermission(input))),
Effect.andThen(
input.action === denyAction
? Effect.fail(
@@ -65,6 +82,7 @@ const reset = () => {
assertions.length = 0
writes.length = 0
formatFile = () => Effect.succeed(false)
afterPermission = () => Effect.void
denyAction = undefined
}
@@ -83,7 +101,11 @@ const environment = Layer.effect(
}),
).pipe(Layer.provide(LayerNode.compile(Environment.node)))
const withTool = <A, E, R>(directory: string, body: (registry: Tool.Interface) => Effect.Effect<A, E, R>) => {
const withTool = <A, E, R>(
directory: string,
body: (registry: Tool.Interface) => Effect.Effect<A, E, R>,
options?: { edit?: boolean },
) => {
const activeLocation = Layer.succeed(
Location.Service,
Location.Service.of(location({ directory: AbsolutePath.make(directory) })),
@@ -93,7 +115,13 @@ const withTool = <A, E, R>(directory: string, body: (registry: Tool.Interface) =
}).pipe(
Effect.provide(
AppNodeBuilder.build(
LayerNode.group([Tool.node, Tool.node, LocationMutation.node, FileMutation.node, writeToolNode]),
LayerNode.group([
Tool.node,
LocationMutation.node,
FileMutation.node,
writeToolNode,
...(options?.edit ? [editToolNode] : []),
]),
[
[Environment.node, environment],
[Location.node, activeLocation],
@@ -111,6 +139,12 @@ const call = (input: typeof WriteTool.Input.Type, id = "call-write") => ({
call: { type: "tool-call" as const, id, name: "write", input },
})
const editCall = (input: typeof EditTool.Input.Type, id = "call-edit") => ({
sessionID,
...toolIdentity,
call: { type: "tool-call" as const, id, name: "edit", input },
})
const it = testEffect(Layer.empty)
describe("WriteTool", () => {
@@ -420,4 +454,59 @@ describe("WriteTool", () => {
),
),
)
it.live("serializes write and edit transactions across Location service instances", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
(tmp) => {
reset()
const target = path.join(tmp.path, "shared.txt")
return Effect.gen(function* () {
yield* Effect.promise(() => fs.writeFile(target, "initial"))
const formatting = yield* Deferred.make<void>()
const releaseFormatting = yield* Deferred.make<void>()
const editApproved = yield* Deferred.make<void>()
let formats = 0
formatFile = () =>
++formats === 1
? Deferred.succeed(formatting, undefined).pipe(
Effect.andThen(Deferred.await(releaseFormatting)),
Effect.as(false),
)
: Effect.succeed(false)
afterPermission = (input) =>
input.source?.id === "call-serialized-edit" && input.action === "edit"
? Deferred.succeed(editApproved, undefined).pipe(Effect.asVoid)
: Effect.void
const write = yield* withTool(
tmp.path,
(registry) =>
executeTool(registry, call({ path: "shared.txt", content: "before" }, "call-serialized-write")),
{ edit: true },
).pipe(Effect.forkChild)
yield* Deferred.await(formatting)
const edit = yield* withTool(
tmp.path,
(registry) =>
executeTool(
registry,
editCall({ path: "shared.txt", oldString: "before", newString: "after" }, "call-serialized-edit"),
),
{ edit: true },
).pipe(Effect.forkChild)
yield* Effect.yieldNow
expect(yield* Deferred.isDone(editApproved)).toBe(false)
yield* Deferred.succeed(releaseFormatting, undefined)
expect((yield* Fiber.join(write)).status).toBe("completed")
expect((yield* Fiber.join(edit)).status).toBe("completed")
expect(yield* Deferred.isDone(editApproved)).toBe(true)
expect(yield* Effect.promise(() => fs.readFile(target, "utf8"))).toBe("after")
})
},
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
})