diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index aba4cbe78..dadf62037 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -150,6 +150,7 @@ describe("OrchestrationEngine", () => { updatedAt: "2026-03-03T00:00:03.000Z", archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 617ef2812..91d99ae39 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -592,6 +592,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti updatedAt: event.payload.updatedAt, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: 0, doneOverride: null, doneOverrideAt: null, lastSeenAt: null, @@ -664,6 +665,21 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.pull-request-automation-changed": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + pullRequestAutoFix: event.payload.autoFix ? 1 : 0, + updatedAt: event.payload.updatedAt, + }); + return; + } + // Inbox filing and read state deliberately leave `updatedAt` alone -- // the client weighs both stamps against the thread's real activity. case "thread.done-override-set": { diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index da61c84a5..622a4cffc 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -377,6 +377,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { updatedAt: "2026-02-24T00:00:03.000Z", archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, @@ -501,6 +502,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { updatedAt: "2026-02-24T00:00:03.000Z", archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, session: { diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index f7ef3e0e1..f869718b2 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -526,6 +526,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updated_at AS "updatedAt", archived_at AS "archivedAt", pinned_at AS "pinnedAt", + pull_request_auto_fix AS "pullRequestAutoFix", done_override AS "doneOverride", done_override_at AS "doneOverrideAt", last_seen_at AS "lastSeenAt", @@ -564,6 +565,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updated_at AS "updatedAt", archived_at AS "archivedAt", pinned_at AS "pinnedAt", + pull_request_auto_fix AS "pullRequestAutoFix", done_override AS "doneOverride", done_override_at AS "doneOverrideAt", last_seen_at AS "lastSeenAt", @@ -604,6 +606,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updated_at AS "updatedAt", archived_at AS "archivedAt", pinned_at AS "pinnedAt", + pull_request_auto_fix AS "pullRequestAutoFix", done_override AS "doneOverride", done_override_at AS "doneOverrideAt", last_seen_at AS "lastSeenAt", @@ -1214,6 +1217,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updated_at AS "updatedAt", archived_at AS "archivedAt", pinned_at AS "pinnedAt", + pull_request_auto_fix AS "pullRequestAutoFix", done_override AS "doneOverride", done_override_at AS "doneOverrideAt", last_seen_at AS "lastSeenAt", @@ -1786,6 +1790,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updatedAt: row.updatedAt, archivedAt: row.archivedAt, pinnedAt: row.pinnedAt, + pullRequestAutoFix: (row.pullRequestAutoFix ?? 0) > 0, doneOverride: mapThreadDoneOverride(row), lastSeenAt: row.lastSeenAt ?? null, deletedAt: row.deletedAt, @@ -2032,6 +2037,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updatedAt: row.updatedAt, archivedAt: row.archivedAt, pinnedAt: row.pinnedAt, + pullRequestAutoFix: (row.pullRequestAutoFix ?? 0) > 0, doneOverride: mapThreadDoneOverride(row), lastSeenAt: row.lastSeenAt ?? null, deletedAt: row.deletedAt, @@ -2180,6 +2186,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updatedAt: row.updatedAt, archivedAt: row.archivedAt, pinnedAt: row.pinnedAt, + pullRequestAutoFix: (row.pullRequestAutoFix ?? 0) > 0, doneOverride: mapThreadDoneOverride(row), lastSeenAt: row.lastSeenAt ?? null, session: sessionByThread.get(row.threadId) ?? null, @@ -2331,6 +2338,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updatedAt: row.updatedAt, archivedAt: row.archivedAt, pinnedAt: row.pinnedAt, + pullRequestAutoFix: (row.pullRequestAutoFix ?? 0) > 0, doneOverride: mapThreadDoneOverride(row), lastSeenAt: row.lastSeenAt ?? null, session: sessionByThread.get(row.threadId) ?? null, @@ -2608,6 +2616,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updatedAt: threadRow.value.updatedAt, archivedAt: threadRow.value.archivedAt, pinnedAt: threadRow.value.pinnedAt, + pullRequestAutoFix: (threadRow.value.pullRequestAutoFix ?? 0) > 0, doneOverride: mapThreadDoneOverride(threadRow.value), lastSeenAt: threadRow.value.lastSeenAt ?? null, session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null, @@ -2721,6 +2730,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updatedAt: threadRow.value.updatedAt, archivedAt: threadRow.value.archivedAt, pinnedAt: threadRow.value.pinnedAt, + pullRequestAutoFix: (threadRow.value.pullRequestAutoFix ?? 0) > 0, doneOverride: mapThreadDoneOverride(threadRow.value), lastSeenAt: threadRow.value.lastSeenAt ?? null, deletedAt: null, diff --git a/apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.test.ts b/apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.test.ts new file mode 100644 index 000000000..5bd2d4caa --- /dev/null +++ b/apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.test.ts @@ -0,0 +1,477 @@ +import { + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationCommand, + type OrchestrationProjectShell, + type OrchestrationThreadShell, + type PullRequestActivity, + type PullRequestCheck, + type PullRequestDetail, + type VcsStatusRemoteResult, +} from "@threadlines/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as ManagedRuntime from "effect/ManagedRuntime"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; +import { afterEach, describe, expect, it } from "vite-plus/test"; + +import { GitManager, type GitManagerShape } from "../../git/GitManager.ts"; +import { + PullRequestService, + type PullRequestServiceShape, +} from "../../pullRequest/PullRequestService.ts"; +import { + OrchestrationEngineService, + type OrchestrationEngineShape, +} from "../Services/OrchestrationEngine.ts"; +import { + ProjectionSnapshotQuery, + type ProjectionSnapshotQueryShape, +} from "../Services/ProjectionSnapshotQuery.ts"; +import { PullRequestAutoFixWatcher } from "../Services/PullRequestAutoFixWatcher.ts"; +import { makePullRequestAutoFixWatcherLive } from "./PullRequestAutoFixWatcher.ts"; + +const NOW_ISO = "2026-05-04T10:00:00.000Z"; +const PROJECT_ID = ProjectId.make("project-auto-fix"); +const THREAD_ID = ThreadId.make("thread-auto-fix"); +const REPOSITORY = "acme/widgets"; +const PR_NUMBER = 42; + +function project(overrides: Partial = {}): OrchestrationProjectShell { + return { + id: PROJECT_ID, + kind: "workspace", + title: "Widgets", + workspaceRoot: "/repos/widgets", + repositoryIdentity: { + canonicalKey: "github|acme/widgets", + locator: { + source: "git-remote", + remoteName: "origin", + remoteUrl: "https://github.com/acme/widgets.git", + }, + displayName: REPOSITORY, + provider: "github", + owner: "acme", + name: "widgets", + }, + defaultModelSelection: null, + scripts: [], + createdAt: NOW_ISO, + updatedAt: NOW_ISO, + ...overrides, + }; +} + +function thread(overrides: Partial = {}): OrchestrationThreadShell { + return { + id: THREAD_ID, + projectId: PROJECT_ID, + title: "Fix the widget", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: "fix-the-widget", + worktreePath: null, + effectiveCwd: null, + goal: null, + latestTurn: null, + createdAt: NOW_ISO, + updatedAt: NOW_ISO, + archivedAt: null, + pinnedAt: null, + pullRequestAutoFix: true, + doneOverride: null, + lastSeenAt: null, + session: null, + latestUserMessageAt: null, + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + cumulativeDiffStat: null, + diffStatBaselineTurnCount: 0, + ...overrides, + }; +} + +const OPEN_PULL_REQUEST: VcsStatusRemoteResult = { + hasUpstream: true, + aheadCount: 0, + behindCount: 0, + pr: { + number: PR_NUMBER, + title: "Fix the widget", + url: `https://github.com/${REPOSITORY}/pull/${PR_NUMBER}`, + baseRef: "main", + headRef: "fix-the-widget", + state: "open", + }, +}; + +function detail(checks: readonly PullRequestCheck[]): PullRequestDetail { + return { + provider: "github", + projectId: PROJECT_ID, + projectTitle: "Widgets", + workspaceRoot: "/repos/widgets", + repository: REPOSITORY, + number: PR_NUMBER, + title: "Fix the widget", + body: "", + url: `https://github.com/${REPOSITORY}/pull/${PR_NUMBER}`, + author: { login: "will", isBot: false, avatarUrl: null }, + state: "open", + isDraft: false, + mergeability: "mergeable", + additions: 1, + deletions: 0, + changedFiles: 1, + headBranch: "fix-the-widget", + baseBranch: "main", + createdAt: NOW_ISO, + updatedAt: NOW_ISO, + mergedAt: null, + closedAt: null, + viewerIsAuthor: true, + reviewers: [], + labels: [], + checks, + checksState: checks.some((check) => check.status === "failure") ? "failure" : "success", + viewer: { canWrite: true, canReview: false, canManage: true }, + mergeMethods: ["squash"], + capabilities: { + diff: true, + comment: true, + actions: [], + mergeMethods: ["squash"], + updateMethods: ["merge"], + reactions: true, + review: { inlineComment: true, reply: true, resolve: true, verdicts: ["comment"] }, + reviewers: { request: true, listCandidates: true }, + edit: { pullRequest: true, comment: true }, + }, + baseComparison: "up-to-date", + behindBy: 0, + autoMergeEnabled: false, + isStacked: false, + defaultBranch: "main", + }; +} + +const PASSING_CHECK: PullRequestCheck = { + name: "typecheck", + status: "success", + description: null, + url: "https://github.com/acme/widgets/runs/1", +}; +const FAILING_CHECK: PullRequestCheck = { ...PASSING_CHECK, status: "failure" }; + +function activity( + comments: PullRequestActivity["comments"] = [], + reviewThreads: PullRequestActivity["reviewThreads"] = [], +): PullRequestActivity { + return { comments, commits: [], reviewThreads, reactions: [] }; +} + +function reviewerComment(id: string, body: string): PullRequestActivity["comments"][number] { + return { + id, + kind: "issue-comment", + author: { login: "dana", isBot: false, avatarUrl: null }, + body, + createdAt: NOW_ISO, + url: null, + reviewState: null, + reactions: [], + viewerIsAuthor: false, + }; +} + +/** Every read the watcher makes, scripted per sweep so a test can change one. */ +interface HostScript { + readonly remote: VcsStatusRemoteResult | null; + readonly detail: PullRequestDetail; + readonly activity: PullRequestActivity; +} + +function makeSnapshotQuery(input: { + readonly readThreads: () => readonly OrchestrationThreadShell[]; + readonly projects: readonly OrchestrationProjectShell[]; +}): ProjectionSnapshotQueryShape { + const shellSnapshot = () => ({ + snapshotSequence: 1, + projects: [...input.projects], + threads: [...input.readThreads()], + updatedAt: NOW_ISO, + }); + return { + getProjectCatalog: () => Effect.die("unused"), + getCommandReadModel: () => Effect.die("unused"), + getSnapshot: () => Effect.die("unused"), + getShellSnapshot: () => Effect.sync(shellSnapshot), + getArchivedShellSnapshot: () => Effect.die("unused"), + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), + getCounts: () => Effect.die("unused"), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), + getProjectShellById: () => Effect.succeed(Option.none()), + getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadCheckpointContext: () => Effect.succeed(Option.none()), + getFullThreadDiffContext: () => Effect.succeed(Option.none()), + listThreadDiffStatBaselines: () => Effect.succeed([]), + listThreadTurnOverlapsSince: () => Effect.succeed([]), + getThreadShellById: () => Effect.succeed(Option.none()), + getThreadDetailById: () => Effect.succeed(Option.none()), + }; +} + +describe("PullRequestAutoFixWatcher", () => { + let runtime: ManagedRuntime.ManagedRuntime | null = null; + + afterEach(async () => { + if (runtime) { + await runtime.dispose(); + } + runtime = null; + }); + + async function createHarness(input: { + readonly threads: readonly OrchestrationThreadShell[]; + readonly projects?: readonly OrchestrationProjectShell[]; + readonly script: HostScript; + }) { + let script = input.script; + let threads = input.threads; + const dispatched: Array> = []; + + const orchestrationEngine: OrchestrationEngineShape = { + readEvents: () => Stream.empty, + getCommandReceipt: () => Effect.succeed(Option.none()), + dispatch: (command) => { + if (command.type !== "thread.turn.start") { + return Effect.die(`Unexpected command: ${command.type}`); + } + return Effect.sync(() => { + dispatched.push(command); + return { sequence: dispatched.length }; + }); + }, + streamDomainEvents: Stream.empty, + subscribeDomainEvents: Effect.succeed(Stream.empty), + }; + + const gitManager: GitManagerShape = { + status: () => Effect.die("unused"), + localStatus: () => Effect.die("unused"), + remoteStatus: () => Effect.sync(() => script.remote), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + resolvePullRequest: () => Effect.die("unused"), + preparePullRequestThread: () => Effect.die("unused"), + runStackedAction: () => Effect.die("unused"), + generateCommitMessage: () => Effect.die("unused"), + }; + + const pullRequestService: PullRequestServiceShape = { + list: () => Effect.die("unused"), + detail: () => Effect.sync(() => script.detail), + activity: () => Effect.sync(() => script.activity), + diff: () => Effect.die("unused"), + comment: () => Effect.die("unused"), + runAction: () => Effect.die("unused"), + submitReview: () => Effect.die("unused"), + replyToThread: () => Effect.die("unused"), + setThreadResolution: () => Effect.die("unused"), + setReaction: () => Effect.die("unused"), + update: () => Effect.die("unused"), + updateComment: () => Effect.die("unused"), + reviewerCandidates: () => Effect.die("unused"), + requestReviewers: () => Effect.die("unused"), + }; + + const layer = makePullRequestAutoFixWatcherLive({ sweepIntervalMs: 60_000 }).pipe( + Layer.provideMerge( + Layer.succeed( + ProjectionSnapshotQuery, + makeSnapshotQuery({ + readThreads: () => threads, + projects: input.projects ?? [project()], + }), + ), + ), + Layer.provideMerge(Layer.succeed(OrchestrationEngineService, orchestrationEngine)), + Layer.provideMerge(Layer.succeed(GitManager, gitManager)), + Layer.provideMerge(Layer.succeed(PullRequestService, pullRequestService)), + ); + runtime = ManagedRuntime.make(layer); + const watcher = await runtime.runPromise(Effect.service(PullRequestAutoFixWatcher)); + const sweep = () => runtime!.runPromise(watcher.sweepNow()); + const setScript = (next: HostScript) => { + script = next; + }; + const setThreads = (next: readonly OrchestrationThreadShell[]) => { + threads = next; + }; + return { dispatched, setScript, setThreads, sweep }; + } + + it("records a baseline on first sight and starts nothing", async () => { + const { dispatched, sweep } = await createHarness({ + threads: [thread()], + script: { + remote: OPEN_PULL_REQUEST, + detail: detail([FAILING_CHECK]), + activity: activity([reviewerComment("c1", "Already said this")]), + }, + }); + + expect(await sweep()).toBe(0); + expect(dispatched).toEqual([]); + }); + + it("starts a turn when a passing check begins to fail", async () => { + const { dispatched, setScript, sweep } = await createHarness({ + threads: [thread()], + script: { + remote: OPEN_PULL_REQUEST, + detail: detail([PASSING_CHECK]), + activity: activity(), + }, + }); + + await sweep(); + setScript({ + remote: OPEN_PULL_REQUEST, + detail: detail([FAILING_CHECK]), + activity: activity(), + }); + + expect(await sweep()).toBe(1); + expect(dispatched).toHaveLength(1); + expect(dispatched[0]?.commandId).toContain(`pull-request-auto-fix:${THREAD_ID}:`); + expect(dispatched[0]?.threadId).toBe(THREAD_ID); + expect(dispatched[0]?.message.text).toBe( + [ + `Pull request #${PR_NUMBER} on ${REPOSITORY} needs attention.`, + "", + "These checks failed:", + "- typecheck (https://github.com/acme/widgets/runs/1)", + "", + "Fix what needs fixing, run the project's checks, commit on this branch, and push so the pull request updates.", + ].join("\n"), + ); + + // The same failure is not news twice. + expect(await sweep()).toBe(0); + expect(dispatched).toHaveLength(1); + }); + + it("starts a turn for a new comment from someone else, once", async () => { + const { dispatched, setScript, sweep } = await createHarness({ + threads: [thread()], + script: { remote: OPEN_PULL_REQUEST, detail: detail([]), activity: activity() }, + }); + + await sweep(); + setScript({ + remote: OPEN_PULL_REQUEST, + detail: detail([]), + activity: activity([reviewerComment("c1", "This leaks.")]), + }); + + expect(await sweep()).toBe(1); + expect(dispatched[0]?.message.text).toContain("New review comments:"); + expect(dispatched[0]?.message.text).toContain("dana wrote:\n> This leaks."); + + expect(await sweep()).toBe(0); + expect(dispatched).toHaveLength(1); + }); + + it("starts nothing for an archived or disarmed thread", async () => { + const { dispatched, setScript, sweep } = await createHarness({ + threads: [ + thread({ id: ThreadId.make("archived"), archivedAt: NOW_ISO }), + thread({ id: ThreadId.make("disarmed"), pullRequestAutoFix: false }), + ], + script: { + remote: OPEN_PULL_REQUEST, + detail: detail([PASSING_CHECK]), + activity: activity(), + }, + }); + + await sweep(); + setScript({ + remote: OPEN_PULL_REQUEST, + detail: detail([FAILING_CHECK]), + activity: activity(), + }); + + expect(await sweep()).toBe(0); + expect(dispatched).toEqual([]); + }); + + it("waits out a busy thread without forgetting what it had already seen", async () => { + const { dispatched, setScript, setThreads, sweep } = await createHarness({ + threads: [thread()], + script: { + remote: OPEN_PULL_REQUEST, + detail: detail([PASSING_CHECK]), + activity: activity(), + }, + }); + + await sweep(); + // The check fails while a turn is running, so this sweep does nothing -- + // and must not re-baseline the failure away. + setThreads([ + thread({ + latestTurn: { + turnId: TurnId.make("turn-1"), + state: "running", + requestedAt: NOW_ISO, + startedAt: NOW_ISO, + completedAt: null, + assistantMessageId: null, + }, + }), + ]); + setScript({ + remote: OPEN_PULL_REQUEST, + detail: detail([FAILING_CHECK]), + activity: activity(), + }); + expect(await sweep()).toBe(0); + expect(dispatched).toEqual([]); + + setThreads([thread()]); + expect(await sweep()).toBe(1); + }); + + it("stops starting turns once the per-pull-request cap is reached", async () => { + const { dispatched, setScript, sweep } = await createHarness({ + threads: [thread()], + script: { remote: OPEN_PULL_REQUEST, detail: detail([]), activity: activity() }, + }); + + await sweep(); + // A fresh comment each sweep: without the cap this would fire every time. + for (let index = 1; index <= 5; index += 1) { + setScript({ + remote: OPEN_PULL_REQUEST, + detail: detail([]), + activity: activity( + Array.from({ length: index }, (_unused, position) => + reviewerComment(`c${position + 1}`, `Remark ${position + 1}`), + ), + ), + }); + await sweep(); + } + + expect(dispatched).toHaveLength(3); + }); +}); diff --git a/apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.ts b/apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.ts new file mode 100644 index 000000000..1c7c65378 --- /dev/null +++ b/apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.ts @@ -0,0 +1,423 @@ +import { randomUUID } from "node:crypto"; + +import { + CommandId, + MessageId, + type OrchestrationProjectShell, + type OrchestrationThreadShell, + type PullRequestActivity, + type PullRequestDetail, + type ThreadId, +} from "@threadlines/contracts"; +import { + buildPullRequestAutoFixPrompt, + type PullRequestAutoFixCheck, + type PullRequestAutoFixComment, +} from "@threadlines/shared/pullRequestAutoFix"; +import { changeRequestRepositoryName } from "@threadlines/shared/sourceControl"; +import { resolveThreadWorkingCwd } from "@threadlines/shared/threadCwd"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; +import * as Semaphore from "effect/Semaphore"; + +import { GitManager } from "../../git/GitManager.ts"; +import { PullRequestService } from "../../pullRequest/PullRequestService.ts"; +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +import { + PullRequestAutoFixWatcher, + type PullRequestAutoFixWatcherShape, +} from "../Services/PullRequestAutoFixWatcher.ts"; + +const DEFAULT_SWEEP_INTERVAL_MS = 120 * 1_000; + +/** + * How many turns one thread's pull request may start on its own before the + * watcher stands down. A fix that keeps failing is a conversation for the user, + * not a loop for the agent. Counted per server lifetime, like the baseline. + */ +const MAX_AUTO_TURNS_PER_PULL_REQUEST = 3; + +/** Only GitHub answers the reads this needs; see `WRITE_ACCESS_HOSTS`. */ +const SUPPORTED_PROVIDER = "github"; + +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +/** What the watcher had already seen for one thread's pull request. */ +interface AutoFixBaseline { + /** Check names failing when the pull request was last looked at. */ + readonly failingCheckNames: ReadonlySet; + /** Every comment id seen, the viewer's own included. */ + readonly commentIds: ReadonlySet; +} + +/** One remark, flattened out of wherever the host filed it. */ +interface AutoFixComment { + readonly id: string; + readonly author: string | null; + readonly body: string; + readonly viewerIsAuthor: boolean; +} + +/** One thread and pull request number: the baseline and the cap are both per pair. */ +function pairKey(threadId: ThreadId, number: number): string { + return `${threadId}#${number}`; +} + +function threadIdFromPairKey(key: string): string { + return key.slice(0, key.lastIndexOf("#")); +} + +/** + * Whether a turn is in flight, or the thread is waiting on the user. Starting a + * turn on top of either would be refused or would bury the question, so both + * mean "come back next sweep". The shell read model says this three ways: the + * latest turn's own state, the live session, and the pending-request flags. + */ +function hasWorkInProgress(thread: OrchestrationThreadShell): boolean { + if (thread.latestTurn?.state === "running") { + return true; + } + if ( + thread.hasPendingApprovals || + thread.hasPendingUserInput || + thread.hasActionableProposedPlan + ) { + return true; + } + const session = thread.session; + if (session === null) { + return false; + } + return ( + session.status === "running" || + session.activeTurnId !== null || + (session.pendingBackgroundTaskCount ?? 0) > 0 + ); +} + +/** Every remark on a pull request: the conversation, then each diff thread's. */ +function listActivityComments(activity: PullRequestActivity): ReadonlyArray { + return [ + ...activity.comments.map((comment) => ({ + id: comment.id, + author: comment.author?.login ?? null, + body: comment.body, + viewerIsAuthor: comment.viewerIsAuthor, + })), + ...activity.reviewThreads.flatMap((thread) => + thread.comments.map((comment) => ({ + id: comment.id, + author: comment.author?.login ?? null, + body: comment.body, + viewerIsAuthor: comment.viewerIsAuthor, + })), + ), + ]; +} + +function failingCheckNames(detail: PullRequestDetail): ReadonlySet { + return new Set( + detail.checks.filter((check) => check.status === "failure").map((check) => check.name), + ); +} + +export interface PullRequestAutoFixWatcherLiveOptions { + readonly sweepIntervalMs?: number; +} + +const makePullRequestAutoFixWatcher = (options?: PullRequestAutoFixWatcherLiveOptions) => + Effect.gen(function* () { + const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + const orchestrationEngine = yield* OrchestrationEngineService; + const gitManager = yield* GitManager; + const pullRequestService = yield* PullRequestService; + + const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS); + // Dropped when a thread stops being a candidate, so turning the switch off + // and on again re-observes instead of firing on what it finds. + const baselinesRef = yield* Ref.make(new Map()); + // Never dropped: the cap is per server lifetime, so flipping the switch + // cannot buy three more turns. + const turnCountsRef = yield* Ref.make(new Map()); + const cappedLoggedRef = yield* Ref.make(new Set()); + // One sweep at a time: the interval fiber and `sweepNow` reach for the same + // baselines, and two sweeps racing would start the same turn twice. + const sweepSemaphore = yield* Semaphore.make(1); + + /** A read that must not take the sweep down with it. */ + const attempt = (operation: string, threadId: ThreadId, effect: Effect.Effect) => + effect.pipe( + Effect.catchCause((cause) => + Effect.logWarning(`pull-request.auto-fix.${operation}-failed`, { + threadId, + cause: Cause.pretty(cause), + }).pipe(Effect.as(null)), + ), + ); + + const startTurn = (input: { + readonly thread: OrchestrationThreadShell; + readonly number: number; + readonly text: string; + readonly trigger: "checks" | "comments" | "checks-and-comments"; + }) => + Effect.gen(function* () { + const createdAt = yield* nowIso; + const started = yield* orchestrationEngine + .dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`pull-request-auto-fix:${input.thread.id}:${randomUUID()}`), + threadId: input.thread.id, + message: { + messageId: MessageId.make(randomUUID()), + role: "user", + text: input.text, + attachments: [], + }, + modelSelection: input.thread.modelSelection, + runtimeMode: input.thread.runtimeMode, + interactionMode: input.thread.interactionMode, + createdAt, + }) + .pipe( + Effect.as(true), + Effect.catchCause((cause) => + Effect.logWarning("pull-request.auto-fix.dispatch-failed", { + threadId: input.thread.id, + number: input.number, + cause: Cause.pretty(cause), + }).pipe(Effect.as(false)), + ), + ); + if (started) { + yield* Effect.logInfo("pull-request.auto-fix.turn-started", { + threadId: input.thread.id, + number: input.number, + trigger: input.trigger, + }); + } + return started; + }); + + /** + * One candidate thread. Answers whether it started a turn, and leaves the + * baseline holding what its pull request looks like now. + */ + const sweepThread = Effect.fn("PullRequestAutoFixWatcher.sweepThread")(function* (input: { + readonly thread: OrchestrationThreadShell; + readonly project: OrchestrationProjectShell; + readonly repository: string; + }) { + const { thread, project, repository } = input; + const cwd = resolveThreadWorkingCwd({ + projectCwd: project.workspaceRoot, + worktreePath: thread.worktreePath, + effectiveCwd: thread.effectiveCwd, + }); + + const remote = yield* attempt("status-read", thread.id, gitManager.remoteStatus({ cwd })); + const pullRequest = remote?.pr ?? null; + if (pullRequest === null || pullRequest.state !== "open") { + return false; + } + + const key = pairKey(thread.id, pullRequest.number); + const turnCount = yield* Ref.get(turnCountsRef).pipe( + Effect.map((counts) => counts.get(key) ?? 0), + ); + if (turnCount >= MAX_AUTO_TURNS_PER_PULL_REQUEST) { + const alreadyLogged = yield* Ref.modify(cappedLoggedRef, (logged) => { + const seen = logged.has(key); + logged.add(key); + return [seen, logged]; + }); + if (!alreadyLogged) { + yield* Effect.logWarning("pull-request.auto-fix.cap-reached", { + threadId: thread.id, + number: pullRequest.number, + cap: MAX_AUTO_TURNS_PER_PULL_REQUEST, + }); + } + return false; + } + + const reference = { + projectId: project.id, + repository, + number: pullRequest.number, + force: true, + } as const; + const detail = yield* attempt("detail-read", thread.id, pullRequestService.detail(reference)); + if (detail === null) { + return false; + } + const activity = yield* attempt( + "activity-read", + thread.id, + pullRequestService.activity(reference), + ); + if (activity === null) { + return false; + } + + const currentFailing = failingCheckNames(detail); + const comments = listActivityComments(activity); + const commentIds = new Set(comments.map((comment) => comment.id)); + const baseline = yield* Ref.get(baselinesRef).pipe(Effect.map((map) => map.get(key))); + + // First sight of this pull request, whether because the server just + // started or because the switch just came on. Record it and act on what + // happens next, not on what was already there. + if (baseline === undefined) { + yield* Ref.update(baselinesRef, (map) => + map.set(key, { failingCheckNames: currentFailing, commentIds }), + ); + return false; + } + + // A check that has stopped failing leaves the baseline, so the same check + // failing again after a push is news again. + const carriedFailing = new Set( + [...baseline.failingCheckNames].filter((name) => currentFailing.has(name)), + ); + // A rollup still running is not a verdict: wait for it rather than send + // the agent after a check that may pass on its own. + const newlyFailing: PullRequestAutoFixCheck[] = + detail.checksState === "pending" + ? [] + : detail.checks + .filter((check) => check.status === "failure" && !carriedFailing.has(check.name)) + .map((check) => ({ name: check.name, url: check.url })); + const newComments: PullRequestAutoFixComment[] = comments + .filter((comment) => !comment.viewerIsAuthor && !baseline.commentIds.has(comment.id)) + .map((comment) => ({ author: comment.author, body: comment.body })); + + const text = buildPullRequestAutoFixPrompt({ + number: pullRequest.number, + repository, + failingChecks: newlyFailing, + comments: newComments, + }); + if (text === null) { + // Nothing new. Keep the checks that are still failing, and remember + // every remark, the viewer's own included, so none of them fires later. + yield* Ref.update(baselinesRef, (map) => + map.set(key, { failingCheckNames: carriedFailing, commentIds }), + ); + return false; + } + + const started = yield* startTurn({ + thread, + number: pullRequest.number, + text, + trigger: + newlyFailing.length > 0 && newComments.length > 0 + ? "checks-and-comments" + : newlyFailing.length > 0 + ? "checks" + : "comments", + }); + // The baseline advances either way: a dispatch the engine refused is not + // one to retry every two minutes. + yield* Ref.update(baselinesRef, (map) => + map.set(key, { + failingCheckNames: newlyFailing.length > 0 ? currentFailing : carriedFailing, + commentIds, + }), + ); + if (started) { + yield* Ref.update(turnCountsRef, (counts) => counts.set(key, turnCount + 1)); + } + return started; + }); + + const runSweep = Effect.gen(function* () { + const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( + Effect.catchCause((cause) => + Effect.logWarning("pull-request.auto-fix.snapshot-query-failed", { + cause: Cause.pretty(cause), + }).pipe(Effect.as(null)), + ), + ); + if (snapshot === null) { + return 0; + } + + const projectsById = new Map(snapshot.projects.map((project) => [project.id, project])); + // A thread that is no longer armed forgets its baseline, so re-arming it + // starts from a fresh observation rather than the state it left behind. + // Being armed is the only test: a thread that is merely busy this sweep + // must keep its baseline, or the failure it should act on would be + // re-observed as the state it started from once the turn ends. + const armedThreadIds = new Set( + snapshot.threads + .filter((thread) => thread.pullRequestAutoFix) + .map((thread) => thread.id as string), + ); + yield* Ref.update(baselinesRef, (map) => { + // Deleting through a Map's own key iterator is defined behaviour, so + // the keys do not need copying first. + for (const key of map.keys()) { + if (!armedThreadIds.has(threadIdFromPairKey(key))) { + map.delete(key); + } + } + return map; + }); + + // Deleted threads never reach the shell snapshot, so archived, unbranched + // and busy are the only ones left to rule out here. + const candidates = snapshot.threads.flatMap((thread) => { + if (!thread.pullRequestAutoFix || thread.archivedAt !== null || thread.branch === null) { + return []; + } + if (hasWorkInProgress(thread)) { + return []; + } + const project = projectsById.get(thread.projectId); + if (project === undefined || project.repositoryIdentity?.provider !== SUPPORTED_PROVIDER) { + return []; + } + const repository = changeRequestRepositoryName(project.repositoryIdentity); + if (repository === null) { + return []; + } + return [{ thread, project, repository }]; + }); + + let startedCount = 0; + for (const candidate of candidates) { + const started = yield* sweepThread(candidate); + if (started) { + startedCount += 1; + } + } + return startedCount; + }); + + const sweep = sweepSemaphore.withPermits(1)(runSweep); + + const sweepNow: PullRequestAutoFixWatcherShape["sweepNow"] = () => sweep; + + const start: PullRequestAutoFixWatcherShape["start"] = () => + Effect.gen(function* () { + yield* Effect.forkScoped( + sweep.pipe(Effect.repeat(Schedule.spaced(Duration.millis(sweepIntervalMs)))), + ); + yield* Effect.logInfo("pull-request.auto-fix.started", { sweepIntervalMs }); + }); + + return { start, sweepNow } satisfies PullRequestAutoFixWatcherShape; + }); + +export const makePullRequestAutoFixWatcherLive = (options?: PullRequestAutoFixWatcherLiveOptions) => + Layer.effect(PullRequestAutoFixWatcher, makePullRequestAutoFixWatcher(options)); + +export const PullRequestAutoFixWatcherLive = makePullRequestAutoFixWatcherLive(); diff --git a/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts b/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts index 7dd8c2dc7..b893412e2 100644 --- a/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts @@ -64,6 +64,7 @@ function thread( updatedAt: daysAgo(45), archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, session: null, diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 870ed7663..8d3ba6ed7 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -11,6 +11,7 @@ import { ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema, ThreadUnarchivedPayload as ContractsThreadUnarchivedPayloadSchema, ThreadUnpinnedPayload as ContractsThreadUnpinnedPayloadSchema, + ThreadPullRequestAutomationChangedPayload as ContractsThreadPullRequestAutomationChangedPayloadSchema, ThreadDoneOverrideSetPayload as ContractsThreadDoneOverrideSetPayloadSchema, ThreadSeenSetPayload as ContractsThreadSeenSetPayloadSchema, ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema, @@ -47,6 +48,8 @@ export const ThreadInteractionModeSetPayload = ContractsThreadInteractionModeSet export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema; export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema; export const ThreadUnpinnedPayload = ContractsThreadUnpinnedPayloadSchema; +export const ThreadPullRequestAutomationChangedPayload = + ContractsThreadPullRequestAutomationChangedPayloadSchema; export const ThreadDoneOverrideSetPayload = ContractsThreadDoneOverrideSetPayloadSchema; export const ThreadSeenSetPayload = ContractsThreadSeenSetPayloadSchema; diff --git a/apps/server/src/orchestration/Services/PullRequestAutoFixWatcher.ts b/apps/server/src/orchestration/Services/PullRequestAutoFixWatcher.ts new file mode 100644 index 000000000..de2b50e47 --- /dev/null +++ b/apps/server/src/orchestration/Services/PullRequestAutoFixWatcher.ts @@ -0,0 +1,22 @@ +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; +import type * as Scope from "effect/Scope"; + +/** + * Watches the pull requests of threads whose "Fix failing checks and review + * comments" switch is on, and starts a turn when a check has just failed or a + * reviewer has just said something. + * + * It runs only while the server does: the baseline it compares against lives in + * memory, so a restart re-observes rather than replaying what it missed. + */ +export interface PullRequestAutoFixWatcherShape { + readonly start: () => Effect.Effect; + /** Runs one sweep and answers how many turns it started. */ + readonly sweepNow: () => Effect.Effect; +} + +export class PullRequestAutoFixWatcher extends Context.Service< + PullRequestAutoFixWatcher, + PullRequestAutoFixWatcherShape +>()("threadlines/orchestration/Services/PullRequestAutoFixWatcher") {} diff --git a/apps/server/src/orchestration/commandInvariants.test.ts b/apps/server/src/orchestration/commandInvariants.test.ts index 90abf1e41..82ac5d7c8 100644 --- a/apps/server/src/orchestration/commandInvariants.test.ts +++ b/apps/server/src/orchestration/commandInvariants.test.ts @@ -73,6 +73,7 @@ const readModel: OrchestrationReadModel = { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, latestTurn: null, @@ -102,6 +103,7 @@ const readModel: OrchestrationReadModel = { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, latestTurn: null, diff --git a/apps/server/src/orchestration/decider.checkoutSwitch.test.ts b/apps/server/src/orchestration/decider.checkoutSwitch.test.ts index bbcc99f2f..46ab3eea1 100644 --- a/apps/server/src/orchestration/decider.checkoutSwitch.test.ts +++ b/apps/server/src/orchestration/decider.checkoutSwitch.test.ts @@ -87,6 +87,7 @@ function makeReadModel(input: { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/decider.diffStatRebase.test.ts b/apps/server/src/orchestration/decider.diffStatRebase.test.ts index 39755d253..4489b90d1 100644 --- a/apps/server/src/orchestration/decider.diffStatRebase.test.ts +++ b/apps/server/src/orchestration/decider.diffStatRebase.test.ts @@ -40,6 +40,7 @@ function makeReadModel(diffStatBaselineTurnCount: number): OrchestrationReadMode updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/decider.followUp.test.ts b/apps/server/src/orchestration/decider.followUp.test.ts index e71e8631b..12b69d5e0 100644 --- a/apps/server/src/orchestration/decider.followUp.test.ts +++ b/apps/server/src/orchestration/decider.followUp.test.ts @@ -47,6 +47,7 @@ const readModelAfterProviderDelivery: OrchestrationReadModel = { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/decider.generalChats.test.ts b/apps/server/src/orchestration/decider.generalChats.test.ts index 79ea4f8a3..b1d52ef1f 100644 --- a/apps/server/src/orchestration/decider.generalChats.test.ts +++ b/apps/server/src/orchestration/decider.generalChats.test.ts @@ -260,6 +260,7 @@ function makeThread(input: { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/decider.inboxLifecycle.test.ts b/apps/server/src/orchestration/decider.inboxLifecycle.test.ts index 1b554c82a..5ff06e4f7 100644 --- a/apps/server/src/orchestration/decider.inboxLifecycle.test.ts +++ b/apps/server/src/orchestration/decider.inboxLifecycle.test.ts @@ -39,6 +39,7 @@ function makeReadModel(): OrchestrationReadModel { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, @@ -102,6 +103,48 @@ describe("decider inbox lifecycle", () => { expect(event?.payload).not.toHaveProperty("updatedAt"); }); + it("records the pull request auto-fix switch, and refuses it on an archived thread", async () => { + const decided = await Effect.runPromise( + decideOrchestrationCommand({ + command: { + type: "thread.pull-request-automation.set", + commandId: CommandId.make("cmd-auto-fix-on"), + threadId, + autoFix: true, + }, + readModel: makeReadModel(), + }), + ); + const event = Array.isArray(decided) ? decided[0] : decided; + + expect(event).toMatchObject({ + type: "thread.pull-request-automation-changed", + aggregateKind: "thread", + aggregateId: threadId, + payload: { threadId, autoFix: true }, + }); + // Unlike filing, arming the watcher is work on the thread, so it stamps. + expect(event?.payload).toHaveProperty("updatedAt"); + + const archived = makeReadModel(); + await expect( + Effect.runPromise( + decideOrchestrationCommand({ + command: { + type: "thread.pull-request-automation.set", + commandId: CommandId.make("cmd-auto-fix-archived"), + threadId, + autoFix: true, + }, + readModel: { + ...archived, + threads: archived.threads.map((thread) => ({ ...thread, archivedAt: now })), + }, + }), + ), + ).rejects.toThrow("already archived"); + }); + it("rejects filing a thread that does not exist", async () => { await expect( Effect.runPromise( diff --git a/apps/server/src/orchestration/decider.proposedPlan.test.ts b/apps/server/src/orchestration/decider.proposedPlan.test.ts index e77a0ae69..157457bbf 100644 --- a/apps/server/src/orchestration/decider.proposedPlan.test.ts +++ b/apps/server/src/orchestration/decider.proposedPlan.test.ts @@ -57,6 +57,7 @@ function makeReadModel(proposedPlan: OrchestrationProposedPlan): OrchestrationRe updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 583860d18..674c8c12d 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -515,6 +515,32 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + // Setting the switch to the value it already holds still emits: the event + // is the record of the user's word, and the projector writes the same + // value either way. There is no "no change" outcome in this decider. + case "thread.pull-request-automation.set": { + yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + const occurredAt = yield* nowIso; + return { + ...withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + }), + type: "thread.pull-request-automation-changed", + payload: { + threadId: command.threadId, + autoFix: command.autoFix, + updatedAt: occurredAt, + }, + }; + } + case "thread.done-override.set": { yield* requireThread({ readModel, diff --git a/apps/server/src/orchestration/decider.turnRetry.test.ts b/apps/server/src/orchestration/decider.turnRetry.test.ts index 8432eef98..9d510296c 100644 --- a/apps/server/src/orchestration/decider.turnRetry.test.ts +++ b/apps/server/src/orchestration/decider.turnRetry.test.ts @@ -60,6 +60,7 @@ function makeThread(overrides: Partial = {}): Orchestration updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index fe8e8787b..99c602e79 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -93,6 +93,7 @@ describe("orchestration projector", () => { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, @@ -524,6 +525,57 @@ describe("orchestration projector", () => { expect(unpinned.threads[0]?.pinnedAt).toBeNull(); }); + it("applies thread.pull-request-automation-changed events", async () => { + const now = "2026-01-01T00:00:00.000Z"; + const later = "2026-01-01T00:00:01.000Z"; + const created = await Effect.runPromise( + projectEvent( + createEmptyReadModel(now), + makeEvent({ + sequence: 1, + type: "thread.created", + aggregateKind: "thread", + aggregateId: "thread-1", + occurredAt: now, + commandId: "cmd-thread-create", + payload: { + threadId: "thread-1", + projectId: "project-1", + title: "demo", + modelSelection: { + provider: ProviderDriverKind.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }), + ), + ); + expect(created.threads[0]?.pullRequestAutoFix).toBe(false); + + const armed = await Effect.runPromise( + projectEvent( + created, + makeEvent({ + sequence: 2, + type: "thread.pull-request-automation-changed", + aggregateKind: "thread", + aggregateId: "thread-1", + occurredAt: later, + commandId: "cmd-thread-auto-fix", + payload: { threadId: "thread-1", autoFix: true, updatedAt: later }, + }), + ), + ); + expect(armed.threads[0]?.pullRequestAutoFix).toBe(true); + expect(armed.threads[0]?.updatedAt).toBe(later); + }); + it("keeps projector forward-compatible for unhandled event types", async () => { const now = "2026-01-01T00:00:00.000Z"; const model = createEmptyReadModel(now); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 909232ecf..0bd2983e4 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -35,6 +35,7 @@ import { ThreadSeenSetPayload, ThreadUnarchivedPayload, ThreadUnpinnedPayload, + ThreadPullRequestAutomationChangedPayload, ThreadRevertedPayload, ThreadSessionSetPayload, ThreadRealtimeStateSetPayload, @@ -301,6 +302,7 @@ export function projectEvent( updatedAt: payload.updatedAt, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, @@ -377,6 +379,22 @@ export function projectEvent( })), ); + case "thread.pull-request-automation-changed": + return decodeForEvent( + ThreadPullRequestAutomationChangedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + pullRequestAutoFix: payload.autoFix, + updatedAt: payload.updatedAt, + }), + })), + ); + // Neither lifecycle event touches `updatedAt`: filing or reading a thread // is not work on it, and the inbox weighs these stamps against activity. case "thread.done-override-set": diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index d1949ee04..a28df1cc0 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -50,6 +50,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { updated_at, archived_at, pinned_at, + pull_request_auto_fix, done_override, done_override_at, last_seen_at, @@ -79,6 +80,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.updatedAt}, ${row.archivedAt}, ${row.pinnedAt}, + ${row.pullRequestAutoFix ?? 0}, ${row.doneOverride ?? null}, ${row.doneOverrideAt ?? null}, ${row.lastSeenAt ?? null}, @@ -108,6 +110,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { updated_at = excluded.updated_at, archived_at = excluded.archived_at, pinned_at = excluded.pinned_at, + pull_request_auto_fix = excluded.pull_request_auto_fix, done_override = excluded.done_override, done_override_at = excluded.done_override_at, last_seen_at = excluded.last_seen_at, @@ -144,6 +147,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { updated_at AS "updatedAt", archived_at AS "archivedAt", pinned_at AS "pinnedAt", + pull_request_auto_fix AS "pullRequestAutoFix", done_override AS "doneOverride", done_override_at AS "doneOverrideAt", last_seen_at AS "lastSeenAt", @@ -182,6 +186,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { updated_at AS "updatedAt", archived_at AS "archivedAt", pinned_at AS "pinnedAt", + pull_request_auto_fix AS "pullRequestAutoFix", done_override AS "doneOverride", done_override_at AS "doneOverrideAt", last_seen_at AS "lastSeenAt", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 9b7a34c12..0bba4b6fa 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -70,6 +70,7 @@ import Migration0048 from "./Migrations/048_BackfillThreadSubagents.ts"; import Migration0049 from "./Migrations/049_ProjectionThreadSubagentsBackgrounded.ts"; import Migration0050 from "./Migrations/050_ProjectionTranscriptEventSequence.ts"; import Migration0051 from "./Migrations/051_ProjectionThreadsBlockingUserInput.ts"; +import Migration0052 from "./Migrations/052_ProjectionThreadsPullRequestAutoFix.ts"; /** * Migration loader with all migrations defined inline. @@ -133,6 +134,7 @@ export const migrationEntries = [ [49, "ProjectionThreadSubagentsBackgrounded", Migration0049], [50, "ProjectionTranscriptEventSequence", Migration0050], [51, "ProjectionThreadsBlockingUserInput", Migration0051], + [52, "ProjectionThreadsPullRequestAutoFix", Migration0052], ] as const; /** Highest id in the full registry, so the "n of total" count is the real total. */ diff --git a/apps/server/src/persistence/Migrations/052_ProjectionThreadsPullRequestAutoFix.ts b/apps/server/src/persistence/Migrations/052_ProjectionThreadsPullRequestAutoFix.ts new file mode 100644 index 000000000..5aa52fa62 --- /dev/null +++ b/apps/server/src/persistence/Migrations/052_ProjectionThreadsPullRequestAutoFix.ts @@ -0,0 +1,21 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +/** + * Whether the server watches this thread's pull request and starts a turn when + * a check fails or a review comment arrives. Off for every existing thread, so + * the column's default is the whole migration: no rows are rewritten. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + if (!columns.some((column) => column.name === "pull_request_auto_fix")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN pull_request_auto_fix INTEGER NOT NULL DEFAULT 0 + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index 41e9d500f..b025beb39 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -55,6 +55,12 @@ export const ProjectionThread = Schema.Struct({ updatedAt: IsoDateTime, archivedAt: Schema.NullOr(IsoDateTime), pinnedAt: Schema.NullOr(IsoDateTime), + /** + * Whether the server watches this thread's pull request. Optional so rows + * written before migration 052 decode; absent reads as 0 (off), which is + * what the migration's column default gives every existing row. + */ + pullRequestAutoFix: Schema.optional(NonNegativeInt), /** * The user's explicit inbox filing and its stamp. Optional so rows written * before migration 042 decode; absent reads as "never filed". Kept as two diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 08c740d10..91cd692e7 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -110,6 +110,7 @@ function makeReadModel( updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, latestUserMessageAt: null, diff --git a/apps/server/src/provider/ProviderReviewCoordinator.test.ts b/apps/server/src/provider/ProviderReviewCoordinator.test.ts index 3a07d9362..12112aaf2 100644 --- a/apps/server/src/provider/ProviderReviewCoordinator.test.ts +++ b/apps/server/src/provider/ProviderReviewCoordinator.test.ts @@ -47,6 +47,7 @@ function makeThreadShell( updatedAt: NOW, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, session: null, diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 60db81eb5..10a4128e1 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -191,6 +191,7 @@ const makeDefaultOrchestrationReadModel = () => { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, latestTurn: null, @@ -245,6 +246,7 @@ const makeDefaultOrchestrationThreadShell = ( updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, session: null, @@ -3723,6 +3725,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { updatedAt: now, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, latestTurn: null, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 5ac9c39f2..89881d288 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -26,6 +26,7 @@ import { ProviderEventLoggersLive } from "./provider/Layers/ProviderEventLoggers import { ProviderServiceLive } from "./provider/Layers/ProviderService.ts"; import { ProviderSessionReaperLive } from "./provider/Layers/ProviderSessionReaper.ts"; import { ThreadAutoArchiveSweeperLive } from "./orchestration/Layers/ThreadAutoArchiveSweeper.ts"; +import { PullRequestAutoFixWatcherLive } from "./orchestration/Layers/PullRequestAutoFixWatcher.ts"; import { BootstrapTurnStartRunsLive } from "./orchestration/Layers/BootstrapTurnStartRuns.ts"; import { CheckpointDiffQueryLive } from "./checkpointing/Layers/CheckpointDiffQuery.ts"; import { CheckpointRevertLive } from "./checkpointing/Layers/CheckpointRevert.ts"; @@ -195,6 +196,7 @@ const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(StorageMaintenanceDaemonLive), Layer.provideMerge(RuntimeReceiptBusLive), Layer.provideMerge(AutomaticGitFetchSupervisorLive), + Layer.provideMerge(PullRequestAutoFixWatcherLive), ); const ProviderSessionDirectoryLayerLive = ProviderSessionDirectoryLive.pipe( diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index a805093de..9689ca31f 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -30,6 +30,7 @@ import { OrchestrationEngineService } from "./orchestration/Services/Orchestrati import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor.ts"; import { ThreadAutoArchiveSweeper } from "./orchestration/Services/ThreadAutoArchiveSweeper.ts"; +import { PullRequestAutoFixWatcher } from "./orchestration/Services/PullRequestAutoFixWatcher.ts"; import { AutomaticGitFetchSupervisor } from "./vcs/AutomaticGitFetchSupervisor.ts"; import { pauseActiveThreadGoalForStop } from "./orchestration/threadGoalLifecycle.ts"; import { ServerLifecycleEvents } from "./serverLifecycleEvents.ts"; @@ -329,6 +330,7 @@ export const makeServerRuntimeStartup = Effect.gen(function* () { const orchestrationReactor = yield* OrchestrationReactor; const providerSessionReaper = yield* ProviderSessionReaper; const threadAutoArchiveSweeper = yield* ThreadAutoArchiveSweeper; + const pullRequestAutoFixWatcher = yield* PullRequestAutoFixWatcher; const automaticGitFetchSupervisor = yield* AutomaticGitFetchSupervisor; const sleepInhibitor = yield* SleepInhibitor; const lifecycleEvents = yield* ServerLifecycleEvents; @@ -388,6 +390,7 @@ export const makeServerRuntimeStartup = Effect.gen(function* () { yield* orchestrationReactor.start().pipe(Scope.provide(reactorScope)); yield* providerSessionReaper.start().pipe(Scope.provide(reactorScope)); yield* threadAutoArchiveSweeper.start().pipe(Scope.provide(reactorScope)); + yield* pullRequestAutoFixWatcher.start().pipe(Scope.provide(reactorScope)); yield* automaticGitFetchSupervisor.start().pipe(Scope.provide(reactorScope)); yield* sleepInhibitor.start().pipe(Scope.provide(reactorScope)); }), diff --git a/apps/web/src/components/ChatView.browser.tsx b/apps/web/src/components/ChatView.browser.tsx index 4cec8a7b4..603d5b816 100644 --- a/apps/web/src/components/ChatView.browser.tsx +++ b/apps/web/src/components/ChatView.browser.tsx @@ -356,6 +356,7 @@ function createSnapshotForTargetUser(options: { targetText: string; targetAttachmentCount?: number; threadPinnedAt?: string | null; + threadPullRequestAutoFix?: boolean; sessionActiveTurnId?: TurnId | null; sessionStatus?: OrchestrationSessionStatus; }): OrchestrationReadModel { @@ -432,6 +433,7 @@ function createSnapshotForTargetUser(options: { updatedAt: NOW_ISO, archivedAt: null, pinnedAt: options.threadPinnedAt ?? null, + pullRequestAutoFix: options.threadPullRequestAutoFix ?? false, doneOverride: null, lastSeenAt: null, deletedAt: null, @@ -648,6 +650,7 @@ function addThreadToSnapshot( updatedAt: NOW_ISO, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, @@ -1213,6 +1216,7 @@ function createSnapshotWithSecondaryProject(options?: { updatedAt: isoAt(31), deletedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, messages: [], @@ -1253,6 +1257,7 @@ function createSnapshotWithSecondaryProject(options?: { updatedAt: isoAt(25), deletedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, messages: [], diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 531d78d04..dcb83b277 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -181,6 +181,7 @@ export function mergeLocalDraftThreadWithServerThread( createdAt: serverThread.createdAt, archivedAt: serverThread.archivedAt, pinnedAt: serverThread.pinnedAt, + pullRequestAutoFix: serverThread.pullRequestAutoFix ?? false, doneOverride: serverThread.doneOverride, lastSeenAt: serverThread.lastSeenAt, updatedAt: serverThread.updatedAt, diff --git a/apps/web/src/components/KeybindingsToast.browser.tsx b/apps/web/src/components/KeybindingsToast.browser.tsx index 8f4778ec4..7aec7da25 100644 --- a/apps/web/src/components/KeybindingsToast.browser.tsx +++ b/apps/web/src/components/KeybindingsToast.browser.tsx @@ -193,6 +193,7 @@ function createMinimalSnapshot(): OrchestrationReadModel { updatedAt: NOW_ISO, archivedAt: null, pinnedAt: null, + pullRequestAutoFix: false, doneOverride: null, lastSeenAt: null, deletedAt: null, @@ -255,6 +256,7 @@ function toShellSnapshot(snapshot: OrchestrationReadModel) { updatedAt: thread.updatedAt, archivedAt: thread.archivedAt, pinnedAt: thread.pinnedAt, + pullRequestAutoFix: false, doneOverride: thread.doneOverride, lastSeenAt: thread.lastSeenAt, session: thread.session, diff --git a/apps/web/src/components/chat/ComposerPullRequestRow.tsx b/apps/web/src/components/chat/ComposerPullRequestRow.tsx index 4f4699f9e..2d5a2e1b4 100644 --- a/apps/web/src/components/chat/ComposerPullRequestRow.tsx +++ b/apps/web/src/components/chat/ComposerPullRequestRow.tsx @@ -6,8 +6,8 @@ * everything about it; neither is in view while you are writing the next * message, which is exactly when "did the checks pass" decides what you type. * The row is one line: which pull request, on which branch, how big, and how - * its checks are going, with the two switches that decide what happens when - * they pass. + * its checks are going, with the switches that decide what happens when they + * pass, and what happens when they do not. * * @module ComposerPullRequestRow */ @@ -18,7 +18,7 @@ import type { PullRequestRef, } from "@threadlines/contracts"; import { useMutation, useQueryClient } from "@tanstack/react-query"; -import { ChevronDownIcon, ExternalLinkIcon, XIcon } from "lucide-react"; +import { ChevronDownIcon, ExternalLinkIcon, WrenchIcon, XIcon } from "lucide-react"; import { useState } from "react"; import { isElectron } from "../../env"; @@ -37,8 +37,10 @@ import { CHECK_TONES } from "../pull-requests/pullRequestPresentation"; import { pullRequestBadgeTone, type ThreadPullRequest } from "../pull-requests/pullRequests.logic"; import { Checkbox } from "../ui/checkbox"; import { Popover, PopoverPopup, PopoverTrigger } from "../ui/popover"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { DiffStatLabel } from "./DiffStatLabel"; import { + composerAutoFixOffered, composerAutoMergeControl, composerPullRequestCheckBuckets, composerPullRequestRow, @@ -46,6 +48,9 @@ import { type ComposerPullRequestChipTone, } from "./composerPullRequest.logic"; +/** Said the same way on the marker and its tooltip, so they read as one thing. */ +const AUTO_FIX_MARKER_LABEL = "Fixes failing checks and review comments on its own"; + /** Everything the thread route hands the composer about its pull request. */ export interface ComposerPullRequest { readonly environmentId: EnvironmentId; @@ -60,6 +65,9 @@ export interface ComposerPullRequest { readonly onOpen: () => void; /** Closes the row for this pull request in this thread. */ readonly onDismiss: () => void; + /** The thread's own switch: the server watches this pull request while it is on. */ + readonly autoFix: boolean; + readonly onAutoFixChange: (next: boolean) => void; } const CHIP_TONE_CLASS: Readonly< @@ -135,6 +143,24 @@ export function ComposerPullRequestRow({ /> ) : null} + {pullRequest.autoFix ? ( + + + + + } + /> + + {AUTO_FIX_MARKER_LABEL} + + + ) : null} ) : null} + {composerAutoFixOffered(detail) ? ( + + ) : null}