Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 126 additions & 0 deletions apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@ import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as PlatformError from "effect/PlatformError";
import * as Queue from "effect/Queue";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";
Expand Down Expand Up @@ -152,6 +154,7 @@ describe("ThreadPullRequestServiceV2 reads", () => {
SecureRandom.make((size) => new Uint8Array(size).fill(1)),
),
FileSystem.layerNoop({}),
Path.layer,
);
yield* Effect.gen(function* () {
const service = yield* ThreadPullRequestService.make;
Expand All @@ -176,6 +179,127 @@ describe("ThreadPullRequestServiceV2 reads", () => {
),
);

it.effect.each([
{
worktree: "a linked worktree",
existingPaths: ["/workspace/worktree", "/workspace/worktree/.git"],
unreadablePaths: [],
cwd: "/workspace/worktree",
},
// A folder that outlived its .git makes git fail with not_a_repository.
{
worktree: "a folder without .git",
existingPaths: ["/workspace/worktree"],
unreadablePaths: [],
cwd: "/workspace/project",
},
{
worktree: "a folder whose .git cannot be checked",
existingPaths: ["/workspace/worktree"],
unreadablePaths: ["/workspace/worktree/.git"],
cwd: "/workspace/worktree",
},
{
worktree: "a missing folder",
existingPaths: [],
unreadablePaths: [],
cwd: "/workspace/project",
},
] as const)(
"looks up a branch whose worktree is $worktree in $cwd",
({ existingPaths, unreadablePaths, cwd }) =>
Effect.scoped(
Effect.gen(function* () {
const thread = {
...threadShell("worktree-thread"),
branch: "feature/worktree",
worktreePath: "/workspace/worktree",
};
const project: OrchestrationProjectShell = {
id: thread.projectId,
title: "Project",
workspaceRoot: "/workspace/project",
defaultModelSelection: null,
scripts: [],
createdAt: "2026-08-01T00:00:00.000Z",
updatedAt: "2026-08-01T00:00:00.000Z",
};
const activation = yield* Deferred.make<void>();
const sweepStarted = yield* Deferred.make<void>();
const lookupCwds: Array<string> = [];
const dependencies = Layer.mergeAll(
Layer.mock(Orchestrator.OrchestratorV2)({
streamDomainEvents: Stream.never,
getShellSnapshot: () =>
Deferred.succeed(sweepStarted, undefined).pipe(
Effect.as({
schemaVersion: 2,
snapshotSequence: 1,
threads: [thread],
archivedThreads: [],
}),
),
}),
Layer.mock(ProjectStore.ProjectStoreV2)({
listShells: () => Effect.succeed([project]),
getShell: () => Effect.succeedSome(project),
}),
Layer.mock(GitManager.GitManager)({
branchPullRequest: (input) =>
Effect.sync(() => {
lookupCwds.push(input.cwd);
return null;
}),
}),
Layer.mock(PullRequestService.PullRequestService)({}),
Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
resolve: () =>
Effect.succeed({
canonicalKey: "github.com/acme/project",
locator: {
source: "git-remote" as const,
remoteName: "origin",
remoteUrl: "git@github.com:acme/project.git",
},
provider: SourceControlProviderKind.make("github"),
owner: "acme",
name: "project",
}),
}),
Layer.mock(BackgroundPolicy.BackgroundPolicy)({}),
Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
Layer.succeed(
SecureRandom.SecureRandom,
SecureRandom.make((size) => new Uint8Array(size).fill(1)),
),
FileSystem.layerNoop({
exists: (path) =>
unreadablePaths.some((entry) => entry === path)
? Effect.fail(
PlatformError.systemError({
_tag: "PermissionDenied",
module: "FileSystem",
method: "exists",
pathOrDescriptor: path,
}),
)
: Effect.succeed(existingPaths.some((entry) => entry === path)),
}),
Path.layer,
);
yield* Effect.gen(function* () {
const service = yield* ThreadPullRequestService.make;
yield* service.start();
yield* Deferred.succeed(activation, undefined);
yield* Deferred.await(sweepStarted);
yield* service.drain;
}).pipe(Effect.provide(dependencies));

expect(lookupCwds).toEqual([cwd]);
}),
),
);

