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
42 changes: 38 additions & 4 deletions apps/server/src/orchestration-v2/PullRequestWatchReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schedule from "effect/Schedule";
import * as Schema from "effect/Schema";
import type * as Scope from "effect/Scope";

Expand All @@ -38,6 +37,13 @@ import { evaluatePullRequestWatch, pullRequestWatchMessage } from "./pullRequest

/** Checks take minutes; faster sweeps waste the account's shared host quota. */
const SWEEP_MINUTES = 2;
/**
* While checks run, a sweep of conditional fingerprints spends no quota when nothing moved, so
* it can follow CI closely. GitHub's secondary limits count requests, hence not faster.
*/
const ACTIVE_SWEEP_SECONDS = 30;
/** A push starts checks that may not be listed yet, so keep the quick pace for a while. */
const RECENT_PUSH_MS = 5 * 60_000;
const READ_FAILURE_LIMIT = 8;
/** Without fingerprints, reread quiet PRs on snapshot changes or this interval to catch edits. */
const QUIET_REREAD_MS = 10 * 60_000;
Expand Down Expand Up @@ -78,6 +84,7 @@ interface LastRead {
readonly at: number;
readonly fingerprint: string;
readonly inFlight: boolean;
readonly conditional: boolean;
readonly remarksComplete: boolean;
readonly watches: ReadonlySet<string>;
readonly status: string | null;
Expand Down Expand Up @@ -151,7 +158,10 @@ function planRead(
last.remarks !== fingerprint.remarks ||
!last.remarksComplete ||
now - last.at >= FINGERPRINT_REREAD_MS;
return { detail: activity || last.status !== fingerprint.status || last.inFlight, activity };
// A conditional fingerprint already carries check and mergeability state, so in-flight work
// shows up as a status change there.
const settling = last.inFlight && fingerprint.conditional !== true;
return { detail: activity || last.status !== fingerprint.status || settling, activity };
}

function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatch): boolean {
Expand All @@ -176,6 +186,8 @@ export class PullRequestWatchReactor extends Context.Service<
{
readonly start: () => Effect.Effect<void, never, Scope.Scope>;
readonly sweep: Effect.Effect<void>;
/** How long to wait after the latest sweep before the next one. */
readonly sweepDelayMs: Effect.Effect<number>;
}
>()("t3/orchestration-v2/PullRequestWatchReactor") {}

Expand Down Expand Up @@ -522,6 +534,7 @@ export const make = Effect.gen(function* () {
? (threadTails.get(group.key)?.fetchedAt ?? now)
: last.at,
fingerprint: group.fingerprint,
conditional: fingerprint?.conditional === true,
inFlight:
detail.mergeability === "unknown" ||
detail.checks.some((check) => check.status === "pending"),
Expand Down Expand Up @@ -555,6 +568,8 @@ export const make = Effect.gen(function* () {
});
});

let nextSweepMs = SWEEP_MINUTES * 60_000;

const sweep = Effect.gen(function* () {
const threads = yield* projections.getThreadsWithPullRequests();
const targets = threads.flatMap((thread) =>
Expand Down Expand Up @@ -606,6 +621,7 @@ export const make = Effect.gen(function* () {
const fingerprints = yield* Effect.forEach(groups, fingerprintOf, {
concurrency: "unbounded",
});
const paused = fingerprints.includes("paused");
yield* Effect.forEach(
groups,
(group, index) => {
Expand All @@ -620,17 +636,35 @@ export const make = Effect.gen(function* () {
},
{ concurrency: 4, discard: true },
);
const now = yield* Clock.currentTimeMillis;
const quick =
!paused &&
groups.some((group) => {
const last = lastReads.get(group.key);
return (
last?.conditional === true &&
(last.inFlight ||
group.targets.some((target) => now - lifeOf(target).pushedAt < RECENT_PUSH_MS))
);
});
nextSweepMs = quick ? ACTIVE_SWEEP_SECONDS * 1000 : SWEEP_MINUTES * 60_000;
Comment thread
yordis marked this conversation as resolved.
Resolved
}).pipe(
Effect.catchCause(logFailure("pull request watch sweep failed", {})),
Effect.withSpan("PullRequestWatchReactor.sweep"),
);

const sweepDelayMs = Effect.sync(() => nextSweepMs);

const start: PullRequestWatchReactor["Service"]["start"] = () =>
forkParked(
sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid),
sweep.pipe(
Effect.andThen(Effect.flatMap(sweepDelayMs, Effect.sleep)),
Effect.forever,
Effect.asVoid,
),
);

return { start, sweep } satisfies PullRequestWatchReactor["Service"];
return { start, sweep, sweepDelayMs } satisfies PullRequestWatchReactor["Service"];
});

export const layer = Layer.effect(PullRequestWatchReactor, make);
129 changes: 129 additions & 0 deletions apps/server/src/orchestration-v2/runtimeLayer.test.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
import { SourceControlProviderKind } from "@t3tools/contracts";
import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts";
import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
Expand Down Expand Up @@ -3452,6 +3453,134 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
}),
);

