Skip to content

Recover stale worktree checkouts #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ENK0DED wants to merge 1 commit into from
Closed

Recover stale worktree checkouts #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
5 changes: 4 additions & 1 deletion apps/server/integration/providerService.integration.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,10 @@ const makeIntegrationFixture = Effect.gen(function* () {
Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers),
).pipe(Layer.provide(SqlitePersistenceMemory));

const layer = makeProviderServiceLive().pipe(Layer.provide(shared));
const layer = makeProviderServiceLive().pipe(
Layer.provide(shared),
Layer.provide(NodeServices.layer),
);

return {
cwd,
Expand Down
140 changes: 99 additions & 41 deletions apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -610,14 +610,18 @@ describe("ProviderCommandReactor", () => {
it("generates a worktree branch name for the first turn", async () => {
const harness = await createHarness();
const now = "2026-01-01T00:00:00.000Z";
const worktreePath = NodeFS.mkdtempSync(
NodePath.join(NodeOS.tmpdir(), "t3-provider-branch-worktree-"),
);
createdBaseDirs.add(worktreePath);

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.meta.update",
commandId: CommandId.make("cmd-thread-branch"),
threadId: ThreadId.make("thread-1"),
branch: "t3code/1234abcd",
worktreePath: "/tmp/provider-project-worktree",
worktreePath,
}),
);

Expand Down Expand Up @@ -658,7 +662,7 @@ describe("ProviderCommandReactor", () => {
expect(harness.generateBranchName.mock.calls[0]?.[0]).toMatchObject({
message: "Add a safer reconnect backoff.",
});
expect(harness.refreshStatus.mock.calls[0]?.[0]).toBe("/tmp/provider-project-worktree");
expect(harness.refreshStatus.mock.calls[0]?.[0]).toBe(worktreePath);
});

it("forwards codex model options through session start and turn send", async () => {
Expand Down Expand Up @@ -1130,6 +1134,10 @@ describe("ProviderCommandReactor", () => {
},
});
const now = "2026-01-01T00:00:00.000Z";
const worktreePath = NodeFS.mkdtempSync(
NodePath.join(NodeOS.tmpdir(), "t3-provider-session-worktree-"),
);
createdBaseDirs.add(worktreePath);

await Effect.runPromise(
harness.engine.dispatch({
Expand Down Expand Up @@ -1159,7 +1167,7 @@ describe("ProviderCommandReactor", () => {
type: "thread.meta.update",
commandId: CommandId.make("cmd-thread-worktree-change"),
threadId: ThreadId.make("thread-1"),
worktreePath: "/tmp/provider-project-worktree",
worktreePath,
}),
);

Expand All @@ -1185,7 +1193,7 @@ describe("ProviderCommandReactor", () => {
expect(harness.stopSession.mock.calls.length).toBe(0);
expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({
threadId: ThreadId.make("thread-1"),
cwd: "/tmp/provider-project-worktree",
cwd: worktreePath,
resumeCursor: { opaque: "resume-1" },
modelSelection: {
instanceId: ProviderInstanceId.make("claudeAgent"),
Expand All @@ -1195,6 +1203,56 @@ describe("ProviderCommandReactor", () => {
});
});

it("falls back to the project checkout when the thread worktree path is gone", async () => {
const harness = await createHarness();
const now = "2026-01-01T00:00:00.000Z";
const staleWorktreePath = NodeFS.mkdtempSync(
NodePath.join(NodeOS.tmpdir(), "t3-provider-stale-worktree-"),
);

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.meta.update",
commandId: CommandId.make("cmd-thread-stale-worktree"),
threadId: ThreadId.make("thread-1"),
worktreePath: staleWorktreePath,
}),
);
NodeFS.rmSync(staleWorktreePath, { recursive: true, force: true });

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.turn.start",
commandId: CommandId.make("cmd-turn-start-stale-worktree"),
threadId: ThreadId.make("thread-1"),
message: {
messageId: asMessageId("user-message-stale-worktree"),
role: "user",
text: "continue after deleted worktree",
attachments: [],
},
interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
runtimeMode: "approval-required",
createdAt: now,
}),
);

await waitFor(() => harness.startSession.mock.calls.length === 1);
await waitFor(() => harness.sendTurn.mock.calls.length === 1);

const readModel = await harness.readModel();
const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
expect(thread?.branch).toBeNull();
expect(thread?.worktreePath).toBeNull();
expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({
threadId: ThreadId.make("thread-1"),
cwd: "/tmp/provider-project",
});
expect(
thread?.activities.find((activity) => activity.kind === "provider.turn.start.failed"),
).toBeUndefined();
});

it("restarts claude sessions when claude effort changes", async () => {
const harness = await createHarness({
threadModelSelection: {
Expand Down Expand Up @@ -2053,45 +2111,45 @@ describe("ProviderCommandReactor", () => {
expect(resolvedActivity).toBeUndefined();
});

it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => {
const harness = await createHarness();
const now = "2026-01-01T00:00:00.000Z";
effectIt(
"reacts to thread.session.stop by stopping provider session and clearing thread session state",
() =>
Effect.gen(function* () {
const harness = yield* Effect.promise(() => createHarness());
const now = "2026-01-01T00:00:00.000Z";

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.make("cmd-session-set-for-stop"),
threadId: ThreadId.make("thread-1"),
session: {
yield* harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.make("cmd-session-set-for-stop"),
threadId: ThreadId.make("thread-1"),
status: "ready",
providerName: "codex",
providerInstanceId: ProviderInstanceId.make("codex_work"),
runtimeMode: "approval-required",
activeTurnId: null,
lastError: null,
updatedAt: now,
},
createdAt: now,
}),
);
session: {
threadId: ThreadId.make("thread-1"),
status: "ready",
providerName: "codex",
providerInstanceId: ProviderInstanceId.make("codex_work"),
runtimeMode: "approval-required",
activeTurnId: null,
lastError: null,
updatedAt: now,
},
createdAt: now,
});

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.stop",
commandId: CommandId.make("cmd-session-stop"),
threadId: ThreadId.make("thread-1"),
createdAt: now,
}),
);
yield* harness.engine.dispatch({
type: "thread.session.stop",
commandId: CommandId.make("cmd-session-stop"),
threadId: ThreadId.make("thread-1"),
createdAt: now,
});

await waitFor(() => harness.stopSession.mock.calls.length === 1);
const readModel = await harness.readModel();
const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
expect(thread?.session).not.toBeNull();
expect(thread?.session?.status).toBe("stopped");
expect(thread?.session?.threadId).toBe("thread-1");
expect(thread?.session?.providerInstanceId).toBe(ProviderInstanceId.make("codex_work"));
expect(thread?.session?.activeTurnId).toBeNull();
});
yield* Effect.promise(() => waitFor(() => harness.stopSession.mock.calls.length === 1));
const readModel = yield* Effect.promise(() => harness.readModel());
const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
expect(thread?.session).not.toBeNull();
expect(thread?.session?.status).toBe("stopped");
expect(thread?.session?.threadId).toBe("thread-1");
expect(thread?.session?.providerInstanceId).toBe(ProviderInstanceId.make("codex_work"));
expect(thread?.session?.activeTurnId).toBeNull();
}),
);
});
73 changes: 68 additions & 5 deletions apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import * as Crypto from "effect/Crypto";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Equal from "effect/Equal";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
Expand Down Expand Up @@ -196,6 +197,7 @@ const make = Effect.gen(function* () {
const vcsStatusBroadcaster = yield* VcsStatusBroadcaster;
const textGeneration = yield* TextGeneration;
const serverSettingsService = yield* ServerSettingsService;
const fileSystem = yield* FileSystem.FileSystem;
const serverCommandId = (tag: string) =>
crypto.randomUUIDv4.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`)));
const serverEventId = () => crypto.randomUUIDv4.pipe(Effect.map(EventId.make));
Expand Down Expand Up @@ -349,6 +351,65 @@ const make = Effect.gen(function* () {
});
});

const resolveAvailableThreadWorkspaceCwd = Effect.fnUntraced(function* (input: {
readonly thread: {
readonly id: ThreadId;
readonly projectId: ProjectId;
readonly worktreePath: string | null;
readonly modelSelection: ModelSelection;
readonly session: OrchestrationSession | null;
};
readonly projects: ReadonlyArray<{
readonly id: ProjectId;
readonly workspaceRoot: string;
}>;
}) {
const fallbackCwd = resolveThreadWorkspaceCwd({
thread: {
projectId: input.thread.projectId,
worktreePath: null,
},
projects: input.projects,
});

if (!input.thread.worktreePath) {
return fallbackCwd;
}

const stat = yield* fileSystem
.stat(input.thread.worktreePath)
.pipe(Effect.orElseSucceed(() => null));
if (stat?.type === "Directory") {
return input.thread.worktreePath;
}

if (!fallbackCwd) {
return yield* new ProviderAdapterRequestError({
provider: providerErrorLabelFromInstanceHint({
modelSelectionInstanceId: String(input.thread.modelSelection.instanceId),
sessionProvider: input.thread.session?.providerName ?? undefined,
}),
method: "thread.turn.start",
detail: `Thread '${input.thread.id}' worktree path is unavailable and the project checkout could not be resolved: ${input.thread.worktreePath}.`,
});
}

yield* Effect.logWarning("provider command reactor falling back from stale worktree path", {
threadId: input.thread.id,
worktreePath: input.thread.worktreePath,
fallbackCwd,
reason: stat === null ? "missing" : `not-${stat.type}`,
});
yield* orchestrationEngine.dispatch({
type: "thread.meta.update",
commandId: yield* serverCommandId("stale-worktree-fallback"),
threadId: input.thread.id,
branch: null,
worktreePath: null,
});
return fallbackCwd;
});

const ensureSessionForThread = Effect.fn("ensureSessionForThread")(function* (
threadId: ThreadId,
createdAt: string,
Expand Down Expand Up @@ -466,7 +527,7 @@ const make = Effect.gen(function* () {
}
}
const project = yield* resolveProject(thread.projectId);
const effectiveCwd = resolveThreadWorkspaceCwd({
const effectiveCwd = yield* resolveAvailableThreadWorkspaceCwd({
thread,
projects: project ? [project] : [],
});
Expand Down Expand Up @@ -775,10 +836,12 @@ const make = Effect.gen(function* () {
if (isFirstUserMessageTurn) {
const project = yield* resolveProject(thread.projectId);
const generationCwd =
resolveThreadWorkspaceCwd({
(yield* resolveAvailableThreadWorkspaceCwd({
thread,
projects: project ? [project] : [],
}) ?? process.cwd();
})) ?? process.cwd();
const generationWorktreePath =
thread.worktreePath && generationCwd === thread.worktreePath ? thread.worktreePath : null;
const generationInput = {
messageText: message.text,
...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
Expand All @@ -787,8 +850,8 @@ const make = Effect.gen(function* () {

yield* maybeGenerateAndRenameWorktreeBranchForFirstTurn({
threadId: event.payload.threadId,
branch: thread.branch,
worktreePath: thread.worktreePath,
branch: generationWorktreePath ? thread.branch : null,
worktreePath: generationWorktreePath,
...generationInput,
}).pipe(Effect.forkScoped);

Expand Down
Loading
Loading