diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 61929131e9d72..609903fc71f5c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1104,6 +1104,13 @@ function isStalePullRequestWatchStamp( candidate.source !== "stack-dismissed" && threadPullRequestKeysEqual(candidate, stamp), ); if (link === undefined) return true; + if (stamp.readBlocker !== undefined) { + return ( + link.watch === undefined || + link.watch.readBlocker !== stamp.readBlocker || + pullRequestSnapshotEndsWatch(link.snapshot, link.watch.startedAt) + ); + } if (stamp.watchEnded === true) { // The final notice outlives its watch, but a replacement watch supersedes it. const latestWatch = link.watch ?? link.endedWatch; @@ -3974,9 +3981,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } if (wake === undefined || outdated[0]) return; const { sections, ...wakeMessage } = wake; - const stamps = entries.map(({ entry, key }, index) => ({ + const stamps = entries.map(({ entry, key, link }, index) => ({ ...key, watchStartedAt: entry.startedAt, + ...(entry.watch?.readBlocker !== undefined && link?.watch?.readBlocker === undefined + ? { readBlocker: entry.watch.readBlocker } + : {}), headSha: entry.includesRemarks === true ? null : (entry.watch?.headSha ?? null), ...(entry.watch === null ? { watchEnded: true } : {}), ...(entries.length > 1 && sections?.[index] !== undefined @@ -5092,6 +5102,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(existing.endedWatch?.checksAwaitedSince !== undefined && { checksAwaitedSince: existing.endedWatch.checksAwaitedSince, }), + ...(existing.endedWatch?.readBlocker !== undefined && { + readBlocker: existing.endedWatch.readBlocker, + }), remarksThrough: startedAt, remarkIds: [], wakes: 0, diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 5b69a7efa3377..300b458c7255f 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -216,6 +216,7 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc left.rebaseRequired === right.rebaseRequired && left.mergeQueueRemovalId === right.mergeQueueRemovalId && left.wakes === right.wakes && + left.readBlocker === right.readBlocker && left.awaitingBaseline === right.awaitingBaseline && left.checksAwaitedSince === right.checksAwaitedSince && left.checksPending === right.checksPending @@ -615,25 +616,70 @@ export const make = Effect.gen(function* () { ], { concurrency: 2 }, ); - if (Exit.isFailure(detailRead)) { - if (Cause.hasInterruptsOnly(detailRead.cause)) - return yield* Effect.failCause(detailRead.cause); + const activityAccessFailure = + Exit.isFailure(activityRead) && + ["forbidden", "unauthenticated"].includes(providerFailureReason(activityRead.cause) ?? ""); + const readFailure = Exit.isFailure(detailRead) + ? activityAccessFailure && + providerFailureReason(detailRead.cause) === "failed" && + Exit.isFailure(activityRead) + ? activityRead + : detailRead + : detailRead.value.state === "open" && activityAccessFailure && Exit.isFailure(activityRead) + ? activityRead + : null; + let newReadBlocker: string | undefined; + let blockerWake: ((target: WatchTarget) => Wake) | undefined; + if (readFailure !== null) { + if (Cause.hasInterruptsOnly(readFailure.cause)) + return yield* Effect.failCause(readFailure.cause); lastReads.delete(group.key); - const reason = providerFailureReason(detailRead.cause); + const reason = providerFailureReason(readFailure.cause); if (reason === "rate-limited") return; // Transport and host failures can outlast the failure limit; keep the watch for recovery. if (reason === "failed") { readFailures.delete(group.key); - return yield* Effect.failCause(detailRead.cause); + return yield* Effect.failCause(readFailure.cause); } const failures = (readFailures.get(group.key) ?? 0) + 1; readFailures.set(group.key, failures); - if (failures >= READ_FAILURE_LIMIT && (yield* eachTarget(group, giveUp))) { - readFailures.delete(group.key); + if (reason === "forbidden" || reason === "unauthenticated") { + if (failures >= READ_FAILURE_LIMIT) { + const operation = readFailure === detailRead ? "detail" : "activity"; + newReadBlocker = `${operation}:${reason}:${yield* random.uuidV4()}`; + const readBlocker = newReadBlocker; + blockerWake = (target) => ({ + text: `T3 Code cannot read pull request #${target.link.number} (${target.link.url}): host ${operation} access is ${reason}. The watch remains active and will resume after access recovers. Report this blocker once; do not restart the watch or merge without check and review evidence.`, + notification: { + source: { kind: "monitor" }, + outcome: "failed", + summary: `#${target.link.number}: host access blocked, watching for recovery`, + }, + }); + if (Exit.isFailure(detailRead)) { + const wake = blockerWake; + yield* eachTarget(group, (target) => + target.watch.readBlocker !== undefined + ? Effect.void + : hold({ + groupKey: group.key, + target, + next: { ...target.watch, readBlocker }, + wake: wake(target), + }), + ); + } + } + } else { + if (failures >= READ_FAILURE_LIMIT && (yield* eachTarget(group, giveUp))) { + readFailures.delete(group.key); + } + return yield* Effect.failCause(readFailure.cause); } - return yield* Effect.failCause(detailRead.cause); + } else { + readFailures.delete(group.key); } - readFailures.delete(group.key); + if (Exit.isFailure(detailRead)) return yield* Effect.failCause(detailRead.cause); const detail = detailRead.value; for (const target of group.targets) { const life = lifeOf(target); @@ -696,7 +742,15 @@ export const make = Effect.gen(function* () { remarks: fingerprint?.remarks ?? null, }); yield* eachTarget(group, (target) => { - const report = evaluatePullRequestWatch(target.watch, detail, remarks, now); + const evaluated = evaluatePullRequestWatch(target.watch, detail, remarks, now); + // Check news must survive recovery; defer a new blocker notice to the next quiet pass. + const readBlocker = + target.watch.readBlocker ?? (evaluated.changes.length === 0 ? newReadBlocker : undefined); + const report = + remarks === null && readBlocker !== undefined + ? { ...evaluated, next: { ...evaluated.next, readBlocker } } + : evaluated; + const accessWake = target.watch.readBlocker === undefined ? blockerWake?.(target) : undefined; if (report.changes.length > 0) { const wake = { ...pullRequestWatchMessage({ @@ -722,6 +776,9 @@ export const make = Effect.gen(function* () { : { groupKey: group.key, target, next: report.next, wake }, ); } + if (accessWake !== undefined) { + return hold({ groupKey: group.key, target, next: report.next, wake: accessWake }); + } return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next); }); }); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index d488c76f313b1..4939b6cc3bbf7 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3969,20 +3969,25 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { ); it.effect.each([ - { reason: "failed", activityFails: false }, - { reason: "not-found", activityFails: false }, - { reason: "failed", activityFails: true }, - { reason: "not-found", activityFails: true }, - { reason: "forbidden", activityFails: true }, + { reason: "failed", activityFails: false, activityFailure: "failed" }, + { reason: "not-found", activityFails: false, activityFailure: "failed" }, + { reason: "failed", activityFails: true, activityFailure: "failed" }, + { reason: "not-found", activityFails: true, activityFailure: "failed" }, + { reason: "forbidden", activityFails: false, activityFailure: "failed" }, + { reason: "forbidden", activityFails: true, activityFailure: "failed" }, + { reason: "unauthenticated", activityFails: false, activityFailure: "failed" }, + { reason: "failed", activityFails: true, activityFailure: "forbidden" }, + { reason: "failed", activityFails: true, activityFailure: "unauthenticated" }, + { reason: "activity-only", activityFails: true, activityFailure: "forbidden" }, + { reason: "activity-only", activityFails: true, activityFailure: "unauthenticated" }, ] as const)( - "handles repeated host read failures: $reason (activity fails: $activityFails)", - ({ reason, activityFails }) => + "handles repeated host read failures: $reason (activity fails: $activityFails, $activityFailure)", + ({ reason, activityFails, activityFailure }) => Effect.gen(function* () { + const scenario = `${reason}-${activityFails}-${activityFailure}`; const orchestrator = yield* Orchestrator.OrchestratorV2; - const threadId = ThreadId.make( - `runtime-pull-request-watch-unreadable-${reason}-${activityFails}`, - ); - const projectId = ProjectId.make(`pr-watch-unreadable-project-${reason}-${activityFails}`); + const threadId = ThreadId.make(`runtime-pull-request-watch-unreadable-${scenario}`); + const projectId = ProjectId.make(`pr-watch-unreadable-project-${scenario}`); yield* seedProject({ projectId, title: "Watch unreadable", @@ -3994,7 +3999,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { type: "thread.create", createdBy: "user", creationSource: "web", - commandId: CommandId.make(`pr-watch-unreadable-create-${reason}-${activityFails}`), + commandId: CommandId.make(`pr-watch-unreadable-create-${scenario}`), threadId, projectId, title: "Watch unreadable", @@ -4006,7 +4011,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }); yield* orchestrator.dispatch({ type: "thread.pull-request.watch", - commandId: CommandId.make(`pr-watch-unreadable-start-${reason}-${activityFails}`), + commandId: CommandId.make(`pr-watch-unreadable-start-${scenario}`), threadId, host: "github.com", repository: "pingdotgg/t3code", @@ -4017,19 +4022,30 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { let disconnected = false; let passed = false; let genericFailure = false; + let accessChanged = false; + let remarksIncomplete = false; const activityFailures = yield* Queue.unbounded(); - const readFailure = (operation: string, failure: "failed" | "not-found" | "forbidden") => + const readFailure = ( + operation: string, + failure: "failed" | "not-found" | "forbidden" | "unauthenticated", + ) => new PullRequestOperationError({ operation, detail: "host read failed", cause: new PullRequestProviderError({ provider: SourceControlProviderKind.make("github"), operation, - reason: failure, + reason: accessChanged + ? failure === "forbidden" + ? "unauthenticated" + : failure === "unauthenticated" + ? "forbidden" + : failure + : failure, detail: "host read failed", }), }); - const reactor = yield* PullRequestWatchReactor.make.pipe( + const makeReactor = PullRequestWatchReactor.make.pipe( Effect.provide( Layer.mergeAll( NodeServices.layer, @@ -4039,7 +4055,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { detail: () => genericFailure ? Effect.die("host read failed") - : disconnected + : disconnected && reason !== "activity-only" ? (activityFails ? Queue.take(activityFailures) : Effect.void).pipe( Effect.andThen(Effect.fail(readFailure("detail", reason))), ) @@ -4063,12 +4079,13 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { activity: () => disconnected && activityFails ? Queue.offer(activityFailures, undefined).pipe( - Effect.andThen(Effect.fail(readFailure("activity", "failed"))), + Effect.andThen(Effect.fail(readFailure("activity", activityFailure))), ) : Effect.succeed({ comments: [], commentCount: 0, - commentsTruncated: false, + commentsTruncated: remarksIncomplete, + reviewThreadsTruncated: remarksIncomplete, reviewThreads: [], commits: [], }), @@ -4076,6 +4093,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { ), ), ); + const reactor = yield* makeReactor; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; const sweep = reactor.sweep.pipe(Effect.andThen(worker.drain())); const summaries = Effect.map( @@ -4086,30 +4104,113 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { assert.deepEqual(yield* summaries, []); disconnected = true; for (let pass = 0; pass < 15; pass += 1) { - genericFailure = reason === "failed" && (pass === 0 || (pass >= 2 && pass <= 8)); + genericFailure = + reason === "failed" && + activityFailure === "failed" && + (pass === 0 || (pass >= 2 && pass <= 8)); + if (reason === "activity-only" && pass === 7) passed = true; yield* TestClock.adjust("2 minutes"); yield* sweep; + if (reason === "activity-only" && pass === 7) { + assert.deepEqual(yield* summaries, ["#8: checks passed"]); + assert.isUndefined( + (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch?.readBlocker, + ); + } } const thread = yield* orchestrator.getThreadShell(threadId); - if (reason !== "failed") { + if (reason === "not-found") { assert.isUndefined(thread?.pullRequests?.[0]?.watch); assert.deepEqual(yield* summaries, ["#8: stopped watching, could not read it"]); return; } assert.isDefined(thread?.pullRequests?.[0]?.watch); - assert.deepEqual(yield* summaries, []); + const blocked = [reason, activityFailure].some( + (failure) => failure === "forbidden" || failure === "unauthenticated", + ); + const blockerExpected = blocked ? ["#8: host access blocked, watching for recovery"] : []; + const expected = + reason === "activity-only" ? ["#8: checks passed", ...blockerExpected] : blockerExpected; + const recoveredExpected = + reason === "activity-only" ? expected : [...expected, "#8: checks passed"]; + assert.deepEqual(yield* summaries, expected); + if (blocked) { + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`pr-watch-blocked-stop-${scenario}`), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number: 8, + watching: false, + }); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`pr-watch-blocked-resume-${scenario}`), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number: 8, + watching: true, + }); + const restarted = yield* makeReactor; + for (let pass = 0; pass < 10; pass += 1) { + yield* TestClock.adjust("2 minutes"); + yield* restarted.sweep; + yield* worker.drain(); + } + assert.deepEqual(yield* summaries, expected); + accessChanged = true; + for (let pass = 0; pass < 10; pass += 1) { + yield* TestClock.adjust("2 minutes"); + yield* restarted.sweep; + yield* worker.drain(); + } + assert.deepEqual(yield* summaries, expected); + const blocker = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch + ?.readBlocker; + disconnected = false; + passed = true; + remarksIncomplete = true; + yield* TestClock.adjust("2 minutes"); + yield* sweep; + assert.equal( + (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch?.readBlocker, + blocker, + ); + assert.deepEqual(yield* summaries, recoveredExpected); + disconnected = true; + for (let pass = 0; pass < 10; pass += 1) { + yield* TestClock.adjust("2 minutes"); + yield* sweep; + } + assert.deepEqual(yield* summaries, recoveredExpected); + } disconnected = false; genericFailure = false; passed = true; + remarksIncomplete = false; yield* sweep; - assert.deepEqual(yield* summaries, ["#8: checks passed"]); + assert.deepEqual(yield* summaries, recoveredExpected); + assert.isUndefined( + (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch?.readBlocker, + ); yield* TestClock.adjust("10 minutes"); yield* sweep; - assert.deepEqual(yield* summaries, ["#8: checks passed"]); + assert.deepEqual(yield* summaries, recoveredExpected); + if (blocked) { + disconnected = true; + accessChanged = false; + for (let pass = 0; pass < 15; pass += 1) { + yield* TestClock.adjust("2 minutes"); + yield* sweep; + } + assert.deepEqual(yield* summaries, [...recoveredExpected, ...blockerExpected]); + } yield* orchestrator.dispatch({ type: "thread.pull-request.watch", - commandId: CommandId.make(`pr-watch-unreadable-stop-${reason}-${activityFails}`), + commandId: CommandId.make(`pr-watch-unreadable-stop-${scenario}`), threadId, host: "github.com", repository: "pingdotgg/t3code", @@ -5802,6 +5903,12 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }); it.effect.each([ + "blocker still watching", + "blocker new watch", + "blocker recovered", + "blocker new outage", + "blocker unwatch", + "blocker newer head", "unwatch", "unlink", "new watch", @@ -5822,9 +5929,18 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { `queued-${mode}`, mode === "reopened", ); + const accessNotice = mode.startsWith("blocker "); const finalNotice = mode === "closed notice" || mode === "closed notice unlinked and relinked"; + const wakeSummary = finalNotice + ? "#7: closed, stopped watching" + : accessNotice + ? "#7: host access blocked, watching for recovery" + : "#7: checks failed"; const deliverable = + mode === "blocker still watching" || + mode === "blocker new watch" || + mode === "blocker newer head" || mode === "still watching" || mode === "closed notice" || mode === "reopened" || @@ -5843,7 +5959,12 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { dispatchMode: { type: "start_immediately" }, }); const wakeMessageId = MessageId.make(`${threadId}-wake`); - const recordedWatch = { ...watch, headSha: "abc123", wakes: 1 }; + const recordedWatch = { + ...watch, + headSha: "abc123", + wakes: 1, + ...(accessNotice ? { readBlocker: "detail:forbidden" } : {}), + }; yield* orchestrator.dispatch({ type: "thread.pull-request-watch.sync", commandId: CommandId.make(`${threadId}-wake`), @@ -5853,11 +5974,11 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { watch: finalNotice ? null : recordedWatch, wake: { messageId: wakeMessageId, - text: finalNotice ? "Closed, stopped watching" : "Checks failed", + text: wakeSummary, notification: { source: { kind: "monitor" }, outcome: "updated", - summary: finalNotice ? "#7: closed, stopped watching" : "#7: checks failed", + summary: wakeSummary, }, ...(mode === "comments then newer head" || mode === "comment limit" ? { includesRemarks: true } @@ -5887,7 +6008,12 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { assert.equal(wakeRun.status, "queued"); assert.equal(userRun.status, "queued"); - if (mode === "unwatch" || mode === "new watch") { + if ( + mode === "unwatch" || + mode === "new watch" || + mode === "blocker unwatch" || + mode === "blocker new watch" + ) { yield* orchestrator.dispatch({ type: "thread.pull-request.watch", commandId: CommandId.make(`${threadId}-unwatch`), @@ -5895,7 +6021,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { ...key, watching: false, }); - if (mode === "new watch") { + if (mode === "new watch" || mode === "blocker new watch") { yield* TestClock.adjust("1 millis"); yield* orchestrator.dispatch({ type: "thread.pull-request.watch", @@ -5977,7 +6103,24 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { includesRemarks: true, }, }); - } else if (mode === "newer head" || mode === "comments then newer head") { + } else if (mode === "blocker recovered" || mode === "blocker new outage") { + const { readBlocker: _blocker, ...recovered } = recordedWatch; + yield* orchestrator.dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make(`${threadId}-recovered`), + threadId, + ...key, + startedAt: watch.startedAt, + watch: + mode === "blocker new outage" + ? { ...recovered, readBlocker: "detail:forbidden:next" } + : recovered, + }); + } else if ( + mode === "newer head" || + mode === "comments then newer head" || + mode === "blocker newer head" + ) { yield* orchestrator.dispatch({ type: "thread.pull-request-watch.sync", commandId: CommandId.make(`${threadId}-advance`), @@ -6036,7 +6179,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { after.messages .filter((message) => message.id === wakeMessageId) .map((message) => message.notification?.summary), - [finalNotice ? "#7: closed, stopped watching" : "#7: checks failed"], + [wakeSummary], ); }), ); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 44e9aa8290e9a..59759b4957a6c 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1318,6 +1318,7 @@ const PullRequestWakeSection = Schema.Struct({ summary: TrimmedNonEmptyString, }); const MonitorPullRequestStamp = Schema.Struct({ + readBlocker: Schema.optional(TrimmedNonEmptyString), ...ThreadPullRequestKey.fields, watchStartedAt: IsoDateTime, /** Null when the wake is not about one head, so a newer push does not outdate it. */ diff --git a/packages/contracts/src/threadPullRequest.ts b/packages/contracts/src/threadPullRequest.ts index 0c8d7a61fe92d..d4fc3173e9e9c 100644 --- a/packages/contracts/src/threadPullRequest.ts +++ b/packages/contracts/src/threadPullRequest.ts @@ -113,6 +113,7 @@ export const ThreadPullRequestWatch = Schema.Struct({ * none did. Null once either happened; absent on watches saved before it existed. */ checksAwaitedSince: Schema.optional(Schema.NullOr(IsoDateTime)), + readBlocker: Schema.optional(TrimmedNonEmptyString), /** * Whether the host was still running checks on the head at the last read. Only then does the * watch keep its thread waiting; absent on watches not read since it existed.