it.effect("sweeps quickly only while checks run and the fingerprint is conditional", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const threadId = ThreadId.make("runtime-pull-request-watch-cadence");
const projectId = ProjectId.make("pr-watch-cadence-project");
yield* seedProject({
projectId,
title: "Watch cadence",
workspaceRoot: "/workspace/watch-cadence",
defaultModelSelection: null,
createdAt: "2026-10-01T00:00:00.000Z",
});
yield* orchestrator.dispatch({
type: "thread.create",
createdBy: "user",
creationSource: "web",
commandId: CommandId.make("pr-watch-cadence-create"),
threadId,
projectId,
title: "Watch cadence",
modelSelection,
runtimeMode: "full-access",
interactionMode: "default",
branch: null,
worktreePath: null,
});
yield* orchestrator.dispatch({
type: "thread.pull-request.watch",
commandId: CommandId.make("pr-watch-cadence-start"),
threadId,
host: "github.com",
repository: "pingdotgg/t3code",
number: 15,
watching: true,
link: { url: "https://github.com/pingdotgg/t3code/pull/15", source: "agent" },
});

let checkStatus: "pending" | "success" = "pending";
let fingerprint: "conditional" | "graphql" | "paused" = "conditional";
let detailReads = 0;
const reactor = yield* PullRequestWatchReactor.make.pipe(
Effect.provide(
Layer.mergeAll(
NodeServices.layer,
Layer.mock(PullRequestService.PullRequestService)({
watchFingerprint: () =>
fingerprint === "paused"
? Effect.fail(
new PullRequestOperationError({
operation: "watchFingerprint",
detail: "Rate limited",
cause: new PullRequestProviderError({
provider: SourceControlProviderKind.make("github"),
operation: "getChangeRequestWatchFingerprint",
reason: "rate-limited",
detail: "Rate limited",
}),
}),
)
: Effect.succeed({
status: checkStatus,
remarks: "0",
...(fingerprint === "conditional" ? { conditional: true as const } : {}),
}),
invalidate: () => Effect.void,
detail: () =>
Effect.sync(() => {
detailReads += 1;
return {
...watchedPullRequestDetail({
projectId,
number: 15,
at: "2026-10-02T12:00:00.000Z",
}),
headSha: "6074dd48ff",
checks: [
{
name: "lint",
status: checkStatus,
description: null,
url: null,
required: true,
},
],
};
}),
activity: () =>
Effect.succeed({
comments: [],
commentCount: 0,
commentsTruncated: false,
reviewThreads: [],
commits: [],
}),
}),
),
),
);
const delayAfterSweep = Effect.andThen(reactor.sweep, reactor.sweepDelayMs);

assert.equal(yield* delayAfterSweep, 30_000, "checks are running");
assert.equal(yield* delayAfterSweep, 30_000, "checks are still running");
assert.equal(detailReads, 1, "an unchanged conditional fingerprint needs no reread");
fingerprint = "graphql";
assert.equal(
yield* delayAfterSweep,
120_000,
"reading costs quota without a conditional fingerprint",
);
fingerprint = "paused";
assert.equal(yield* delayAfterSweep, 120_000, "the host is rate limited");
fingerprint = "conditional";
checkStatus = "success";
yield* TestClock.adjust("10 minutes");
assert.equal(yield* delayAfterSweep, 120_000, "nothing is in flight");

yield* orchestrator.dispatch({
type: "thread.pull-request.watch",
commandId: CommandId.make("pr-watch-cadence-end"),
threadId,
host: "github.com",
repository: "pingdotgg/t3code",
number: 15,
watching: false,
});
}),
);

it.effect("wakes for checks that complete after the no-checks notice on the same head", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@ export class PullRequestProviderError extends Schema.TaggedError<PullRequestProv
export interface ProviderChangeRequestWatchFingerprint {
readonly status: string;
readonly remarks: string;
/** Read without spending quota while nothing changed, so a watch can poll it often. */
readonly conditional?: true;
}

export interface PullRequestProviderFailure {
Expand Down
Loading
Loading