diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index fffba7935982b..3ad406360a09b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -617,6 +617,90 @@ function nextQueuedRun( return queuedRunsInDeliveryOrder(projection)[0]; } +function pullRequestSnapshotEndsWatch( + snapshot: ThreadPullRequestLink["snapshot"], + watchStartedAt: ThreadPullRequestWatch["startedAt"], +): boolean { + return ( + snapshot?.state === "merged" || + // A restarted watch can see a closed snapshot from before the PR reopened. + (snapshot?.state === "closed" && + DateTime.isGreaterThanOrEqualTo( + DateTime.makeUnsafe(snapshot.syncedAt), + DateTime.makeUnsafe(watchStartedAt), + )) + ); +} + +function isStalePullRequestWatchWake( + projection: Pick, + run: OrchestrationV2Run, +): 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), + ); + if (link === undefined) return true; + if (stamp.watchEnded === true) { + // The final notice outlives its watch, but a replacement watch supersedes it. + const latestWatch = link.watch ?? link.endedWatch; + return latestWatch?.startedAt !== stamp.watchStartedAt; + } + return ( + link.watch === undefined || + link.watch.startedAt !== stamp.watchStartedAt || + link.watch.headSha !== stamp.headSha || + pullRequestSnapshotEndsWatch(link.snapshot, link.watch.startedAt) + ); +} + +function queuedRunCancellationEvents( + projection: Pick, + run: OrchestrationV2Run, + now: DateTime.Utc, +): ReadonlyArray> { + const rootNode = projection.nodes.find((node) => node.id === run.rootNodeId); + const attempt = projection.attempts.find((candidate) => candidate.id === run.activeAttemptId); + const events: Array> = [ + { + type: "run.updated", + threadId: run.threadId, + runId: run.id, + ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now }, + }, + ]; + if (attempt !== undefined && rootNode !== undefined) { + events.push({ + type: "run-attempt.updated", + threadId: run.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { ...attempt, status: "cancelled", completedAt: now }, + }); + } + if (rootNode !== undefined) { + events.push({ + type: "node.updated", + threadId: run.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { ...rootNode, status: "cancelled", completedAt: now }, + }); + } + return events; +} + function latestStableRun( projection: Pick, ): OrchestrationV2Run | null { @@ -1117,60 +1201,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly now: DateTime.Utc; }) => Effect.gen(function* () { - const rootNode = - input.run.rootNodeId === null - ? undefined - : input.projection.nodes.find((candidate) => candidate.id === input.run.rootNodeId); - const attempt = - input.run.activeAttemptId === null - ? undefined - : input.projection.attempts.find( - (candidate) => candidate.id === input.run.activeAttemptId, - ); const emitEvent = emit(input.events, input.command); - yield* emitEvent({ - type: "run.updated", - threadId: input.run.threadId, - runId: input.run.id, - ...(input.run.rootNodeId === null ? {} : { nodeId: input.run.rootNodeId }), - providerInstanceId: input.run.providerInstanceId, - occurredAt: input.now, - payload: { - ...input.run, - status: "cancelled", - queuePosition: null, - completedAt: input.now, - }, - }); - if (attempt !== undefined && rootNode !== undefined) { - yield* emitEvent({ - type: "run-attempt.updated", - threadId: input.run.threadId, - runId: input.run.id, - nodeId: rootNode.id, - providerInstanceId: input.run.providerInstanceId, - occurredAt: input.now, - payload: { - ...attempt, - status: "cancelled", - completedAt: input.now, - }, - }); - } - if (rootNode !== undefined) { - yield* emitEvent({ - type: "node.updated", - threadId: input.run.threadId, - runId: input.run.id, - nodeId: rootNode.id, - providerInstanceId: input.run.providerInstanceId, - occurredAt: input.now, - payload: { - ...rootNode, - status: "cancelled", - completedAt: input.now, - }, - }); + for (const event of queuedRunCancellationEvents(input.projection, input.run, input.now)) { + yield* emitEvent(event); } }); @@ -1285,7 +1318,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) => Effect.gen(function* () { if (!(yield* projectionStore.canStartQueuedRun(threadId))) return; - const projection = yield* readCommandProjection(threadId); + let projection = yield* readCommandProjection(threadId); if ( projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || @@ -1307,7 +1340,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (usageLimitBlockedRun(projection.runs, projection.turnItems, sessionError) !== null) { return; } - const queuedRun = nextQueuedRun(projection); + let queuedRun = nextQueuedRun(projection); + while (queuedRun !== undefined && isStalePullRequestWatchWake(projection, queuedRun)) { + yield* writeSystemEvents( + queuedRunCancellationEvents(projection, queuedRun, yield* DateTime.now), + ); + projection = yield* readCommandProjection(threadId); + queuedRun = nextQueuedRun(projection); + } if (queuedRun === undefined) { return; } @@ -2411,6 +2451,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } 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; + } yield* dispatchMessage( { type: "message.dispatch", @@ -2418,7 +2466,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId: command.threadId, messageId: command.wake.messageId, text: command.wake.text, - notification: command.wake.notification, + notification: { + ...command.wake.notification, + source: + command.wake.notification.source.kind === "monitor" + ? { + kind: "monitor", + pullRequest: { + ...key, + watchStartedAt: command.startedAt, + headSha: command.watch?.headSha ?? null, + ...(command.watch === null ? { watchEnded: true } : {}), + }, + } + : command.wake.notification.source, + }, attachments: [], dispatchMode: { type: "queue_after_active" }, createdBy: "agent", @@ -7985,61 +8047,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); return; } - const queuedRootNode = - queuedRun.rootNodeId === null - ? undefined - : projection.nodes.find((candidate) => candidate.id === queuedRun.rootNodeId); - const queuedAttempt = - queuedRun.activeAttemptId === null - ? undefined - : projection.attempts.find((candidate) => candidate.id === queuedRun.activeAttemptId); - - const now = yield* DateTime.now; - const emitEvent = emit(events, command); - yield* emitEvent({ - type: "run.updated", - threadId: command.threadId, - runId: queuedRun.id, - ...(queuedRun.rootNodeId === null ? {} : { nodeId: queuedRun.rootNodeId }), - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { - ...queuedRun, - status: "cancelled", - queuePosition: null, - completedAt: now, - }, + yield* emitQueuedRunCancellation({ + command, + events, + projection, + run: queuedRun, + now: yield* DateTime.now, }); - if (queuedAttempt !== undefined && queuedRootNode !== undefined) { - yield* emitEvent({ - type: "run-attempt.updated", - threadId: command.threadId, - runId: queuedRun.id, - nodeId: queuedRootNode.id, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { - ...queuedAttempt, - status: "cancelled", - completedAt: now, - }, - }); - } - if (queuedRootNode !== undefined) { - yield* emitEvent({ - type: "node.updated", - threadId: command.threadId, - runId: queuedRun.id, - nodeId: queuedRootNode.id, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { - ...queuedRootNode, - status: "cancelled", - completedAt: now, - }, - }); - } }); const dispatchQueuedRunEdit = ( diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 1b6a7de8a944f..eab592666ab5b 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3388,6 +3388,328 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + const createWatchedThread = Effect.fnUntraced(function* (name: string, reopened = false) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make(`pr-wake-${name}`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}-create`), + threadId, + projectId: ProjectId.make(`${threadId}-project`), + title: "Watch wake", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + const key = { host: "github.com", repository: "pingdotgg/t3code", number: 7 }; + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`${threadId}-watch`), + threadId, + ...key, + watching: true, + link: { url: "https://github.com/pingdotgg/t3code/pull/7", source: "agent" }, + }); + let watch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch; + assert.isDefined(watch); + const snapshotSyncedAt = reopened + ? DateTime.formatIso(yield* DateTime.now) + : "2026-10-02T12:00:00.000Z"; + if (reopened) { + yield* orchestrator.dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make(`${threadId}-end-watch`), + threadId, + ...key, + startedAt: watch.startedAt, + watch: null, + }); + yield* TestClock.adjust("1 millis"); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`${threadId}-rewatch`), + threadId, + ...key, + watching: true, + }); + watch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch; + assert.isDefined(watch); + } + return { threadId, key, watch, snapshotSyncedAt }; + }); + + it.effect.each([ + "unwatch", + "unlink", + "new watch", + "merged", + "closed", + "newer head", + "still watching", + "closed notice", + "closed notice unlinked and relinked", + "reopened", + ] as const)("revalidates a queued pull-request wake before promotion: %s", (mode) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const { threadId, key, watch, snapshotSyncedAt } = yield* createWatchedThread( + `queued-${mode}`, + mode === "reopened", + ); + const finalNotice = + mode === "closed notice" || mode === "closed notice unlinked and relinked"; + const deliverable = + mode === "still watching" || mode === "closed notice" || mode === "reopened"; + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}-active`), + threadId, + messageId: MessageId.make(`${threadId}-active`), + text: "Active work", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const wakeMessageId = MessageId.make(`${threadId}-wake`); + const recordedWatch = { ...watch, headSha: "abc123", wakes: 1 }; + yield* orchestrator.dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make(`${threadId}-wake`), + threadId, + ...key, + startedAt: watch.startedAt, + watch: finalNotice ? null : recordedWatch, + wake: { + messageId: wakeMessageId, + text: finalNotice ? "Closed, stopped watching" : "Checks failed", + notification: { + source: { kind: "monitor" }, + outcome: "updated", + summary: finalNotice ? "#7: closed, stopped watching" : "#7: checks failed", + }, + }, + }); + const userMessageId = MessageId.make(`${threadId}-user`); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}-user`), + threadId, + messageId: userMessageId, + text: "User follow-up", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + const before = yield* orchestrator.getThreadProjection(threadId); + const activeRun = before.runs.find((run) => run.status === "starting"); + const wakeRun = before.runs.find((run) => run.userMessageId === wakeMessageId); + const userRun = before.runs.find((run) => run.userMessageId === userMessageId); + assert.isDefined(activeRun); + assert.isDefined(wakeRun); + assert.isDefined(userRun); + assert.equal(wakeRun.status, "queued"); + assert.equal(userRun.status, "queued"); + + if (mode === "unwatch" || mode === "new watch") { + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`${threadId}-unwatch`), + threadId, + ...key, + watching: false, + }); + if (mode === "new watch") { + yield* TestClock.adjust("1 millis"); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`${threadId}-rewatch`), + threadId, + ...key, + watching: true, + }); + const restarted = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0] + ?.watch; + assert.isDefined(restarted); + assert.notEqual(restarted.startedAt, watch.startedAt); + yield* orchestrator.dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make(`${threadId}-same-head`), + threadId, + ...key, + startedAt: restarted.startedAt, + watch: { ...restarted, headSha: recordedWatch.headSha }, + }); + } + } else if (mode === "unlink" || mode === "closed notice unlinked and relinked") { + yield* orchestrator.dispatch({ + type: "thread.pull-request.unlink", + commandId: CommandId.make(`${threadId}-unlink`), + threadId, + ...key, + }); + if (mode === "closed notice unlinked and relinked") { + yield* orchestrator.dispatch({ + type: "thread.pull-request.link", + commandId: CommandId.make(`${threadId}-relink`), + threadId, + ...key, + url: "https://github.com/pingdotgg/t3code/pull/7", + source: "manual", + }); + } + } else if ( + mode === "merged" || + mode === "closed" || + mode === "closed notice" || + mode === "reopened" + ) { + yield* orchestrator.dispatch({ + type: "thread.pull-request-link.sync", + commandId: CommandId.make(`${threadId}-snapshot`), + threadId, + ...key, + snapshot: { + state: mode === "merged" ? "merged" : "closed", + title: "Watched pull request", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: null, + // A delayed sync can still report the closure from before this watch restarted. + syncedAt: snapshotSyncedAt, + }, + stack: null, + }); + } else if (mode === "newer head") { + yield* orchestrator.dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make(`${threadId}-advance`), + threadId, + ...key, + startedAt: watch.startedAt, + watch: { ...recordedWatch, headSha: "def456" }, + }); + } + + const afterSequence = yield* orchestrator.getThreadEventSequence(threadId); + const completedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make(`${threadId}-completed`), + type: "run.updated", + threadId, + runId: activeRun.id, + ...(activeRun.rootNodeId === null ? {} : { nodeId: activeRun.rootNodeId }), + providerInstanceId: activeRun.providerInstanceId, + occurredAt: completedAt, + payload: { ...activeRun, status: "completed", completedAt }, + }, + ], + }); + yield* eventSink.stream({ threadId, afterSequence }).pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && stored.event.payload.status === "starting", + ), + Stream.take(1), + Stream.runDrain, + ); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + after.runs.find((run) => run.id === wakeRun.id)?.status, + deliverable ? "starting" : "cancelled", + ); + assert.equal( + after.runs.find((run) => run.id === userRun.id)?.status, + deliverable ? "queued" : "starting", + ); + if (!deliverable) { + assert.equal( + after.attempts.find((attempt) => attempt.id === wakeRun.activeAttemptId)?.status, + "cancelled", + ); + assert.equal( + after.nodes.find((node) => node.id === wakeRun.rootNodeId)?.status, + "cancelled", + ); + assert.isFalse(after.turnItems.some((item) => item.runId === wakeRun.id)); + } + assert.deepEqual( + after.messages + .filter((message) => message.id === wakeMessageId) + .map((message) => message.notification?.summary), + [finalNotice ? "#7: closed, stopped watching" : "#7: checks failed"], + ); + }), + ); + + it.effect.each(["open", "merged", "closed", "closed notice", "reopened"] as const)( + "revalidates a pull-request wake before idle dispatch: %s", + (mode) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { threadId, key, watch, snapshotSyncedAt } = yield* createWatchedThread( + `idle-${mode}`, + mode === "reopened", + ); + yield* orchestrator.dispatch({ + type: "thread.pull-request-link.sync", + commandId: CommandId.make(`${threadId}-snapshot`), + threadId, + ...key, + snapshot: { + state: mode === "closed notice" || mode === "reopened" ? "closed" : mode, + title: "Watched pull request", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: null, + syncedAt: snapshotSyncedAt, + }, + stack: null, + }); + const wakeMessageId = MessageId.make(`${threadId}-wake`); + yield* orchestrator.dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make(`${threadId}-wake`), + threadId, + ...key, + startedAt: watch.startedAt, + watch: mode === "closed notice" ? null : { ...watch, headSha: "abc123", wakes: 1 }, + wake: { + messageId: wakeMessageId, + text: mode === "closed notice" ? "Closed, stopped watching" : "Checks failed", + notification: { + source: { kind: "monitor" }, + outcome: "updated", + summary: + mode === "closed notice" ? "#7: closed, stopped watching" : "#7: checks failed", + }, + }, + }); + const after = yield* orchestrator.getThreadProjection(threadId); + const deliverable = mode === "open" || mode === "closed notice" || mode === "reopened"; + assert.deepEqual( + after.runs.map((run) => [run.userMessageId, run.status]), + deliverable ? [[wakeMessageId, "starting"]] : [], + ); + assert.deepEqual( + after.messages.map((message) => message.id), + deliverable ? [wakeMessageId] : [], + ); + }), + ); + it.effect("persists rejected command receipts across retries", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..07bc6c5bd6468 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1065,7 +1065,17 @@ export const OrchestrationV2NotificationSource = kindUnionWithFallback( Schema.Struct({ kind: Schema.Literal("background_command").transform("command") }), SubagentNotificationSource, CommandNotificationSource, - Schema.Struct({ kind: Schema.Literal("monitor") }), + Schema.Struct({ + kind: Schema.Literal("monitor"), + pullRequest: Schema.optional( + Schema.Struct({ + ...ThreadPullRequestKey.fields, + watchStartedAt: IsoDateTime, + headSha: Schema.NullOr(TrimmedNonEmptyString), + watchEnded: Schema.optional(Schema.Boolean), + }), + ), + }), Schema.Struct({ kind: Schema.Literal("background_task") }), Schema.Struct({ kind: Schema.Literal("system") }), ],