diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a311d1ed6a7b..be0462812201 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -34,6 +34,7 @@ import { OrchestrationV2Command, type OrchestrationV2InternalCommand, type OrchestrationV2LimitRecoveryUpdate, + type OrchestrationV2Notification, type OrchestrationV2ServerCommand, type ThreadPullRequestLink, type ThreadPullRequestWatch, @@ -80,6 +81,7 @@ import { import * as KeyedLock from "@t3tools/shared/KeyedLock"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import { modelSwitchSince } from "./modelSwitch.ts"; +import { combinePullRequestWatchWakes, type PullRequestWatchWake } from "./pullRequestWatch.ts"; import { recordThreadOpen } from "@t3tools/shared/paletteSearch"; import { backgroundWorkHoldsCompletion, @@ -960,14 +962,21 @@ function pullRequestSnapshotEndsWatch( ); } -function isStalePullRequestWatchWake( +type MonitorPullRequestStamp = NonNullable< + Extract["pullRequest"] +>; + +function monitorPullRequestStamps( + source: OrchestrationV2Notification["source"] | undefined, +): ReadonlyArray { + if (source?.kind !== "monitor" || source.pullRequest === undefined) return []; + return [source.pullRequest, ...(source.otherPullRequests ?? [])]; +} + +function isStalePullRequestWatchStamp( projection: Pick, - run: OrchestrationV2Run, + stamp: MonitorPullRequestStamp, ): boolean { - const source = projection.messages.find((message) => message.id === run.userMessageId) - ?.notification?.source; - if (source?.kind !== "monitor" || source.pullRequest === undefined) return false; - const stamp = source.pullRequest; const link = threadPullRequestsOf(projection.thread).find( (candidate) => candidate.source !== "stack-dismissed" && threadPullRequestKeysEqual(candidate, stamp), @@ -980,15 +989,14 @@ function isStalePullRequestWatchWake( } if (link.watch === undefined) { // A watch the server ended with a final notice still owes the wakes it queued before it. - const endedWithNotice = projection.messages.some((message) => { - const ending = message.notification?.source; - return ( - ending?.kind === "monitor" && - ending.pullRequest?.watchEnded === true && - ending.pullRequest.watchStartedAt === stamp.watchStartedAt && - threadPullRequestKeysEqual(ending.pullRequest, stamp) - ); - }); + const endedWithNotice = projection.messages.some((message) => + monitorPullRequestStamps(message.notification?.source).some( + (ending) => + ending.watchEnded === true && + ending.watchStartedAt === stamp.watchStartedAt && + threadPullRequestKeysEqual(ending, stamp), + ), + ); return ( !endedWithNotice || link.endedWatch?.startedAt !== stamp.watchStartedAt || @@ -1002,6 +1010,58 @@ function isStalePullRequestWatchWake( ); } +function isStalePullRequestWatchWake( + projection: Pick, + run: OrchestrationV2Run, +): boolean { + const stamps = monitorPullRequestStamps( + projection.messages.find((message) => message.id === run.userMessageId)?.notification?.source, + ); + return ( + stamps.length > 0 && stamps.every((stamp) => isStalePullRequestWatchStamp(projection, stamp)) + ); +} + +/** + * A queued wake covering several pull requests, rebuilt from the ones still current, or + * undefined when it needs no change. + */ +function currentPullRequestWatchWake( + projection: Pick, + run: OrchestrationV2Run, +): OrchestrationV2ConversationMessage | undefined { + const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); + const stamps = monitorPullRequestStamps(message?.notification?.source); + if (message === undefined || stamps.length < 2) return undefined; + const current = stamps.filter((stamp) => !isStalePullRequestWatchStamp(projection, stamp)); + const [first, ...rest] = current; + if (first === undefined || rest.length === stamps.length - 1) return undefined; + const sections = current.flatMap(({ section }) => (section === undefined ? [] : [section])); + if (sections.length !== current.length) return undefined; + const wake = combinePullRequestWatchWakes( + sections.map(({ text, outcome, summary }) => ({ + text, + notification: { source: { kind: "monitor" }, outcome, summary }, + })) as [PullRequestWatchWake, ...Array], + ); + return { + ...message, + text: wake.text, + notification: { + ...wake.notification, + source: { + kind: "monitor", + pullRequest: rest.length === 0 ? withoutSection(first) : first, + ...(rest.length === 0 ? {} : { otherPullRequests: rest }), + }, + }, + }; +} + +function withoutSection({ section: _, ...stamp }: MonitorPullRequestStamp) { + return stamp; +} + function queuedRunCancellationEvents( projection: Pick, run: OrchestrationV2Run, @@ -2113,6 +2173,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection = yield* readCommandProjection(threadId); storedQueuedRun = nextQueuedRun(projection); } + // A wake that outlived some of its pull requests tells only the current ones. + const currentWake = + storedQueuedRun === undefined + ? undefined + : currentPullRequestWatchWake(projection, storedQueuedRun); + if (storedQueuedRun !== undefined && currentWake !== undefined) { + const now = yield* DateTime.now; + yield* writeSystemEvents([ + { + type: "message.updated", + threadId, + runId: storedQueuedRun.id, + ...(storedQueuedRun.rootNodeId === null ? {} : { nodeId: storedQueuedRun.rootNodeId }), + providerInstanceId: storedQueuedRun.providerInstanceId, + occurredAt: now, + payload: { ...currentWake, updatedAt: now }, + }, + ]); + const runId = storedQueuedRun.id; + projection = yield* readCommandProjection(threadId); + storedQueuedRun = projection.runs.find((run) => run.id === runId); + } if (storedQueuedRun === undefined) { return; } @@ -3550,57 +3632,79 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), ), ); - const key = normalizeThreadPullRequestKey(command); - const link = threadPullRequestsOf(thread).find( - (candidate) => - candidate.source !== "stack-dismissed" && threadPullRequestKeysEqual(candidate, key), - ); + const { others = [], wake, ...primary } = command; + const entries = [ + { ...primary, includesRemarks: wake?.includesRemarks }, + ...others.map((other) => ({ + type: primary.type, + commandId: primary.commandId, + threadId: primary.threadId, + ...other, + })), + ].map((entry) => { + const key = normalizeThreadPullRequestKey(entry); + const link = threadPullRequestsOf(thread).find( + (candidate) => + candidate.source !== "stack-dismissed" && threadPullRequestKeysEqual(candidate, key), + ); + return { entry, key, link }; + }); const inactive = thread.archivedAt !== null || thread.settledOverride === "settled" || thread.settledAt !== null || isProviderNativeSubagentThread(thread); - if (link?.watch?.startedAt !== command.startedAt || (command.wake && inactive)) { + // An idle thread dispatches immediately, bypassing queued-wake validation. + const outdated = entries.map( + ({ entry, link }) => + wake?.notification.source.kind === "monitor" && + entry.watch !== null && + pullRequestSnapshotEndsWatch(link?.snapshot ?? null, entry.startedAt), + ); + if ( + entries.some(({ entry, link }) => link?.watch?.startedAt !== entry.startedAt) || + (wake && inactive) || + // One wake cannot drop the news of just some of its pull requests; the sender splits it. + (entries.length > 1 && outdated.some(Boolean)) + ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, cause: "The pull request watch ended or its thread settled while it was read.", }); } - yield* dispatchThreadMutation(command, events, effects); - if (command.wake === undefined) return; - // An idle thread dispatches immediately, bypassing queued-wake validation. - if ( - command.wake.notification.source.kind === "monitor" && - command.watch !== null && - pullRequestSnapshotEndsWatch(link.snapshot, command.startedAt) - ) { - return; - } + for (const { entry } of entries) { + const { includesRemarks: _, ...mutation } = entry; + yield* dispatchThreadMutation(mutation, events, effects); + } + if (wake === undefined || outdated[0]) return; + const { sections, ...wakeMessage } = wake; + const stamps = entries.map(({ entry, key }, index) => ({ + ...key, + watchStartedAt: entry.startedAt, + headSha: entry.includesRemarks === true ? null : (entry.watch?.headSha ?? null), + ...(entry.watch === null ? { watchEnded: true } : {}), + ...(entries.length > 1 && sections?.[index] !== undefined + ? { section: sections[index] } + : {}), + })); yield* dispatchMessage( { type: "message.dispatch", commandId: command.commandId, threadId: command.threadId, - messageId: command.wake.messageId, - text: command.wake.text, + messageId: wakeMessage.messageId, + text: wakeMessage.text, notification: { - ...command.wake.notification, + ...wakeMessage.notification, source: - command.wake.notification.source.kind === "monitor" + wake.notification.source.kind === "monitor" ? { kind: "monitor", - pullRequest: { - ...key, - watchStartedAt: command.startedAt, - headSha: - command.wake.includesRemarks === true - ? null - : (command.watch?.headSha ?? null), - ...(command.watch === null ? { watchEnded: true } : {}), - }, + pullRequest: stamps[0]!, + ...(stamps.length > 1 ? { otherPullRequests: stamps.slice(1) } : {}), } - : command.wake.notification.source, + : wake.notification.source, }, attachments: [], dispatchMode: { type: "queue_after_active" }, diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 6fdea57bf9e1..511b65c3b2aa 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -38,6 +38,7 @@ import { forkParked } from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import { + combinePullRequestWatchWakes, evaluatePullRequestWatch, PULL_REQUEST_WATCH_NO_CHECKS_MS, pullRequestWatchMessage, @@ -98,6 +99,22 @@ interface ReadPlan { readonly activity: boolean; } +interface Wake { + readonly text: string; + readonly notification: OrchestrationV2Notification; + readonly includesRemarks?: boolean; +} + +/** A wake a pass holds back so each thread gets one for everything the pass found. */ +interface HeldWake { + readonly groupKey: string; + readonly target: WatchTarget; + readonly next: ThreadPullRequestWatch | null; + readonly endedWatch?: ThreadPullRequestWatch; + readonly wake: Wake; + readonly reason?: WatchEndReason; +} + const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`; type WatchEndReason = @@ -284,11 +301,7 @@ export const make = Effect.gen(function* () { const record = Effect.fn("PullRequestWatchReactor.record")(function* ( target: WatchTarget, next: ThreadPullRequestWatch | null, - wake?: { - readonly text: string; - readonly notification: OrchestrationV2Notification; - readonly includesRemarks?: boolean; - }, + wake?: Wake, endedWatch?: ThreadPullRequestWatch, ) { const uuid = yield* random.uuidV4(); @@ -411,18 +424,102 @@ export const make = Effect.gen(function* () { ), ); - const closed = (target: WatchTarget) => - record(target, null, { - text: `Pull request #${target.link.number} (${target.link.url}) was closed, so T3 Code stopped watching it. Call watch_pull_request if it reopens.`, - notification: { - source: { kind: "monitor" }, - outcome: "updated", - summary: `#${target.link.number}: closed, stopped watching`, + /** Wakes the current pass found, sent by `sendHeld` once every read of the pass is done. */ + let held: Array = []; + const hold = (wake: HeldWake) => Effect.sync(() => void held.push(wake)); + + const sendOne = (item: HeldWake) => + record(item.target, item.next, item.wake, item.endedWatch).pipe( + Effect.tap(() => woke(item.target)), + Effect.tap(() => (item.reason === undefined ? Effect.void : ended(item.target, item.reason))), + Effect.catchCause((cause) => { + lastReads.delete(item.groupKey); + return logFailure("pull request watch update failed", { + threadId: item.target.thread.id, + pullRequest: item.groupKey, + })(cause); + }), + ); + + const sendTogether = Effect.fn("PullRequestWatchReactor.sendTogether")(function* ( + items: readonly [HeldWake, ...ReadonlyArray], + ) { + const [first, ...rest] = items; + const uuid = yield* random.uuidV4(); + yield* engine.dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make(`server:pr-watch:${first.target.thread.id}:${uuid}`), + threadId: first.target.thread.id, + ...identityOf(first.target.link), + startedAt: first.target.watch.startedAt, + watch: first.next, + ...(first.endedWatch === undefined ? {} : { endedWatch: first.endedWatch }), + others: rest.map((item) => ({ + ...identityOf(item.target.link), + startedAt: item.target.watch.startedAt, + watch: item.next, + ...(item.endedWatch === undefined ? {} : { endedWatch: item.endedWatch }), + ...(item.wake.includesRemarks === true ? { includesRemarks: true } : {}), + })), + wake: { + ...combinePullRequestWatchWakes([first.wake, ...rest.map((item) => item.wake)]), + ...(first.wake.includesRemarks === true ? { includesRemarks: true } : {}), + sections: items.map(({ wake }) => ({ + text: wake.text, + outcome: wake.notification.outcome, + summary: wake.notification.summary, + })), + messageId: MessageId.make(`message:pr-watch:${uuid}`), }, - }).pipe( - Effect.tap(() => woke(target)), - Effect.tap(() => ended(target, "closed")), + }); + yield* Effect.forEach( + items, + (item) => + woke(item.target).pipe( + Effect.andThen(item.reason === undefined ? Effect.void : ended(item.target, item.reason)), + ), + { discard: true }, ); + }); + + const sendHeld = Effect.suspend(() => { + const byThread = new Map>(); + for (const item of held) { + byThread.set(item.target.thread.id, [...(byThread.get(item.target.thread.id) ?? []), item]); + } + held = []; + return Effect.forEach( + byThread.values(), + ([first, ...rest]) => + rest.length === 0 + ? sendOne(first!) + : // A watch that ended or moved during the pass fails the whole command; each then + // goes on its own, as the orchestrator decides for it alone. + sendTogether([first!, ...rest]).pipe( + Effect.catchCauseIf( + (cause) => !Cause.hasInterruptsOnly(cause), + () => Effect.forEach([first!, ...rest], sendOne, { discard: true }), + ), + ), + { discard: true }, + ); + }); + + const closed = (groupKey: string, target: WatchTarget) => + hold({ + groupKey, + target, + next: null, + wake: { + text: `Pull request #${target.link.number} (${target.link.url}) was closed, so T3 Code stopped watching it. Call watch_pull_request if it reopens.`, + notification: { + source: { kind: "monitor" }, + outcome: "updated", + summary: `#${target.link.number}: closed, stopped watching`, + }, + }, + reason: "closed", + }); const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined => link.snapshot?.state === "merged" @@ -545,7 +642,7 @@ export const make = Effect.gen(function* () { lastReads.delete(group.key); return yield* eachTarget(group, (target) => detail.state === "closed" - ? closed(target) + ? closed(group.key, target) : record(target, null).pipe(Effect.tap(() => ended(target, "merged"))), ); } @@ -588,13 +685,17 @@ export const make = Effect.gen(function* () { includesRemarks: report.changes.some((change) => change.kind === "remarks"), }; // An ending watch keeps what this pass saw, so watching again starts from it. - return ( + return hold( report.exhausted - ? record(target, null, wake, report.next) - : record(target, report.next, wake) - ).pipe( - Effect.tap(() => woke(target)), - Effect.tap(() => (report.exhausted ? ended(target, "comment-limit") : Effect.void)), + ? { + groupKey: group.key, + target, + next: null, + endedWatch: report.next, + wake, + reason: "comment-limit", + } + : { groupKey: group.key, target, next: report.next, wake }, ); } return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next); @@ -604,6 +705,7 @@ export const make = Effect.gen(function* () { /** Every watch, or with `changed`, only that pull request's after its host reported a change. */ // The sweep and change reads each name their own span. const pass = Effect.fnUntraced(function* (changed: string | undefined) { + held = []; const threads = yield* projections.getThreadsWithPullRequests(); const targets = threads.flatMap((thread) => visibleThreadPullRequests(thread.pullRequests ?? []).flatMap((link) => @@ -697,6 +799,7 @@ export const make = Effect.gen(function* () { ), { concurrency: "unbounded", discard: true }, ); + yield* sendHeld; }, passes.withPermit); const sweep = pass(undefined).pipe( diff --git a/apps/server/src/orchestration-v2/pullRequestWatch.ts b/apps/server/src/orchestration-v2/pullRequestWatch.ts index 046c48e1858b..05b08df73fb4 100644 --- a/apps/server/src/orchestration-v2/pullRequestWatch.ts +++ b/apps/server/src/orchestration-v2/pullRequestWatch.ts @@ -279,6 +279,43 @@ const SUMMARY: Record = { conflicting: "merge conflict", }; +const WATCHING_CLOSING = + "Look into each item and act on it as your task requires. This PR stays linked and watched, so do not link, watch, or list it again. End your turn when you are done, and T3 Code wakes you on the next change. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox."; +const WATCHING_CLOSING_PLURAL = + "Look into each item and act on it as your task requires. Each PR T3 Code still watches stays linked and watched, so do not link, watch, or list them again. End your turn when you are done, and T3 Code wakes you on the next change. When you hand the work back to the user, call unwatch_pull_request for each first so the thread returns to their inbox."; + +export interface PullRequestWatchWake { + readonly text: string; + readonly notification: OrchestrationV2Notification; +} + +/** One wake for news on several pull requests, with the closing instructions said once. */ +export function combinePullRequestWatchWakes( + wakes: readonly [PullRequestWatchWake, ...ReadonlyArray], +): PullRequestWatchWake { + if (wakes.length === 1) return wakes[0]; + const closing = `\n\n${WATCHING_CLOSING}`; + const watching = wakes.some((wake) => wake.text.endsWith(closing)); + const outcomes = wakes.map((wake) => wake.notification.outcome); + return { + text: [ + ...wakes.map((wake) => + wake.text.endsWith(closing) ? wake.text.slice(0, -closing.length) : wake.text, + ), + ...(watching ? [WATCHING_CLOSING_PLURAL] : []), + ].join("\n\n"), + notification: { + source: { kind: "monitor" }, + outcome: outcomes.includes("failed") + ? "failed" + : outcomes.every((outcome) => outcome === "completed") + ? "completed" + : "updated", + summary: wakes.map((wake) => wake.notification.summary).join("; "), + }, + }; +} + export function pullRequestWatchMessage(input: { readonly number: number; readonly url: string; @@ -297,7 +334,7 @@ export function pullRequestWatchMessage(input: { "", exhausted ? `T3 Code stopped watching after ${PULL_REQUEST_WATCH_WAKE_LIMIT} comment-only updates in a row. Call watch_pull_request to watch it again.` - : "Look into each item and act on it as your task requires. This PR stays linked and watched, so do not link, watch, or list it again. End your turn when you are done, and T3 Code wakes you on the next change. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox.", + : WATCHING_CLOSING, ].join("\n"); const failed = changes.some( (change) => change.kind === "checks-failed" || change.kind === "conflicting", diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 954f72e2b5e9..1557e411af93 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4094,6 +4094,135 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect.each([ + ["an idle", false], + ["a busy", true], + ] as const)( + "wakes %s thread once for a pass that changed several of its pull requests", + ([, busy]) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make(`runtime-pull-request-watch-pass-${busy}`); + const projectId = ProjectId.make(`pr-watch-pass-project-${busy}`); + yield* seedProject({ + projectId, + title: "Watch pass", + workspaceRoot: "/workspace/watch-pass", + defaultModelSelection: null, + createdAt: "2026-10-01T00:00:00.000Z", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`pr-watch-pass-create-${busy}`), + threadId, + projectId, + title: "Watch pass", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const numbers = [12, 13, 14]; + for (const number of numbers) { + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`pr-watch-pass-start-${busy}-${number}`), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number, + watching: true, + link: { url: `https://github.com/pingdotgg/t3code/pull/${number}`, source: "agent" }, + }); + } + if (busy) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("pr-watch-pass-busy"), + threadId, + messageId: MessageId.make("pr-watch-pass-busy"), + text: "Keep this run active.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + } + const host = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(PullRequestService.PullRequestService)({ + detail: (input) => + Effect.succeed({ + ...watchedPullRequestDetail({ + projectId, + number: input.number, + at: "2026-10-02T12:00:00.000Z", + }), + headSha: `head${input.number}`, + checks: [ + { name: "test", status: "failure", description: null, url: null, required: true }, + ], + }), + activity: () => + Effect.succeed({ + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }), + }), + ); + + const reactor = yield* PullRequestWatchReactor.make.pipe(Effect.provide(host)); + yield* reactor.sweep; + const { messages } = yield* orchestrator.getThreadRecords(threadId, ["messages"]); + const wakes = messages.filter((message) => message.notification !== undefined); + const stamps = wakes.map(({ notification }) => + notification?.source.kind === "monitor" + ? [notification.source.pullRequest, ...(notification.source.otherPullRequests ?? [])] + .map((stamp) => stamp?.number) + .toSorted() + : [], + ); + assert.deepEqual(stamps, [numbers]); + for (const number of numbers) assert.include(wakes[0]!.text, `#${number}`); + if (!busy) return; + + // A pull request unwatched while the wake waited leaves it; the others still start. + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("pr-watch-pass-stop-12"), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number: 12, + watching: false, + }); + const active = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("pr-watch-pass-interrupt"), + threadId, + runId: active.id, + }); + yield* orchestrator.resumeQueuedRuns; + const after = yield* orchestrator.getThreadProjection(threadId); + const woken = after.messages.find((message) => message.id === wakes[0]!.id)!; + assert.notInclude(woken.text, "#12"); + for (const number of [13, 14]) assert.include(woken.text, `#${number}`); + assert.notEqual( + after.runs.find((run) => run.userMessageId === woken.id)?.status, + "cancelled", + ); + }), + ); + it.effect.each(["pending", "action-required"] as const)("recovers %s watches", (initialStatus) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 8d8cc42c235b..7232eb2eb2b5 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1260,6 +1260,28 @@ const SubagentNotificationSource = Schema.Struct({ childThreadId: Schema.optional(ThreadId), }); const CommandNotificationSource = Schema.Struct({ kind: Schema.Literal("command") }); +const OrchestrationV2NotificationOutcome = Schema.Literals([ + "completed", + "failed", + "cancelled", + "updated", + "unknown", +]); +/** One pull request's part of a wake, which stands alone once the others are outdated. */ +const PullRequestWakeSection = Schema.Struct({ + text: Schema.String, + outcome: OrchestrationV2NotificationOutcome, + summary: TrimmedNonEmptyString, +}); +const MonitorPullRequestStamp = Schema.Struct({ + ...ThreadPullRequestKey.fields, + watchStartedAt: IsoDateTime, + /** Null when the wake is not about one head, so a newer push does not outdate it. */ + headSha: Schema.NullOr(TrimmedNonEmptyString), + watchEnded: Schema.optional(Schema.Boolean), + /** Set on a wake covering several pull requests. */ + section: Schema.optional(PullRequestWakeSection), +}); /** Encode legacy background_command/background_task shapes first so older clients accept command/subagent sources. */ export const OrchestrationV2NotificationSource = kindUnionWithFallback( @@ -1292,15 +1314,9 @@ export const OrchestrationV2NotificationSource = kindUnionWithFallback( CommandNotificationSource, Schema.Struct({ kind: Schema.Literal("monitor"), - pullRequest: Schema.optional( - Schema.Struct({ - ...ThreadPullRequestKey.fields, - watchStartedAt: IsoDateTime, - /** Null when the wake is not about one head, so a newer push does not outdate it. */ - headSha: Schema.NullOr(TrimmedNonEmptyString), - watchEnded: Schema.optional(Schema.Boolean), - }), - ), + pullRequest: Schema.optional(MonitorPullRequestStamp), + /** The other pull requests one wake covers; outdated ones leave it before it starts. */ + otherPullRequests: Schema.optional(Schema.Array(MonitorPullRequestStamp)), }), Schema.Struct({ kind: Schema.Literal("background_task") }), Schema.Struct({ kind: Schema.Literal("system") }), @@ -1312,7 +1328,7 @@ export type OrchestrationV2NotificationSource = typeof OrchestrationV2Notificati export const OrchestrationV2Notification = Schema.Struct({ source: OrchestrationV2NotificationSource, - outcome: Schema.Literals(["completed", "failed", "cancelled", "updated", "unknown"]), + outcome: OrchestrationV2NotificationOutcome, summary: TrimmedNonEmptyString, detail: Schema.optional(Schema.String), }); @@ -3482,6 +3498,18 @@ const OrchestrationV2InternalCommand = Schema.Union([ watch: Schema.NullOr(ThreadPullRequestWatch), /** With `watch: null`, the state the ended watch keeps instead of its last saved one. */ endedWatch: Schema.optional(ThreadPullRequestWatch), + /** More watches of this thread recorded together, which `wake` also reports on. */ + others: Schema.optional( + Schema.Array( + Schema.Struct({ + ...ThreadPullRequestKey.fields, + startedAt: IsoDateTime, + watch: Schema.NullOr(ThreadPullRequestWatch), + endedWatch: Schema.optional(ThreadPullRequestWatch), + includesRemarks: Schema.optional(Schema.Boolean), + }), + ), + ), wake: Schema.optional( Schema.Struct({ messageId: MessageId, @@ -3489,6 +3517,8 @@ const OrchestrationV2InternalCommand = Schema.Union([ notification: OrchestrationV2Notification, /** Comments stay worth reading after a push, so the wake outlives a newer head. */ includesRemarks: Schema.optional(Schema.Boolean), + /** With `others`, each pull request's own part, in the same order. */ + sections: Schema.optional(Schema.Array(PullRequestWakeSection)), }), ), }),