From 71c8725544af40322f672aa66236328020513b05 Mon Sep 17 00:00:00 2001 From: badcuban <108198679+badcuban@users.noreply.github.com> Date: Sun, 6 Sep 2026 13:22:53 -0400 Subject: [PATCH] feat(server): a thread can fix its pull request's failing checks and review comments on its own A third switch in the composer's pull request popover, "Fix failing checks and review comments", arms the thread. While it is on, the server watches the pull request and, when a check starts failing or a reviewer comments, starts a turn in the thread with a message naming the checks and quoting the comments, and asking the agent to fix them, run the checks, commit and push. The switch is an orchestration event on the thread, projected to a new column with no backfill. The watcher is a reactor scoped to the server, so it runs only while the app is open. It records what it first sees and acts only on changes after that, skips threads with a turn in flight or a question pending, and stops after three automatic turns per pull request. The row shows a small wrench while the switch is on. --- .../Layers/OrchestrationEngine.test.ts | 1 + .../Layers/ProjectionPipeline.ts | 16 + .../Layers/ProjectionSnapshotQuery.test.ts | 2 + .../Layers/ProjectionSnapshotQuery.ts | 10 + .../Layers/PullRequestAutoFixWatcher.test.ts | 477 ++++++++++++++++++ .../Layers/PullRequestAutoFixWatcher.ts | 423 ++++++++++++++++ .../Layers/ThreadAutoArchiveSweeper.test.ts | 1 + apps/server/src/orchestration/Schemas.ts | 3 + .../Services/PullRequestAutoFixWatcher.ts | 22 + .../orchestration/commandInvariants.test.ts | 2 + .../decider.checkoutSwitch.test.ts | 1 + .../decider.diffStatRebase.test.ts | 1 + .../orchestration/decider.followUp.test.ts | 1 + .../decider.generalChats.test.ts | 1 + .../decider.inboxLifecycle.test.ts | 43 ++ .../decider.proposedPlan.test.ts | 1 + apps/server/src/orchestration/decider.ts | 26 + .../orchestration/decider.turnRetry.test.ts | 1 + .../src/orchestration/projector.test.ts | 52 ++ apps/server/src/orchestration/projector.ts | 18 + .../persistence/Layers/ProjectionThreads.ts | 5 + apps/server/src/persistence/Migrations.ts | 2 + ...052_ProjectionThreadsPullRequestAutoFix.ts | 21 + .../persistence/Services/ProjectionThreads.ts | 6 + .../Layers/ProviderSessionReaper.test.ts | 1 + .../ProviderReviewCoordinator.test.ts | 1 + apps/server/src/server.test.ts | 3 + apps/server/src/server.ts | 2 + apps/server/src/serverRuntimeStartup.ts | 3 + apps/web/src/components/ChatView.browser.tsx | 5 + apps/web/src/components/ChatView.logic.ts | 1 + .../components/KeybindingsToast.browser.tsx | 2 + .../chat/ComposerPullRequestRow.tsx | 74 ++- .../chat/composerPullRequest.logic.test.ts | 9 + .../chat/composerPullRequest.logic.ts | 10 + .../pull-requests/pullRequests.logic.ts | 9 +- .../service.threadSubscriptions.test.ts | 1 + apps/web/src/lib/threadPullRequestCommands.ts | 26 + .../routes/_chat.$environmentId.$threadId.tsx | 6 + apps/web/src/store.test.ts | 2 + apps/web/src/store.ts | 12 + apps/web/src/types.ts | 7 + packages/contracts/src/orchestration.ts | 34 ++ packages/shared/package.json | 4 + .../shared/src/pullRequestAutoFix.test.ts | 75 +++ packages/shared/src/pullRequestAutoFix.ts | 79 +++ 46 files changed, 1492 insertions(+), 10 deletions(-) create mode 100644 apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.test.ts create mode 100644 apps/server/src/orchestration/Layers/PullRequestAutoFixWatcher.ts create mode 100644 apps/server/src/orchestration/Services/PullRequestAutoFixWatcher.ts create mode 100644 apps/server/src/persistence/Migrations/052_ProjectionThreadsPullRequestAutoFix.ts create mode 100644 apps/web/src/lib/threadPullRequestCommands.ts create mode 100644 packages/shared/src/pullRequestAutoFix.test.ts create mode 100644 packages/shared/src/pullRequestAutoFix.ts 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}