Skip to content
Open
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
2 changes: 2 additions & 0 deletions apps/server/src/mcp/toolkits/pullRequests/handlers.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,7 @@ describe("pull request toolkit handlers", () => {
const harness = yield* makeHarness();
const result = yield* harness.call("watch_pull_request", {
url: "https://github.com/t3tools/t3code/pull/9",
events: ["review", "checks_failed"],
});
expect(result).toMatchObject({ number: 9, watching: false, wasWatching: false });
expect(yield* Ref.get(harness.commands)).toMatchObject([
Expand All @@ -279,6 +280,7 @@ describe("pull request toolkit handlers", () => {
number: 9,
watching: true,
link: { url: "https://github.com/t3tools/t3code/pull/9", source: "agent" },
events: ["review", "checks_failed"],
},
]);
}),
Expand Down
4 changes: 3 additions & 1 deletion apps/server/src/mcp/toolkits/pullRequests/handlers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ import {
PullRequestsToolkit,
type ThreadPullRequestEntry,
CHILD_PULL_REQUEST_REPORT_NEXT,
type WatchPullRequestInput,
} from "./tools.ts";

interface ResolvedTarget {
Expand Down Expand Up @@ -261,7 +262,7 @@ const make = Effect.gen(function* () {
: Effect.fail(new Failure({ cause }));

const setWatching = Effect.fn("PullRequestsToolkit.setWatching")(function* (
input: PullRequestTargetInput,
input: WatchPullRequestInput,
watching: boolean,
) {
const thread = yield* requireThread(PullRequestWatchFailedError, input.threadId);
Expand All @@ -288,6 +289,7 @@ const make = Effect.gen(function* () {
number: target.number,
watching,
...(watching ? { link: { url: target.url, source: "agent" as const } } : {}),
...(input.events === undefined ? {} : { events: input.events }),
})
.pipe(Effect.catchCause(dispatchFailure(PullRequestWatchFailedError)));
const after = yield* requireThread(PullRequestWatchFailedError, thread.id);
Expand Down
14 changes: 13 additions & 1 deletion apps/server/src/mcp/toolkits/pullRequests/tools.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import {
PullRequestReviewGuideRun,
PullRequestState,
ThreadPullRequestLinkSource,
ThreadPullRequestWatchEvents,
TrimmedNonEmptyString,
ThreadId,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -59,6 +60,17 @@ export const PullRequestTargetInput = Schema.Struct({
});
export type PullRequestTargetInput = typeof PullRequestTargetInput.Type;

export const WatchPullRequestInput = Schema.Struct({
...PullRequestTargetInput.fields,
events: Schema.optional(
ThreadPullRequestWatchEvents.annotate({
description:
"Only these changes wake you. Default: all. A closed PR or a watch T3 ends still wakes you.",
}),
),
});
export type WatchPullRequestInput = typeof WatchPullRequestInput.Type;

export class PullRequestUrlInvalidError extends Schema.TaggedError<PullRequestUrlInvalidError>()(
"PullRequestUrlInvalidError",
{},
Expand Down Expand Up @@ -317,7 +329,7 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", {
const WatchPullRequestTool = Tool.make("watch_pull_request", {
description:
"Watch an open PR for this thread, linking it first if needed. T3 wakes you on failed checks, passed required checks, new comments or reviews, or conflicts. Handle old comments first. Only later ones wake you. After this call, end your turn. The watch stays on after each wake, so do not call this again on a wake. A wake is not a merge decision. Check readiness yourself.",
parameters: PullRequestTargetInput,
parameters: WatchPullRequestInput,
success: WatchPullRequestResult,
failure: PullRequestToolError,
dependencies,
Expand Down
33 changes: 32 additions & 1 deletion apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import {
type OrchestrationV2ServerCommand,
type ThreadPullRequestLink,
type ThreadPullRequestWatch,
type ThreadPullRequestWatchEvents,
type OrchestrationV2AppThread,
type OrchestrationV2ContextHandoff,
type OrchestrationV2ContextSourcePoint,
Expand Down Expand Up @@ -654,6 +655,18 @@ function hasLiveRun(projection: Pick<OrchestrationV2ThreadProjection, "runs">):
return projection.runs.some(isLiveRun);
}

/** Sorted and deduplicated, so the same filter in another order is the same watch. */
const sortedWatchEvents = (events: ThreadPullRequestWatchEvents | undefined) =>
events && [...new Set(events)].toSorted();

function withWatchEvents(
watch: ThreadPullRequestWatch,
events: ThreadPullRequestWatchEvents | undefined,
): ThreadPullRequestWatch {
const { events: _events, ...rest } = watch;
return events === undefined ? rest : { ...rest, events };
}

function withPullRequestWatch(
link: ThreadPullRequestLink,
watch: ThreadPullRequestWatch | undefined,
Expand Down Expand Up @@ -2836,6 +2849,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
cause: "The pull request watch ended or its thread settled while it was read.",
});
}
if (
sortedWatchEvents(link.watch.events)?.join() !==
sortedWatchEvents(command.expectedEvents)?.join()
) {
return yield* new OrchestratorDispatchError({
commandId: command.commandId,
commandType: command.type,
cause: "The pull request watch events changed while it was read.",
});
}
yield* dispatchThreadMutation(command, events, effects);
if (command.wake === undefined) return;
yield* dispatchMessage(
Expand Down Expand Up @@ -3724,7 +3747,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
: linked;
const existing = links.find(visible);
if (existing === undefined) return thread;
const watch =
const observed =
command.type === "thread.pull-request-watch.sync"
? existing.watch?.startedAt === command.startedAt
? (command.watch ?? undefined)
Expand Down Expand Up @@ -3756,6 +3779,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
remarkIds: [],
wakes: 0,
});
// The filter is the watcher's choice, not read state, so a sync never changes it.
const events = sortedWatchEvents(
command.type === "thread.pull-request.watch" ? command.events : existing.watch?.events,
);
const watch =
observed === undefined || observed.events?.join() === events?.join()
? observed
: withWatchEvents(observed, events);
if (watch === existing.watch && links === linked) return thread;
const final =
command.type === "thread.pull-request-watch.sync" && watch === undefined
Expand Down
4 changes: 3 additions & 1 deletion apps/server/src/orchestration-v2/PullRequestWatchReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,8 @@ interface ReadPlan {
readonly activity: boolean;
}

const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`;
const watchKey = ({ thread, watch }: WatchTarget) =>
`${thread.id} ${watch.startedAt} ${watch.events?.join() ?? "all"}`;

type WatchEndReason =
| "merged"
Expand Down Expand Up @@ -283,6 +284,7 @@ export const make = Effect.gen(function* () {
threadId: target.thread.id,
...identityOf(target.link),
startedAt: target.watch.startedAt,
...(target.watch.events === undefined ? {} : { expectedEvents: target.watch.events }),
watch: next,
...(endedWatch === undefined ? {} : { endedWatch }),
...(wake === undefined
Expand Down
25 changes: 25 additions & 0 deletions apps/server/src/orchestration-v2/pullRequestWatch.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,31 @@ const remark = (
const noRemarks: ReadonlyArray<PullRequestComment> = [];

describe("evaluatePullRequestWatch", () => {
it("wakes only for the events a watch lists, and still records the rest", () => {
const green = detail({ checks: [check("lint", "success")], mergeability: "conflicting" });
const comment = {
...remark("reviewer", "2026-10-02T12:01:00.000Z"),
kind: "issue-comment" as const,
};
const review = { ...remark("reviewer", "2026-10-02T12:02:00.000Z"), kind: "review" as const };
const remarks: ReadonlyArray<PullRequestComment> = [comment, review];
assert.deepEqual(
evaluatePullRequestWatch(watch(), green, noRemarks).changes.map((change) => change.kind),
["checks-passed", "conflicting"],
);

const quiet = evaluatePullRequestWatch(watch({ events: ["comment"] }), green, noRemarks);
assert.deepEqual(quiet.changes, []);
assert.include(quiet.next, { passed: true, conflicting: true, wakes: 0 });

const reviews = evaluatePullRequestWatch(watch({ events: ["review"] }), green, remarks);
assert.deepEqual(reviews.changes, [{ kind: "remarks", remarks: [review] }]);
assert.equal(reviews.next.remarksThrough, review.createdAt);

const conversation = evaluatePullRequestWatch(watch({ events: ["comment"] }), green, remarks);
assert.deepEqual(conversation.changes, [{ kind: "remarks", remarks: [comment] }]);
});

it("reports each failure at once, even while another check never finishes", () => {
const bot = check("CodeRabbit", "pending");
const first = detail({ checks: [check("lint", "failure"), check("test", "pending"), bot] });
Expand Down
34 changes: 31 additions & 3 deletions apps/server/src/orchestration-v2/pullRequestWatch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ import type {
PullRequestComment,
PullRequestDetail,
ThreadPullRequestWatch,
ThreadPullRequestWatchEvent,
ThreadPullRequestWatchEvents,
} from "@t3tools/contracts";
import * as DateTime from "effect/DateTime";

Expand Down Expand Up @@ -31,6 +33,28 @@ export interface PullRequestWatchReport {
const isFailedCheck = (check: PullRequestCheck) =>
check.status === "failure" || check.status === "cancelled" || check.status === "action-required";

const CHANGE_EVENT: Record<
Exclude<PullRequestWatchChange["kind"], "remarks">,
ThreadPullRequestWatchEvent
> = {
"checks-failed": "checks_failed",
"checks-passed": "checks_passed",
"no-checks": "no_checks",
conflicting: "conflict",
};

const remarkEvent = (remark: PullRequestComment): ThreadPullRequestWatchEvent =>
remark.kind === "issue-comment" ? "comment" : "review";

function wanted(
change: PullRequestWatchChange,
events: ThreadPullRequestWatchEvents,
): Array<PullRequestWatchChange> {
if (change.kind !== "remarks") return events.includes(CHANGE_EVENT[change.kind]) ? [change] : [];
const remarks = change.remarks.filter((remark) => events.includes(remarkEvent(remark)));
return remarks.length === 0 ? [] : [{ kind: "remarks", remarks }];
}

export function evaluatePullRequestWatch(
watch: ThreadPullRequestWatch,
detail: Pick<
Expand Down Expand Up @@ -133,11 +157,15 @@ export function evaluatePullRequestWatch(
const conflicting =
detail.mergeability === "unknown" ? watch.conflicting : detail.mergeability === "conflicting";

const commentsOnly = changes.length > 0 && changes.every((change) => change.kind === "remarks");
const progress = headMoved || (changes.length > 0 && !commentsOnly);
// Filtered changes still advance the state above, so widening the filter never replays them.
const events = watch.events;
const waking =
events === undefined ? changes : changes.flatMap((change) => wanted(change, events));
const commentsOnly = waking.length > 0 && waking.every((change) => change.kind === "remarks");
const progress = headMoved || (waking.length > 0 && !commentsOnly);
const wakes = (progress ? 0 : watch.wakes) + (commentsOnly ? 1 : 0);
return {
changes,
changes: waking,
next: {
startedAt: watch.startedAt,
headSha,
Expand Down
Loading
Loading