[Fix] PR review feedback disappears during pull request creation

apps/bullmq/src/jobs/pr-review-notification.ts

@@ -17,14 +17,16 @@ import {
   PR_REVIEW_NOTIFICATION_MAX_DEFERRALS,
   attachPendingPrReviewActionMessage,
   getCommunicationProviderAdapter,
-  type PrReviewNotificationRequest,
+  type PrReviewNotificationQueueRequest,
   type PrReviewNotificationRoute,
   consumePendingPrReviewActivity,
   dispatchPrReviewFollowUp,
   preparePrReviewNotificationDelivery,
   prReviewNotificationRequestSchema,
+  prReviewAssociationReplayRequestSchema,
   recordPrReviewNotificationDeliveryBestEffort,
   requeuePendingPrReviewActivity,
+  replayPrReviewNotificationAssociation,
   schedulePrReviewNotificationJob,
   setPendingPrReviewAction,
 } from '@roomote/sdk/server';
@@ -39,7 +41,11 @@ import {
   WORKER_HEARTBEAT_STALE_MS,
 } from '@roomote/types';
 
-type PrReviewNotificationJob = Job<PrReviewNotificationRequest, void, string>;
+type PrReviewNotificationJob = Job<
+  PrReviewNotificationQueueRequest,
+  void,
+  string
+>;
 
 function buildPrReviewNotificationPostInput(
   route: PrReviewNotificationRoute,
@@ -212,6 +218,13 @@ async function postPrReviewNotification({
 export const prReviewNotificationJob = async (
   job: PrReviewNotificationJob,
 ): Promise<void> => {
+  const replay = prReviewAssociationReplayRequestSchema.safeParse(job.data);
+
+  if (replay.success) {
+    await replayPrReviewNotificationAssociation(replay.data);
+    return;
+  }
+
   const parsed = prReviewNotificationRequestSchema.safeParse(job.data);
 
   if (!parsed.success) {

apps/bullmq/src/pr-review-notification-queue.ts

@@ -2,16 +2,20 @@ import { Queue, QueueEvents, Worker } from 'bullmq';
 
 import {
   PR_REVIEW_NOTIFICATION_QUEUE_NAME,
-  type PrReviewNotificationRequest,
+  type PrReviewNotificationQueueRequest,
 } from '@roomote/sdk/server';
 
 import { prReviewNotificationJob } from './jobs/pr-review-notification';
 import { getRedis } from './redis';
 
+function formatJobTarget(data: PrReviewNotificationQueueRequest): string {
+  return `${data.repository}#${data.prNumber}`;
+}
+
 export function startPrReviewNotificationQueue() {
   const connection = getRedis();
 
-  const queue = new Queue<PrReviewNotificationRequest, void, string>(
+  const queue = new Queue<PrReviewNotificationQueueRequest, void, string>(
     PR_REVIEW_NOTIFICATION_QUEUE_NAME,
     {
       connection,
@@ -24,15 +28,17 @@ export function startPrReviewNotificationQueue() {
     },
   );
 
-  const worker = new Worker<PrReviewNotificationRequest, void, string>(
+  const worker = new Worker<PrReviewNotificationQueueRequest, void, string>(
     PR_REVIEW_NOTIFICATION_QUEUE_NAME,
     prReviewNotificationJob,
     { connection, concurrency: 5, autorun: true },
   );
 
   worker.on('failed', (job, err) =>
     console.error(
-      `[PrReviewNotificationQueue] job ${job?.id} failed for ${job?.data.repository}#${job?.data.prNumber}:`,
+      `[PrReviewNotificationQueue] job ${job?.id} failed for ${
+        job?.data ? formatJobTarget(job.data) : 'unknown pull request'
+      }:`,
       err.message,
     ),
   );

packages/sdk/src/server/index.ts

@@ -197,21 +197,27 @@ export {
   PR_REVIEW_NOTIFICATION_MAX_DEFERRALS,
   PR_REVIEW_NOTIFICATION_QUEUE_NAME,
   PR_REVIEW_NOTIFICATION_ROOMOTE_FALLBACK_MS,
+  PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS,
   consumePendingPrReviewActivity,
   enqueuePrReviewNotification,
   enqueuePrReviewNotificationInputSchema,
   formatPrReviewActivityMessage,
   getPrReviewCompletedCycleKey,
   hasPrReviewNotificationThreadContext,
   prReviewActivityEventSchema,
+  prReviewAssociationReplayRequestSchema,
+  prReviewNotificationQueueRequestSchema,
   prReviewNotificationRequestSchema,
   requeuePendingPrReviewActivity,
+  replayPrReviewNotificationAssociation,
   resolvePrReviewNotificationRoute,
   schedulePrReviewNotificationJob,
   startPrReviewNotificationCycle,
   startPrReviewNotificationCycleInputSchema,
   type EnqueuePrReviewNotificationInput,
   type PrReviewActivityEvent,
+  type PrReviewAssociationReplayRequest,
+  type PrReviewNotificationQueueRequest,
   type PrReviewNotificationRequest,
   type PrReviewNotificationRoute,
   type StartPrReviewNotificationCycleInput,

packages/sdk/src/server/lib/pull-requests/source-control-pull-requests.ts

@@ -79,6 +79,8 @@ import {
 } from './source-control-pull-request-shared';
 
 const ADO_API_VERSION = '7.1';
+const PR_ASSOCIATION_MAX_ATTEMPTS = 3;
+const PR_ASSOCIATION_RETRY_DELAY_MS = 100;
 
 export const sourceControlPullRequestMutationInputSchema = z.object({
   action: z.literal('create_or_update_pull_request'),
@@ -391,13 +393,15 @@ export async function createOrUpdateSourceControlPullRequestForTaskRun({
     }
   })();
 
-  await persistSourceControlPullRequestAssociation({
+  const associationWarning = await persistSourceControlPullRequestAssociation({
     taskRun,
     result,
     repository,
   });
 
-  return result;
+  return associationWarning
+    ? { ...result, warnings: [...result.warnings, associationWarning] }
+    : result;
 }
 
 async function resolveLiveGitHubAssigneePlan({
@@ -452,8 +456,8 @@ async function resolveEffectivePrAction(taskRun: TaskRun): Promise<PrAction> {
  * delivery path used to create this association by parsing `gh pr create`
  * tool output from the transcript; the server-side mutation path knows the
  * pull request authoritatively for every provider, so it persists the
- * association directly. Association failures must not fail the mutation the
- * agent already performed.
+ * association directly. Transient association failures are retried, but must
+ * not fail the provider mutation the agent already performed.
  */
 async function persistSourceControlPullRequestAssociation({
   taskRun,
@@ -463,48 +467,61 @@ async function persistSourceControlPullRequestAssociation({
   taskRun: TaskRun;
   result: SourceControlPullRequestMutationResult;
   repository: RepositoryRow;
-}): Promise<void> {
+}): Promise<string | null> {
   if (!taskRun.taskId) {
-    return;
+    return null;
   }
 
   const status = result.draft ? 'draft' : 'open';
 
-  try {
-    await db
-      .insert(taskPullRequests)
-      .values({
-        taskId: taskRun.taskId,
-        sourceControlProvider: repository.sourceControlProvider,
-        host: repository.host,
-        repositoryId: repository.id,
-        prUrl: result.url,
-        prNumber: result.number,
-        prTitle: result.title,
-        repository: result.repositoryFullName,
-        status,
-        createdByRoomote: result.action === 'created',
-        prBaseRef: result.targetBranch,
-      })
-      .onConflictDoUpdate({
-        target: [taskPullRequests.taskId, taskPullRequests.prUrl],
-        set: {
+  for (let attempt = 1; attempt <= PR_ASSOCIATION_MAX_ATTEMPTS; attempt += 1) {
+    try {
+      await db
+        .insert(taskPullRequests)
+        .values({
+          taskId: taskRun.taskId,
           sourceControlProvider: repository.sourceControlProvider,
           host: repository.host,
           repositoryId: repository.id,
+          prUrl: result.url,
+          prNumber: result.number,
           prTitle: result.title,
+          repository: result.repositoryFullName,
           status,
+          createdByRoomote: result.action === 'created',
           prBaseRef: result.targetBranch,
-          updatedAt: new Date(),
-        },
-      });
-  } catch (error) {
-    console.warn(
-      `[persistSourceControlPullRequestAssociation] Failed to associate ${result.repositoryFullName}#${result.number} with task ${taskRun.taskId}: ${
-        error instanceof Error ? error.message : String(error)
-      }`,
-    );
+        })
+        .onConflictDoUpdate({
+          target: [taskPullRequests.taskId, taskPullRequests.prUrl],
+          set: {
+            sourceControlProvider: repository.sourceControlProvider,
+            host: repository.host,
+            repositoryId: repository.id,
+            prTitle: result.title,
+            status,
+            prBaseRef: result.targetBranch,
+            updatedAt: new Date(),
+          },
+        });
+      return null;
+    } catch (error) {
+      if (attempt < PR_ASSOCIATION_MAX_ATTEMPTS) {
+        await new Promise((resolve) =>
+          setTimeout(resolve, PR_ASSOCIATION_RETRY_DELAY_MS),
+        );
+        continue;
+      }
+
+      console.warn(
+        `[persistSourceControlPullRequestAssociation] Failed to associate ${result.repositoryFullName}#${result.number} with task ${taskRun.taskId} after ${attempt} attempts: ${
+          error instanceof Error ? error.message : String(error)
+        }`,
+      );
+      return `The pull request was ${result.action}, but Roomote could not link it to this task after ${attempt} attempts. Review feedback may not reach the task until create_or_update_pull_request is retried.`;
+    }
   }
+
+  return null;
 }
 
 async function createOrUpdateGitHubPullRequest({