diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..26091236091f8fb4b2f87a14d6a7af5a074f7b52 --- /dev/null +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.test.ts @@ -0,0 +1,319 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + SourceControlProviderKind, + ThreadId, + type PullRequestDetail, + type MergebotPullRequestDetails, + type ThreadPullRequestSnapshot, +} from "@t3tools/contracts"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as MergebotQueue from "../mergebot/MergebotQueue.ts"; +import * as TestClock from "effect/testing/TestClock"; + +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as PullRequestService from "../pullRequest/PullRequestService.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; + +const threadId = ThreadId.make("watch-merge"); +const projectId = ProjectId.make("watch-project"); +const instanceId = ProviderInstanceId.make("codex"); +const key = { host: "github.com", repository: "owner/repository", number: 42 }; +const url = "https://github.com/owner/repository/pull/42"; +const timestamp = "2026-10-14T12:00:00.000Z"; +const adapter = { + instanceId, + driver: ProviderDriverKind.make("codex"), + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), + openSession: () => Effect.die("Merge-watch tests do not start providers"), +} as ProviderAdapter.ProviderAdapterV2["Service"]; +const database = SqlitePersistence.layerMemory; +const layerTest = Layer.mergeAll( + NodeCrypto.layer, + ProjectionStore.layer.pipe(Layer.provide(database)), + OrchestrationEventStore.layer.pipe(Layer.provide(database)), + ProviderReplayHarness.layerWithRegistry( + { name: "pull-request-watch" }, + ProviderAdapterRegistry.layerFromAdapters([adapter]), + { databaseLayer: database, runEffectWorker: false }, + ), +); + +const detail: PullRequestDetail = { + provider: SourceControlProviderKind.make("github"), + capabilities: { + diff: false, + comment: false, + actions: [], + mergeMethods: [], + search: false, + review: { inlineComment: false, reply: false, resolve: false, verdicts: [] }, + reviewers: { request: false, listCandidates: false }, + }, + viewerPermissions: { + actions: [], + comment: false, + resolve: false, + verdicts: [], + requestReviewers: false, + }, + projectId, + projectTitle: "Project", + workspaceRoot: "/workspace", + repository: key.repository, + number: key.number, + title: "Change", + body: "", + url, + author: null, + state: "merged", + isDraft: false, + mergeability: "mergeable", + additions: 0, + deletions: 0, + changedFiles: 0, + headBranch: "feature", + baseBranch: "main", + createdAt: timestamp, + updatedAt: timestamp, + mergedAt: timestamp, + closedAt: null, + reviewers: [], + labels: [], + checks: [], + mergeCapabilities: { merge: false, squash: false, rebase: false }, +}; + +const makeHarness = Effect.fn("makeMergeWatchHarness")(function* (options: { + readonly mergebot?: MergebotQueue.MergebotQueue["Service"]; + readonly snapshotState: ThreadPullRequestSnapshot["state"]; + readonly beforeRead?: (engine: Orchestrator.OrchestratorV2["Service"]) => Effect.Effect; +}) { + const engine = yield* Orchestrator.OrchestratorV2; + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId, + title: "Wait for merge", + modelSelection: { instanceId, model: "gpt-5.1-codex" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* engine.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("watch"), + threadId, + ...key, + watching: true, + link: { url, source: "agent" }, + }); + yield* engine.dispatch({ + type: "thread.pull-request-link.sync", + commandId: CommandId.make("snapshot"), + threadId, + ...key, + snapshot: { + state: options.snapshotState, + title: "Change", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: timestamp, + syncedAt: timestamp, + }, + stack: null, + }); + const reads = yield* Ref.make(0); + const reactor = yield* PullRequestWatchReactor.make.pipe( + Effect.provide( + Layer.merge( + options.mergebot + ? Layer.succeed(MergebotQueue.MergebotQueue, options.mergebot) + : Layer.mock(MergebotQueue.MergebotQueue)({ + configuration: null, + streamChanges: Stream.empty, + }), + Layer.mock(PullRequestService.PullRequestService)({ + watchFingerprint: () => Effect.succeed(null), + detail: () => + Ref.update(reads, (count) => count + 1).pipe( + Effect.andThen(options.beforeRead?.(engine) ?? Effect.void), + Effect.as(detail), + ), + activity: () => + Effect.succeed({ + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }), + }), + ), + ), + ); + return { engine, reactor, reads }; +}); + +it.effect.each(["merged", "open"] as const)( + "wakes once and ends the watch when the %s snapshot leads to a merged PR", + (snapshotState) => + Effect.gen(function* () { + const { engine, reactor, reads } = yield* makeHarness({ snapshotState }); + yield* reactor.sweep; + const first = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]); + assert.isUndefined(first.thread.pullRequests?.[0]?.watch); + assert.equal(first.runs.length, 1); + const notice = first.turnItems.find((item) => item.type === "notification"); + assert.ok(notice?.type === "notification"); + assert.equal(notice.summary, "Pull request #42 merged"); + assert.equal(notice.outcome, "completed"); + assert.deepEqual(notice.source, { kind: "monitor" }); + assert.equal(yield* Ref.get(reads), snapshotState === "merged" ? 0 : 1); + yield* reactor.sweep; + const second = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]); + assert.equal(second.runs.length, 1); + assert.equal(second.turnItems.filter((item) => item.type === "notification").length, 1); + }).pipe(Effect.provide(layerTest)), +); + +it.effect.each(["stop", "settle", "archive", "restart"] as const)( + "a %s during the host read prevents a stale merged wake", + (action) => + Effect.gen(function* () { + const { engine, reactor } = yield* makeHarness({ + snapshotState: "open", + beforeRead: (engine) => + Effect.gen(function* () { + if (action === "settle" || action === "archive") { + yield* engine.dispatch({ + type: action === "settle" ? "thread.settle" : "thread.archive", + commandId: CommandId.make(action), + threadId, + }); + return; + } + yield* engine.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("stop"), + threadId, + ...key, + watching: false, + }); + if (action === "restart") { + yield* TestClock.adjust(1); + yield* engine.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("restart"), + threadId, + ...key, + watching: true, + }); + } + }).pipe(Effect.orDie), + }); + yield* reactor.sweep; + const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]); + assert.equal(records.runs.length, 0); + assert.equal(records.turnItems.filter((item) => item.type === "notification").length, 0); + if (action === "restart") assert.isDefined(records.thread.pullRequests?.[0]?.watch); + else assert.isUndefined(records.thread.pullRequests?.[0]?.watch); + if (action === "settle") assert.equal(records.thread.settledOverride, "settled"); + if (action === "archive") assert.isNotNull(records.thread.archivedAt); + }).pipe(Effect.provide(layerTest)), +); + +it.effect.each(["settle", "archive"] as const)( + "a thread already %s does not wake on a merged snapshot", + (action) => + Effect.gen(function* () { + const { engine, reactor, reads } = yield* makeHarness({ snapshotState: "merged" }); + yield* engine.dispatch({ + type: action === "settle" ? "thread.settle" : "thread.archive", + commandId: CommandId.make(action), + threadId, + }); + yield* reactor.sweep; + const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]); + assert.equal(records.runs.length, 0); + assert.equal(records.turnItems.filter((item) => item.type === "notification").length, 0); + assert.equal(yield* Ref.get(reads), 0); + }).pipe(Effect.provide(layerTest)), +); + +it.effect( + "work PR watches verify Mergebot and never call GitHub detail, activity or fingerprint reads", + () => + Effect.gen(function* () { + let nativeReads = 0; + const value: MergebotPullRequestDetails = { + number: key.number, + headSha: "head", + lifecycle: "merged", + queueStatus: "not_queued", + position: null, + mergeEtaAt: null, + title: "Change", + body: null, + url: "https://mergebot.example/owner/repository/pull/42", + createdAt: timestamp, + updatedAt: timestamp, + mergedAt: timestamp, + closedAt: timestamp, + author: null, + additions: 3, + deletions: 1, + changedFilesCount: 1, + labels: [], + comments: [], + observedAt: Date.parse(timestamp), + draft: false, + baseBranch: "main", + headBranch: "feature", + ciStatus: "passed", + hasConflicts: false, + checks: [], + }; + const { engine, reactor, reads } = yield* makeHarness({ + snapshotState: "merged", + mergebot: { + configuration: { origin: "https://mergebot.example", repositories: [key.repository] }, + read: () => Effect.succeed(null), + watchQueue: () => Stream.empty, + findPullRequest: () => Effect.succeed(null), + readPullRequest: () => + Effect.sync(() => { + nativeReads++; + return value; + }), + streamChanges: Stream.empty, + }, + }); + yield* reactor.sweep; + const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]); + assert.equal(records.runs.length, 1); + assert.isUndefined(records.thread.pullRequests?.[0]?.watch); + assert.equal(yield* Ref.get(reads), 0); + assert.isAbove(nativeReads, 0); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 4bbfbb27eb62b1ad76a33ea5af9e49333853c210..236072bfe004aad4732730840af58d0bf3ebb651 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -1,8 +1,10 @@ import { CommandId, + isMergebotRepository, MessageId, type OrchestrationV2Notification, type PullRequestActivity, + type PullRequestDetail, type PullRequestComment, type PullRequestRef, type PullRequestThreadCommentsResult, @@ -23,6 +25,8 @@ 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 Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; @@ -30,6 +34,12 @@ import { type ProviderChangeRequestWatchFingerprint, PullRequestProviderError, } from "@t3tools/source-control-core/server/PullRequestProvider"; +import * as MergebotQueue from "../mergebot/MergebotQueue.ts"; +import { + mergebotChecks, + mergebotMergeability, + mergebotWatchFingerprint, +} from "../mergebot/pullRequestRead.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { forkParked } from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -217,9 +227,9 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc * of a project that watch it. Where the host has a fingerprint, one batched read of those tells * the pass which pull requests moved, and only those are read; elsewhere a pull request is read * unless its sync snapshot has not moved while nothing is in flight. - * Settling or archiving a thread ends its watches, and a merged or closed pull request ends - * its watch. Each ended watch is logged once with why it ended and how long it went without - * a push, for debugging watches that live too long. + * Settling or archiving a thread ends its watches. A merged or closed pull request wakes its + * active thread once and ends its watch. Each ended watch is logged once with why it ended + * and how long it went without a push, for debugging watches that live too long. */ export class PullRequestWatchReactor extends Context.Service< PullRequestWatchReactor, @@ -235,6 +245,8 @@ export const make = Effect.gen(function* () { const engine = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; const pullRequests = yield* PullRequestService.PullRequestService; + const mergebot = yield* MergebotQueue.MergebotQueue; + const sweepLock = yield* Semaphore.make(1); const crypto = yield* Crypto.Crypto; const bootedAt = yield* Clock.currentTimeMillis; const lives = new Map(); @@ -406,16 +418,32 @@ export const make = Effect.gen(function* () { Effect.tap(() => ended(target, "closed")), ); + const merged = (target: WatchTarget) => + record(target, null, { + text: `Pull request #${target.link.number} (${target.link.url}) merged. T3 Code stopped watching it. Continue any work you were waiting to do after this merge.`, + notification: { + source: { kind: "monitor" }, + outcome: "completed", + summary: `Pull request #${target.link.number} merged`, + }, + }).pipe( + Effect.tap(() => woke(target)), + Effect.tap(() => ended(target, "merged")), + ); + /** Why a watch ends without a host read; the rest are read once per pull request. */ + const isWorkLink = (link: ThreadPullRequestLink) => + normalizeThreadPullRequestKey(link).host === "github.com" && + isMergebotRepository(mergebot.configuration ?? undefined, link.repository); + const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined => - // A merged pull request cannot reopen. Settling and archiving end watches, and a subagent - // cannot start one; a watch left from before those rules ends here. - link.snapshot?.state === "merged" - ? "merged" - : thread.settledOverride === "settled" || thread.settledAt !== null - ? "settled" - : thread.lineage.relationshipToParent === "subagent" - ? "subagent" + // Inactive threads discard even a merged watch without waking. + thread.settledOverride === "settled" || thread.settledAt !== null + ? "settled" + : thread.lineage.relationshipToParent === "subagent" + ? "subagent" + : link.snapshot?.state === "merged" && !isWorkLink(link) + ? "merged" : undefined; /** @@ -451,8 +479,14 @@ export const make = Effect.gen(function* () { * sync-snapshot gating; "paused" skips the pass while the host is rate limited, since every * read would be refused. */ - const fingerprintOf = (group: WatchGroup) => - pullRequests.watchFingerprint(referenceOf(group)).pipe( + const fingerprintOf = Effect.fnUntraced(function* ( + group: WatchGroup, + ): Effect.fn.Return { + if (isWorkLink(group.targets[0]!.link)) { + const value = yield* mergebot.readPullRequest(referenceOf(group)); + return value === null ? "paused" : mergebotWatchFingerprint(value); + } + return yield* pullRequests.watchFingerprint(referenceOf(group)).pipe( Effect.catchCause( (cause): Effect.Effect => Cause.hasInterruptsOnly(cause) @@ -467,6 +501,7 @@ export const make = Effect.gen(function* () { }).pipe(Effect.as(null)), ), ); + }); const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* ( group: WatchGroup, @@ -479,18 +514,42 @@ export const make = Effect.gen(function* () { const reference = referenceOf(group); // The reads are recorded against this fingerprint, so a cached answer older than it must not // stand in for them. Clients are not told: this only drops this pull request's cached reads. - if (fingerprint !== null) yield* pullRequests.invalidate({ reference }); - const read = yield* Effect.exit( - Effect.all( - [ - pullRequests.detail({ ...reference, allowStale: false }), - plan.activity - ? pullRequests.activity(reference).pipe(Effect.map(Option.some)) - : Effect.succeed(Option.none()), - ], - { concurrency: 2 }, - ), - ); + const work = isWorkLink(group.targets[0]!.link); + const native = work ? yield* mergebot.readPullRequest(reference) : null; + if (work && native === null) return; + if (!work && fingerprint !== null) yield* pullRequests.invalidate({ reference }); + const reading: Effect.Effect< + readonly [ + Pick< + PullRequestDetail, + "state" | "headSha" | "baseBranch" | "mergeability" | "checks" | "author" | "viewer" + >, + Option.Option, + ], + PullRequestService.PullRequestError + > = + native !== null + ? Effect.succeed([ + { + state: native.lifecycle, + headSha: native.headSha, + baseBranch: native.baseBranch, + mergeability: mergebotMergeability(native), + checks: mergebotChecks(native), + author: native.author, + }, + Option.none(), + ] as const) + : Effect.all( + [ + pullRequests.detail({ ...reference, allowStale: false }), + plan.activity + ? pullRequests.activity(reference).pipe(Effect.map(Option.some)) + : Effect.succeed(Option.none()), + ], + { concurrency: 2 }, + ); + const read = yield* Effect.exit(reading); if (Exit.isFailure(read)) { if (Cause.hasInterruptsOnly(read.cause)) return yield* Effect.failCause(read.cause); lastReads.delete(group.key); @@ -524,17 +583,18 @@ export const make = Effect.gen(function* () { if (detail.state !== "open") { lastReads.delete(group.key); return yield* eachTarget(group, (target) => - detail.state === "closed" - ? closed(target) - : record(target, null).pipe(Effect.tap(() => ended(target, "merged"))), + detail.state === "closed" ? closed(target) : merged(target), ); } // Never advance the remark watermark past comments an incomplete read could have missed. // A pass that did not read the activity has no remarks to report. - const remarks = Option.isSome(activity) - ? yield* readRemarks(group.key, reference, activity.value) - : null; + const remarks = + native !== null + ? native.comments + : Option.isSome(activity) + ? yield* readRemarks(group.key, reference, activity.value) + : null; lastReads.set(group.key, { at: plan.activity || last === undefined ? now : last.at, fingerprint: group.fingerprint, @@ -608,8 +668,10 @@ export const make = Effect.gen(function* () { yield* Effect.forEach( ending, ([target, reason]) => - record(target, null).pipe( - Effect.tap(() => ended(target, reason)), + (reason === "merged" + ? merged(target) + : record(target, null).pipe(Effect.tap(() => ended(target, reason))) + ).pipe( Effect.catchCause( logFailure("pull request watch stop failed", { threadId: target.thread.id, @@ -639,11 +701,18 @@ export const make = Effect.gen(function* () { }).pipe( Effect.catchCause(logFailure("pull request watch sweep failed", {})), Effect.withSpan("PullRequestWatchReactor.sweep"), + sweepLock.withPermit, ); const start: PullRequestWatchReactor["Service"]["start"] = () => forkParked( - sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid), + Effect.all( + [ + sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid), + mergebot.streamChanges.pipe(Stream.runForEach(() => sweep)), + ], + { concurrency: "unbounded", discard: true }, + ), ); return { start, sweep } satisfies PullRequestWatchReactor["Service"]; diff --git a/docs/user/source-control.md b/docs/user/source-control.md index 4a222b9102f21eaf9926167f7ab6d2d0bb4e7103..b4b47b302e93d0697abf36d885da8355bf8b31ab 100644 --- a/docs/user/source-control.md +++ b/docs/user/source-control.md @@ -153,6 +153,12 @@ Open **Pull requests** to review changes and comments, request reviewers, check or merge. You can edit review titles and descriptions and your own comments where the host allows it. GitLab calls these merge requests. +Configured work repositories use Mergebot for PR status, checks, and live updates. +Their PR links open Mergebot inside T3 Code's Browser, with its queue and review actions. +Open **Mergebot** from the sidebar or Pull requests page to see the queue without a chat. +On mobile, these links use the environment's server Browser and require browser access. +If Mergebot disconnects, PR decisions wait for a verified update after it reconnects. + On mobile, **Pull requests** lists reviews across your connected environments. Open a review to read its description, activity, commits, and changed files. Line comments stay as drafts until you submit the review. **Work on PR** opens a chat with the review attached; sending @@ -279,8 +285,9 @@ the thread is active, the server checks the pull request every two minutes and w check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict. Threads in a project that watch the same pull request share one check. On GitHub, a check first asks whether anything changed and reads the pull request only when it did, which keeps -watching inside GitHub's rate limit. Comments from your own account do not wake it. Watching ends -when the pull request merges or closes, after 10 wakes in a row that bring only comments, after 8 +watching inside GitHub's rate limit. Comments from your own account do not wake it. A merge wakes +the agent once to finish any follow-up work and ends the watch. Watching also ends +when the pull request closes, after 10 wakes in a row that bring only comments, after 8 failed reads in a row, or when you press Stop on the thread. A rate limit only pauses watching. Settling or archiving a thread also ends all its watches. Unsettle the thread before starting a new watch. Subagents cannot watch pull requests; the thread that delegated to them does. To start or stop