diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 568025d523b3..c9e6a9cd8527 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -631,6 +631,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "thread.pull-request-link.sync": case "thread.pull-request.watch": case "thread.pull-request-watch.sync": + case "thread.pull-request-watch.checks-reopen": case "thread.pull-request.sync": case "thread.title.regeneration.complete": case "thread.project.adopt-folder": @@ -782,6 +783,26 @@ function withPullRequestWatch( return ended === undefined ? rest : { ...rest, endedWatch: ended }; } +function syncedPullRequestWatch( + existing: ThreadPullRequestWatch | undefined, + command: { + readonly watch: ThreadPullRequestWatch | null; + readonly checksPendingBefore?: boolean | undefined; + }, +): ThreadPullRequestWatch | undefined { + if (command.watch === null) return undefined; + // A push cleared the answer while this read was in flight, so its answer is the old head's. + if (command.checksPendingBefore === undefined || existing?.checksPending !== undefined) { + return command.watch; + } + return withoutChecksAnswer(command.watch); +} + +const withoutChecksAnswer = ({ + checksPending: _answer, + ...watch +}: ThreadPullRequestWatch): ThreadPullRequestWatch => watch; + function legacyPullRequestLink( thread: OrchestrationV2AppThread, linked: ThreadLinkedPullRequest, @@ -3871,6 +3892,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | "thread.pull-request-link.sync" | "thread.pull-request.watch" | "thread.pull-request-watch.sync" + | "thread.pull-request-watch.checks-reopen" | "thread.pull-request.sync" | "thread.title.regeneration.complete" | "thread.runtime-mode.set" @@ -4669,6 +4691,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedAt: command.type === "thread.pull-request-link.sync" ? thread.updatedAt : now, }; } + case "thread.pull-request-watch.checks-reopen": { + const linked = threadPullRequestsOf(thread); + if (!linked.some((link) => link.watch?.checksPending !== undefined)) return thread; + return { + ...thread, + pullRequests: linked.map((link) => + link.watch?.checksPending === undefined + ? link + : { ...link, watch: withoutChecksAnswer(link.watch) }, + ), + }; + } case "thread.pull-request.watch": case "thread.pull-request-watch.sync": { const key = normalizeThreadPullRequestKey(command); @@ -4705,7 +4739,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const watch = command.type === "thread.pull-request-watch.sync" ? existing.watch?.startedAt === command.startedAt - ? (command.watch ?? undefined) + ? syncedPullRequestWatch(existing.watch, command) : existing.watch : !command.watching ? undefined @@ -4736,6 +4770,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(existing.endedWatch?.approvalAnnounced !== undefined && { approvalAnnounced: existing.endedWatch.approvalAnnounced, }), + // A finished-checks answer may be an older head's; a fresh read decides. + ...(existing.endedWatch?.checksPending === true && { checksPending: true }), ...(existing.endedWatch?.checksAwaitedSince !== undefined && { checksAwaitedSince: existing.endedWatch.checksAwaitedSince, }), @@ -4823,6 +4859,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }; } })(); + if (command.type === "thread.pull-request-watch.checks-reopen") { + if (updatedThread === thread) return; + yield* emit( + events, + command, + )({ + type: "thread.pull-request-synced", + threadId: command.threadId, + providerInstanceId: updatedThread.providerInstanceId, + occurredAt: now, + payload: updatedThread, + }); + return; + } if ( command.type === "thread.pull-request-link.sync" || command.type === "thread.pull-request-watch.sync" @@ -14481,6 +14531,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "provider.switch": yield* dispatchThreadMutation(command, events, effects); break; + case "thread.pull-request-watch.checks-reopen": + yield* dispatchThreadMutation(command, events, effects); + break; case "thread.pull-request-watch.sync": yield* dispatchPullRequestWatchSync(command, events, effects); break; @@ -14838,6 +14891,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio planned.events.length > 0 || isUnobservedAcknowledgement(command) || command.type === "thread.background-work.settle" || + // A read can answer a watch again before a push's reopen lands; nothing is left to clear. + command.type === "thread.pull-request-watch.checks-reopen" || command.type === "thread.interrupt.unresponsive" || command.type === "thread.stop" || command.type === "subagent.stop" || diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 02a8e43c1d0e..8ec62d07e75e 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -12,6 +12,7 @@ import { ThreadId, TurnItemId, type OrchestrationV2Command as OrchestrationCommand, + type OrchestrationV2InternalCommand, type OrchestrationV2DomainEvent, type OrchestrationV2Run, type OrchestrationProjectShell, @@ -51,6 +52,10 @@ type SyncCommand = Extract< { readonly type: "thread.pull-request-link.sync" } >; type LinkCommand = Extract; +type ChecksReopenCommand = Extract< + OrchestrationV2InternalCommand, + { readonly type: "thread.pull-request-watch.checks-reopen" } +>; const testCrypto = SecureRandom.make((size) => new Uint8Array(size).fill(1)); @@ -189,6 +194,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: const shellSnapshotReads = yield* Ref.make(0); const syncCommands = yield* Ref.make>([]); const linkCommands = yield* Ref.make>([]); + const reopenCommands = yield* Queue.unbounded(); const summaryCalls = yield* Ref.make>([]); const stackCalls = yield* Ref.make>([]); const domainEvents = yield* Queue.unbounded(); @@ -223,6 +229,11 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: Effect.as({ sequence: 1, storedEvents: [] }), ); } + if (command.type === "thread.pull-request-watch.checks-reopen") { + return Queue.offer(reopenCommands, command).pipe( + Effect.as({ sequence: 1, storedEvents: [] }), + ); + } return Effect.die(new Error(`Unexpected command: ${command.type}`)); }; @@ -284,6 +295,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: shellSnapshotReads, syncCommands, linkCommands, + reopenCommands, summaryCalls, stackCalls, domainEvents, @@ -1880,6 +1892,56 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("reopens a watch's checks when its thread's agent pushes", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const watch = { + startedAt: NOW, + headSha: "aaaaaaaaaa", + failedChecks: [], + passed: true, + passedChecks: ["test"], + remarksThrough: NOW, + remarkIds: [], + conflicting: false, + wakes: 0, + }; + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { + pullRequests: [ + makeLink(7, { state: "open" }, { watch: { ...watch, checksPending: false } }), + ], + }), + makeThread("unread", { + pullRequests: [makeLink(8, { state: "open" }, { watch })], + }), + ]), + }); + + yield* Effect.gen(function* () { + yield* startAndSweep(fixture); + const agent = ThreadId.make("agent"); + yield* Queue.offerAll(fixture.domainEvents, [ + commandRan(agent, "git status && echo 'then git push'", undefined, "status"), + // A watch that has not read its checks yet has no answer to forget. + commandRan(ThreadId.make("unread"), "git push", undefined, "unread-push"), + commandRan(agent, "cd app && git -C . push --force-with-lease", undefined, "push"), + commandRan(agent, '/bin/zsh -lc "gh stack submit --auto"', undefined, "stack"), + ]); + for (const item of ["push", "stack"]) { + const command = yield* Queue.take(fixture.reopenCommands); + assert.deepStrictEqual( + [command.threadId, command.commandId], + [agent, `server:pr-push-checks:agent:item:agent:${item}`], + ); + } + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("reads open links again when the run ends, after `merge --auto` left them open", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 7573a95bb436..b654364d3a7f 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -72,6 +72,11 @@ const CREATE_COMMANDS = sourceControlDefinitions.definitions.flatMap( ); const createCommandCount = (code: string) => CREATE_COMMANDS.reduce((count, { run }) => count + [...code.matchAll(run)].length, 0); +// A push, directly or through a stack tool, that can start checks on a linked pull request. +const PUSH_COMMAND = new RegExp( + `${COMMAND_START}(?:git(?:\\s+-[cC]\\s+\\S+)*\\s+push|gh\\s+stack\\s+(?:push|submit)|cafe\\s+stack\\s+submit|jj\\s+git\\s+push)(?=$|[\\s;&|)])`, + "u", +); // Codex runs each command as `/bin/zsh -lc