it.effect("leaves delegated child threads to their own refresh instead of the sweep", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down Expand Up @@ -264,6 +388,7 @@ describe("ThreadPullRequestServiceV2 reads", () => {
SecureRandom.make((size) => new Uint8Array(size).fill(1)),
),
FileSystem.layerNoop({}),
Path.layer,
);
yield* Effect.gen(function* () {
const service = yield* ThreadPullRequestService.make;
Expand Down Expand Up @@ -387,6 +512,7 @@ describe("ThreadPullRequestServiceV2 reads", () => {
SecureRandom.make((size) => new Uint8Array(size).fill(1)),
),
FileSystem.layerNoop({}),
Path.layer,
);
yield* Effect.gen(function* () {
const service = yield* ThreadPullRequestService.make;
Expand Down
11 changes: 8 additions & 3 deletions apps/server/src/orchestration-v2/ThreadPullRequestService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import * as SecureRandom from "effect/crypto/SecureRandom";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as Schedule from "effect/Schedule";
import type * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
Expand Down Expand Up @@ -102,6 +103,7 @@ export const make = Effect.gen(function* () {
const repositoryIdentities = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
const random = yield* SecureRandom.SecureRandom;
const fileSystem = yield* FileSystem.FileSystem;
const path = yield* Path.Path;
const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
const pendingBackfill = new Map<ThreadId, number>();

Expand Down Expand Up @@ -188,10 +190,13 @@ export const make = Effect.gen(function* () {
refresh: request.refresh,
});
if (first.branch !== null && repository === null) return finishBackfill(group);
const worktreeExists =
first.worktreePath !== null && (yield* fileSystem.exists(first.worktreePath));
// A worktree folder can outlive its .git, and git there fails with not_a_repository.
// An unreadable .git keeps the worktree rather than failing the whole group.
const cwd =
worktreeExists && first.worktreePath !== null
first.worktreePath !== null &&
(yield* fileSystem
.exists(path.join(first.worktreePath, ".git"))
.pipe(Effect.orElseSucceed(() => true)))
? first.worktreePath
: project.workspaceRoot;
const detected =
Expand Down
127 changes: 126 additions & 1 deletion apps/server/src/orchestration-v2/ThreadSettlementService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Path from "effect/Path";
import * as PlatformError from "effect/PlatformError";
import * as PubSub from "effect/PubSub";
import * as Queue from "effect/Queue";
import * as Ref from "effect/Ref";
Expand Down Expand Up @@ -553,6 +555,7 @@ interface HarnessOptions {
readonly branchPullRequest?: GitManager.GitManager["Service"]["branchPullRequest"];
readonly pullRequestSummary?: PullRequestService.PullRequestService["Service"]["summary"];
readonly existingWorktreePaths?: ReadonlyArray<string>;
readonly unreadablePaths?: ReadonlyArray<string>;
readonly onDispatch?: (command: AutoSettleCommand) => Effect.Effect<void>;
readonly onScriptRun?: (
input: ProjectSetupScriptRunner.ProjectSetupScriptRunnerInput,
Expand Down Expand Up @@ -702,8 +705,19 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
Layer.succeed(SecureRandom.SecureRandom, testCrypto),
FileSystem.layerNoop({
exists: (path) => Effect.succeed(options.existingWorktreePaths?.includes(path) ?? false),
exists: (path) =>
options.unreadablePaths?.includes(path)
? Effect.fail(
PlatformError.systemError({
_tag: "PermissionDenied",
module: "FileSystem",
method: "exists",
pathOrDescriptor: path,
}),
)
: Effect.succeed(options.existingWorktreePaths?.includes(path) ?? false),
}),
Path.layer,
);

return {
Expand Down Expand Up @@ -897,6 +911,117 @@ describe("ThreadSettlementServiceV2 worker", () => {
),
);

it.effect("looks up a branch in its worktree only while the worktree has .git", () =>
Effect.scoped(
Effect.gen(function* () {
yield* TestClock.setTime(Date.parse(NOW));
const fixture = yield* makeHarness({
snapshot: makeSnapshot([
makeThread("linked-worktree", {
branch: "linked",
worktreePath: "/worktrees/linked",
latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"),
}),
// A folder that outlived its .git makes git fail with not_a_repository.
makeThread("emptied-worktree", {
branch: "emptied",
worktreePath: "/worktrees/emptied",
latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"),
}),
]),
existingWorktreePaths: [
"/worktrees/linked",
"/worktrees/linked/.git",
"/worktrees/emptied",
],
});

yield* Effect.gen(function* () {
const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2;
yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
const cwdByBranch = Object.fromEntries(
(yield* Ref.get(fixture.branchCalls)).map((call) => [call.branch, call.cwd]),
);
expect(cwdByBranch).toEqual({
linked: "/worktrees/linked",
emptied: "/workspace/project",
});
}).pipe(Effect.provide(fixture.layer));
}),
),
);

it.effect("does not settle a thread whose worktree cannot be checked", () =>
Effect.scoped(
Effect.gen(function* () {
yield* TestClock.setTime(Date.parse(NOW));
const fixture = yield* makeHarness({
snapshot: makeSnapshot(
[
makeThread("unreadable-worktree", {
branch: "unreadable",
worktreePath: "/worktrees/unreadable",
latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"),
}),
// A merged linked pull request still needs the branch recheck before settling.
makeThread("unreadable-linked-worktree", {
branch: "feature",
worktreePath: "/worktrees/unreadable-linked",
latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"),
linkedPullRequest: {
projectId: PROJECT_ID,
repository: "owner/repository",
number: 41,
url: "https://example.test/owner/repository/pull/41",
},
}),
makeThread("readable-worktree", {
branch: "readable",
worktreePath: "/worktrees/readable",
latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"),
}),
],
[
{
...makeProject(),
repositoryIdentity: {
canonicalKey: "example.test/owner/repository",
locator: {
source: "git-remote",
remoteName: "origin",
remoteUrl: "https://example.test/owner/repository.git",
},
provider: SourceControlProviderKind.make("github"),
owner: "owner",
name: "repository",
},
},
],
),
existingWorktreePaths: ["/worktrees/readable", "/worktrees/readable/.git"],
unreadablePaths: ["/worktrees/unreadable/.git", "/worktrees/unreadable-linked/.git"],
pullRequestSummary: (input) =>
Effect.succeed({
...makePullRequestSummary({ ...input, state: "merged" }),
mergedAt: NOW,
}),
branchPullRequest: (input) =>
input.cwd.startsWith("/worktrees/unreadable")
? Effect.die(new Error("git: permission denied"))
: Effect.succeed(makeBranchPullRequest("merged")),
});

yield* Effect.gen(function* () {
const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2;
yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
expect((yield* Ref.get(fixture.commands)).map((command) => command.threadId)).toEqual([
ThreadId.make("readable-worktree"),
]);
}).pipe(Effect.provide(fixture.layer));
}),
),
);

it.effect("reevaluates immediately after a pull request merge", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down
12 changes: 8 additions & 4 deletions apps/server/src/orchestration-v2/ThreadSettlementService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as FiberSet from "effect/FiberSet";
import * as FileSystem from "effect/FileSystem";
import * as Path from "effect/Path";
import * as Schedule from "effect/Schedule";
import * as Semaphore from "effect/Semaphore";
import type * as Scope from "effect/Scope";
Expand Down Expand Up @@ -297,6 +298,7 @@ export const make = Effect.gen(function* () {
const pullRequests = yield* PullRequestService.PullRequestService;
const random = yield* SecureRandom.SecureRandom;
const fileSystem = yield* FileSystem.FileSystem;
const path = yield* Path.Path;
// A shell merge or close sends no event, and the branch lookup caches an open pull request for
// a minute, so these threads' next lookup reads the host fresh.
const shellClosedThreads = new Set<ThreadId>();
Expand Down Expand Up @@ -394,12 +396,14 @@ export const make = Effect.gen(function* () {
Effect.gen(function* () {
const project = projects.get(thread.projectId);
if (project === undefined || thread.branch === null) return;
const worktreeExists =
thread.worktreePath !== null &&
(yield* fileSystem.exists(thread.worktreePath).pipe(Effect.orElseSucceed(() => false)));
// A worktree folder can outlive its .git, and git there fails with not_a_repository.
// An unreadable .git keeps the worktree, so git fails there and the thread is skipped.
lookupCwdByThreadId.set(
thread.id,
worktreeExists && thread.worktreePath !== null
thread.worktreePath !== null &&
(yield* fileSystem
.exists(path.join(thread.worktreePath, ".git"))
.pipe(Effect.orElseSucceed(() => true)))
? thread.worktreePath
: project.workspaceRoot,
);
Expand Down
Loading