Deduplicate poll-path activity events
This commit is contained in:
whoisasx
parent
df8223ef9d
commit
f8bc9d1ea2
|
|
@ -57,10 +57,10 @@ export function setExecGhAsync(
|
|||
* Configuration constants for cache sizing.
|
||||
* LRU cache automatically evicts oldest entries when these limits are reached.
|
||||
*/
|
||||
const MAX_PR_LIST_ETAGS = 100; // Number of repos to cache
|
||||
const MAX_COMMIT_STATUS_ETAGS = 500; // Number of commits to cache
|
||||
const MAX_REVIEW_COMMENTS_ETAGS = 500; // Number of PRs to cache review ETags
|
||||
const MAX_PR_METADATA = 200; // Number of PRs to cache full data
|
||||
const MAX_PR_LIST_ETAGS = 100; // Number of repos to cache
|
||||
const MAX_COMMIT_STATUS_ETAGS = 500; // Number of commits to cache
|
||||
const MAX_REVIEW_COMMENTS_ETAGS = 500; // Number of PRs to cache review ETags
|
||||
const MAX_PR_METADATA = 200; // Number of PRs to cache full data
|
||||
|
||||
/**
|
||||
* ETag cache for REST API endpoints.
|
||||
|
|
@ -126,11 +126,7 @@ export function getPRListETag(owner: string, repo: string): string | undefined {
|
|||
/**
|
||||
* Get commit status ETag for a specific commit.
|
||||
*/
|
||||
export function getCommitStatusETag(
|
||||
owner: string,
|
||||
repo: string,
|
||||
sha: string,
|
||||
): string | undefined {
|
||||
export function getCommitStatusETag(owner: string, repo: string, sha: string): string | undefined {
|
||||
return etagCache.commitStatus.get(`${owner}/${repo}#${sha}`);
|
||||
}
|
||||
|
||||
|
|
@ -146,12 +142,7 @@ export function setPRListETag(owner: string, repo: string, etag: string): void {
|
|||
* Set commit status ETag for a specific commit.
|
||||
* Exported for testing.
|
||||
*/
|
||||
export function setCommitStatusETag(
|
||||
owner: string,
|
||||
repo: string,
|
||||
sha: string,
|
||||
etag: string,
|
||||
): void {
|
||||
export function setCommitStatusETag(owner: string, repo: string, sha: string, etag: string): void {
|
||||
etagCache.commitStatus.set(`${owner}/${repo}#${sha}`, etag);
|
||||
}
|
||||
|
||||
|
|
@ -162,10 +153,9 @@ export function setCommitStatusETag(
|
|||
*
|
||||
* Uses LRU eviction to ensure bounded memory usage.
|
||||
*/
|
||||
const prMetadataCache = new LRUCache<
|
||||
string,
|
||||
{ headSha: string | null; ciStatus: CIStatus }
|
||||
>(MAX_PR_METADATA);
|
||||
const prMetadataCache = new LRUCache<string, { headSha: string | null; ciStatus: CIStatus }>(
|
||||
MAX_PR_METADATA,
|
||||
);
|
||||
|
||||
/**
|
||||
* Cache for full PR enrichment data.
|
||||
|
|
@ -242,7 +232,11 @@ export async function shouldRefreshPREnrichment(
|
|||
}
|
||||
|
||||
if (repos.size === 0) {
|
||||
return { shouldRefresh: false, details: ["No repos to check"], prListUnchangedRepos: new Set() };
|
||||
return {
|
||||
shouldRefresh: false,
|
||||
details: ["No repos to check"],
|
||||
prListUnchangedRepos: new Set(),
|
||||
};
|
||||
}
|
||||
|
||||
// Guard 1: Check PR list ETag for each repository
|
||||
|
|
@ -298,9 +292,7 @@ export async function shouldRefreshPREnrichment(
|
|||
);
|
||||
if (statusChanged) {
|
||||
shouldRefresh = true;
|
||||
details.push(
|
||||
`CI status changed for ${pr.owner}/${pr.repo}#${pr.number} (Guard 2)`,
|
||||
);
|
||||
details.push(`CI status changed for ${pr.owner}/${pr.repo}#${pr.number} (Guard 2)`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -311,10 +303,7 @@ export async function shouldRefreshPREnrichment(
|
|||
/**
|
||||
* Get cached PR metadata for testing.
|
||||
*/
|
||||
export function getPRMetadataCache(): Map<
|
||||
string,
|
||||
{ headSha: string | null; ciStatus: CIStatus }
|
||||
> {
|
||||
export function getPRMetadataCache(): Map<string, { headSha: string | null; ciStatus: CIStatus }> {
|
||||
return prMetadataCache.toMap();
|
||||
}
|
||||
|
||||
|
|
@ -354,6 +343,7 @@ interface ErrorWithCause extends Error {
|
|||
// Module-level guard so we only emit gh_unavailable once per process.
|
||||
// The error is system-wide (gh missing globally), not session-specific.
|
||||
let ghUnavailableEmitted = false;
|
||||
const batchEnrichPRFailedEmitted = new Set<string>();
|
||||
|
||||
/**
|
||||
* Pre-flight check to verify gh CLI is available and authenticated.
|
||||
|
|
@ -390,6 +380,11 @@ export function _resetGhUnavailableEmittedForTesting(): void {
|
|||
ghUnavailableEmitted = false;
|
||||
}
|
||||
|
||||
/** Test-only: reset the once-per-PR batch extraction failure guard. */
|
||||
export function _resetBatchEnrichPRFailedEmittedForTesting(): void {
|
||||
batchEnrichPRFailedEmitted.clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* Maximum number of PRs to query in a single GraphQL batch.
|
||||
* GitHub has limits on query complexity and we stay well under this limit.
|
||||
|
|
@ -557,7 +552,10 @@ async function checkCommitStatusETag(
|
|||
if (is304(errorMsg)) {
|
||||
return false;
|
||||
}
|
||||
observer?.log("warn", `[ETag Guard 2] Commit status check failed for ${commitKey}: ${errorMsg}`);
|
||||
observer?.log(
|
||||
"warn",
|
||||
`[ETag Guard 2] Commit status check failed for ${commitKey}: ${errorMsg}`,
|
||||
);
|
||||
return true; // Assume changed to be safe
|
||||
}
|
||||
}
|
||||
|
|
@ -620,7 +618,10 @@ export async function checkReviewCommentsETag(
|
|||
if (is304(errorMsg)) {
|
||||
return false;
|
||||
}
|
||||
observer?.log("warn", `[ETag Guard 3] Review comments check failed for ${cacheKey}: ${errorMsg}`);
|
||||
observer?.log(
|
||||
"warn",
|
||||
`[ETag Guard 3] Review comments check failed for ${cacheKey}: ${errorMsg}`,
|
||||
);
|
||||
return true; // Assume changed to be safe
|
||||
}
|
||||
}
|
||||
|
|
@ -728,9 +729,7 @@ export function generateBatchQuery(prs: PRInfo[]): {
|
|||
*
|
||||
* @throws Error if the query fails with GraphQL errors or parsing issues.
|
||||
*/
|
||||
async function executeBatchQuery(
|
||||
prs: PRInfo[],
|
||||
): Promise<Record<string, unknown>> {
|
||||
async function executeBatchQuery(prs: PRInfo[]): Promise<Record<string, unknown>> {
|
||||
const { query, variables } = generateBatchQuery(prs);
|
||||
|
||||
// Handle empty array - no query needed
|
||||
|
|
@ -890,9 +889,7 @@ function parseCheckContexts(contexts: unknown): CICheck[] {
|
|||
* Uses only the top-level aggregate state to determine overall CI status.
|
||||
* Individual check details are parsed separately via parseCheckContexts().
|
||||
*/
|
||||
function parseCIState(
|
||||
statusCheckRollup: unknown,
|
||||
): CIStatus {
|
||||
function parseCIState(statusCheckRollup: unknown): CIStatus {
|
||||
if (!statusCheckRollup || typeof statusCheckRollup !== "object") {
|
||||
return "none";
|
||||
}
|
||||
|
|
@ -909,8 +906,7 @@ function parseCIState(
|
|||
if (state === "PENDING" || state === "EXPECTED") return "pending";
|
||||
if (state === "TIMED_OUT" || state === "CANCELLED" || state === "ACTION_REQUIRED")
|
||||
return "failing";
|
||||
if (state === "QUEUED" || state === "IN_PROGRESS" || state === "WAITING")
|
||||
return "pending";
|
||||
if (state === "QUEUED" || state === "IN_PROGRESS" || state === "WAITING") return "pending";
|
||||
|
||||
return "none";
|
||||
}
|
||||
|
|
@ -951,11 +947,7 @@ function extractPREnrichment(
|
|||
const pr = pullRequest as Record<string, unknown>;
|
||||
|
||||
// Check for at least one required field to validate this is a valid PR object
|
||||
if (
|
||||
pr["state"] === undefined &&
|
||||
pr["title"] === undefined &&
|
||||
pr["commits"] === undefined
|
||||
) {
|
||||
if (pr["state"] === undefined && pr["title"] === undefined && pr["commits"] === undefined) {
|
||||
return null;
|
||||
}
|
||||
|
||||
|
|
@ -978,9 +970,7 @@ function extractPREnrichment(
|
|||
// Extract merge info
|
||||
const mergeable = pr["mergeable"];
|
||||
const mergeStateStatus =
|
||||
typeof pr["mergeStateStatus"] === "string"
|
||||
? pr["mergeStateStatus"].toUpperCase()
|
||||
: "";
|
||||
typeof pr["mergeStateStatus"] === "string" ? pr["mergeStateStatus"].toUpperCase() : "";
|
||||
const hasConflicts = mergeable === "CONFLICTING";
|
||||
const isBehind = mergeStateStatus === "BEHIND";
|
||||
|
||||
|
|
@ -998,9 +988,7 @@ function extractPREnrichment(
|
|||
// contexts(first: 20) silently truncates PRs with >20 checks — when truncated,
|
||||
// the failing check may be missing, so we set ciChecks to undefined to force
|
||||
// the getCIChecks() REST fallback in maybeDispatchCIFailureDetails.
|
||||
const contextsField = statusCheckRollup?.["contexts"] as
|
||||
| Record<string, unknown>
|
||||
| undefined;
|
||||
const contextsField = statusCheckRollup?.["contexts"] as Record<string, unknown> | undefined;
|
||||
const pageInfo = contextsField?.["pageInfo"];
|
||||
const contextsHasNextPage =
|
||||
pageInfo !== null &&
|
||||
|
|
@ -1008,15 +996,12 @@ function extractPREnrichment(
|
|||
typeof pageInfo === "object" &&
|
||||
(pageInfo as Record<string, unknown>)["hasNextPage"] === true;
|
||||
const ciChecks =
|
||||
contextsField && !contextsHasNextPage
|
||||
? parseCheckContexts(contextsField)
|
||||
: undefined;
|
||||
contextsField && !contextsHasNextPage ? parseCheckContexts(contextsField) : undefined;
|
||||
|
||||
// Build blockers list
|
||||
const blockers: string[] = [];
|
||||
if (ciStatus === "failing") blockers.push("CI is failing");
|
||||
if (reviewDecision === "changes_requested")
|
||||
blockers.push("Changes requested in review");
|
||||
if (reviewDecision === "changes_requested") blockers.push("Changes requested in review");
|
||||
if (reviewDecision === "pending") blockers.push("Review required");
|
||||
if (hasConflicts) blockers.push("Merge conflicts");
|
||||
if (isBehind) blockers.push("Branch is behind base branch");
|
||||
|
|
@ -1154,18 +1139,21 @@ export async function enrichSessionsPRBatch(
|
|||
// GraphQL returned a PR object but extractPREnrichment couldn't
|
||||
// parse it (missing fields, schema drift). Distinct from the
|
||||
// whole-batch failure D02 catches further down.
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.batch_enrich_pr_failed",
|
||||
level: "warn",
|
||||
summary: `batch enrich extraction failed for PR #${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-github",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
},
|
||||
});
|
||||
if (!batchEnrichPRFailedEmitted.has(prKey)) {
|
||||
batchEnrichPRFailedEmitted.add(prKey);
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.batch_enrich_pr_failed",
|
||||
level: "warn",
|
||||
summary: `batch enrich extraction failed for PR #${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-github",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// PR not found (deleted/closed/permission issue)
|
||||
|
|
@ -1187,7 +1175,10 @@ export async function enrichSessionsPRBatch(
|
|||
durationMs: batchDuration,
|
||||
};
|
||||
observer?.recordSuccess(successData);
|
||||
observer?.log("info", `[GraphQL Batch Success] Batch ${batchIndex + 1}/${batches.length} succeeded: added ${prCountAfter - prCountBefore} PRs to cache (${batchDuration}ms)`);
|
||||
observer?.log(
|
||||
"info",
|
||||
`[GraphQL Batch Success] Batch ${batchIndex + 1}/${batches.length} succeeded: added ${prCountAfter - prCountBefore} PRs to cache (${batchDuration}ms)`,
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
// Calculate duration even on failure
|
||||
|
|
|
|||
|
|
@ -63,6 +63,12 @@ const BOT_AUTHORS = new Set([
|
|||
]);
|
||||
|
||||
const CI_FAILURE_LOG_TAIL_LINES = 120;
|
||||
const ciSummaryFailClosedEmitted = new Set<string>();
|
||||
|
||||
/** Test-only: reset once-per-PR activity event guards. */
|
||||
export function _resetGitHubActivityEventDedupeForTesting(): void {
|
||||
ciSummaryFailClosedEmitted.clear();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
|
|
@ -238,10 +244,7 @@ async function getFailedJobLog(
|
|||
]);
|
||||
} catch (err) {
|
||||
if (!runReference.jobId) throw err;
|
||||
return gh([
|
||||
"api",
|
||||
`repos/${pr.owner}/${pr.repo}/actions/jobs/${runReference.jobId}/logs`,
|
||||
]);
|
||||
return gh(["api", `repos/${pr.owner}/${pr.repo}/actions/jobs/${runReference.jobId}/logs`]);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -523,6 +526,10 @@ function repoFlag(pr: PRInfo): string {
|
|||
return `${pr.owner}/${pr.repo}`;
|
||||
}
|
||||
|
||||
function prEventKey(pr: PRInfo): string {
|
||||
return `${repoFlag(pr)}#${pr.number}`;
|
||||
}
|
||||
|
||||
function parseDate(val: string | undefined | null): Date {
|
||||
if (!val) return new Date(0);
|
||||
const d = new Date(val);
|
||||
|
|
@ -946,20 +953,24 @@ function createGitHubSCM(): SCM {
|
|||
// Fail closed for open PRs: report as failing rather than
|
||||
// "none" (which getMergeability treats as passing). Emit so RCA
|
||||
// can distinguish "really failing" from "we couldn't tell".
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.ci_summary_failclosed",
|
||||
level: "warn",
|
||||
summary: `getCISummary failed-closed for PR #${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-github",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
errorMessage,
|
||||
},
|
||||
});
|
||||
const eventKey = prEventKey(pr);
|
||||
if (!ciSummaryFailClosedEmitted.has(eventKey)) {
|
||||
ciSummaryFailClosedEmitted.add(eventKey);
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.ci_summary_failclosed",
|
||||
level: "warn",
|
||||
summary: `getCISummary failed-closed for PR #${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-github",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
errorMessage,
|
||||
},
|
||||
});
|
||||
}
|
||||
return "failing";
|
||||
}
|
||||
if (checks.length === 0) return "none";
|
||||
|
|
@ -1051,16 +1062,16 @@ function createGitHubSCM(): SCM {
|
|||
try {
|
||||
// Use GraphQL with variables to get review threads with actual isResolved status
|
||||
const raw = await gh([
|
||||
"api",
|
||||
"graphql",
|
||||
"-f",
|
||||
`owner=${pr.owner}`,
|
||||
"-f",
|
||||
`name=${pr.repo}`,
|
||||
"-F",
|
||||
`number=${pr.number}`,
|
||||
"-f",
|
||||
`query=query($owner: String!, $name: String!, $number: Int!) {
|
||||
"api",
|
||||
"graphql",
|
||||
"-f",
|
||||
`owner=${pr.owner}`,
|
||||
"-f",
|
||||
`name=${pr.repo}`,
|
||||
"-F",
|
||||
`number=${pr.number}`,
|
||||
"-f",
|
||||
`query=query($owner: String!, $name: String!, $number: Int!) {
|
||||
repository(owner: $owner, name: $name) {
|
||||
pullRequest(number: $number) {
|
||||
reviewThreads(first: 100) {
|
||||
|
|
@ -1083,58 +1094,58 @@ function createGitHubSCM(): SCM {
|
|||
}
|
||||
}
|
||||
}`,
|
||||
]);
|
||||
]);
|
||||
|
||||
const data: {
|
||||
data: {
|
||||
repository: {
|
||||
pullRequest: {
|
||||
reviewThreads: {
|
||||
nodes: Array<{
|
||||
id: string;
|
||||
isResolved: boolean;
|
||||
comments: {
|
||||
nodes: Array<{
|
||||
id: string;
|
||||
author: { login: string } | null;
|
||||
body: string;
|
||||
path: string | null;
|
||||
line: number | null;
|
||||
url: string;
|
||||
createdAt: string;
|
||||
}>;
|
||||
};
|
||||
}>;
|
||||
const data: {
|
||||
data: {
|
||||
repository: {
|
||||
pullRequest: {
|
||||
reviewThreads: {
|
||||
nodes: Array<{
|
||||
id: string;
|
||||
isResolved: boolean;
|
||||
comments: {
|
||||
nodes: Array<{
|
||||
id: string;
|
||||
author: { login: string } | null;
|
||||
body: string;
|
||||
path: string | null;
|
||||
line: number | null;
|
||||
url: string;
|
||||
createdAt: string;
|
||||
}>;
|
||||
};
|
||||
}>;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
} = JSON.parse(raw);
|
||||
} = JSON.parse(raw);
|
||||
|
||||
const threads = data.data.repository.pullRequest.reviewThreads.nodes;
|
||||
const threads = data.data.repository.pullRequest.reviewThreads.nodes;
|
||||
|
||||
return threads
|
||||
.filter((t) => {
|
||||
if (t.isResolved) return false; // only pending (unresolved) threads
|
||||
const c = t.comments.nodes[0];
|
||||
if (!c) return false; // skip threads with no comments
|
||||
const author = c.author?.login ?? "";
|
||||
return !BOT_AUTHORS.has(author);
|
||||
})
|
||||
.map((t) => {
|
||||
const c = t.comments.nodes[0];
|
||||
return {
|
||||
id: c.id,
|
||||
threadId: t.id,
|
||||
author: c.author?.login ?? "unknown",
|
||||
body: c.body,
|
||||
path: c.path || undefined,
|
||||
line: c.line ?? undefined,
|
||||
isResolved: t.isResolved,
|
||||
createdAt: parseDate(c.createdAt),
|
||||
url: c.url,
|
||||
};
|
||||
});
|
||||
return threads
|
||||
.filter((t) => {
|
||||
if (t.isResolved) return false; // only pending (unresolved) threads
|
||||
const c = t.comments.nodes[0];
|
||||
if (!c) return false; // skip threads with no comments
|
||||
const author = c.author?.login ?? "";
|
||||
return !BOT_AUTHORS.has(author);
|
||||
})
|
||||
.map((t) => {
|
||||
const c = t.comments.nodes[0];
|
||||
return {
|
||||
id: c.id,
|
||||
threadId: t.id,
|
||||
author: c.author?.login ?? "unknown",
|
||||
body: c.body,
|
||||
path: c.path || undefined,
|
||||
line: c.line ?? undefined,
|
||||
isResolved: t.isResolved,
|
||||
createdAt: parseDate(c.createdAt),
|
||||
url: c.url,
|
||||
};
|
||||
});
|
||||
} catch (err) {
|
||||
throw new Error("Failed to fetch pending comments", { cause: err });
|
||||
}
|
||||
|
|
@ -1145,7 +1156,12 @@ function createGitHubSCM(): SCM {
|
|||
const cacheKey = `${pr.owner}/${pr.repo}#${pr.number}`;
|
||||
|
||||
// Guard 3: check if review comments changed via REST ETag
|
||||
const reviewsChanged = await checkReviewCommentsETag(pr.owner, pr.repo, pr.number, instanceObserver);
|
||||
const reviewsChanged = await checkReviewCommentsETag(
|
||||
pr.owner,
|
||||
pr.repo,
|
||||
pr.number,
|
||||
instanceObserver,
|
||||
);
|
||||
if (!reviewsChanged) {
|
||||
const cached = reviewThreadsCache.get(cacheKey);
|
||||
if (cached) return cached;
|
||||
|
|
|
|||
|
|
@ -21,9 +21,11 @@ vi.mock("@aoagents/ao-core", async () => {
|
|||
import {
|
||||
enrichSessionsPRBatch,
|
||||
setExecFileAsync,
|
||||
setExecGhAsync,
|
||||
clearETagCache,
|
||||
clearPRMetadataCache,
|
||||
_resetGhUnavailableEmittedForTesting,
|
||||
_resetBatchEnrichPRFailedEmittedForTesting,
|
||||
} from "../src/graphql-batch.js";
|
||||
import type { PRInfo } from "@aoagents/ao-core";
|
||||
|
||||
|
|
@ -45,6 +47,7 @@ beforeEach(() => {
|
|||
clearETagCache();
|
||||
clearPRMetadataCache();
|
||||
_resetGhUnavailableEmittedForTesting();
|
||||
_resetBatchEnrichPRFailedEmittedForTesting();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
|
|
@ -100,3 +103,50 @@ describe("scm.gh_unavailable (MUST emit)", () => {
|
|||
expect(ghUnavailableCalls).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("scm.batch_enrich_pr_failed (poll-path emit)", () => {
|
||||
it("emits exactly once per PR across repeated extraction failures", async () => {
|
||||
const execFileMock = vi.fn().mockResolvedValue({ stdout: "gh version", stderr: "" });
|
||||
setExecFileAsync(execFileMock as unknown as Parameters<typeof setExecFileAsync>[0]);
|
||||
|
||||
const execGhMock = vi.fn(
|
||||
async (_args: string[], _timeout: number, operation: string): Promise<string> => {
|
||||
if (operation === "gh.api.guard-pr-list") {
|
||||
return 'HTTP/2 200 OK\netag: W/"pr-list"\n\n[]';
|
||||
}
|
||||
if (operation === "gh.api.graphql-batch") {
|
||||
return `HTTP/2 200 OK\n\n${JSON.stringify({
|
||||
data: {
|
||||
pr0: {
|
||||
pullRequest: { unexpectedShape: true },
|
||||
},
|
||||
},
|
||||
})}`;
|
||||
}
|
||||
throw new Error(`Unexpected gh operation: ${operation}`);
|
||||
},
|
||||
);
|
||||
setExecGhAsync(execGhMock);
|
||||
|
||||
await enrichSessionsPRBatch(samplePRs);
|
||||
await enrichSessionsPRBatch(samplePRs);
|
||||
|
||||
const extractionFailureCalls = recordActivityEventMock.mock.calls.filter(
|
||||
([event]) => event.kind === "scm.batch_enrich_pr_failed",
|
||||
);
|
||||
expect(extractionFailureCalls).toHaveLength(1);
|
||||
expect(extractionFailureCalls[0]?.[0]).toEqual(
|
||||
expect.objectContaining({
|
||||
source: "scm",
|
||||
kind: "scm.batch_enrich_pr_failed",
|
||||
level: "warn",
|
||||
data: expect.objectContaining({
|
||||
plugin: "scm-github",
|
||||
prNumber: 42,
|
||||
prOwner: "octocat",
|
||||
prRepo: "hello-world",
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -4,7 +4,10 @@ import { describe, it, expect, beforeEach, vi } from "vitest";
|
|||
// Mock node:child_process — gh CLI calls go through execFileAsync = promisify(execFile)
|
||||
// vi.hoisted ensures the mock fn is available when vi.mock factory runs (hoisted above imports)
|
||||
// ---------------------------------------------------------------------------
|
||||
const { ghMock } = vi.hoisted(() => ({ ghMock: vi.fn() }));
|
||||
const { ghMock, recordActivityEventMock } = vi.hoisted(() => ({
|
||||
ghMock: vi.fn(),
|
||||
recordActivityEventMock: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("node:child_process", () => {
|
||||
// Attach the custom promisify symbol so `promisify(execFile)` returns ghMock
|
||||
|
|
@ -14,8 +17,24 @@ vi.mock("node:child_process", () => {
|
|||
return { execFile };
|
||||
});
|
||||
|
||||
import { create, manifest } from "../src/index.js";
|
||||
import { _clearProcessCacheForTests, createActivitySignal, type PreflightContext, type PRInfo, type SCMWebhookRequest, type Session, type ProjectConfig } from "@aoagents/ao-core";
|
||||
vi.mock("@aoagents/ao-core", async () => {
|
||||
const actual = (await vi.importActual("@aoagents/ao-core")) as Record<string, unknown>;
|
||||
return {
|
||||
...actual,
|
||||
recordActivityEvent: recordActivityEventMock,
|
||||
};
|
||||
});
|
||||
|
||||
import { create, manifest, _resetGitHubActivityEventDedupeForTesting } from "../src/index.js";
|
||||
import {
|
||||
_clearProcessCacheForTests,
|
||||
createActivitySignal,
|
||||
type PreflightContext,
|
||||
type PRInfo,
|
||||
type SCMWebhookRequest,
|
||||
type Session,
|
||||
type ProjectConfig,
|
||||
} from "@aoagents/ao-core";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Fixtures
|
||||
|
|
@ -101,6 +120,7 @@ describe("scm-github plugin", () => {
|
|||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
_resetGitHubActivityEventDedupeForTesting();
|
||||
scm = create();
|
||||
delete process.env["GITHUB_WEBHOOK_SECRET"];
|
||||
});
|
||||
|
|
@ -665,13 +685,10 @@ describe("scm-github plugin", () => {
|
|||
url: "https://github.com/acme/repo/actions/runs/124/job/457",
|
||||
},
|
||||
];
|
||||
const logLines = Array.from(
|
||||
{ length: 125 },
|
||||
(_, index) => {
|
||||
const step = index < 100 ? "Install dependencies" : "Run pnpm test";
|
||||
return `build\t${step}\t2026-05-12T00:00:00Z line ${index + 1}`;
|
||||
},
|
||||
);
|
||||
const logLines = Array.from({ length: 125 }, (_, index) => {
|
||||
const step = index < 100 ? "Install dependencies" : "Run pnpm test";
|
||||
return `build\t${step}\t2026-05-12T00:00:00Z line ${index + 1}`;
|
||||
});
|
||||
mockGhRaw(logLines.join("\n"));
|
||||
|
||||
const summary = await scm.getCIFailureSummary?.(pr, checks);
|
||||
|
|
@ -774,6 +791,34 @@ describe("scm-github plugin", () => {
|
|||
expect(await scm.getCISummary(pr)).toBe("failing");
|
||||
});
|
||||
|
||||
it("dedupes fail-closed activity events per PR", async () => {
|
||||
mockGhError("checks failed");
|
||||
mockGhError("state failed");
|
||||
mockGhError("checks failed again");
|
||||
mockGhError("state failed again");
|
||||
|
||||
expect(await scm.getCISummary(pr)).toBe("failing");
|
||||
expect(await scm.getCISummary(pr)).toBe("failing");
|
||||
|
||||
const failClosedCalls = recordActivityEventMock.mock.calls.filter(
|
||||
([event]) => event.kind === "scm.ci_summary_failclosed",
|
||||
);
|
||||
expect(failClosedCalls).toHaveLength(1);
|
||||
expect(failClosedCalls[0]?.[0]).toEqual(
|
||||
expect.objectContaining({
|
||||
source: "scm",
|
||||
kind: "scm.ci_summary_failclosed",
|
||||
level: "warn",
|
||||
data: expect.objectContaining({
|
||||
plugin: "scm-github",
|
||||
prNumber: 42,
|
||||
prOwner: "acme",
|
||||
prRepo: "repo",
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('returns "none" when all checks are skipped', async () => {
|
||||
mockGh([
|
||||
{ name: "a", state: "SKIPPED" },
|
||||
|
|
|
|||
|
|
@ -46,6 +46,14 @@ const BOT_AUTHORS = new Set([
|
|||
"sonarcloud[bot]",
|
||||
"snyk-bot",
|
||||
]);
|
||||
const ciSummaryFailClosedEmitted = new Set<string>();
|
||||
const reviewFetchFailedEmitted = new Set<string>();
|
||||
|
||||
/** Test-only: reset once-per-PR activity event guards. */
|
||||
export function _resetGitLabActivityEventDedupeForTesting(): void {
|
||||
ciSummaryFailClosedEmitted.clear();
|
||||
reviewFetchFailedEmitted.clear();
|
||||
}
|
||||
|
||||
function isBot(username: string): boolean {
|
||||
return (
|
||||
|
|
@ -61,6 +69,10 @@ function repoFlag(pr: PRInfo): string {
|
|||
return `${pr.owner}/${pr.repo}`;
|
||||
}
|
||||
|
||||
function prEventKey(pr: PRInfo): string {
|
||||
return `${repoFlag(pr)}#${pr.number}`;
|
||||
}
|
||||
|
||||
function parseDate(val: string | undefined | null): Date {
|
||||
if (!val) return new Date(0);
|
||||
const d = new Date(val);
|
||||
|
|
@ -107,7 +119,6 @@ function mapPRState(state: string): PRState {
|
|||
return "open";
|
||||
}
|
||||
|
||||
|
||||
function getGitLabWebhookConfig(project: ProjectConfig) {
|
||||
const webhook = project.scm?.webhook;
|
||||
return {
|
||||
|
|
@ -586,20 +597,24 @@ function createGitLabSCM(config?: Record<string, unknown>): SCM {
|
|||
`getCISummary: PR state fallback also failed for MR !${pr.number}: ${(innerErr as Error).message}`,
|
||||
);
|
||||
}
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.ci_summary_failclosed",
|
||||
level: "warn",
|
||||
summary: `getCISummary failed-closed for MR !${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-gitlab",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
errorMessage,
|
||||
},
|
||||
});
|
||||
const eventKey = prEventKey(pr);
|
||||
if (!ciSummaryFailClosedEmitted.has(eventKey)) {
|
||||
ciSummaryFailClosedEmitted.add(eventKey);
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.ci_summary_failclosed",
|
||||
level: "warn",
|
||||
summary: `getCISummary failed-closed for MR !${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-gitlab",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
errorMessage,
|
||||
},
|
||||
});
|
||||
}
|
||||
return "failing";
|
||||
}
|
||||
if (checks.length === 0) return "none";
|
||||
|
|
@ -657,20 +672,24 @@ function createGitLabSCM(config?: Record<string, unknown>): SCM {
|
|||
console.warn(
|
||||
`getReviews: discussions fetch failed for MR !${pr.number}: ${(err as Error).message}`,
|
||||
);
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.review_fetch_failed",
|
||||
level: "warn",
|
||||
summary: `getReviews discussions fetch failed for MR !${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-gitlab",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
errorMessage,
|
||||
},
|
||||
});
|
||||
const eventKey = prEventKey(pr);
|
||||
if (!reviewFetchFailedEmitted.has(eventKey)) {
|
||||
reviewFetchFailedEmitted.add(eventKey);
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
recordActivityEvent({
|
||||
source: "scm",
|
||||
kind: "scm.review_fetch_failed",
|
||||
level: "warn",
|
||||
summary: `getReviews discussions fetch failed for MR !${pr.number}`,
|
||||
data: {
|
||||
plugin: "scm-gitlab",
|
||||
prNumber: pr.number,
|
||||
prOwner: pr.owner,
|
||||
prRepo: pr.repo,
|
||||
errorMessage,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return reviews;
|
||||
|
|
|
|||
|
|
@ -3,7 +3,10 @@ import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
|
|||
// ---------------------------------------------------------------------------
|
||||
// Mock node:child_process — glab CLI calls go through execFileAsync
|
||||
// ---------------------------------------------------------------------------
|
||||
const { glabMock } = vi.hoisted(() => ({ glabMock: vi.fn() }));
|
||||
const { glabMock, recordActivityEventMock } = vi.hoisted(() => ({
|
||||
glabMock: vi.fn(),
|
||||
recordActivityEventMock: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("node:child_process", () => {
|
||||
const execFile = Object.assign(vi.fn(), {
|
||||
|
|
@ -12,8 +15,22 @@ vi.mock("node:child_process", () => {
|
|||
return { execFile };
|
||||
});
|
||||
|
||||
import { create, manifest } from "../src/index.js";
|
||||
import { createActivitySignal, type PRInfo, type Session, type ProjectConfig, type SCMWebhookRequest } from "@aoagents/ao-core";
|
||||
vi.mock("@aoagents/ao-core", async () => {
|
||||
const actual = (await vi.importActual("@aoagents/ao-core")) as Record<string, unknown>;
|
||||
return {
|
||||
...actual,
|
||||
recordActivityEvent: recordActivityEventMock,
|
||||
};
|
||||
});
|
||||
|
||||
import { create, manifest, _resetGitLabActivityEventDedupeForTesting } from "../src/index.js";
|
||||
import {
|
||||
createActivitySignal,
|
||||
type PRInfo,
|
||||
type Session,
|
||||
type ProjectConfig,
|
||||
type SCMWebhookRequest,
|
||||
} from "@aoagents/ao-core";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Fixtures
|
||||
|
|
@ -104,6 +121,7 @@ describe("scm-gitlab plugin", () => {
|
|||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
_resetGitLabActivityEventDedupeForTesting();
|
||||
warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
scm = create();
|
||||
delete process.env["GITLAB_WEBHOOK_SECRET"];
|
||||
|
|
@ -701,6 +719,34 @@ describe("scm-gitlab plugin", () => {
|
|||
expect(await scm.getCISummary(pr)).toBe("failing");
|
||||
});
|
||||
|
||||
it("dedupes fail-closed activity events per MR", async () => {
|
||||
mockGlabError("pipeline error");
|
||||
mockGlabError("network error");
|
||||
mockGlabError("pipeline error again");
|
||||
mockGlabError("network error again");
|
||||
|
||||
expect(await scm.getCISummary(pr)).toBe("failing");
|
||||
expect(await scm.getCISummary(pr)).toBe("failing");
|
||||
|
||||
const failClosedCalls = recordActivityEventMock.mock.calls.filter(
|
||||
([event]) => event.kind === "scm.ci_summary_failclosed",
|
||||
);
|
||||
expect(failClosedCalls).toHaveLength(1);
|
||||
expect(failClosedCalls[0]?.[0]).toEqual(
|
||||
expect.objectContaining({
|
||||
source: "scm",
|
||||
kind: "scm.ci_summary_failclosed",
|
||||
level: "warn",
|
||||
data: expect.objectContaining({
|
||||
plugin: "scm-gitlab",
|
||||
prNumber: 42,
|
||||
prOwner: "acme",
|
||||
prRepo: "repo",
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('returns "none" when all jobs are skipped', async () => {
|
||||
mockGlab([{ id: 1 }]);
|
||||
mockGlab([
|
||||
|
|
@ -807,6 +853,34 @@ describe("scm-gitlab plugin", () => {
|
|||
);
|
||||
});
|
||||
|
||||
it("dedupes review fetch failed activity events per MR", async () => {
|
||||
mockGlab({ approved_by: [] });
|
||||
mockGlabError("discussions fetch failed");
|
||||
mockGlab({ approved_by: [] });
|
||||
mockGlabError("discussions fetch failed again");
|
||||
|
||||
expect(await scm.getReviews(pr)).toEqual([]);
|
||||
expect(await scm.getReviews(pr)).toEqual([]);
|
||||
|
||||
const reviewFetchCalls = recordActivityEventMock.mock.calls.filter(
|
||||
([event]) => event.kind === "scm.review_fetch_failed",
|
||||
);
|
||||
expect(reviewFetchCalls).toHaveLength(1);
|
||||
expect(reviewFetchCalls[0]?.[0]).toEqual(
|
||||
expect.objectContaining({
|
||||
source: "scm",
|
||||
kind: "scm.review_fetch_failed",
|
||||
level: "warn",
|
||||
data: expect.objectContaining({
|
||||
plugin: "scm-gitlab",
|
||||
prNumber: 42,
|
||||
prOwner: "acme",
|
||||
prRepo: "repo",
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("filters bot authors from discussions", async () => {
|
||||
mockGlab({ approved_by: [] });
|
||||
mockGlab([
|
||||
|
|
|
|||
|
|
@ -6,8 +6,13 @@
|
|||
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const { recordActivityEventMock } = vi.hoisted(() => ({
|
||||
const { recordActivityEventMock, requestMock } = vi.hoisted(() => ({
|
||||
recordActivityEventMock: vi.fn(),
|
||||
requestMock: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("node:https", () => ({
|
||||
request: requestMock,
|
||||
}));
|
||||
|
||||
vi.mock("@aoagents/ao-core", async () => {
|
||||
|
|
@ -27,6 +32,8 @@ import type { ProjectConfig } from "@aoagents/ao-core";
|
|||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
recordActivityEventMock.mockReset();
|
||||
requestMock.mockReset();
|
||||
_resetDepMissingEmittedForTesting();
|
||||
process.env.COMPOSIO_API_KEY = "test-key";
|
||||
process.env.COMPOSIO_ENTITY_ID = "test-entity";
|
||||
|
|
@ -36,6 +43,8 @@ beforeEach(() => {
|
|||
afterEach(() => {
|
||||
delete process.env.COMPOSIO_API_KEY;
|
||||
delete process.env.COMPOSIO_ENTITY_ID;
|
||||
delete process.env.LINEAR_API_KEY;
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
function makeProject(): ProjectConfig {
|
||||
|
|
@ -85,3 +94,50 @@ describe("tracker.dep_missing (MUST emit)", () => {
|
|||
expect(depMissingCalls).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("tracker.api_timeout", () => {
|
||||
it("rejects direct transport timeouts even when activity logging throws", async () => {
|
||||
delete process.env.COMPOSIO_API_KEY;
|
||||
delete process.env.COMPOSIO_ENTITY_ID;
|
||||
process.env.LINEAR_API_KEY = "linear-key";
|
||||
vi.useFakeTimers();
|
||||
|
||||
const req = {
|
||||
setTimeout: vi.fn(),
|
||||
on: vi.fn(),
|
||||
write: vi.fn(),
|
||||
end: vi.fn(),
|
||||
destroy: vi.fn(),
|
||||
};
|
||||
req.setTimeout.mockImplementation((_timeoutMs: number, cb: () => void) => {
|
||||
setTimeout(cb, 0);
|
||||
return req;
|
||||
});
|
||||
req.on.mockReturnValue(req);
|
||||
requestMock.mockReturnValue(req);
|
||||
recordActivityEventMock.mockImplementationOnce(() => {
|
||||
throw new Error("activity sink failed");
|
||||
});
|
||||
|
||||
const tracker = create();
|
||||
const timeoutExpectation = expect(tracker.getIssue("TEST-1", makeProject())).rejects.toThrow(
|
||||
"Linear API request timed out after 30s",
|
||||
);
|
||||
|
||||
await vi.runAllTimersAsync();
|
||||
await timeoutExpectation;
|
||||
expect(req.destroy).toHaveBeenCalled();
|
||||
expect(recordActivityEventMock).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
source: "tracker",
|
||||
kind: "tracker.api_timeout",
|
||||
level: "warn",
|
||||
data: expect.objectContaining({
|
||||
plugin: "tracker-linear",
|
||||
transport: "direct",
|
||||
timeoutMs: 30_000,
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -30,6 +30,14 @@ export function _resetDepMissingEmittedForTesting(): void {
|
|||
depMissingEmitted = false;
|
||||
}
|
||||
|
||||
function recordTransportActivityEvent(event: Parameters<typeof recordActivityEvent>[0]): void {
|
||||
try {
|
||||
recordActivityEvent(event);
|
||||
} catch {
|
||||
// Activity logging must never prevent timeout promises from settling.
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Transport abstraction
|
||||
// ---------------------------------------------------------------------------
|
||||
|
|
@ -121,7 +129,7 @@ function createDirectTransport(): GraphQLTransport {
|
|||
req.setTimeout(30_000, () => {
|
||||
settle(() => {
|
||||
req.destroy();
|
||||
recordActivityEvent({
|
||||
recordTransportActivityEvent({
|
||||
source: "tracker",
|
||||
kind: "tracker.api_timeout",
|
||||
level: "warn",
|
||||
|
|
@ -211,7 +219,7 @@ function createComposioTransport(apiKey: string, entityId: string): GraphQLTrans
|
|||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const timeoutPromise = new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(() => {
|
||||
recordActivityEvent({
|
||||
recordTransportActivityEvent({
|
||||
source: "tracker",
|
||||
kind: "tracker.api_timeout",
|
||||
level: "warn",
|
||||
|
|
|
|||
Loading…
Reference in New Issue