fix: Windows e2e stability (CrossSpawnSpawner, snapshot isolation, session race guards) (#19163)

This commit is contained in:
Kit Langton
2026-03-25 19:49:14 -04:00
committed by GitHub
parent 5179b87aef
commit 8864fdce2f
5 changed files with 246 additions and 172 deletions
+2 -1
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 serverPort = process.env.PLAYWRIGHT_SERVER_PORT ?? "4096"
const command = `bun run dev -- --host 0.0.0.0 --port ${port}` const command = `bun run dev -- --host 0.0.0.0 --port ${port}`
const reuse = !process.env.CI 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({ export default defineConfig({
testDir: "./e2e", testDir: "./e2e",
+3 -2
View File
@@ -1,4 +1,5 @@
import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node" import { NodeFileSystem, NodePath } from "@effect/platform-node"
import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
import { Effect, Layer, ServiceMap, Stream } from "effect" import { Effect, Layer, ServiceMap, Stream } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process" import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { makeRunPromise } from "@/effect/run-service" import { makeRunPromise } from "@/effect/run-service"
@@ -258,7 +259,7 @@ export namespace Git {
) )
export const defaultLayer = layer.pipe( export const defaultLayer = layer.pipe(
Layer.provide(NodeChildProcessSpawner.layer), Layer.provide(CrossSpawnSpawner.layer),
Layer.provide(NodeFileSystem.layer), Layer.provide(NodeFileSystem.layer),
Layer.provide(NodePath.layer), Layer.provide(NodePath.layer),
) )
+6 -1
View File
@@ -13,6 +13,7 @@ import { fn } from "@/util/fn"
import { Agent } from "@/agent/agent" import { Agent } from "@/agent/agent"
import { Plugin } from "@/plugin" import { Plugin } from "@/plugin"
import { Config } from "@/config/config" import { Config } from "@/config/config"
import { NotFoundError } from "@/storage/db"
import { ProviderTransform } from "@/provider/transform" import { ProviderTransform } from "@/provider/transform"
import { ModelID, ProviderID } from "@/provider/schema" import { ModelID, ProviderID } from "@/provider/schema"
@@ -60,7 +61,11 @@ export namespace SessionCompaction {
const config = await Config.get() const config = await Config.get()
if (config.compaction?.prune === false) return if (config.compaction?.prune === false) return
log.info("pruning") log.info("pruning")
const msgs = await Session.messages({ sessionID: input.sessionID }) const msgs = await Session.messages({ sessionID: input.sessionID }).catch((err) => {
if (NotFoundError.isInstance(err)) return undefined
throw err
})
if (!msgs) return
let total = 0 let total = 0
let pruned = 0 let pruned = 0
const toPrune = [] const toPrune = []
+38 -19
View File
@@ -4,6 +4,15 @@ import { Session } from "./index"
import { MessageV2 } from "./message-v2" import { MessageV2 } from "./message-v2"
import { SessionTable, MessageTable, PartTable } from "./session.sql" import { SessionTable, MessageTable, PartTable } from "./session.sql"
import { ProjectTable } from "../project/project.sql" import { ProjectTable } from "../project/project.sql"
import { Log } from "../util/log"
const log = Log.create({ service: "session.projector" })
function foreign(err: unknown) {
if (typeof err !== "object" || err === null) return false
if ("code" in err && err.code === "SQLITE_CONSTRAINT_FOREIGNKEY") return true
return "message" in err && typeof err.message === "string" && err.message.includes("FOREIGN KEY constraint failed")
}
export type DeepPartial<T> = T extends object ? { [K in keyof T]?: DeepPartial<T[K]> | null } : T export type DeepPartial<T> = T extends object ? { [K in keyof T]?: DeepPartial<T[K]> | null } : T
@@ -76,15 +85,20 @@ export default [
const time_created = data.info.time.created const time_created = data.info.time.created
const { id, sessionID, ...rest } = data.info const { id, sessionID, ...rest } = data.info
db.insert(MessageTable) try {
.values({ db.insert(MessageTable)
id, .values({
session_id: sessionID, id,
time_created, session_id: sessionID,
data: rest, time_created,
}) data: rest,
.onConflictDoUpdate({ target: MessageTable.id, set: { data: rest } }) })
.run() .onConflictDoUpdate({ target: MessageTable.id, set: { data: rest } })
.run()
} catch (err) {
if (!foreign(err)) throw err
log.warn("ignored late message update", { messageID: id, sessionID })
}
}), }),
SyncEvent.project(MessageV2.Event.Removed, (db, data) => { SyncEvent.project(MessageV2.Event.Removed, (db, data) => {
@@ -102,15 +116,20 @@ export default [
SyncEvent.project(MessageV2.Event.PartUpdated, (db, data) => { SyncEvent.project(MessageV2.Event.PartUpdated, (db, data) => {
const { id, messageID, sessionID, ...rest } = data.part const { id, messageID, sessionID, ...rest } = data.part
db.insert(PartTable) try {
.values({ db.insert(PartTable)
id, .values({
message_id: messageID, id,
session_id: sessionID, message_id: messageID,
time_created: data.time, session_id: sessionID,
data: rest, time_created: data.time,
}) data: rest,
.onConflictDoUpdate({ target: PartTable.id, set: { data: rest } }) })
.run() .onConflictDoUpdate({ target: PartTable.id, set: { data: rest } })
.run()
} catch (err) {
if (!foreign(err)) throw err
log.warn("ignored late part update", { partID: id, messageID, sessionID })
}
}), }),
] ]
+197 -149
View File
@@ -1,5 +1,5 @@
import { NodeFileSystem, NodePath } from "@effect/platform-node" import { NodeFileSystem, NodePath } from "@effect/platform-node"
import { Cause, Duration, Effect, Layer, Schedule, ServiceMap, Stream } from "effect" import { Cause, Duration, Effect, Layer, Schedule, Semaphore, ServiceMap, Stream } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process" import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import path from "path" import path from "path"
import z from "zod" import z from "zod"
@@ -7,6 +7,7 @@ import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
import { InstanceState } from "@/effect/instance-state" import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service" import { makeRunPromise } from "@/effect/run-service"
import { AppFileSystem } from "@/filesystem" import { AppFileSystem } from "@/filesystem"
import { Hash } from "@/util/hash"
import { Config } from "../config/config" import { Config } from "../config/config"
import { Global } from "../global" import { Global } from "../global"
import { Log } from "../util/log" import { Log } from "../util/log"
@@ -38,7 +39,6 @@ export namespace Snapshot {
const core = ["-c", "core.longpaths=true", "-c", "core.symlinks=true"] const core = ["-c", "core.longpaths=true", "-c", "core.symlinks=true"]
const cfg = ["-c", "core.autocrlf=false", ...core] const cfg = ["-c", "core.autocrlf=false", ...core]
const quote = [...cfg, "-c", "core.quotepath=false"] const quote = [...cfg, "-c", "core.quotepath=false"]
interface GitResult { interface GitResult {
readonly code: ChildProcessSpawner.ExitCode readonly code: ChildProcessSpawner.ExitCode
readonly text: string readonly text: string
@@ -66,12 +66,23 @@ export namespace Snapshot {
Effect.gen(function* () { Effect.gen(function* () {
const fs = yield* AppFileSystem.Service const fs = yield* AppFileSystem.Service
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
const locks = new Map<string, Semaphore.Semaphore>()
const lock = (key: string) => {
const hit = locks.get(key)
if (hit) return hit
const next = Semaphore.makeUnsafe(1)
locks.set(key, next)
return next
}
const state = yield* InstanceState.make<State>( const state = yield* InstanceState.make<State>(
Effect.fn("Snapshot.state")(function* (ctx) { Effect.fn("Snapshot.state")(function* (ctx) {
const state = { const state = {
directory: ctx.directory, directory: ctx.directory,
worktree: ctx.worktree, worktree: ctx.worktree,
gitdir: path.join(Global.Path.data, "snapshot", ctx.project.id), gitdir: path.join(Global.Path.data, "snapshot", ctx.project.id, Hash.fast(ctx.worktree)),
vcs: ctx.project.vcs, vcs: ctx.project.vcs,
} }
@@ -108,6 +119,7 @@ export namespace Snapshot {
const exists = (file: string) => fs.exists(file).pipe(Effect.orDie) const exists = (file: string) => fs.exists(file).pipe(Effect.orDie)
const read = (file: string) => fs.readFileString(file).pipe(Effect.catch(() => Effect.succeed(""))) const read = (file: string) => fs.readFileString(file).pipe(Effect.catch(() => Effect.succeed("")))
const remove = (file: string) => fs.remove(file).pipe(Effect.catch(() => Effect.void)) const remove = (file: string) => fs.remove(file).pipe(Effect.catch(() => Effect.void))
const locked = <A, E, R>(fx: Effect.Effect<A, E, R>) => lock(state.gitdir).withPermits(1)(fx)
const enabled = Effect.fnUntraced(function* () { const enabled = Effect.fnUntraced(function* () {
if (state.vcs !== "git") return false if (state.vcs !== "git") return false
@@ -190,175 +202,211 @@ export namespace Snapshot {
}) })
const cleanup = Effect.fnUntraced(function* () { const cleanup = Effect.fnUntraced(function* () {
if (!(yield* enabled())) return return yield* locked(
if (!(yield* exists(state.gitdir))) return Effect.gen(function* () {
const result = yield* git(args(["gc", `--prune=${prune}`]), { cwd: state.directory }) if (!(yield* enabled())) return
if (result.code !== 0) { if (!(yield* exists(state.gitdir))) return
log.warn("cleanup failed", { const result = yield* git(args(["gc", `--prune=${prune}`]), { cwd: state.directory })
exitCode: result.code, if (result.code !== 0) {
stderr: result.stderr, log.warn("cleanup failed", {
}) exitCode: result.code,
return stderr: result.stderr,
} })
log.info("cleanup", { prune }) return
}
log.info("cleanup", { prune })
}),
)
}) })
const track = Effect.fnUntraced(function* () { const track = Effect.fnUntraced(function* () {
if (!(yield* enabled())) return return yield* locked(
const existed = yield* exists(state.gitdir) Effect.gen(function* () {
yield* fs.ensureDir(state.gitdir).pipe(Effect.orDie) if (!(yield* enabled())) return
if (!existed) { const existed = yield* exists(state.gitdir)
yield* git(["init"], { yield* fs.ensureDir(state.gitdir).pipe(Effect.orDie)
env: { GIT_DIR: state.gitdir, GIT_WORK_TREE: state.worktree }, if (!existed) {
}) yield* git(["init"], {
yield* git(["--git-dir", state.gitdir, "config", "core.autocrlf", "false"]) env: { GIT_DIR: state.gitdir, GIT_WORK_TREE: state.worktree },
yield* git(["--git-dir", state.gitdir, "config", "core.longpaths", "true"]) })
yield* git(["--git-dir", state.gitdir, "config", "core.symlinks", "true"]) yield* git(["--git-dir", state.gitdir, "config", "core.autocrlf", "false"])
yield* git(["--git-dir", state.gitdir, "config", "core.fsmonitor", "false"]) yield* git(["--git-dir", state.gitdir, "config", "core.longpaths", "true"])
log.info("initialized") yield* git(["--git-dir", state.gitdir, "config", "core.symlinks", "true"])
} yield* git(["--git-dir", state.gitdir, "config", "core.fsmonitor", "false"])
yield* add() log.info("initialized")
const result = yield* git(args(["write-tree"]), { cwd: state.directory }) }
const hash = result.text.trim() yield* add()
log.info("tracking", { hash, cwd: state.directory, git: state.gitdir }) const result = yield* git(args(["write-tree"]), { cwd: state.directory })
return hash const hash = result.text.trim()
log.info("tracking", { hash, cwd: state.directory, git: state.gitdir })
return hash
}),
)
}) })
const patch = Effect.fnUntraced(function* (hash: string) { const patch = Effect.fnUntraced(function* (hash: string) {
yield* add() return yield* locked(
const result = yield* git( Effect.gen(function* () {
[...quote, ...args(["diff", "--cached", "--no-ext-diff", "--name-only", hash, "--", "."])], yield* add()
{ const result = yield* git(
cwd: state.directory, [...quote, ...args(["diff", "--cached", "--no-ext-diff", "--name-only", hash, "--", "."])],
}, {
cwd: state.directory,
},
)
if (result.code !== 0) {
log.warn("failed to get diff", { hash, exitCode: result.code })
return { hash, files: [] }
}
return {
hash,
files: result.text
.trim()
.split("\n")
.map((x) => x.trim())
.filter(Boolean)
.map((x) => path.join(state.worktree, x).replaceAll("\\", "/")),
}
}),
) )
if (result.code !== 0) {
log.warn("failed to get diff", { hash, exitCode: result.code })
return { hash, files: [] }
}
return {
hash,
files: result.text
.trim()
.split("\n")
.map((x) => x.trim())
.filter(Boolean)
.map((x) => path.join(state.worktree, x).replaceAll("\\", "/")),
}
}) })
const restore = Effect.fnUntraced(function* (snapshot: string) { const restore = Effect.fnUntraced(function* (snapshot: string) {
log.info("restore", { commit: snapshot }) return yield* locked(
const result = yield* git([...core, ...args(["read-tree", snapshot])], { cwd: state.worktree }) Effect.gen(function* () {
if (result.code === 0) { log.info("restore", { commit: snapshot })
const checkout = yield* git([...core, ...args(["checkout-index", "-a", "-f"])], { cwd: state.worktree }) const result = yield* git([...core, ...args(["read-tree", snapshot])], { cwd: state.worktree })
if (checkout.code === 0) return if (result.code === 0) {
log.error("failed to restore snapshot", { const checkout = yield* git([...core, ...args(["checkout-index", "-a", "-f"])], {
snapshot, cwd: state.worktree,
exitCode: checkout.code, })
stderr: checkout.stderr, if (checkout.code === 0) return
}) log.error("failed to restore snapshot", {
return snapshot,
} exitCode: checkout.code,
log.error("failed to restore snapshot", { stderr: checkout.stderr,
snapshot, })
exitCode: result.code, return
stderr: result.stderr, }
}) log.error("failed to restore snapshot", {
snapshot,
exitCode: result.code,
stderr: result.stderr,
})
}),
)
}) })
const revert = Effect.fnUntraced(function* (patches: Snapshot.Patch[]) { const revert = Effect.fnUntraced(function* (patches: Snapshot.Patch[]) {
const seen = new Set<string>() return yield* locked(
for (const item of patches) { Effect.gen(function* () {
for (const file of item.files) { const seen = new Set<string>()
if (seen.has(file)) continue for (const item of patches) {
seen.add(file) for (const file of item.files) {
log.info("reverting", { file, hash: item.hash }) if (seen.has(file)) continue
const result = yield* git([...core, ...args(["checkout", item.hash, "--", file])], { seen.add(file)
cwd: state.worktree, log.info("reverting", { file, hash: item.hash })
}) const result = yield* git([...core, ...args(["checkout", item.hash, "--", file])], {
if (result.code !== 0) { cwd: state.worktree,
const rel = path.relative(state.worktree, file) })
const tree = yield* git([...core, ...args(["ls-tree", item.hash, "--", rel])], { if (result.code !== 0) {
cwd: state.worktree, const rel = path.relative(state.worktree, file)
}) const tree = yield* git([...core, ...args(["ls-tree", item.hash, "--", rel])], {
if (tree.code === 0 && tree.text.trim()) { cwd: state.worktree,
log.info("file existed in snapshot but checkout failed, keeping", { file }) })
} else { if (tree.code === 0 && tree.text.trim()) {
log.info("file did not exist in snapshot, deleting", { file }) log.info("file existed in snapshot but checkout failed, keeping", { file })
yield* remove(file) } else {
log.info("file did not exist in snapshot, deleting", { file })
yield* remove(file)
}
}
} }
} }
} }),
} )
}) })
const diff = Effect.fnUntraced(function* (hash: string) { const diff = Effect.fnUntraced(function* (hash: string) {
yield* add() return yield* locked(
const result = yield* git([...quote, ...args(["diff", "--cached", "--no-ext-diff", hash, "--", "."])], { Effect.gen(function* () {
cwd: state.worktree, yield* add()
}) const result = yield* git(
if (result.code !== 0) { [...quote, ...args(["diff", "--cached", "--no-ext-diff", hash, "--", "."])],
log.warn("failed to get diff", { {
hash, cwd: state.worktree,
exitCode: result.code, },
stderr: result.stderr, )
}) if (result.code !== 0) {
return "" log.warn("failed to get diff", {
} hash,
return result.text.trim() exitCode: result.code,
stderr: result.stderr,
})
return ""
}
return result.text.trim()
}),
)
}) })
const diffFull = Effect.fnUntraced(function* (from: string, to: string) { const diffFull = Effect.fnUntraced(function* (from: string, to: string) {
const result: Snapshot.FileDiff[] = [] return yield* locked(
const status = new Map<string, "added" | "deleted" | "modified">() Effect.gen(function* () {
const result: Snapshot.FileDiff[] = []
const status = new Map<string, "added" | "deleted" | "modified">()
const statuses = yield* git( const statuses = yield* git(
[...quote, ...args(["diff", "--no-ext-diff", "--name-status", "--no-renames", from, to, "--", "."])], [
{ cwd: state.directory }, ...quote,
...args(["diff", "--no-ext-diff", "--name-status", "--no-renames", from, to, "--", "."]),
],
{ cwd: state.directory },
)
for (const line of statuses.text.trim().split("\n")) {
if (!line) continue
const [code, file] = line.split("\t")
if (!code || !file) continue
status.set(file, code.startsWith("A") ? "added" : code.startsWith("D") ? "deleted" : "modified")
}
const numstat = yield* git(
[...quote, ...args(["diff", "--no-ext-diff", "--no-renames", "--numstat", from, to, "--", "."])],
{
cwd: state.directory,
},
)
for (const line of numstat.text.trim().split("\n")) {
if (!line) continue
const [adds, dels, file] = line.split("\t")
if (!file) continue
const binary = adds === "-" && dels === "-"
const [before, after] = binary
? ["", ""]
: yield* Effect.all(
[
git([...cfg, ...args(["show", `${from}:${file}`])]).pipe(Effect.map((item) => item.text)),
git([...cfg, ...args(["show", `${to}:${file}`])]).pipe(Effect.map((item) => item.text)),
],
{ concurrency: 2 },
)
const additions = binary ? 0 : parseInt(adds)
const deletions = binary ? 0 : parseInt(dels)
result.push({
file,
before,
after,
additions: Number.isFinite(additions) ? additions : 0,
deletions: Number.isFinite(deletions) ? deletions : 0,
status: status.get(file) ?? "modified",
})
}
return result
}),
) )
for (const line of statuses.text.trim().split("\n")) {
if (!line) continue
const [code, file] = line.split("\t")
if (!code || !file) continue
status.set(file, code.startsWith("A") ? "added" : code.startsWith("D") ? "deleted" : "modified")
}
const numstat = yield* git(
[...quote, ...args(["diff", "--no-ext-diff", "--no-renames", "--numstat", from, to, "--", "."])],
{
cwd: state.directory,
},
)
for (const line of numstat.text.trim().split("\n")) {
if (!line) continue
const [adds, dels, file] = line.split("\t")
if (!file) continue
const binary = adds === "-" && dels === "-"
const [before, after] = binary
? ["", ""]
: yield* Effect.all(
[
git([...cfg, ...args(["show", `${from}:${file}`])]).pipe(Effect.map((item) => item.text)),
git([...cfg, ...args(["show", `${to}:${file}`])]).pipe(Effect.map((item) => item.text)),
],
{ concurrency: 2 },
)
const additions = binary ? 0 : parseInt(adds)
const deletions = binary ? 0 : parseInt(dels)
result.push({
file,
before,
after,
additions: Number.isFinite(additions) ? additions : 0,
deletions: Number.isFinite(deletions) ? deletions : 0,
status: status.get(file) ?? "modified",
})
}
return result
}) })
yield* cleanup().pipe( yield* cleanup().pipe(