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
144 changes: 144 additions & 0 deletions apps/server/src/orchestration-v2/ThreadFork.execution.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import {
} from "@t3tools/contracts";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as TestClock from "effect/testing/TestClock";

import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts";
Expand Down Expand Up @@ -54,6 +55,149 @@ const forkCases = (["codex", "claudeAgent"] as const).flatMap((driverName) => {
}));
});

it.effect("a persisted fork leaves PR watching with its parent until explicitly opted in", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const eventSink = yield* EventSink.EventSinkV2;
const now = yield* DateTime.now;
const { instanceId, modelSelection } = forkCases[0]!;
const sourceThreadId = ThreadId.make("fork-watch-source");
const targetThreadId = ThreadId.make("fork-watch-target");
const sourceRunId = RunId.make("fork-watch-source-run");
const key = { host: "github.com", repository: "example/repo", number: 1 };
const link = { url: "https://github.com/example/repo/pull/1", source: "agent" as const };

yield* orchestrator.dispatch({
type: "thread.create",
commandId: CommandId.make("fork-watch-create"),
threadId: sourceThreadId,
projectId: ProjectId.make("fork-watch-project"),
title: "Watching source",
modelSelection,
runtimeMode: "full-access",
interactionMode: "default",
branch: null,
worktreePath: null,
createdBy: "user",
creationSource: "web",
});
yield* eventSink.write({
events: [
{
id: EventId.make("fork-watch-source-run"),
type: "run.created",
threadId: sourceThreadId,
runId: sourceRunId,
occurredAt: now,
payload: {
id: sourceRunId,
threadId: sourceThreadId,
ordinal: 1,
providerInstanceId: instanceId,
modelSelection,
providerThreadId: null,
userMessageId: MessageId.make("fork-watch-source-message"),
rootNodeId: null,
activeAttemptId: null,
status: "completed",
requestedAt: now,
startedAt: now,
completedAt: now,
checkpointId: null,
contextHandoffId: null,
},
},
],
});
yield* orchestrator.dispatch({
type: "thread.pull-request.watch",
commandId: CommandId.make("fork-watch-parent-start"),
threadId: sourceThreadId,
...key,
watching: true,
link,
});
const parentBefore = yield* orchestrator.getThreadShell(sourceThreadId);
const parentWatch = parentBefore?.pullRequests?.[0]?.watch;
assert.isDefined(parentWatch);
assert.lengthOf(parentBefore?.pendingBackgroundTasks ?? [], 1);

yield* orchestrator.dispatch({
type: "thread.fork",
commandId: CommandId.make("fork-watch-fork"),
sourceThreadId,
targetThreadId,
sourcePoint: { type: "run", runId: sourceRunId },
createdBy: "user",
creationSource: "mcp",
});
const fork = yield* orchestrator.getThreadShell(targetThreadId);
assert.isNotNull(fork);
assert.deepEqual(fork.pullRequests, [
{
...key,
...link,
linkedAt: parentWatch.startedAt,
snapshot: null,
stack: null,
},
]);
assert.deepEqual(fork.pendingBackgroundTasks, []);
assert.deepEqual(yield* orchestrator.getThreadShell(sourceThreadId), parentBefore);

// A delayed watcher update cannot reinstate the parent's subscription on the fork.
const stale = yield* orchestrator
.dispatch({
type: "thread.pull-request-watch.sync",
commandId: CommandId.make("fork-watch-stale-sync"),
threadId: targetThreadId,
...key,
startedAt: parentWatch.startedAt,
watch: parentWatch,
wake: {
messageId: MessageId.make("fork-watch-unwanted-wake"),
text: "A check failed",
notification: {
source: { kind: "monitor" },
outcome: "updated",
summary: "A check failed",
},
},
})
.pipe(Effect.flip);
assert.equal(stale._tag, "OrchestratorDispatchError");
assert.deepEqual((yield* orchestrator.getThreadProjection(targetThreadId)).runs, []);

// Advance virtual time only to distinguish the independent watch's start time.
yield* TestClock.adjust("1 second");
yield* orchestrator.dispatch({
type: "thread.pull-request.watch",
commandId: CommandId.make("fork-watch-opt-in"),
threadId: targetThreadId,
...key,
watching: true,
link,
});
const watchingFork = yield* orchestrator.getThreadShell(targetThreadId);
assert.isDefined(watchingFork?.pullRequests?.[0]?.watch);
assert.notEqual(watchingFork.pullRequests[0].watch.startedAt, parentWatch.startedAt);
assert.lengthOf(watchingFork.pendingBackgroundTasks ?? [], 1);
assert.deepEqual(yield* orchestrator.getThreadShell(sourceThreadId), parentBefore);

yield* orchestrator.dispatch({
type: "thread.pull-request.watch",
commandId: CommandId.make("fork-watch-opt-out"),
threadId: targetThreadId,
...key,
watching: false,
});
const stoppedFork = yield* orchestrator.getThreadShell(targetThreadId);
assert.deepEqual(stoppedFork?.pullRequests, fork.pullRequests);
assert.deepEqual(stoppedFork?.pendingBackgroundTasks, []);
assert.deepEqual(yield* orchestrator.getThreadShell(sourceThreadId), parentBefore);
}).pipe(Effect.provide(forkCases[0]!.layer)),
);

