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
124 changes: 88 additions & 36 deletions apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -979,6 +979,49 @@ describe("PullRequestSyncReactor", () => {
),
);

it.effect("stops reading a pull request a day after it closed, until a refresh asks", () =>
Effect.scoped(
Effect.gen(function* () {
yield* TestClock.setTime(Date.parse(NOW));
const closed = (closedAt: string | null) => ({ state: "closed" as const, closedAt });
const fixture = yield* makeHarness({
snapshot: makeSnapshot([
makeThread("closed", {
pullRequests: [
makeLink(1, closed("2026-08-20T00:00:00.000Z")),
makeLink(2, closed("2026-08-26T00:00:00.000Z")),
makeLink(3, closed("2026-08-28T11:00:00.000Z")),
makeLink(4, closed(null)),
],
}),
]),
summary: (input) =>
Effect.succeed(makeSummary(input, { state: "closed", closedAt: NOW })),
});
const reads = Ref.get(fixture.summaryCalls).pipe(
Effect.map((calls) => calls.map((call) => call.number).toSorted()),
);

yield* Effect.gen(function* () {
const reactor = yield* startAndSweep(fixture);
// Closed within the day, or with no close time, keeps the fifteen-minute cadence.
assert.deepStrictEqual(yield* reads, [3, 4]);
for (let index = 0; index < 15; index += 1) yield* sweepAgain(fixture, reactor);
assert.deepStrictEqual(yield* reads, [3, 3, 4, 4]);

yield* reactor.requestSync({
host: "github.com",
repository: "owner/repository",
number: 1,
});
yield* Queue.take(fixture.snapshotReads);
yield* reactor.drain;
assert.deepStrictEqual(yield* reads, [1, 3, 3, 4, 4]);
}).pipe(Effect.provide(fixture.layer));
}),
),
);

it.effect("re-reads a merged pull request until the host supplies its merge time", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down Expand Up @@ -1579,46 +1622,55 @@ describe("PullRequestSyncReactor", () => {
),
);

it.effect("reads open links fresh as soon as a merge command ends, before its run ends", () =>
Effect.scoped(
Effect.gen(function* () {
yield* TestClock.setTime(Date.parse(NOW));
const state = yield* Ref.make<PullRequestSummary["state"]>("open");
const fixture = yield* makeHarness({
snapshot: makeSnapshot([
makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }),
]),
summary: (input) =>
Ref.get(state).pipe(
Effect.map((current) =>
makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}),
it.effect.each(["open", "closed"] as const)(
"reads %s links fresh as soon as a merge command ends, before its run ends",
(initialState) =>
Effect.scoped(
Effect.gen(function* () {
yield* TestClock.setTime(Date.parse(NOW));
const state = yield* Ref.make<PullRequestSummary["state"]>(initialState);
const initialSnapshot = {
state: initialState,
closedAt: initialState === "closed" ? "2026-08-20T00:00:00.000Z" : null,
};
const fixture = yield* makeHarness({
snapshot: makeSnapshot([
makeThread("agent", { pullRequests: [makeLink(7, initialSnapshot)] }),
]),
summary: (input) =>
Ref.get(state).pipe(
Effect.map((current) =>
makeSummary(
input,
current === "merged" ? { state: current, mergedAt: NOW } : initialSnapshot,
),
),
),
),
});
});

yield* Effect.gen(function* () {
const reactor = yield* startAndSweep(fixture);
yield* Ref.set(state, "merged");
yield* Ref.set(fixture.summaryCalls, []);
yield* Effect.gen(function* () {
const reactor = yield* startAndSweep(fixture);
yield* Ref.set(state, "merged");
yield* Ref.set(fixture.summaryCalls, []);

yield* Queue.offer(
fixture.domainEvents,
commandRan(ThreadId.make("agent"), "gh pr merge 7"),
);
yield* Queue.take(fixture.snapshotReads);
yield* Queue.take(fixture.snapshotReads);
yield* reactor.drain;
yield* Queue.offer(
fixture.domainEvents,
commandRan(ThreadId.make("agent"), "gh pr merge 7"),
);
yield* Queue.take(fixture.snapshotReads);
yield* Queue.take(fixture.snapshotReads);
yield* reactor.drain;

assert.deepStrictEqual(
(yield* Ref.get(fixture.syncCommands)).map((command) => [
command.number,
command.snapshot.state,
]),
[[7, "merged"]],
);
}).pipe(Effect.provide(fixture.layer));
}),
),
assert.deepStrictEqual(
(yield* Ref.get(fixture.syncCommands)).map((command) => [
command.number,
command.snapshot.state,
]),
[[7, "merged"]],
);
}).pipe(Effect.provide(fixture.layer));
}),
),
);

