防止历史 PR 补查误配和重复推进

cloudflare/automation/migrations/0004_action_log_task_lookup.sql

@@ -0,0 +1,2 @@
+CREATE INDEX IF NOT EXISTS idx_action_log_task_lookup
+  ON action_log(action_type, status, action_key COLLATE NOCASE);

cloudflare/automation/src/index.js

@@ -165,13 +165,15 @@ export function taskCompletionState(outcome) {
   return outcome?.review?.decision === "approve" ? "已完成" : "进行中";
 }
 
-export function taskUpdateActionKey(outcome, previousResponse = null, retryBucket = Math.floor(Date.now() / (10 * 60 * 1000))) {
+export function taskUpdateActionKey(outcome, previousResponse = null, retryBucket = Math.floor(Date.now() / (10 * 60 * 1000)), reconciliationId = "") {
   const baseKey = `task-pr:${outcome.repository}#${outcome.number}:${outcome.headSha || "closed"}:${outcomeStateKey(outcome)}`;
+  if (reconciliationId) return `${baseKey}:reconcile:${reconciliationId}`;
   if (previousResponse?.matched === false && previousResponse.reason === "task_not_linked") return `${baseKey}:link-retry:${retryBucket}`;
   return baseKey;
 }
 
-async function taskUpdateActionKeyForEvent(env, outcome) {
+async function taskUpdateActionKeyForEvent(env, outcome, reconciliationId = "") {
+  if (reconciliationId) return taskUpdateActionKey(outcome, null, undefined, reconciliationId);
   const baseKey = taskUpdateActionKey(outcome);
   const existing = await env.STATE_DB.prepare("SELECT status, response_json FROM action_log WHERE action_key = ?").bind(baseKey).first();
   const previousResponse = existing?.status === "completed" && existing.response_json ? JSON.parse(existing.response_json) : null;
@@ -342,7 +344,7 @@ async function processGithubClosed(env, event) {
   const reference = pullRequestReference(event);
   const payload = eventPayload(event);
   if (!reference || payload.pull_request?.merged !== true) return { handled: false, reason: "pull_request_not_merged" };
-  return { handled: true, repository: reference.repository, number: reference.number, url: `https://github.com/${reference.repository}/pull/${reference.number}`, title: payload.pull_request.title || `PR #${reference.number}`, author_login: payload.pull_request.user?.login || null, headSha: payload.pull_request.merge_commit_sha || payload.pull_request.head?.sha || null, review: { summary: "GitHub 已确认该 PR 合并。", decision: "approve", blocking_findings: [], requested_changes: [], non_blocking_findings: [], test_evidence: [] }, merge: { merged: true, sha: payload.pull_request.merge_commit_sha || null } };
+  return { handled: true, repository: reference.repository, number: reference.number, url: `https://github.com/${reference.repository}/pull/${reference.number}`, title: payload.pull_request.title || `PR #${reference.number}`, author_login: payload.pull_request.user?.login || null, headRefName: payload.pull_request.head?.ref || null, headSha: payload.pull_request.merge_commit_sha || payload.pull_request.head?.sha || null, task_sync: payload.task_sync === true, review: { summary: "GitHub 已确认该 PR 合并。", decision: "approve", blocking_findings: [], requested_changes: [], non_blocking_findings: [], test_evidence: [] }, merge: { merged: true, sha: payload.pull_request.merge_commit_sha || null } };
 }
 
 function pushRecord(event) {
@@ -365,7 +367,8 @@ async function processGithubEffects(env, event, reviewResult) {
   if (event.event_type === "pull_request" && githubAction(event) === "closed") {
     const closed = await processGithubClosed(env, event);
     if (!closed.handled) return closed;
-    const task = await runAction(env, event, "task_update", await taskUpdateActionKeyForEvent(env, closed), () => applyGithubOutcome(env, closed));
+    const reconciliationId = event.payload?.task_sync === true ? event.event_id : "";
+    const task = await runAction(env, event, "task_update", await taskUpdateActionKeyForEvent(env, closed, reconciliationId), () => applyGithubOutcome(env, closed));
     return { ...closed, task };
   }
   if (!reviewResult) return null;
@@ -503,6 +506,7 @@ async function processInternalReconcile(env, event) {
       repository: { full_name: repository },
       pull_request: merged ? { ...fresh, merged: true } : fresh,
       force_review: event.payload.force_review === true,
+      task_sync: merged,
     },
   };
   if (merged) return processGithubEffects(env, synthetic, null);

cloudflare/automation/src/online-runtime.js

@@ -4,8 +4,8 @@ import {
   ensureNextStageTask,
   ensureStageOneTask,
   listProjectRecords,
+  mergedPullRequestTaskSyncCandidate,
   projectForRepository,
-  shouldReconcileMergedPullRequest,
   taskRecordSnapshot,
 } from "./task-board.js";
 import { formatDailyReport } from "./online.js";
@@ -51,7 +51,31 @@ export function mergedPullRequestSyncKey(repository, pullRequest, date = new Dat
   const number = Number(pullRequest?.number);
   const mergeId = String(pullRequest?.merge_commit_sha || pullRequest?.head?.sha || pullRequest?.merged_at || "").trim();
   if (!Number.isInteger(number) || number < 1 || !mergeId) throw new Error("合并 PR 的编号或提交标识缺失");
-  return `merged-pr-task-sync:${repository}#${number}:${mergeId}:${shanghaiDateKey(date)}`;
+  return `merged-pr-task-sync:v2:${repository}#${number}:${mergeId}:${shanghaiDateKey(date)}`;
+}
+
+export function mergedPullRequestScanActionKey(projectKey, date = new Date()) {
+  return `online-merged-pr-scan:v2:${projectKey}:${shanghaiDateKey(date)}`;
+}
+
+export function selectUnambiguousMergedPullRequestCandidates(candidates) {
+  const groups = new Map();
+  for (const candidate of Array.isArray(candidates) ? candidates : []) {
+    if (!candidate?.record_id || !candidate?.pullRequest) continue;
+    const group = groups.get(candidate.record_id) || [];
+    group.push(candidate);
+    groups.set(candidate.record_id, group);
+  }
+  const selected = [];
+  for (const group of groups.values()) {
+    if (group.length === 1) {
+      selected.push(group[0]);
+      continue;
+    }
+    const explicitlyLinked = group.filter((candidate) => candidate.association !== "github_identity");
+    if (explicitlyLinked.length === 1) selected.push(explicitlyLinked[0]);
+  }
+  return selected;
 }
 
 async function enqueueReviewForOpenPullRequest({ env, project, pullRequest, enqueueEvent, runAction, eventId }) {
@@ -87,7 +111,7 @@ async function enqueueMergedPullRequestTaskSync({ env, project, pullRequest, enq
       source: "internal",
       event_type: "reconcile",
       delivery_id: `merged-pr:${project.repository}#${number}:${pullRequest.merge_commit_sha || pullRequest.head?.sha || pullRequest.merged_at}`,
-      payload: { repository: project.repository, number },
+      payload: { repository: project.repository, number, task_sync: true },
     }).then(() => ({ queued: true, repository: project.repository, number })),
   );
 }
@@ -163,7 +187,7 @@ async function continueCompletedTasks({ env, project, records, runAction, eventI
 export async function runOnlineReconciliation({ env, enqueueEvent, runAction } = {}) {
   if (typeof enqueueEvent !== "function" || typeof runAction !== "function") throw new Error("online reconciliation dependencies are missing");
   const eventId = `cron-reconcile:${reconciliationBucket()}`;
-  const summary = { projects: 0, members: 0, stageOneTasks: 0, nextStageTasks: 0, nextStageSkipped: {}, messages: 0, reviewsQueued: 0, openPullRequests: 0, mergedPullRequests: 0, mergedTaskSyncQueued: 0 };
+  const summary = { projects: 0, members: 0, stageOneTasks: 0, nextStageTasks: 0, nextStageSkipped: {}, messages: 0, reviewsQueued: 0, openPullRequests: 0, mergedPullRequests: 0, mergedTaskSyncQueued: 0, mergedTaskSyncAmbiguous: 0 };
   for (const repository of PROJECT_REPOSITORIES) {
     const project = projectForRepository(env, repository);
     if (!project) throw new Error(`project is not configured: ${repository}`);
@@ -190,25 +214,30 @@ export async function runOnlineReconciliation({ env, enqueueEvent, runAction } =
       if (result.queued) summary.reviewsQueued += 1;
     }
     const syncDate = new Date();
-    const syncDay = shanghaiDateKey(syncDate);
     const mergedScan = await runAction(
       env,
       eventForCron(eventId),
       "online_merged_pr_scan",
-      `online-merged-pr-scan:${project.key}:${syncDay}`,
+      mergedPullRequestScanActionKey(project.key, syncDate),
       async () => {
         const mergedPullRequests = await listMergedPullRequests(env, project.repository);
-        let queued = 0;
+        const candidates = [];
         for (const pullRequest of mergedPullRequests) {
-          if (!await shouldReconcileMergedPullRequest(env, project, records, pullRequest)) continue;
+          const candidate = await mergedPullRequestTaskSyncCandidate(env, project, records, pullRequest);
+          if (candidate) candidates.push({ ...candidate, pullRequest });
+        }
+        const selectedCandidates = selectUnambiguousMergedPullRequestCandidates(candidates);
+        let queued = 0;
+        for (const { pullRequest } of selectedCandidates) {
           const result = await enqueueMergedPullRequestTaskSync({ env, project, pullRequest, enqueueEvent, runAction, eventId, date: syncDate });
           if (result.queued) queued += 1;
         }
-        return { mergedPullRequests: mergedPullRequests.length, queued };
+        return { mergedPullRequests: mergedPullRequests.length, queued, ambiguous: candidates.length - selectedCandidates.length };
       },
     );
     summary.mergedPullRequests += Number(mergedScan.mergedPullRequests || 0);
     summary.mergedTaskSyncQueued += Number(mergedScan.queued || 0);
+    summary.mergedTaskSyncAmbiguous += Number(mergedScan.ambiguous || 0);
     summary.projects += 1;
   }
   return summary;

cloudflare/automation/src/task-board.js

@@ -182,6 +182,16 @@ export function selectTaskForGithubIdentity(records, assigneeOpenId) {
   return candidates.length === 1 ? candidates[0] : null;
 }
 
+export function selectTaskForMergedPullRequestFallback(records, assigneeOpenId, outcome) {
+  const { stage, conflict } = pullRequestStageEvidence(outcome);
+  if (conflict || !stage) return null;
+  const candidate = selectTaskForGithubIdentity(records, assigneeOpenId);
+  if (!candidate) return null;
+  return parseStageNumber(valueText(candidate.fields[FIELD_NAMES.title])) === stage
+    ? candidate
+    : null;
+}
+
 export function memberTaskPlan(records, memberOpenId) {
   const memberRecords = (Array.isArray(records) ? records : [])
     .map((record) => ({ record, snapshot: taskRecordSnapshot(record) }))
@@ -233,14 +243,30 @@ export function parseStageNumber(title) {
   return digits[chinese[1]] ?? null;
 }
 
+function pullRequestStageValues(value) {
+  const pattern = /(?:阶段\s*(\d+|[零一二三四五六七八九十百]+)|\bstage[\s_-]*(\d+))/gi;
+  const stages = [...String(value || "").matchAll(pattern)]
+    .map((match) => match[2] ? Number(match[2]) : parseStageNumber(`阶段${match[1]}`))
+    .filter((stage) => Number.isInteger(stage) && stage > 0);
+  return [...new Set(stages)];
+}
+
 export function parsePullRequestStage(value) {
-  const match = String(value || "").match(/(?:阶段|stage)[\s_-]*(\d+)/i);
-  return match ? Number(match[1]) : null;
+  const stages = pullRequestStageValues(value);
+  return stages.length === 1 ? stages[0] : null;
+}
+
+function pullRequestStageEvidence(outcome) {
+  const sources = [outcome?.headRefName, outcome?.title].map(pullRequestStageValues);
+  const stages = [...new Set(sources.flat())];
+  const conflict = sources.some((values) => values.length > 1) || stages.length > 1;
+  return { stage: !conflict && stages.length === 1 ? stages[0] : null, conflict };
 }
 
 export function pullRequestStageMatchesTask(taskTitle, outcome) {
   const taskStage = parseStageNumber(taskTitle);
-  const pullRequestStage = parsePullRequestStage([outcome?.headRefName, outcome?.title].filter(Boolean).join(" "));
+  const { stage: pullRequestStage, conflict } = pullRequestStageEvidence(outcome);
+  if (conflict) return false;
   return !pullRequestStage || !taskStage || pullRequestStage === taskStage;
 }
 
@@ -508,13 +534,16 @@ export async function ensureNextStageTask(env, project, record, records = []) {
 }
 
 function selectLinkedTaskRecord(records, prUrl, number) {
-  return (Array.isArray(records) ? records : [])
-    .filter((record) => prMatches(record, prUrl, number))
-    .sort((left, right) => {
-      const leftStage = parseStageNumber(valueText(left.fields?.[FIELD_NAMES.title])) || Number.MAX_SAFE_INTEGER;
-      const rightStage = parseStageNumber(valueText(right.fields?.[FIELD_NAMES.title])) || Number.MAX_SAFE_INTEGER;
-      return leftStage - rightStage || String(left.record_id).localeCompare(String(right.record_id));
-    })[0] || null;
+  const matches = (Array.isArray(records) ? records : [])
+    .filter((record) => prMatches(record, prUrl, number));
+  return {
+    record: matches.length === 1 ? matches[0] : null,
+    ambiguous: matches.length > 1,
+  };
+}
+
+export function isSafeMergedPullRequestAssociation(association) {
+  return association === "task_link" || association === "saved_action";
 }
 
 export function selectPreviouslyLinkedTaskRecord(records, responses) {
@@ -528,49 +557,67 @@ export function selectPreviouslyLinkedTaskRecord(records, responses) {
 
 async function previouslyLinkedTaskRecord(env, project, records, outcome) {
   const number = Number(outcome?.number);
-  if (!env.STATE_DB || !Number.isInteger(number) || number < 1) return null;
-  const prefix = `task-pr:${project.repository}#${number}:%`;
+  if (!env.STATE_DB || !Number.isInteger(number) || number < 1) return { record: null, ambiguous: false };
+  const prefix = `task-pr:${project.repository}#${number}:`;
   const result = await env.STATE_DB.prepare(
-    "SELECT response_json FROM action_log WHERE action_type = 'task_update' AND status = 'completed' AND action_key LIKE ? ORDER BY updated_at DESC LIMIT 100",
-  ).bind(prefix).all();
-  const responses = (result.results || []).map((row) => JSON.parse(row.response_json || "{}"));
-  return selectPreviouslyLinkedTaskRecord(records, responses);
-}
-
-async function findTaskRecordForOutcome(env, project, records, outcome, prUrl) {
-  let record = selectLinkedTaskRecord(records, prUrl, outcome.number);
+    "SELECT DISTINCT json_extract(response_json, '$.record_id') AS record_id FROM action_log WHERE action_type = 'task_update' AND status = 'completed' AND action_key COLLATE NOCASE >= ? AND action_key COLLATE NOCASE < ? AND response_json IS NOT NULL AND json_extract(response_json, '$.matched') = 1 AND json_extract(response_json, '$.record_id') IS NOT NULL LIMIT 2",
+  ).bind(prefix, `${prefix}\uffff`).all();
+  const recordIds = (result.results || []).map((row) => String(row.record_id));
+  if (recordIds.length > 1) return { record: null, ambiguous: true };
+  if (!recordIds.length) return { record: null, ambiguous: false };
+  const record = (Array.isArray(records) ? records : [])
+    .find((item) => item?.record_id === recordIds[0]) || null;
+  return { record, ambiguous: !record };
+}
+
+async function findTaskRecordForOutcome(env, project, records, outcome, prUrl, { requireExplicitStage = false, allowGithubIdentityFallback = true } = {}) {
+  const linked = selectLinkedTaskRecord(records, prUrl, outcome.number);
+  if (linked.ambiguous) return { record: null, linkedAutomatically: false, association: null, ambiguous: true };
+  let record = linked.record;
   let linkedAutomatically = false;
+  let association = record ? "task_link" : null;
   if (!record) {
-    record = await previouslyLinkedTaskRecord(env, project, records, outcome);
+    const previous = await previouslyLinkedTaskRecord(env, project, records, outcome);
+    if (previous.ambiguous) return { record: null, linkedAutomatically: false, association: null, ambiguous: true };
+    record = previous.record;
     linkedAutomatically = Boolean(record);
+    if (record) association = "saved_action";
   }
   const githubLogin = String(outcome.author_login || "").trim();
-  if (!record && githubLogin) {
+  if (!record && githubLogin && allowGithubIdentityFallback) {
     const identity = await lookupGithubMember(env, project.repository, githubLogin);
-    const candidate = identity ? selectTaskForGithubIdentity(records, identity.assignee_open_id) : null;
+    const candidate = identity
+      ? requireExplicitStage
+        ? selectTaskForMergedPullRequestFallback(records, identity.assignee_open_id, outcome)
+        : selectTaskForGithubIdentity(records, identity.assignee_open_id)
+      : null;
     if (candidate && pullRequestStageMatchesTask(valueText(candidate.fields[FIELD_NAMES.title]), outcome)) {
       record = candidate;
       linkedAutomatically = true;
+      association = "github_identity";
     }
   }
-  return { record, linkedAutomatically };
+  if (record && !pullRequestStageMatchesTask(valueText(record.fields[FIELD_NAMES.title]), outcome)) {
+    return { record: null, linkedAutomatically: false, association: null, ambiguous: false, stageConflict: true };
+  }
+  return { record, linkedAutomatically, association, ambiguous: false, stageConflict: false };
 }
 
-export async function shouldReconcileMergedPullRequest(env, project, records, pullRequest) {
+export async function mergedPullRequestTaskSyncCandidate(env, project, records, pullRequest) {
   const number = Number(pullRequest?.number);
-  if (!Number.isInteger(number) || number < 1) return false;
+  if (!Number.isInteger(number) || number < 1) return null;
   const prUrl = `https://github.com/${project.repository}/pull/${number}`;
   const outcome = {
     number,
     author_login: pullRequest.user?.login || pullRequest.author_login || "",
     headRefName: pullRequest.head?.ref || "",
     title: pullRequest.title || "",
   };
-  const { record } = await findTaskRecordForOutcome(env, project, records, outcome, prUrl);
-  if (!record) return false;
+  const { record, association } = await findTaskRecordForOutcome(env, project, records, outcome, prUrl, { requireExplicitStage: true, allowGithubIdentityFallback: false });
+  if (!record || !isSafeMergedPullRequestAssociation(association)) return null;
   const snapshot = taskRecordSnapshot(record);
-  if (snapshot.status !== "已完成") return true;
-  return nextStageEligibility(records, snapshot).eligible;
+  if (snapshot.status === "已完成" && !nextStageEligibility(records, snapshot).eligible) return null;
+  return { record_id: record.record_id, association };
 }
 
 function outcomeText(outcome) {
@@ -591,8 +638,15 @@ export async function applyGithubOutcome(env, outcome) {
   const records = await listRecords(env, project);
   const prUrl = `https://github.com/${project.repository}/pull/${Number(outcome.number)}`;
   const githubLogin = String(outcome.author_login || "").trim();
-  const { record, linkedAutomatically } = await findTaskRecordForOutcome(env, project, records, outcome, prUrl);
-  if (!record) return { matched: false, reason: "task_not_linked", project: project.key };
+  const { record, linkedAutomatically, ambiguous, stageConflict } = await findTaskRecordForOutcome(
+    env,
+    project,
+    records,
+    outcome,
+    prUrl,
+    { requireExplicitStage: outcome.task_sync === true, allowGithubIdentityFallback: outcome.task_sync !== true },
+  );
+  if (!record) return { matched: false, reason: stageConflict ? "task_stage_conflict" : ambiguous ? "task_association_ambiguous" : "task_not_linked", project: project.key };
   const assignee = assigneeId(record);
   const assigneeName = valueText(record.fields[FIELD_NAMES.assignee]) || "成员";
   if (githubLogin && assignee) await rememberGithubMember(env, project.repository, githubLogin, assignee, assigneeName);