Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
40 changes: 40 additions & 0 deletions apps/server/src/orchestration-v2/ThreadLaunchService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -375,6 +375,7 @@ it.effect("retains automation and sender attribution while a message waits in th
command: "command:automation:queue",
thread: "thread:automation:queue",
message: "First message",
workspace: { type: "worktree", baseRef: "main" },
}),
);
const scheduledTaskId = ScheduledTaskId.make("scheduled-task:queued");
Expand Down Expand Up @@ -492,13 +493,15 @@ it.effect("provisions independent launches concurrently instead of behind a glob
command: "command:launch:concurrent-a",
thread: "thread:launch:concurrent-a",
message: "First",
workspace: { type: "worktree", baseRef: "main" },
}),
),
launches.launch(
launchInput({
command: "command:launch:concurrent-b",
thread: "thread:launch:concurrent-b",
message: "Second",
workspace: { type: "worktree", baseRef: "main" },
}),
),
],
Expand Down Expand Up @@ -1111,6 +1114,37 @@ it.effect("runs a Scratch thread launched at the root in its own folder", () =>
}),
);

it.effect("starts a thread in the main checkout without running the setup action", () =>
Effect.gen(function* () {
const harness = makeHarness();
yield* Effect.gen(function* () {
const launches = yield* ThreadLaunch.ThreadLaunchService;
const threads = yield* ThreadManagement.ThreadManagementService;
const input = launchInput({
command: "command:launch:main-checkout",
thread: "thread:launch:main-checkout",
message: "Work in the main checkout",
});
const launched = yield* launches.launch(input);
assert.isNull(launched.projection.thread.worktreePath);
const events = yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe(
Stream.takeUntil(
(stored) =>
stored.commandId === CommandId.make(`${input.commandId}:release`) &&
stored.event.type === "run.updated",
),
Stream.runCollect,
);
assert.isFalse(
events.some(
(stored) => stored.commandId === CommandId.make(`${input.commandId}:progress:setup`),
),
);
assert.equal(harness.runSetup.mock.calls.length, 0);
}).pipe(Effect.provide(harness.layer));
}),
);

it.effect("names the worktree itself when the client provides no branch", () =>
Effect.gen(function* () {
const harness = makeHarness();
Expand Down Expand Up @@ -1739,6 +1773,7 @@ it.effect("replays a server-allocated launch", () =>
command: "command:launch:allocated-retry",
thread: "unused",
message: "Only once",
workspace: { type: "worktree", baseRef: "main" },
});
const first = yield* launches.launch(rest);
yield* Deferred.await(setupEntered);
Expand Down Expand Up @@ -1973,6 +2008,7 @@ it.effect("bounds concurrent first launches to one thread per command", () =>
command: "command:launch:concurrent-allocated",
thread: "unused",
message: "Race me",
workspace: { type: "worktree", baseRef: "main" },
});
// The command receipt is reserved atomically with the winning create, so
// a loser either replays the winner's stored events or surfaces a
Expand Down Expand Up @@ -2031,6 +2067,7 @@ it.effect("deduplicates retried launch side effects in-process", () =>
command: "command:launch:retry",
thread: "thread:launch:retry",
message: "Only once",
workspace: { type: "worktree", baseRef: "main" },
});
const [first, retry] = yield* Effect.all([launches.launch(input), launches.launch(input)], {
concurrency: "unbounded",
Expand Down Expand Up @@ -2067,6 +2104,7 @@ it.effect("does not let a failing same-command caller strand a concurrent durabl
command,
thread: "thread:launch:failed-owner-race",
message: "This invalid reuse fails",
workspace: { type: "worktree", baseRef: "main" },
}),
reuseExistingThread: true,
})
Expand All @@ -2076,6 +2114,7 @@ it.effect("does not let a failing same-command caller strand a concurrent durabl
command,
thread: "thread:launch:successful-peer",
message: "This peer persists",
workspace: { type: "worktree", baseRef: "main" },
}),
),
],
Expand Down Expand Up @@ -2111,6 +2150,7 @@ it.effect("schedules an accepted preparing message exactly once across concurren
command: "command:launch:accepted-before-fork",
thread: "thread:launch:accepted-before-fork",
message: "Resume preparation",
workspace: { type: "worktree", baseRef: "main" },
});
const messageId = MessageId.make("message:launch:accepted-before-fork");

Expand Down
46 changes: 24 additions & 22 deletions apps/server/src/orchestration-v2/ThreadLaunchService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -535,8 +535,7 @@ const make = Effect.gen(function* () {
);
}

const cwd = worktreePath ?? project.workspaceRoot;
if (runId !== null) {
if (runId !== null && worktreePath !== null) {
yield* threads
.dispatch({
type: "prepared-run.progress",
Expand All @@ -548,27 +547,30 @@ const make = Effect.gen(function* () {
.pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
}
yield* setupTracker.stageStatus(threadId, "setup-script", "running");
const setup = yield* setupScripts
.runForThread({
threadId,
projectId: input.projectId,
projectCwd: project.workspaceRoot,
worktreePath: cwd,
...(tracked
? {
observeCompletion: {
onOutputLine: (line: string) =>
setupTracker.appendTail(threadId, "setup-script", line),
const setup =
worktreePath === null
? ({ status: "no-script" } as const)
: yield* setupScripts
.runForThread({
threadId,
projectId: input.projectId,
projectCwd: project.workspaceRoot,
worktreePath,
...(tracked
? {
observeCompletion: {
onOutputLine: (line: string) =>
setupTracker.appendTail(threadId, "setup-script", line),
},
}
: {}),
project: {
id: project.id,
workspaceRoot: project.workspaceRoot,
scripts: project.scripts,
},
}
: {}),
project: {
id: project.id,
workspaceRoot: project.workspaceRoot,
scripts: project.scripts,
},
})
.pipe(Effect.mapError(mapError(input, "run-setup-script", threadId)));
})
.pipe(Effect.mapError(mapError(input, "run-setup-script", threadId)));

let awaitAsyncSetup = Effect.void;
if (setup.status === "started") {
Expand Down
Loading