diff --git a/packages/core/src/activity-events.ts b/packages/core/src/activity-events.ts index 6065c3d41..0c1072e88 100644 --- a/packages/core/src/activity-events.ts +++ b/packages/core/src/activity-events.ts @@ -20,6 +20,9 @@ export type ActivityEventSource = | "scm" | "runtime" | "agent" + | "tracker" + | "workspace" + | "notifier" | "reaction" | "report-watcher"; @@ -41,6 +44,20 @@ export type ActivityEventKind = | "runtime.probe_failed" | "agent.process_probe_failed" | "agent.activity_probe_failed" + // Plugin-internal failure shapes (issue #1659) + | "scm.gh_unavailable" + | "scm.batch_enrich_pr_failed" + | "scm.ci_summary_failclosed" + | "workspace.post_create_failed" + | "workspace.branch_collision" + | "workspace.destroy_fell_back" + | "workspace.corrupt_clone_skipped" + | "tracker.dep_missing" + | "tracker.api_timeout" + | "notifier.auth_failed" + | "notifier.unreachable" + | "notifier.rate_limited" + | "notifier.dep_missing" // Reaction lifecycle | "reaction.escalated" | "reaction.send_to_agent_failed" diff --git a/packages/plugins/notifier-composio/src/index.ts b/packages/plugins/notifier-composio/src/index.ts index 08bb21c53..d3dd456ab 100644 --- a/packages/plugins/notifier-composio/src/index.ts +++ b/packages/plugins/notifier-composio/src/index.ts @@ -1,12 +1,21 @@ -import type { - PluginModule, - Notifier, - OrchestratorEvent, - NotifyAction, - NotifyContext, - EventPriority, +import { + recordActivityEvent, + type EventPriority, + type Notifier, + type NotifyAction, + type NotifyContext, + type OrchestratorEvent, + type PluginModule, } from "@aoagents/ao-core"; +// Module-level guard so we only emit notifier.dep_missing once per process. +let depMissingEmitted = false; + +/** Test-only: reset the once-per-process dep_missing guard. */ +export function _resetDepMissingEmittedForTesting(): void { + depMissingEmitted = false; +} + export const manifest = { name: "composio", slot: "notifier" as const, @@ -71,6 +80,22 @@ async function loadComposioSDK(apiKey: string): Promise message.includes("MODULE_NOT_FOUND") || code === "ERR_MODULE_NOT_FOUND" ) { + // User-actionable. Emit once per process so RCA can answer + // "why is the composio notifier silent?" without spamming on every notify call. + if (!depMissingEmitted) { + depMissingEmitted = true; + recordActivityEvent({ + source: "notifier", + kind: "notifier.dep_missing", + level: "error", + summary: "Composio SDK (composio-core) is not installed", + data: { + plugin: "notifier-composio", + package: "composio-core", + installHint: "pnpm add composio-core", + }, + }); + } return null; } throw err; diff --git a/packages/plugins/notifier-discord/src/index.ts b/packages/plugins/notifier-discord/src/index.ts index 7134fa2e2..00191239e 100644 --- a/packages/plugins/notifier-discord/src/index.ts +++ b/packages/plugins/notifier-discord/src/index.ts @@ -1,4 +1,5 @@ import { + recordActivityEvent, validateUrl, type PluginModule, type Notifier, @@ -129,6 +130,17 @@ async function postWithRetry( // Rate-limit budget exhausted — fail immediately rather than falling through // to the error retry path (which would compound the two counters). const body = await response.text().catch(() => ""); + recordActivityEvent({ + source: "notifier", + kind: "notifier.rate_limited", + level: "warn", + summary: `Discord webhook rate-limit retry budget exhausted`, + data: { + plugin: "notifier-discord", + status: 429, + rateLimitRetries, + }, + }); lastError = new Error(`Discord webhook rate-limited (HTTP 429)${body ? `: ${body.trim()}` : ""}`); throw lastError; } diff --git a/packages/plugins/notifier-openclaw/src/activity-events.test.ts b/packages/plugins/notifier-openclaw/src/activity-events.test.ts new file mode 100644 index 000000000..da89f5e2f --- /dev/null +++ b/packages/plugins/notifier-openclaw/src/activity-events.test.ts @@ -0,0 +1,174 @@ +/** + * Regression tests for plugin-internal activity events (issue #1659). + * + * Covers notifier.auth_failed (MUST) and notifier.unreachable (SHOULD) — + * the two failure shapes RCA needs to distinguish. + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { OrchestratorEvent } from "@aoagents/ao-core"; + +const { recordActivityEventMock } = vi.hoisted(() => ({ + recordActivityEventMock: vi.fn(), +})); + +vi.mock("@aoagents/ao-core", async () => { + const actual = (await vi.importActual("@aoagents/ao-core")) as Record; + return { + ...actual, + recordActivityEvent: recordActivityEventMock, + }; +}); + +import { create } from "./index.js"; + +function makeEvent(overrides: Partial = {}): OrchestratorEvent { + return { + id: "evt-1", + type: "reaction.escalated", + priority: "urgent", + sessionId: "ao-5", + projectId: "ao", + timestamp: new Date("2026-03-08T12:00:00Z"), + message: "Reaction escalated", + data: {}, + ...overrides, + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.restoreAllMocks(); + delete process.env.OPENCLAW_HOOKS_TOKEN; +}); + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("notifier.auth_failed (MUST emit)", () => { + it("emits on 401 (distinct from notifier.unreachable on ECONNREFUSED)", async () => { + const fetchMock = vi + .fn() + .mockResolvedValue({ ok: false, status: 401, text: () => Promise.resolve("unauthorized") }); + vi.stubGlobal("fetch", fetchMock); + + const notifier = create({ token: "tok", retries: 0 }); + await expect(notifier.notify(makeEvent())).rejects.toThrow(/OpenClaw rejected the auth token/); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "notifier", + kind: "notifier.auth_failed", + level: "error", + sessionId: "ao-5", + data: expect.objectContaining({ + plugin: "notifier-openclaw", + status: 401, + }), + }), + ); + }); + + it("emits on 403", async () => { + const fetchMock = vi + .fn() + .mockResolvedValue({ ok: false, status: 403, text: () => Promise.resolve("forbidden") }); + vi.stubGlobal("fetch", fetchMock); + + const notifier = create({ token: "tok", retries: 0 }); + await expect(notifier.notify(makeEvent())).rejects.toThrow(); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "notifier.auth_failed", + data: expect.objectContaining({ status: 403 }), + }), + ); + }); +}); + +describe("notifier.unreachable (SHOULD emit)", () => { + it.each(["ECONNREFUSED", "ETIMEDOUT", "ENOTFOUND", "ENETUNREACH"])( + "emits on %s (distinct from notifier.auth_failed)", + async (code) => { + const fetchMock = vi.fn().mockRejectedValue(new Error(`fetch failed: ${code}`)); + vi.stubGlobal("fetch", fetchMock); + + const notifier = create({ token: "tok", retries: 0 }); + await expect(notifier.notify(makeEvent())).rejects.toThrow(/Can't reach OpenClaw gateway/); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "notifier", + kind: "notifier.unreachable", + level: "warn", + sessionId: "ao-5", + data: expect.objectContaining({ + plugin: "notifier-openclaw", + errorMessage: expect.stringContaining(code), + }), + }), + ); + + // Critically: should NOT also fire auth_failed — distinct shapes. + const authFailedCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "notifier.auth_failed", + ); + expect(authFailedCalls).toHaveLength(0); + }, + ); + + it.each(["ETIMEDOUT", "ENOTFOUND", "ENETUNREACH"])( + "does not emit on transient %s when a retry succeeds", + async (code) => { + const fetchMock = vi + .fn() + .mockRejectedValueOnce(new Error(`fetch failed: ${code}`)) + .mockResolvedValueOnce({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + + const notifier = create({ token: "tok", retries: 1, retryDelayMs: 0 }); + await notifier.notify(makeEvent()); + + expect(fetchMock).toHaveBeenCalledTimes(2); + const unreachableCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "notifier.unreachable", + ); + expect(unreachableCalls).toHaveLength(0); + }, + ); + + it.each(["ETIMEDOUT", "ENOTFOUND", "ENETUNREACH"])( + "emits on transient %s only after retry budget is exhausted", + async (code) => { + const fetchMock = vi.fn().mockRejectedValue(new Error(`fetch failed: ${code}`)); + vi.stubGlobal("fetch", fetchMock); + + const notifier = create({ token: "tok", retries: 1, retryDelayMs: 0 }); + await expect(notifier.notify(makeEvent())).rejects.toThrow(/Can't reach OpenClaw gateway/); + + expect(fetchMock).toHaveBeenCalledTimes(2); + const unreachableCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "notifier.unreachable", + ); + expect(unreachableCalls).toHaveLength(1); + expect(unreachableCalls[0]?.[0].data).toMatchObject({ + errorMessage: expect.stringContaining(code), + }); + }, + ); + + it("does not emit unreachable for unrelated network errors", async () => { + const fetchMock = vi.fn().mockRejectedValue(new Error("fetch failed: certificate expired")); + vi.stubGlobal("fetch", fetchMock); + + const notifier = create({ token: "tok", retries: 0 }); + await expect(notifier.notify(makeEvent())).rejects.toThrow(/fetch failed: certificate expired/); + + const unreachableCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "notifier.unreachable", + ); + expect(unreachableCalls).toHaveLength(0); + }); +}); diff --git a/packages/plugins/notifier-openclaw/src/index.ts b/packages/plugins/notifier-openclaw/src/index.ts index fcd5ae6cf..ced31e7cf 100644 --- a/packages/plugins/notifier-openclaw/src/index.ts +++ b/packages/plugins/notifier-openclaw/src/index.ts @@ -9,6 +9,7 @@ import { type OrchestratorEvent, type PluginModule, getObservabilityBaseDir, + recordActivityEvent, } from "@aoagents/ao-core"; import { isRetryableHttpStatus, normalizeRetryConfig, validateUrl } from "@aoagents/ao-core/utils"; @@ -39,6 +40,13 @@ export const manifest = { }; const DEFAULT_TIMEOUT_MS = 10_000; +const UNREACHABLE_NETWORK_ERROR_CODES = [ + "ECONNREFUSED", + "ETIMEDOUT", + "ENOTFOUND", + "ENETUNREACH", +] as const; +type UnreachableNetworkErrorCode = (typeof UNREACHABLE_NETWORK_ERROR_CODES)[number]; type WakeMode = "now" | "next-heartbeat"; @@ -117,6 +125,10 @@ function recordHealthFailure(path: string | null, error: unknown): void { writeHealthSummary(path, summary); } +function getUnreachableNetworkErrorCode(error: Error): UnreachableNetworkErrorCode | undefined { + return UNREACHABLE_NETWORK_ERROR_CODES.find((code) => error.message.includes(code)); +} + async function postWithRetry( url: string, payload: OpenClawWebhookPayload, @@ -130,6 +142,7 @@ async function postWithRetry( for (let attempt = 0; attempt <= retries; attempt++) { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), DEFAULT_TIMEOUT_MS); + let shouldRethrowResponseError = false; try { const response = await fetch(url, { method: "POST", @@ -143,17 +156,33 @@ async function postWithRetry( const body = await response.text(); if (response.status === 401 || response.status === 403) { + // User-actionable: distinct from generic 5xx — token expired or wrong. + recordActivityEvent({ + sessionId: context.sessionId, + source: "notifier", + kind: "notifier.auth_failed", + level: "error", + summary: `OpenClaw rejected auth token (HTTP ${response.status})`, + data: { + plugin: "notifier-openclaw", + status: response.status, + url, + fixHint: "ao setup openclaw", + }, + }); lastError = new Error( `OpenClaw rejected the auth token (HTTP ${response.status}).\n` + ` Check that hooks.token in your OpenClaw config matches the token configured for AO.\n` + ` Reconfigure: ao setup openclaw`, ); + shouldRethrowResponseError = true; throw lastError; } lastError = new Error(`OpenClaw webhook failed (${response.status}): ${body}`); if (!isRetryableHttpStatus(response.status)) { + shouldRethrowResponseError = true; throw lastError; } @@ -163,10 +192,24 @@ async function postWithRetry( ); } } catch (err) { - if (err === lastError) throw err; + if (shouldRethrowResponseError && err === lastError) throw err; lastError = err instanceof Error ? err : new Error(String(err)); - if (lastError.message.includes("ECONNREFUSED")) { + const unreachableCode = getUnreachableNetworkErrorCode(lastError); + if (unreachableCode && (unreachableCode === "ECONNREFUSED" || attempt >= retries)) { + recordActivityEvent({ + sessionId: context.sessionId, + source: "notifier", + kind: "notifier.unreachable", + level: "warn", + summary: `OpenClaw gateway unreachable at ${url}`, + data: { + plugin: "notifier-openclaw", + url, + errorMessage: lastError.message, + fixHint: "openclaw status", + }, + }); throw new Error( `Can't reach OpenClaw gateway at ${url}.\n` + ` Is OpenClaw running? Check: openclaw status\n` + diff --git a/packages/plugins/scm-github/src/graphql-batch.ts b/packages/plugins/scm-github/src/graphql-batch.ts index 108d47168..cf92add29 100644 --- a/packages/plugins/scm-github/src/graphql-batch.ts +++ b/packages/plugins/scm-github/src/graphql-batch.ts @@ -9,6 +9,7 @@ import { execFile } from "node:child_process"; import { promisify } from "node:util"; import { execGhObserved, + recordActivityEvent, type BatchObserver, type CICheck, type CIStatus, @@ -56,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. @@ -125,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}`); } @@ -145,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); } @@ -161,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( + MAX_PR_METADATA, +); /** * Cache for full PR enrichment data. @@ -241,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 @@ -297,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)`); } } } @@ -310,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 { return prMetadataCache.toMap(); } @@ -350,6 +340,11 @@ interface ErrorWithCause extends Error { cause?: unknown; } +// 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(); + /** * Pre-flight check to verify gh CLI is available and authenticated. * This prevents silent failures during GraphQL batch queries. @@ -357,7 +352,21 @@ interface ErrorWithCause extends Error { async function verifyGhCLI(): Promise { try { await execFileAsync("gh", ["--version"], { timeout: 5000 }); - } catch { + } catch (err) { + if (!ghUnavailableEmitted) { + ghUnavailableEmitted = true; + const errorMessage = err instanceof Error ? err.message : String(err); + recordActivityEvent({ + source: "scm", + kind: "scm.gh_unavailable", + level: "error", + summary: "gh CLI not available or not authenticated", + data: { + plugin: "scm-github", + errorMessage, + }, + }); + } const error = new Error( "gh CLI not available or not authenticated. GraphQL batch enrichment requires gh CLI to be installed and configured.", ) as ErrorWithCause; @@ -366,6 +375,16 @@ async function verifyGhCLI(): Promise { } } +/** Test-only: reset the once-per-process gh_unavailable guard. */ +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. @@ -533,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 } } @@ -596,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 } } @@ -704,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> { +async function executeBatchQuery(prs: PRInfo[]): Promise> { const { query, variables } = generateBatchQuery(prs); // Handle empty array - no query needed @@ -866,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"; } @@ -885,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"; } @@ -927,11 +947,7 @@ function extractPREnrichment( const pr = pullRequest as Record; // 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; } @@ -954,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"; @@ -974,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 - | undefined; + const contextsField = statusCheckRollup?.["contexts"] as Record | undefined; const pageInfo = contextsField?.["pageInfo"]; const contextsHasNextPage = pageInfo !== null && @@ -984,15 +996,12 @@ function extractPREnrichment( typeof pageInfo === "object" && (pageInfo as Record)["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"); @@ -1126,6 +1135,25 @@ export async function enrichSessionsPRBatch( result.set(prKey, enrichment); // Update PR metadata cache for future ETag checks updatePRMetadataCache(prKey, enrichment, headSha); + } else { + // 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. + 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) @@ -1147,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 diff --git a/packages/plugins/scm-github/src/index.ts b/packages/plugins/scm-github/src/index.ts index 8d065db01..fa59cecdc 100644 --- a/packages/plugins/scm-github/src/index.ts +++ b/packages/plugins/scm-github/src/index.ts @@ -11,6 +11,7 @@ import { CI_STATUS, execGhObserved, memoizeAsync, + recordActivityEvent, type PluginModule, type PreflightContext, type SCM, @@ -62,6 +63,12 @@ const BOT_AUTHORS = new Set([ ]); const CI_FAILURE_LOG_TAIL_LINES = 120; +const ciSummaryFailClosedEmitted = new Set(); + +/** Test-only: reset once-per-PR activity event guards. */ +export function _resetGitHubActivityEventDedupeForTesting(): void { + ciSummaryFailClosedEmitted.clear(); +} // --------------------------------------------------------------------------- // Helpers @@ -237,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`]); } } @@ -522,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); @@ -932,7 +940,7 @@ function createGitHubSCM(): SCM { let checks: CICheck[]; try { checks = await this.getCIChecks(pr); - } catch { + } catch (err) { // Before fail-closing, check if the PR is merged/closed — // GitHub may not return check data for those, and reporting // "failing" for a merged PR is wrong. @@ -943,7 +951,26 @@ function createGitHubSCM(): SCM { // Can't determine state either; fall through to fail-closed. } // Fail closed for open PRs: report as failing rather than - // "none" (which getMergeability treats as passing). + // "none" (which getMergeability treats as passing). Emit so RCA + // can distinguish "really failing" from "we couldn't tell". + 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"; @@ -1035,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) { @@ -1067,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 }); } @@ -1129,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; diff --git a/packages/plugins/scm-github/test/activity-events.test.ts b/packages/plugins/scm-github/test/activity-events.test.ts new file mode 100644 index 000000000..ad54976df --- /dev/null +++ b/packages/plugins/scm-github/test/activity-events.test.ts @@ -0,0 +1,152 @@ +/** + * Regression tests for plugin-internal activity events (issue #1659). + * + * Covers scm.gh_unavailable (MUST emit, deduped once-per-process). + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { recordActivityEventMock } = vi.hoisted(() => ({ + recordActivityEventMock: vi.fn(), +})); + +vi.mock("@aoagents/ao-core", async () => { + const actual = (await vi.importActual("@aoagents/ao-core")) as Record; + return { + ...actual, + recordActivityEvent: recordActivityEventMock, + }; +}); + +import { + enrichSessionsPRBatch, + setExecFileAsync, + setExecGhAsync, + clearETagCache, + clearPRMetadataCache, + _resetGhUnavailableEmittedForTesting, + _resetBatchEnrichPRFailedEmittedForTesting, +} from "../src/graphql-batch.js"; +import type { PRInfo } from "@aoagents/ao-core"; + +const samplePRs: PRInfo[] = [ + { + owner: "octocat", + repo: "hello-world", + number: 42, + url: "https://github.com/octocat/hello-world/pull/42", + title: "Add new feature", + branch: "feature/new", + baseBranch: "main", + isDraft: false, + }, +]; + +beforeEach(() => { + vi.clearAllMocks(); + clearETagCache(); + clearPRMetadataCache(); + _resetGhUnavailableEmittedForTesting(); + _resetBatchEnrichPRFailedEmittedForTesting(); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("scm.gh_unavailable (MUST emit)", () => { + it("emits when verifyGhCLI fails because gh is missing/unauthenticated", async () => { + const execFileMock = vi.fn().mockImplementation((file: string) => { + if (file === "gh") { + const err = new Error("spawn gh ENOENT") as Error & { code?: string }; + err.code = "ENOENT"; + return Promise.reject(err); + } + return Promise.resolve({ stdout: "", stderr: "" }); + }); + setExecFileAsync(execFileMock as unknown as Parameters[0]); + + // The batch-level try/catch swallows verifyGhCLI's throw — but the event + // fires before the throw, which is what we care about for RCA. + const result = await enrichSessionsPRBatch(samplePRs); + expect(result.enrichment.size).toBe(0); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "scm", + kind: "scm.gh_unavailable", + level: "error", + data: expect.objectContaining({ + plugin: "scm-github", + errorMessage: expect.any(String), + }), + }), + ); + }); + + it("emits exactly once across multiple gh-missing failures (deduped per-process)", async () => { + const execFileMock = vi.fn().mockImplementation((file: string) => { + if (file === "gh") { + return Promise.reject(new Error("gh ENOENT")); + } + return Promise.resolve({ stdout: "", stderr: "" }); + }); + setExecFileAsync(execFileMock as unknown as Parameters[0]); + + await enrichSessionsPRBatch(samplePRs); + await enrichSessionsPRBatch(samplePRs); + await enrichSessionsPRBatch(samplePRs); + + const ghUnavailableCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "scm.gh_unavailable", + ); + 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[0]); + + const execGhMock = vi.fn( + async (_args: string[], _timeout: number, operation: string): Promise => { + 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", + }), + }), + ); + }); +}); diff --git a/packages/plugins/scm-github/test/index.test.ts b/packages/plugins/scm-github/test/index.test.ts index 501ff4e09..7fee69578 100644 --- a/packages/plugins/scm-github/test/index.test.ts +++ b/packages/plugins/scm-github/test/index.test.ts @@ -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; + 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" }, diff --git a/packages/plugins/scm-gitlab/src/index.ts b/packages/plugins/scm-gitlab/src/index.ts index df9511244..31402653a 100644 --- a/packages/plugins/scm-gitlab/src/index.ts +++ b/packages/plugins/scm-gitlab/src/index.ts @@ -7,6 +7,7 @@ import { createHash, timingSafeEqual } from "node:crypto"; import { CI_STATUS, + recordActivityEvent, type PluginModule, type SCM, type SCMWebhookEvent, @@ -45,6 +46,14 @@ const BOT_AUTHORS = new Set([ "sonarcloud[bot]", "snyk-bot", ]); +const ciSummaryFailClosedEmitted = new Set(); +const reviewFetchFailedEmitted = new Set(); + +/** Test-only: reset once-per-PR activity event guards. */ +export function _resetGitLabActivityEventDedupeForTesting(): void { + ciSummaryFailClosedEmitted.clear(); + reviewFetchFailedEmitted.clear(); +} function isBot(username: string): boolean { return ( @@ -60,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); @@ -106,7 +119,6 @@ function mapPRState(state: string): PRState { return "open"; } - function getGitLabWebhookConfig(project: ProjectConfig) { const webhook = project.scm?.webhook; return { @@ -585,6 +597,24 @@ function createGitLabSCM(config?: Record): SCM { `getCISummary: PR state fallback also failed for MR !${pr.number}: ${(innerErr as Error).message}`, ); } + 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"; @@ -642,6 +672,24 @@ function createGitLabSCM(config?: Record): SCM { console.warn( `getReviews: discussions fetch failed for MR !${pr.number}: ${(err as Error).message}`, ); + 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; diff --git a/packages/plugins/scm-gitlab/test/index.test.ts b/packages/plugins/scm-gitlab/test/index.test.ts index 9dad4d0eb..c72d93ee0 100644 --- a/packages/plugins/scm-gitlab/test/index.test.ts +++ b/packages/plugins/scm-gitlab/test/index.test.ts @@ -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; + 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([ diff --git a/packages/plugins/tracker-linear/src/activity-events.test.ts b/packages/plugins/tracker-linear/src/activity-events.test.ts new file mode 100644 index 000000000..06d552ee4 --- /dev/null +++ b/packages/plugins/tracker-linear/src/activity-events.test.ts @@ -0,0 +1,143 @@ +/** + * Regression tests for plugin-internal activity events (issue #1659). + * + * Covers tracker.dep_missing (MUST emit, deduped once-per-process). + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { recordActivityEventMock, requestMock } = vi.hoisted(() => ({ + recordActivityEventMock: vi.fn(), + requestMock: vi.fn(), +})); + +vi.mock("node:https", () => ({ + request: requestMock, +})); + +vi.mock("@aoagents/ao-core", async () => { + const actual = (await vi.importActual("@aoagents/ao-core")) as Record; + return { + ...actual, + recordActivityEvent: recordActivityEventMock, + }; +}); + +// @composio/core is intentionally not installed — the real dynamic import +// will fail with ERR_MODULE_NOT_FOUND, which is exactly the dep_missing +// shape we want to exercise. + +import { create, _resetDepMissingEmittedForTesting } from "./index.js"; +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"; + delete process.env.LINEAR_API_KEY; +}); + +afterEach(() => { + delete process.env.COMPOSIO_API_KEY; + delete process.env.COMPOSIO_ENTITY_ID; + delete process.env.LINEAR_API_KEY; + vi.useRealTimers(); +}); + +function makeProject(): ProjectConfig { + return { + name: "test-project", + repo: "test/repo", + path: "/repo/path", + defaultBranch: "main", + sessionPrefix: "test", + tracker: { teamId: "TEAM-1" }, + }; +} + +describe("tracker.dep_missing (MUST emit)", () => { + it("emits when Composio SDK is not installed", async () => { + const tracker = create(); + + // Any tracker call routes through the composio transport, which will + // fail to load the missing SDK on first use. + await expect(tracker.getIssue("TEST-1", makeProject())).rejects.toThrow( + /Composio SDK.*not installed/, + ); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "tracker", + kind: "tracker.dep_missing", + level: "error", + data: expect.objectContaining({ + plugin: "tracker-linear", + package: "@composio/core", + }), + }), + ); + }); + + it("emits exactly once across multiple calls (deduped per-process)", async () => { + const tracker = create(); + + await expect(tracker.getIssue("TEST-1", makeProject())).rejects.toThrow(); + await expect(tracker.getIssue("TEST-2", makeProject())).rejects.toThrow(); + await expect(tracker.getIssue("TEST-3", makeProject())).rejects.toThrow(); + + const depMissingCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "tracker.dep_missing", + ); + 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, + }), + }), + ); + }); +}); diff --git a/packages/plugins/tracker-linear/src/index.ts b/packages/plugins/tracker-linear/src/index.ts index fd93e3eb8..6101ac1c3 100644 --- a/packages/plugins/tracker-linear/src/index.ts +++ b/packages/plugins/tracker-linear/src/index.ts @@ -9,17 +9,35 @@ */ import { request } from "node:https"; -import type { - PluginModule, - Tracker, - Issue, - IssueFilters, - IssueUpdate, - CreateIssueInput, - ProjectConfig, +import { + recordActivityEvent, + type CreateIssueInput, + type Issue, + type IssueFilters, + type IssueUpdate, + type PluginModule, + type ProjectConfig, + type Tracker, } from "@aoagents/ao-core"; import type { Composio } from "@composio/core"; +// Module-level guard so we only emit tracker.dep_missing once per process +// even if multiple sessions trigger the missing-SDK path. +let depMissingEmitted = false; + +/** Test-only: reset the once-per-process dep_missing guard. */ +export function _resetDepMissingEmittedForTesting(): void { + depMissingEmitted = false; +} + +function recordTransportActivityEvent(event: Parameters[0]): void { + try { + recordActivityEvent(event); + } catch { + // Activity logging must never prevent timeout promises from settling. + } +} + // --------------------------------------------------------------------------- // Transport abstraction // --------------------------------------------------------------------------- @@ -111,6 +129,17 @@ function createDirectTransport(): GraphQLTransport { req.setTimeout(30_000, () => { settle(() => { req.destroy(); + recordTransportActivityEvent({ + source: "tracker", + kind: "tracker.api_timeout", + level: "warn", + summary: "Linear API request timed out after 30s", + data: { + plugin: "tracker-linear", + transport: "direct", + timeoutMs: 30_000, + }, + }); reject(new Error("Linear API request timed out after 30s")); }); }); @@ -147,6 +176,21 @@ function createComposioTransport(apiKey: string, entityId: string): GraphQLTrans msg.includes("Cannot find package") || msg.includes("ERR_MODULE_NOT_FOUND") ) { + // User-actionable, system-wide. Emit once per process. + if (!depMissingEmitted) { + depMissingEmitted = true; + recordActivityEvent({ + source: "tracker", + kind: "tracker.dep_missing", + level: "error", + summary: "Composio SDK (@composio/core) is not installed", + data: { + plugin: "tracker-linear", + package: "@composio/core", + installHint: "pnpm add @composio/core", + }, + }); + } throw new Error( "Composio SDK (@composio/core) is not installed. " + "Install it with: pnpm add @composio/core", @@ -175,6 +219,17 @@ function createComposioTransport(apiKey: string, entityId: string): GraphQLTrans let timer: ReturnType | undefined; const timeoutPromise = new Promise((_resolve, reject) => { timer = setTimeout(() => { + recordTransportActivityEvent({ + source: "tracker", + kind: "tracker.api_timeout", + level: "warn", + summary: "Composio Linear API request timed out after 30s", + data: { + plugin: "tracker-linear", + transport: "composio", + timeoutMs: 30_000, + }, + }); reject(new Error("Composio Linear API request timed out after 30s")); }, 30_000); }); diff --git a/packages/plugins/workspace-clone/src/__tests__/index.test.ts b/packages/plugins/workspace-clone/src/__tests__/index.test.ts index 2b194ccd3..0c2b1da45 100644 --- a/packages/plugins/workspace-clone/src/__tests__/index.test.ts +++ b/packages/plugins/workspace-clone/src/__tests__/index.test.ts @@ -3,6 +3,20 @@ import * as childProcess from "node:child_process"; import * as fs from "node:fs"; import type { ProjectConfig } from "@aoagents/ao-core"; +const { getShellMock, recordActivityEventMock } = vi.hoisted(() => ({ + getShellMock: vi.fn(() => ({ cmd: "sh", args: (c: string) => ["-c", c] })), + recordActivityEventMock: vi.fn(), +})); + +vi.mock("@aoagents/ao-core", async () => { + const actual = (await vi.importActual("@aoagents/ao-core")) as Record; + return { + ...actual, + getShell: getShellMock, + recordActivityEvent: recordActivityEventMock, + }; +}); + // Mock node:child_process with custom promisify support vi.mock("node:child_process", () => { const mockExecFile = vi.fn(); @@ -10,10 +24,6 @@ vi.mock("node:child_process", () => { return { execFile: mockExecFile }; }); -vi.mock("@aoagents/ao-core", () => ({ - getShell: vi.fn(() => ({ cmd: "sh", args: (c: string) => ["-c", c] })), -})); - // Mock node:fs vi.mock("node:fs", () => ({ existsSync: vi.fn(), @@ -267,9 +277,48 @@ describe("workspace.create()", () => { cwd: "/mock-home/.ao-clones/proj/sess", }); + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "workspace", + kind: "workspace.branch_collision", + level: "warn", + projectId: "proj", + sessionId: "sess", + data: expect.objectContaining({ + plugin: "workspace-clone", + branch: "feat/existing", + errorMessage: expect.stringContaining("already exists"), + }), + }), + ); expect(info.branch).toBe("feat/existing"); }); + it("does not emit branch_collision when checkout -b fails for a non-collision reason", async () => { + const workspace = create(); + + mockGitSuccess("https://github.com/test/repo.git"); + (fs.existsSync as ReturnType).mockReturnValue(false); + mockGitSuccess(""); + // git checkout -b fails for a non-collision reason + mockGitError("fatal: cannot lock ref 'refs/heads/feat/locked': Permission denied"); + // git checkout (plain) succeeds + mockGitSuccess(""); + + const info = await workspace.create({ + projectId: "proj", + sessionId: "sess", + branch: "feat/locked", + project: makeProject(), + }); + + expect(info.branch).toBe("feat/locked"); + const branchCollisionCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "workspace.branch_collision", + ); + expect(branchCollisionCalls).toHaveLength(0); + }); + it("cleans up partial clone on clone failure", async () => { const workspace = create(); @@ -588,6 +637,41 @@ describe("workspace.list()", () => { warnSpy.mockRestore(); }); + it("emits corrupt_clone_skipped only once per clone path", async () => { + const workspace = create(); + (fs.existsSync as ReturnType).mockReturnValue(true); + + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + + (fs.readdirSync as ReturnType).mockReturnValue([ + { name: "corrupt-repeat", isDirectory: () => true }, + ]); + + mockGitError("fatal: not a git repository"); + mockGitError("fatal: not a git repository"); + + await expect(workspace.list("myproject")).resolves.toEqual([]); + await expect(workspace.list("myproject")).resolves.toEqual([]); + + const corruptCloneCalls = recordActivityEventMock.mock.calls.filter( + ([event]) => event.kind === "workspace.corrupt_clone_skipped", + ); + expect(corruptCloneCalls).toHaveLength(1); + expect(corruptCloneCalls[0][0]).toEqual( + expect.objectContaining({ + projectId: "myproject", + sessionId: "corrupt-repeat", + source: "workspace", + kind: "workspace.corrupt_clone_skipped", + data: expect.objectContaining({ + clonePath: "/mock-home/.ao-clones/myproject/corrupt-repeat", + }), + }), + ); + + warnSpy.mockRestore(); + }); + it("rejects invalid projectId with special characters", async () => { const workspace = create(); diff --git a/packages/plugins/workspace-clone/src/index.ts b/packages/plugins/workspace-clone/src/index.ts index 075fc3119..71100644a 100644 --- a/packages/plugins/workspace-clone/src/index.ts +++ b/packages/plugins/workspace-clone/src/index.ts @@ -3,7 +3,15 @@ import { promisify } from "node:util"; import { existsSync, rmSync, mkdirSync, readdirSync } from "node:fs"; import { join } from "node:path"; import { homedir } from "node:os"; -import { getShell, type PluginModule, type Workspace, type WorkspaceCreateConfig, type WorkspaceInfo, type ProjectConfig } from "@aoagents/ao-core"; +import { + getShell, + recordActivityEvent, + type PluginModule, + type ProjectConfig, + type Workspace, + type WorkspaceCreateConfig, + type WorkspaceInfo, +} from "@aoagents/ao-core"; const execFileAsync = promisify(execFile); @@ -37,6 +45,16 @@ function expandPath(p: string): string { return p; } +function getErrorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +function isBranchAlreadyExistsError(err: unknown): boolean { + return getErrorMessage(err).toLowerCase().includes("already exists"); +} + +const emittedCorruptClonePaths = new Set(); + export function create(config?: Record): Workspace { const cloneBaseDir = config?.cloneDir ? expandPath(config.cloneDir as string) @@ -97,8 +115,24 @@ export function create(config?: Record): Workspace { // Create and checkout the feature branch try { await git(clonePath, "checkout", "-b", cfg.branch); - } catch { - // Branch may exist on remote — try plain checkout + } catch (branchErr: unknown) { + // Branch may exist on remote — try plain checkout, but only label it + // as a branch_collision when git reported the distinct collision shape. + if (isBranchAlreadyExistsError(branchErr)) { + recordActivityEvent({ + projectId: cfg.projectId, + sessionId: cfg.sessionId, + source: "workspace", + kind: "workspace.branch_collision", + level: "warn", + summary: `branch "${cfg.branch}" already exists; falling back to checkout`, + data: { + plugin: "workspace-clone", + branch: cfg.branch, + errorMessage: getErrorMessage(branchErr), + }, + }); + } try { await git(clonePath, "checkout", cfg.branch); } catch (checkoutErr: unknown) { @@ -142,10 +176,27 @@ export function create(config?: Record): Workspace { try { branch = await git(clonePath, "branch", "--show-current"); } catch (err: unknown) { - // Warn about corrupted clones instead of silently skipping + // Warn about corrupted clones instead of silently skipping. + // RCA: "session shows up on disk but isn't returned by list()". const msg = err instanceof Error ? err.message : String(err); // eslint-disable-next-line no-console -- expected diagnostic for corrupted clones console.warn(`[workspace-clone] Skipping "${entry.name}": not a valid git repo (${msg})`); + if (!emittedCorruptClonePaths.has(clonePath)) { + emittedCorruptClonePaths.add(clonePath); + recordActivityEvent({ + projectId, + sessionId: entry.name, + source: "workspace", + kind: "workspace.corrupt_clone_skipped", + level: "warn", + summary: `skipped corrupt clone "${entry.name}"`, + data: { + plugin: "workspace-clone", + clonePath, + errorMessage: msg, + }, + }); + } continue; } diff --git a/packages/plugins/workspace-worktree/src/__tests__/activity-events.test.ts b/packages/plugins/workspace-worktree/src/__tests__/activity-events.test.ts new file mode 100644 index 000000000..3a2f6039c --- /dev/null +++ b/packages/plugins/workspace-worktree/src/__tests__/activity-events.test.ts @@ -0,0 +1,192 @@ +/** + * Regression tests for plugin-internal activity events (issue #1659). + * + * Covers the MUST emit: workspace.post_create_failed, plus the SHOULDs + * workspace.branch_collision and workspace.destroy_fell_back. + */ + +import { describe, it, expect, beforeEach, vi } from "vitest"; + +// --------------------------------------------------------------------------- +// Mocks — declared before any import that uses the mocked modules +// --------------------------------------------------------------------------- + +const { recordActivityEventMock } = vi.hoisted(() => ({ + recordActivityEventMock: vi.fn(), +})); + +vi.mock("@aoagents/ao-core", async () => { + const actual = (await vi.importActual("@aoagents/ao-core")) as Record; + return { + ...actual, + recordActivityEvent: recordActivityEventMock, + }; +}); + +vi.mock("node:child_process", () => { + const mockExecFile = vi.fn(); + (mockExecFile as unknown as Record)[Symbol.for("nodejs.util.promisify.custom")] = + vi.fn(); + return { execFile: mockExecFile }; +}); + +vi.mock("node:fs", () => ({ + cpSync: vi.fn(), + existsSync: vi.fn(() => false), + linkSync: vi.fn(), + lstatSync: vi.fn(), + statSync: vi.fn(), + symlinkSync: vi.fn(), + rmSync: vi.fn(), + mkdirSync: vi.fn(), + readdirSync: vi.fn(), +})); + +vi.mock("node:os", () => ({ homedir: () => "/mock-home" })); + +import * as childProcess from "node:child_process"; +import { create } from "../index.js"; +import type { ProjectConfig, WorkspaceCreateConfig, WorkspaceInfo } from "@aoagents/ao-core/types"; + +const mockExecFileAsync = (childProcess.execFile as unknown as Record)[ + Symbol.for("nodejs.util.promisify.custom") +] as ReturnType; + +function makeProject(overrides?: Partial): ProjectConfig { + return { + name: "test-project", + repo: "test/repo", + path: "/repo/path", + defaultBranch: "main", + sessionPrefix: "test", + ...overrides, + }; +} + +function makeCreateConfig(overrides?: Partial): WorkspaceCreateConfig { + return { + projectId: "myproject", + project: makeProject(), + sessionId: "session-1", + branch: "feat/TEST-1", + ...overrides, + }; +} + +function makeWorkspaceInfo(): WorkspaceInfo { + return { + path: "/mock-home/.worktrees/myproject/session-1", + branch: "feat/TEST-1", + sessionId: "session-1", + projectId: "myproject", + }; +} + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe("workspace.post_create_failed (MUST emit)", () => { + it("emits when a postCreate command fails, then rethrows", async () => { + const ws = create(); + const project = makeProject({ postCreate: ["pnpm install"] }); + + mockExecFileAsync.mockRejectedValueOnce(new Error("Command failed: exit 127")); + + await expect(ws.postCreate!(makeWorkspaceInfo(), project)).rejects.toThrow( + "Command failed: exit 127", + ); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "workspace", + kind: "workspace.post_create_failed", + level: "error", + sessionId: "session-1", + projectId: "myproject", + data: expect.objectContaining({ + plugin: "workspace-worktree", + command: "pnpm install", + errorMessage: expect.stringContaining("exit 127"), + }), + }), + ); + }); + + it("does NOT emit when postCreate command succeeds", async () => { + const ws = create(); + const project = makeProject({ postCreate: ["echo hi"] }); + + mockExecFileAsync.mockResolvedValueOnce({ stdout: "hi\n", stderr: "" }); + + await ws.postCreate!(makeWorkspaceInfo(), project); + + expect(recordActivityEventMock).not.toHaveBeenCalled(); + }); +}); + +describe("workspace.branch_collision (SHOULD emit)", () => { + it("emits when worktree add -b fails because branch already exists", async () => { + const ws = create(); + + // git remote get-url origin + mockExecFileAsync.mockResolvedValueOnce({ stdout: "origin\n", stderr: "" }); + // git fetch origin --quiet + mockExecFileAsync.mockResolvedValueOnce({ stdout: "", stderr: "" }); + // resolveBaseRef -> rev-parse origin/main + mockExecFileAsync.mockResolvedValueOnce({ stdout: "abc\n", stderr: "" }); + // git worktree add -b ... → already exists + mockExecFileAsync.mockRejectedValueOnce( + new Error("fatal: a branch named 'feat/TEST-1' already exists"), + ); + // rev-parse baseRef for stale-branch comparison + mockExecFileAsync.mockResolvedValueOnce({ stdout: "abc\n", stderr: "" }); + // refExists(branchRef) -> true + mockExecFileAsync.mockResolvedValueOnce({ stdout: "refs/heads/feat/TEST-1\n", stderr: "" }); + // rev-parse existing branch -> same as base, so reuse it + mockExecFileAsync.mockResolvedValueOnce({ stdout: "abc\n", stderr: "" }); + // git worktree add (without -b) — succeeds + mockExecFileAsync.mockResolvedValueOnce({ stdout: "", stderr: "" }); + + await ws.create(makeCreateConfig()); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "workspace", + kind: "workspace.branch_collision", + level: "warn", + projectId: "myproject", + sessionId: "session-1", + data: expect.objectContaining({ + plugin: "workspace-worktree", + branch: "feat/TEST-1", + errorMessage: expect.stringContaining("already exists"), + }), + }), + ); + }); +}); + +describe("workspace.destroy_fell_back (SHOULD emit)", () => { + it("emits when destroy() falls back to rmSync after git failure", async () => { + const ws = create(); + + // git rev-parse --git-common-dir → fails + mockExecFileAsync.mockRejectedValueOnce(new Error("not a git repository")); + + await ws.destroy("/some/stale/path"); + + expect(recordActivityEventMock).toHaveBeenCalledWith( + expect.objectContaining({ + source: "workspace", + kind: "workspace.destroy_fell_back", + level: "warn", + data: expect.objectContaining({ + plugin: "workspace-worktree", + workspacePath: "/some/stale/path", + errorMessage: expect.stringContaining("not a git repository"), + }), + }), + ); + }); +}); diff --git a/packages/plugins/workspace-worktree/src/__tests__/index.test.ts b/packages/plugins/workspace-worktree/src/__tests__/index.test.ts index f4837fd62..c39afd6d9 100644 --- a/packages/plugins/workspace-worktree/src/__tests__/index.test.ts +++ b/packages/plugins/workspace-worktree/src/__tests__/index.test.ts @@ -5,6 +5,10 @@ import type { ProjectConfig, WorkspaceCreateConfig, WorkspaceInfo } from "@aoage // Mocks — must be declared before any import that uses the mocked modules // --------------------------------------------------------------------------- +const { recordActivityEventMock } = vi.hoisted(() => ({ + recordActivityEventMock: vi.fn(), +})); + vi.mock("node:child_process", () => { const mockExecFile = vi.fn(); // Set custom promisify so `promisify(execFile)` returns { stdout, stderr } @@ -27,6 +31,7 @@ vi.mock("node:fs", () => ({ vi.mock("@aoagents/ao-core", () => ({ getShell: vi.fn(() => ({ cmd: "sh", args: (c: string) => ["-c", c] })), isWindows: vi.fn(() => false), + recordActivityEvent: recordActivityEventMock, })); vi.mock("node:os", () => ({ diff --git a/packages/plugins/workspace-worktree/src/index.ts b/packages/plugins/workspace-worktree/src/index.ts index 7e0b6f61e..1f5c32840 100644 --- a/packages/plugins/workspace-worktree/src/index.ts +++ b/packages/plugins/workspace-worktree/src/index.ts @@ -16,6 +16,7 @@ import { homedir } from "node:os"; import { getShell, isWindows, + recordActivityEvent, type PluginModule, type Workspace, type WorkspaceCreateConfig, @@ -361,7 +362,21 @@ export function create(config?: Record): Workspace { // Branch already exists. It may be a stale session branch left behind // from an earlier spawn, so compare it with the freshly-resolved base - // before reusing it. + // before reusing it. Surface the collision shape for RCA before the + // recovery path decides whether to reuse or reset the local branch. + recordActivityEvent({ + projectId: cfg.projectId, + sessionId: cfg.sessionId, + source: "workspace", + kind: "workspace.branch_collision", + level: "warn", + summary: `branch "${cfg.branch}" already exists; falling back to worktree recovery`, + data: { + plugin: "workspace-worktree", + branch: cfg.branch, + errorMessage: msg, + }, + }); const baseSha = await git(repoPath, "rev-parse", baseRef); const branchRef = `refs/heads/${cfg.branch}`; const existingBranchSha = (await refExists(repoPath, branchRef)) @@ -453,8 +468,24 @@ export function create(config?: Record): Workspace { // pre-existing local branches unrelated to this workspace (any branch // containing "/" would have been deleted). Stale branches can be // cleaned up separately via `git branch --merged` or similar. - } catch { + } catch (err) { // If git commands fail, try to clean up the directory. + // The worktree metadata may be left stale in `git worktree list` + // because we couldn't run `worktree remove`. Surface so RCA can + // explain why a path was deleted but `git worktree list` still + // references it. + const errorMessage = err instanceof Error ? err.message : String(err); + recordActivityEvent({ + source: "workspace", + kind: "workspace.destroy_fell_back", + level: "warn", + summary: "destroy fell back to rmSync; git worktree metadata may be stale", + data: { + plugin: "workspace-worktree", + workspacePath, + errorMessage, + }, + }); // On Windows, retry with backoff for the file-handle drain race // (just-killed pty-host children still hold handles inside the worktree). if (existsSync(workspacePath)) { @@ -661,10 +692,31 @@ export function create(config?: Record): Workspace { if (project.postCreate) { const shell = getShell(); for (const command of project.postCreate) { - await execFileAsync(shell.cmd, shell.args(command), { - cwd: info.path, - windowsHide: true, - }); + try { + await execFileAsync(shell.cmd, shell.args(command), { + cwd: info.path, + windowsHide: true, + }); + } catch (err) { + // Surface which postCreate command failed. Lifecycle records + // a generic spawn_failed but loses the specific command and + // its sanitized error output. + const errorMessage = err instanceof Error ? err.message : String(err); + recordActivityEvent({ + projectId: info.projectId, + sessionId: info.sessionId, + source: "workspace", + kind: "workspace.post_create_failed", + level: "error", + summary: `postCreate command failed for session ${info.sessionId}`, + data: { + plugin: "workspace-worktree", + command, + errorMessage, + }, + }); + throw err; + } } } },