it.effect.each([
Expand Down
26 changes: 14 additions & 12 deletions apps/server/src/orchestration-v2/PullRequestSyncReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;
// requested refreshes, watches and a host CLI's merge command still read it at once.
const QUIET_AFTER_MS = 30 * 60 * 1_000;
const QUIET_SYNC_INTERVAL_MS = 5 * 60 * 1_000;
// A pull request closed this long ago is rarely reopened, so the sweep stops reading it, as it
// does merged ones. A reader that sees it open, or a requested refresh, still reads it at once.
const CLOSED_FINAL_AFTER_MS = 24 * 60 * 60 * 1_000;

const isChangeRequestCloseCommand = (command: string) =>
sourceControlDefinitions.definitions.some(
Expand Down Expand Up @@ -351,13 +354,12 @@ export const make = Effect.gen(function* () {
if (entries.some((entry) => entry.link.snapshot === null)) return true;
const active = entries.filter((entry) => isUnsettled(entry.thread));
// Merged snapshots without mergedAt cannot settle threads; reread slowly until supplied.
if (
active.every(
(entry) => entry.link.snapshot?.state === "merged" && entry.link.snapshot.mergedAt != null,
)
) {
return false;
}
const isFinal = ({ link: { snapshot } }: LinkEntry) =>
snapshot?.state === "merged"
? snapshot.mergedAt != null
: snapshot?.state === "closed" &&
nowMs - Date.parse(snapshot.closedAt ?? "") >= CLOSED_FINAL_AFTER_MS;
if (active.every(isFinal)) return false;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
const last = lastSyncedAt.get(key);
const sinceLastMs = last === undefined ? Number.POSITIVE_INFINITY : nowMs - last;
const open = active.filter((entry) => entry.link.snapshot?.state === "open");
Expand Down Expand Up @@ -627,13 +629,13 @@ export const make = Effect.gen(function* () {

// Threads that ran a close command this run, with the commands already read for.
const closeCommands = new Map<ThreadId, Set<string>>();
const refreshOpenLinks = (threadId: ThreadId) =>
const refreshUnmergedLinks = (threadId: ThreadId) =>
projections.getThreadsWithPullRequests(threadId).pipe(
Effect.flatMap((threads) =>
Effect.forEach(
threads.flatMap((thread) =>
visibleThreadPullRequests(thread.pullRequests ?? []).filter(
(link) => link.snapshot?.state === "open",
(link) => link.snapshot?.state === "open" || link.snapshot?.state === "closed",
),
),
requestSync,
Expand Down Expand Up @@ -698,7 +700,7 @@ export const make = Effect.gen(function* () {
{ discard: true },
);
// An agent can merge or close its pull request from a shell with its host's CLI, which
// sends no merge notification. Read the thread's open links fresh once that command
// sends no merge notification. Read the thread's unmerged links fresh once that command
// ends, while the agent usually still replies, so settlement can land before the run
// ends. Read again when the run ends, since `merge --auto` returns before it merges.
case "turn-item.updated": {
Expand All @@ -710,11 +712,11 @@ export const make = Effect.gen(function* () {
closeCommands.set(event.threadId, ended);
if (isOrchestrationV2WorkActive(item.status) || ended.has(item.id)) return linked;
ended.add(item.id);
return Effect.andThen(linked, refreshOpenLinks(event.threadId));
return Effect.andThen(linked, refreshUnmergedLinks(event.threadId));
}
case "run.updated":
return isTerminalRunStatus(event.payload.status) && closeCommands.delete(event.threadId)
? refreshOpenLinks(event.threadId)
? refreshUnmergedLinks(event.threadId)
: Effect.void;
default:
return Effect.void;
Expand Down
Loading