it.effect.each(forkCases)(
"bounds $driver context when continuing a fork of a $status run",
({ driver, status, instanceId, modelSelection, layer }) =>
Expand Down
79 changes: 77 additions & 2 deletions apps/server/src/orchestration-v2/ThreadForkService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
ProviderThreadId,
RunId,
ThreadId,
type ThreadPullRequestLink,
} from "@t3tools/contracts";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
Expand Down Expand Up @@ -106,11 +107,14 @@ function makeSourceProjection(sourceRun: OrchestrationV2Run): OrchestrationV2Thr
};
}

const planFork = (sourceRun: OrchestrationV2Run) =>
const planFork = (
sourceRun: OrchestrationV2Run,
sourceProjection = makeSourceProjection(sourceRun),
) =>
Effect.gen(function* () {
const service = yield* ThreadForkService.ThreadForkServiceV2;
return yield* service.plan({
sourceProjection: makeSourceProjection(sourceRun),
sourceProjection,
sourceRun,
sourceProviderThread: undefined,
canonicalSourcePoint: {
Expand Down Expand Up @@ -179,6 +183,77 @@ it.effect("forks from a usage-limited failed run", () =>
}),
);

it.effect("preserves pull request context without inheriting any watches", () =>
Effect.gen(function* () {
const sourceRun = makeSourceRun("completed");
const watch = {
startedAt: DateTime.formatIso(sourceCreatedAt),
headSha: "source-head",
failedChecks: ["lint"],
passed: false,
passedChecks: [],
remarksThrough: DateTime.formatIso(snoozedAt),
remarkIds: ["review-1"],
conflicting: true,
wakes: 2,
};
const links = (["agent", "stack", "manual", "stack-dismissed"] as const).map(
(source, index) =>
({
host: "github.com",
repository: "example/repo",
number: index + 1,
url: `https://github.com/example/repo/pull/${index + 1}`,
source,
linkedAt: DateTime.formatIso(sourceCreatedAt),
snapshot: {
state: "open",
title: `Pull request ${index + 1}`,
headBranch: `feature-${index + 1}`,
baseBranch: "main",
isDraft: false,
updatedAt: null,
syncedAt: DateTime.formatIso(snoozedAt),
},
stack: {
kind: "native",
id: "stack-1",
number: 1,
url: "https://github.com/example/repo/stack/1",
base: "main",
layers: [{ number: 1, headBranch: "feature-1", state: "open" }],
},
}) satisfies ThreadPullRequestLink,
);
const watchedLinks: ReadonlyArray<ThreadPullRequestLink> = links.map((link, index) =>
index < 2 ? { ...link, watch } : link,
);
const legacyLink = {
projectId: makeSourceThread().projectId,
repository: "example/repo",
number: 1,
url: links[0]!.url,
};
const sourceProjection = {
...makeSourceProjection(sourceRun),
thread: {
...makeSourceThread(),
pullRequests: watchedLinks,
linkedPullRequest: legacyLink,
branchPullRequest: legacyLink,
},
};
const result = yield* planFork(sourceRun, sourceProjection);

assert.deepEqual(result.targetThread.pullRequests, links);
assert.deepEqual(result.targetThread.linkedPullRequest, legacyLink);
assert.deepEqual(result.targetThread.branchPullRequest, legacyLink);
assert.deepEqual(sourceProjection.thread.pullRequests, watchedLinks);
assert.deepEqual(sourceProjection.thread.pullRequests[0]?.watch, watch);
assert.deepEqual(sourceProjection.thread.pullRequests[1]?.watch, watch);
}),
);

it.effect("rejects in-progress and rolled-back fork sources", () =>
Effect.gen(function* () {
for (const status of ["running", "rolled_back"] as const) {
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/orchestration-v2/ThreadForkService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,10 @@ export const layer: Layer.Layer<ThreadForkServiceV2> = Layer.succeed(
}
const targetThread: OrchestrationV2AppThread = {
...input.sourceProjection.thread,
// PR context follows the conversation; watching remains the source thread's responsibility.
pullRequests: input.sourceProjection.thread.pullRequests?.map(
({ watch: _watch, ...link }) => link,
),
createdBy: input.createdBy,
creationSource: input.creationSource,
id: input.targetThreadId,
Expand Down
Loading