From c20a6d42c7a12c080cab679a9e876b8c6773d1cd Mon Sep 17 00:00:00 2001 From: cleanjunc Date: Wed, 22 Jul 2026 16:33:10 +0000 Subject: [PATCH] feat(notifications): extend badge notifications to AMS attempt, governor-pause, and PR-outcome events (#7657) --- apps/loopover-ui/public/openapi.json | 117 ++++++++ .../loopover-miner/lib/ams-notifications.ts | 202 +++++++++++++ packages/loopover-miner/lib/attempt-cli.ts | 43 +++ .../loopover-miner/lib/governor-pause-cli.ts | 67 ++++- packages/loopover-miner/lib/loop-cli.ts | 3 +- packages/loopover-miner/lib/pr-outcome.ts | 27 +- src/api/routes.ts | 48 ++- src/db/schema.ts | 4 + src/notifications/ams-events.ts | 161 ++++++++++ src/notifications/service.ts | 81 ++++- src/openapi/schemas.ts | 8 + src/openapi/spec.ts | 41 +++ src/types.ts | 12 +- test/unit/miner-ams-notifications.test.ts | 279 ++++++++++++++++++ test/unit/miner-attempt-cli.test.ts | 129 ++++++++ test/unit/miner-governor-pause-cli.test.ts | 182 +++++++++++- test/unit/miner-pr-outcome.test.ts | 75 ++++- test/unit/notifications-ams-events.test.ts | 204 +++++++++++++ test/unit/notifications-service.test.ts | 80 +++++ test/unit/routes-ams-notifications.test.ts | 150 ++++++++++ 20 files changed, 1897 insertions(+), 16 deletions(-) create mode 100644 packages/loopover-miner/lib/ams-notifications.ts create mode 100644 src/notifications/ams-events.ts create mode 100644 test/unit/miner-ams-notifications.test.ts create mode 100644 test/unit/notifications-ams-events.test.ts create mode 100644 test/unit/routes-ams-notifications.test.ts diff --git a/apps/loopover-ui/public/openapi.json b/apps/loopover-ui/public/openapi.json index 06f0ddd1f9..b1124bae41 100644 --- a/apps/loopover-ui/public/openapi.json +++ b/apps/loopover-ui/public/openapi.json @@ -14610,6 +14610,25 @@ "recommendation", "summary" ] + }, + "AmsNotificationsAccepted": { + "type": "object", + "properties": { + "login": { + "type": "string" + }, + "accepted": { + "type": "number" + }, + "enqueued": { + "type": "number" + } + }, + "required": [ + "login", + "accepted", + "enqueued" + ] } }, "parameters": {}, @@ -19082,6 +19101,104 @@ } ] } + }, + "/v1/contributors/{login}/ams-notifications": { + "post": { + "summary": "Ingest AMS notification events for a contributor", + "parameters": [ + { + "schema": { + "type": "string" + }, + "required": true, + "name": "login", + "in": "path" + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "events": { + "type": "array", + "items": { + "type": "object", + "properties": { + "eventType": { + "type": "string", + "enum": [ + "ams_attempt_started", + "ams_attempt_failed", + "ams_governor_paused", + "ams_pr_outcome" + ] + }, + "repoFullName": { + "type": "string" + }, + "pullNumber": { + "type": "integer", + "minimum": 0 + }, + "dedupKey": { + "type": "string" + }, + "deeplink": { + "type": "string" + }, + "detectedAt": { + "type": "string" + } + }, + "required": [ + "eventType", + "repoFullName", + "pullNumber", + "dedupKey", + "deeplink", + "detectedAt" + ] + }, + "minItems": 1, + "maxItems": 20 + } + }, + "required": [ + "events" + ] + } + } + } + }, + "responses": { + "200": { + "description": "Accepts AMS-relevant notification events (attempt start/fail, governor pause, PR outcome) posted by the contributor's own AMS miner and evaluates them through the existing notify-evaluate → notify-deliver path (#7657).", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AmsNotificationsAccepted" + } + } + } + }, + "400": { + "description": "Invalid AMS notification body" + }, + "403": { + "description": "Forbidden when login does not match the authenticated session" + } + }, + "security": [ + { + "LoopOverBearer": [] + }, + { + "LoopOverSessionCookie": [] + } + ] + } } }, "servers": [ diff --git a/packages/loopover-miner/lib/ams-notifications.ts b/packages/loopover-miner/lib/ams-notifications.ts new file mode 100644 index 0000000000..06b9aec0a4 --- /dev/null +++ b/packages/loopover-miner/lib/ams-notifications.ts @@ -0,0 +1,202 @@ +// AMS → hosted badge notifications (#7657). Builds the AMS notification-event payloads and POSTs them to +// POST /v1/contributors/:login/ams-notifications, where they run through the hosted +// evaluateNotificationEvent → notify-deliver path (the same handoff src/queue/job-dispatch.ts uses for +// webhook-detected kinds). Fail-soft by design: a missing session, a slow backend, or a network blip must +// never fail or slow the miner's real work — every failure collapses to a structured no-op result. +// +// dedupKey/deeplink layouts are mirrored from src/notifications/ams-events.ts (the hosted twin — this +// package cannot import src/); change eventType strings or dedupKey layouts in BOTH places or not at all. + +import { resolveLoopoverBackendSession } from "./github-token-resolution.js"; + +export type AmsNotificationEventPayload = { + eventType: "ams_attempt_started" | "ams_attempt_failed" | "ams_governor_paused" | "ams_pr_outcome"; + recipientLogin: string; + repoFullName: string; + pullNumber: number; + dedupKey: string; + deeplink: string; + detectedAt: string; +}; + +export type AmsNotificationPublishResult = { sent: number; error?: string }; + +export type AmsNotificationFetch = ( + url: string, + init?: { method?: string; headers?: Record; body?: string; signal?: AbortSignal }, +) => Promise; + +export type PublishAmsNotificationEventsOptions = { + env?: Record; + fetchFn?: AmsNotificationFetch; + timeoutMs?: number; +}; + +export const DEFAULT_AMS_NOTIFICATION_TIMEOUT_MS = 10_000; + +function normalizeLogin(login: string): string { + return login.trim().toLowerCase(); +} + +function nowIso(): string { + return new Date().toISOString(); +} + +function issueDeeplink(repoFullName: string, issueNumber: number): string { + return `https://github.com/${repoFullName}/issues/${issueNumber}`; +} + +function pullDeeplink(repoFullName: string, pullNumber: number): string { + return `https://github.com/${repoFullName}/pull/${pullNumber}`; +} + +/** Attempt start — `pullNumber` carries the ISSUE number (hosted-twin convention). */ +export function buildAmsAttemptStartedPayload(input: { + recipientLogin: string; + repoFullName: string; + issueNumber: number; + attemptId: string; + detectedAt?: string; +}): AmsNotificationEventPayload { + const detectedAt = input.detectedAt ?? nowIso(); + return { + eventType: "ams_attempt_started", + recipientLogin: normalizeLogin(input.recipientLogin), + repoFullName: input.repoFullName, + pullNumber: input.issueNumber, + dedupKey: `ams_attempt_started:${input.repoFullName}#${input.issueNumber}:${input.attemptId}`, + deeplink: issueDeeplink(input.repoFullName, input.issueNumber), + detectedAt, + }; +} + +/** Attempt fail — the failure reason folds into the dedupKey so distinct failure modes of one attempt + * notify distinctly while a redelivered identical failure stays deduped. */ +export function buildAmsAttemptFailedPayload(input: { + recipientLogin: string; + repoFullName: string; + issueNumber: number; + attemptId: string; + reason?: string | null | undefined; + detectedAt?: string; +}): AmsNotificationEventPayload { + const detectedAt = input.detectedAt ?? nowIso(); + const reasonKey = input.reason?.trim() ? `:${input.reason.trim().slice(0, 80)}` : ""; + return { + eventType: "ams_attempt_failed", + recipientLogin: normalizeLogin(input.recipientLogin), + repoFullName: input.repoFullName, + pullNumber: input.issueNumber, + dedupKey: `ams_attempt_failed:${input.repoFullName}#${input.issueNumber}:${input.attemptId}${reasonKey}`, + deeplink: issueDeeplink(input.repoFullName, input.issueNumber), + detectedAt, + }; +} + +/** Governor pause — miner-global, not repo-scoped: synthetic `ams/governor` scope, `pullNumber` 0. */ +export function buildAmsGovernorPausedPayload(input: { + recipientLogin: string; + reason?: string | null | undefined; + pausedAt?: string | null | undefined; + detectedAt?: string; +}): AmsNotificationEventPayload { + const recipientLogin = normalizeLogin(input.recipientLogin); + const detectedAt = input.detectedAt ?? nowIso(); + const pausedAt = input.pausedAt ?? detectedAt; + const reasonKey = input.reason?.trim() ? `:${input.reason.trim().slice(0, 80)}` : ""; + return { + eventType: "ams_governor_paused", + recipientLogin, + repoFullName: "ams/governor", + pullNumber: 0, + dedupKey: `ams_governor_paused:${recipientLogin}:${pausedAt}${reasonKey}`, + deeplink: "https://github.com/JSONbored/loopover", + detectedAt, + }; +} + +/** Miner-local PR outcome. dedupKey layout: ams_pr_outcome:{merged|closed}:{repo}#{n}:{closedAt} — the + * hosted content builder reads the decision back out of it. */ +export function buildAmsPrOutcomePayload(input: { + recipientLogin: string; + repoFullName: string; + pullNumber: number; + decision: "merged" | "closed"; + closedAt?: string | null | undefined; + detectedAt?: string; +}): AmsNotificationEventPayload { + const detectedAt = input.detectedAt ?? nowIso(); + const closedAt = input.closedAt?.trim() || detectedAt; + return { + eventType: "ams_pr_outcome", + recipientLogin: normalizeLogin(input.recipientLogin), + repoFullName: input.repoFullName, + pullNumber: input.pullNumber, + dedupKey: `ams_pr_outcome:${input.decision}:${input.repoFullName}#${input.pullNumber}:${closedAt}`, + deeplink: pullDeeplink(input.repoFullName, input.pullNumber), + detectedAt, + }; +} + +/** + * POST a batch of AMS notification events to the hosted ingest for their (single, shared) recipient. + * Requires a loopover-mcp session on disk (resolveLoopoverBackendSession); without one this is a silent + * no-op — badge notifications are an opt-in nicety, not miner infrastructure. Never throws. + */ +export async function publishAmsNotificationEvents( + events: AmsNotificationEventPayload[], + options: PublishAmsNotificationEventsOptions = {}, +): Promise { + if (events.length === 0) return { sent: 0 }; + const env = options.env ?? process.env; + const session = resolveLoopoverBackendSession(env as NodeJS.ProcessEnv); + if (!session) return { sent: 0, error: "no_session" }; + + const recipientLogin = normalizeLogin(events[0]!.recipientLogin); + if (!recipientLogin) return { sent: 0, error: "missing_recipient" }; + // The ingest is self-scoped per login; a mixed batch would silently re-stamp someone else's event. + if (events.some((event) => normalizeLogin(event.recipientLogin) !== recipientLogin)) { + return { sent: 0, error: "mixed_recipients" }; + } + + const fetchFn = options.fetchFn ?? (fetch as AmsNotificationFetch); + const url = `${session.apiUrl}/v1/contributors/${encodeURIComponent(recipientLogin)}/ams-notifications`; + const body = JSON.stringify({ + // recipientLogin stays out of the wire payload (it's the URL); the server re-stamps recipient AND actor + // from the authenticated session either way. + events: events.map(({ eventType, repoFullName, pullNumber, dedupKey, deeplink, detectedAt }) => ({ + eventType, + repoFullName, + pullNumber, + dedupKey, + deeplink, + detectedAt, + })), + }); + + try { + const response = await fetchFn(url, { + method: "POST", + headers: { + authorization: `Bearer ${session.sessionToken}`, + "content-type": "application/json", + accept: "application/json", + }, + body, + signal: AbortSignal.timeout(options.timeoutMs ?? DEFAULT_AMS_NOTIFICATION_TIMEOUT_MS), + }); + if (!response.ok) return { sent: 0, error: `http_${response.status}` }; + return { sent: events.length }; + } catch (error) { + return { sent: 0, error: error instanceof Error ? error.message.slice(0, 160) : "network_failed" }; + } +} + +/** Fire-and-forget wrapper for call sites that must never await into their critical path. */ +export function scheduleAmsNotificationEvents( + events: AmsNotificationEventPayload[], + options: PublishAmsNotificationEventsOptions = {}, +): void { + // publishAmsNotificationEvents never rejects; void just detaches the promise from the caller. + void publishAmsNotificationEvents(events, options); +} diff --git a/packages/loopover-miner/lib/attempt-cli.ts b/packages/loopover-miner/lib/attempt-cli.ts index 2367774f46..5963763079 100644 --- a/packages/loopover-miner/lib/attempt-cli.ts +++ b/packages/loopover-miner/lib/attempt-cli.ts @@ -49,6 +49,7 @@ import { buildCodingTaskSpec } from "./coding-task-spec.js"; import type { buildCodingTaskSpec as BuildCodingTaskSpecFn } from "./coding-task-spec.js"; import { resolveAmsPolicy } from "./ams-policy.js"; import type { resolveAmsPolicy as ResolveAmsPolicyFn } from "./ams-policy.js"; +import { buildAmsAttemptFailedPayload, buildAmsAttemptStartedPayload, scheduleAmsNotificationEvents } from "./ams-notifications.js"; import { checkMinerKillSwitch, recordMinerKillSwitchTransition } from "./governor-kill-switch.js"; import type { checkMinerKillSwitch as CheckMinerKillSwitchFn } from "./governor-kill-switch.js"; import { captureMinerError } from "./sentry.js"; @@ -141,6 +142,8 @@ export type RunAttemptOptions = { /** Hosted soft-claim coordination at work-start/work-end, when the plane is enabled (#7168). Defaults to * discovery-index-client.js's own submitSoftClaim. */ submitSoftClaim?: typeof SubmitSoftClaimFn; + /** AMS badge notifications (#7657). Defaults to scheduleAmsNotificationEvents (fire-and-forget session POST). */ + scheduleAmsNotifications?: typeof scheduleAmsNotificationEvents; /** Invoked with the real structured result at every return point, in addition to (never instead of) the * plain exit-code return -- the loop orchestrator's real hook into what actually happened. */ onResult?: (result: AttemptCliResult) => void; @@ -666,6 +669,19 @@ export async function runAttempt(args: string[], options: RunAttemptOptions = {} } const runAttemptPipeline = options.runMinerAttempt ?? runMinerAttempt; + const scheduleAmsNotifications = options.scheduleAmsNotifications ?? scheduleAmsNotificationEvents; + // AMS badge notify (#7657): attempt start. Fire-and-forget — a notify miss never blocks the attempt. + scheduleAmsNotifications( + [ + buildAmsAttemptStartedPayload({ + recipientLogin: parsed.minerLogin, + repoFullName: parsed.repoFullName, + issueNumber: parsed.issueNumber, + attemptId, + }), + ], + { env }, + ); let result; try { result = await runAttemptPipeline( @@ -691,10 +707,37 @@ export async function runAttempt(args: string[], options: RunAttemptOptions = {} // `undefined` and the finally block's `?? true` default (meant for the earlier blocked paths that never // ran anything in the worktree) deleted it -- inverting shouldRetainWorktree's documented policy. worktreeResult.attemptOk = false; + scheduleAmsNotifications( + [ + buildAmsAttemptFailedPayload({ + recipientLogin: parsed.minerLogin, + repoFullName: parsed.repoFullName, + issueNumber: parsed.issueNumber, + attemptId, + reason: "attempt_crashed", + }), + ], + { env }, + ); throw error; } worktreeResult.attemptOk = result.outcome === "submitted"; + // AMS badge notify (#7657): any non-submitted terminal outcome is a failed attempt from the miner's side. + if (result.outcome !== "submitted") { + scheduleAmsNotifications( + [ + buildAmsAttemptFailedPayload({ + recipientLogin: parsed.minerLogin, + repoFullName: parsed.repoFullName, + issueNumber: parsed.issueNumber, + attemptId, + reason: result.outcome, + }), + ], + { env }, + ); + } // Real claim-conflict resolution (#4848): only meaningful once a real PR exists, so this only ever runs // on a real "submitted" outcome. checkSubmissionFreshness (inside runMinerAttempt) already caught the diff --git a/packages/loopover-miner/lib/governor-pause-cli.ts b/packages/loopover-miner/lib/governor-pause-cli.ts index 2f28fc6a14..9a82c2966d 100644 --- a/packages/loopover-miner/lib/governor-pause-cli.ts +++ b/packages/loopover-miner/lib/governor-pause-cli.ts @@ -9,6 +9,8 @@ import { argsWantJson, describeCliError, reportCliFailure } from "./cli-error.js"; import { openGovernorState } from "./governor-state.js"; import type { GovernorPauseState, GovernorState } from "./governor-state.js"; +import { buildAmsGovernorPausedPayload, publishAmsNotificationEvents } from "./ams-notifications.js"; +import { resolveLoopoverBackendSession } from "./github-token-resolution.js"; const GOVERNOR_PAUSE_USAGE = "Usage: loopover-miner governor pause [--reason ] [--dry-run] [--json]"; const GOVERNOR_RESUME_USAGE = "Usage: loopover-miner governor resume [--dry-run] [--json]"; @@ -24,6 +26,11 @@ export type ParsedGovernorNoArgsSubcommand = { json: boolean } | { error: string export type GovernorPauseCliOptions = { openGovernorState?: () => GovernorState; + env?: Record; + /** AMS badge notify on a real pause (#7657). Defaults to publishAmsNotificationEvents. */ + publishAmsNotifications?: typeof publishAmsNotificationEvents; + /** Resolve the session's login (defaults to GET /v1/auth/session with the on-disk session). */ + fetchSessionLogin?: () => Promise; }; export function parseGovernorPauseArgs(args: string[]): ParsedGovernorPauseArgs { @@ -101,6 +108,45 @@ function renderPauseState(pauseState: GovernorPauseState): string { return `governor is PAUSED since ${pauseState.pausedAt}${reason}`; } +// AMS badge notify (#7657) needs a recipient, and this CLI has no --miner-login flag (pausing is not +// attempt-scoped work) -- resolve the login the same place the ingest will re-check it: the on-disk +// loopover-mcp session, via GET /v1/auth/session. No session (or any failure) resolves null = skip notify. +async function fetchSessionLoginFromDisk(env: Record): Promise { + const session = resolveLoopoverBackendSession(env as NodeJS.ProcessEnv); + if (!session) return null; + try { + const response = await fetch(`${session.apiUrl}/v1/auth/session`, { + headers: { authorization: `Bearer ${session.sessionToken}`, accept: "application/json" }, + signal: AbortSignal.timeout(10_000), + }); + if (!response.ok) return null; + const payload = (await response.json().catch(() => null)) as { login?: unknown } | null; + return typeof payload?.login === "string" && payload.login.trim() ? payload.login.trim() : null; + } catch { + return null; + } +} + +// Best-effort: a notify miss (no session, backend down) must never fail the pause that already persisted. +async function notifyGovernorPaused(pauseState: GovernorPauseState, options: GovernorPauseCliOptions): Promise { + const env = options.env ?? process.env; + const login = await (options.fetchSessionLogin ?? (() => fetchSessionLoginFromDisk(env)))(); + if (!login) return; + const publish = options.publishAmsNotifications ?? publishAmsNotificationEvents; + await publish( + [ + buildAmsGovernorPausedPayload({ + recipientLogin: login, + reason: pauseState.reason, + // Always a fresh string right after savePauseState({ paused: true }); the builder's own + // `?? detectedAt` fallback absorbs the type-level null. + pausedAt: pauseState.pausedAt, + }), + ], + { env }, + ); +} + export async function runGovernorPause(args: string[], options: GovernorPauseCliOptions = {}): Promise { const parsed = parseGovernorPauseArgs(args); if ("error" in parsed) { @@ -119,15 +165,18 @@ export async function runGovernorPause(args: string[], options: GovernorPauseCli } try { - return await withGovernorState(options, (governorState) => { - const pauseState = governorState.savePauseState({ paused: true, reason: parsed.reason }); - if (parsed.json) { - console.log(JSON.stringify(pauseState)); - } else { - console.log(renderPauseState(pauseState)); - } - return 0; - }); + const pauseState = await withGovernorState(options, (governorState) => + governorState.savePauseState({ paused: true, reason: parsed.reason }), + ); + // AMS badge notify (#7657) AFTER the persisted write, so the notification never claims a pause that + // failed to save; a notify miss is swallowed (the pause itself already succeeded). + await notifyGovernorPaused(pauseState, options).catch(() => undefined); + if (parsed.json) { + console.log(JSON.stringify(pauseState)); + } else { + console.log(renderPauseState(pauseState)); + } + return 0; } catch (error) { return reportCliFailure(parsed.json, describeCliError(error)); } diff --git a/packages/loopover-miner/lib/loop-cli.ts b/packages/loopover-miner/lib/loop-cli.ts index 676362879a..8a8efc26a9 100644 --- a/packages/loopover-miner/lib/loop-cli.ts +++ b/packages/loopover-miner/lib/loop-cli.ts @@ -567,7 +567,8 @@ export async function runLoop(args: string[], options: RunLoopOptions = {}): Pro decision: prDisposition.merged ? "merged" : "closed", closedAt: prDisposition.closedAt, }, - { eventLedger }, + // recipientLogin routes the recorded outcome into an AMS badge notification too (#7657). + { eventLedger, recipientLogin: parsed.minerLogin, env }, ); // Real per-repo reputation history (#5675): a resolved terminal outcome updates the decided/unfavorable // counts the Governor's self-reputation throttle reads on this repo's next attempt. `decided` always; diff --git a/packages/loopover-miner/lib/pr-outcome.ts b/packages/loopover-miner/lib/pr-outcome.ts index 274df98712..df2cb08f72 100644 --- a/packages/loopover-miner/lib/pr-outcome.ts +++ b/packages/loopover-miner/lib/pr-outcome.ts @@ -11,6 +11,7 @@ import { REJECTION_REASONS } from "./rejection-templates.js"; import type { AppendEventInput, LedgerEntry } from "./event-ledger.js"; +import { buildAmsPrOutcomePayload, scheduleAmsNotificationEvents } from "./ams-notifications.js"; /** Event-ledger vocabulary for a miner-local PR outcome. */ export const MINER_PR_OUTCOME_EVENT = "pr_outcome" as const; @@ -40,6 +41,11 @@ export type RecordPrOutcomeOptions = { * writer throws `invalid_event_ledger` at runtime when this is absent or lacks `appendEvent`. Reuses the * real EventLedger#appendEvent signature so a genuine EventLedger (not just a same-shaped stub) type-checks. */ eventLedger?: { appendEvent(event: AppendEventInput): LedgerEntry }; + /** AMS badge notify (#7657): when the recording miner's login is known (loop-cli passes --miner-login), the + * recorded outcome also fires a fire-and-forget badge notification. Absent = ledger write only. */ + recipientLogin?: string; + env?: Record; + scheduleAmsNotifications?: typeof scheduleAmsNotificationEvents; }; export type PrOutcomeLedgerReader = { @@ -97,7 +103,26 @@ export function recordPrOutcomeSnapshot(input: PrOutcomeInput, options: RecordPr reason: input.reason, }); if (!payload) return null; - return eventLedger.appendEvent({ type: MINER_PR_OUTCOME_EVENT, repoFullName, payload }); + const entry = eventLedger.appendEvent({ type: MINER_PR_OUTCOME_EVENT, repoFullName, payload }); + // AMS badge notify (#7657): only once the ledger write above succeeded, so the notification never claims + // an outcome the miner's own record doesn't hold. Fire-and-forget; skipped when no recipient is known. + const recipientLogin = typeof options.recipientLogin === "string" ? options.recipientLogin.trim() : ""; + if (recipientLogin) { + const schedule = options.scheduleAmsNotifications ?? scheduleAmsNotificationEvents; + schedule( + [ + buildAmsPrOutcomePayload({ + recipientLogin, + repoFullName, + pullNumber: payload.prNumber, + decision: payload.decision, + closedAt: payload.closedAt, + }), + ], + { env: options.env ?? process.env }, + ); + } + return entry; } /** diff --git a/src/api/routes.ts b/src/api/routes.ts index 8055f95372..2e7da2e2fd 100644 --- a/src/api/routes.ts +++ b/src/api/routes.ts @@ -284,7 +284,8 @@ import { attachDataQuality, buildCoreSignalFidelity, buildFreshnessSloReport, bu import { buildContributorOpenPrMonitor } from "../signals/contributor-open-pr-monitor"; import { buildContributorPrOutcomes } from "../signals/contributor-pr-outcomes"; import { buildReviewRiskExplanation } from "../signals/review-risk"; -import { buildNotificationFeed } from "../notifications/service"; +import { buildNotificationFeed, evaluateAndEnqueueNotificationDeliveries } from "../notifications/service"; +import { AMS_NOTIFICATION_EVENT_TYPES, normalizeAmsNotificationEventInput } from "../notifications/ams-events"; import { buildPullRequestReviewability, type PullRequestReviewability } from "../signals/reward-risk"; import { buildLocalBranchAnalysis, findCurrentBranchPullRequest } from "../signals/local-branch"; import { buildIssueSlopAssessment } from "../signals/issue-slop"; @@ -469,6 +470,25 @@ const markNotificationsReadBodySchema = z.object({ ids: z.array(z.string().min(1).max(MAX_NOTIFICATION_DELIVERY_ID_LENGTH)).max(MAX_NOTIFICATION_MARK_READ_IDS).optional(), }); +// #7657: body of POST /v1/contributors/:login/ams-notifications. The miner posts AMS-kind events shaped like +// DetectedNotificationEvent minus recipient/actor — both are forced to the authenticated path login server-side +// (normalizeAmsNotificationEventInput), so a payload can never notify or impersonate someone else. +const amsNotificationsBodySchema = z.object({ + events: z + .array( + z.object({ + eventType: z.enum(AMS_NOTIFICATION_EVENT_TYPES), + repoFullName: z.string().min(1).max(200), + pullNumber: z.number().int().min(0), + dedupKey: z.string().min(1).max(500), + deeplink: z.string().min(1).max(2000), + detectedAt: z.string().min(1).max(64), + }), + ) + .min(1) + .max(20), +}); + // #6746: body of POST/DELETE /v1/contributors/:login/watches. Mirrors watchIssuesShape (src/mcp/server.ts) minus // `login` (path param) and `action` (the HTTP verb). `labels` is POST-only (a DELETE ignores it). const watchSubscriptionBodySchema = z.object({ @@ -3594,6 +3614,25 @@ export function createApp() { return c.json({ login: login.toLowerCase(), marked }); }); + // #7657: the AMS miner posts its own AMS-relevant notification events (attempt start/fail, governor pause, + // PR outcome). Self-scoped via requireContributorAccess; every event is re-stamped onto the authenticated + // login and evaluated through evaluateAndEnqueueNotificationDeliveries — the same + // evaluateNotificationEvent → notify-deliver handoff job-dispatch.ts uses for webhook-detected kinds. + app.post("/v1/contributors/:login/ams-notifications", async (c) => { + const login = c.req.param("login"); + const unauthorized = await requireContributorAccess(c, login); + if (unauthorized) return unauthorized; + const parsed = amsNotificationsBodySchema.safeParse(await c.req.json().catch(() => null)); + if (!parsed.success) return c.json({ error: "invalid_ams_notifications", issues: parsed.error.issues }, 400); + const events = parsed.data.events + .map((raw) => normalizeAmsNotificationEventInput(raw, login)) + .filter((event): event is NonNullable => event !== null); + // Reachable past zod: `.min(1)` admits whitespace-only strings that normalize's `.trim()` checks reject. + if (events.length === 0) return c.json({ error: "invalid_ams_notifications", detail: "no_valid_events" }, 400); + const deliveries = await evaluateAndEnqueueNotificationDeliveries(c.env, events); + return c.json({ login: login.toLowerCase(), accepted: events.length, enqueued: deliveries.length }); + }); + // #6746: REST mirror of the `loopover_watch_issues` MCP tool (LoopoverMcp.watchIssues) — manage a contributor's // own issue-watch subscriptions. The MCP tool's `action` enum splits across the HTTP verbs: GET=list, POST=watch, // DELETE=unwatch. Every verb is self-scoped via requireContributorAccess (a session may only touch its own @@ -6478,9 +6517,16 @@ function canSessionAccessPath(env: Env, identity: Extract/*`; the handler's // requireContributorAccess then enforces actor === login (self-only). if (isExtensionContributorContextPath(path) && isExtensionContributorScopedSession(identity)) return true; + // #7657: the AMS miner posts its own notification events with its loopover-mcp session token; the + // handler's requireContributorAccess enforces actor === login (self-only). + if (isContributorAmsNotificationsPath(path)) return true; return false; } +function isContributorAmsNotificationsPath(path: string): boolean { + return /^\/v1\/contributors\/[^/]+\/ams-notifications$/.test(path); +} + function isRepoSettingsPath(path: string): boolean { return /^\/v1\/repos\/[^/]+\/[^/]+\/settings$/.test(path); } diff --git a/src/db/schema.ts b/src/db/schema.ts index 901c7da958..fe5dc07072 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -1133,6 +1133,10 @@ export const notificationSubscriptions = sqliteTable( }), ); +// event_type carries the NotificationEventType vocabulary as free text: the webhook-detected kinds +// (pull_request_changes_requested / pull_request_merged / issue_watch_match) plus the AMS-ingested kinds +// (#7657: ams_attempt_started / ams_attempt_failed / ams_governor_paused / ams_pr_outcome). Subscriptions +// stay channel-scoped; no schema change was needed for the AMS kinds. export const notificationDeliveries = sqliteTable( "notification_deliveries", { diff --git a/src/notifications/ams-events.ts b/src/notifications/ams-events.ts new file mode 100644 index 0000000000..361925f09b --- /dev/null +++ b/src/notifications/ams-events.ts @@ -0,0 +1,161 @@ +// AMS → badge notification bridge (#7657). Pure builders for the AMS-relevant DetectedNotificationEvent +// kinds (attempt start/fail, governor pause, the miner's own PR-outcome change) plus the ingest-side +// validator the session-authenticated route uses. Everything here feeds the EXISTING +// evaluateNotificationEvent → notify-deliver path (src/queue/job-dispatch.ts) — no parallel delivery store. +// +// The payload/dedupKey layout is mirrored in packages/loopover-miner/lib/ams-notifications.ts (the miner +// cannot import src/); change eventType strings or dedupKey layouts in BOTH places or not at all. + +import type { DetectedNotificationEvent, NotificationEventType } from "../types"; +import { nowIso } from "../utils/json"; + +export const AMS_NOTIFICATION_EVENT_TYPES = [ + "ams_attempt_started", + "ams_attempt_failed", + "ams_governor_paused", + "ams_pr_outcome", +] as const satisfies readonly NotificationEventType[]; + +export type AmsNotificationEventType = (typeof AMS_NOTIFICATION_EVENT_TYPES)[number]; + +const AMS_EVENT_TYPE_SET = new Set(AMS_NOTIFICATION_EVENT_TYPES); + +export function isAmsNotificationEventType(value: unknown): value is AmsNotificationEventType { + return typeof value === "string" && AMS_EVENT_TYPE_SET.has(value); +} + +function normalizeLogin(login: string): string { + return login.trim().toLowerCase(); +} + +function issueDeeplink(repoFullName: string, issueNumber: number): string { + return `https://github.com/${repoFullName}/issues/${issueNumber}`; +} + +function pullDeeplink(repoFullName: string, pullNumber: number): string { + return `https://github.com/${repoFullName}/pull/${pullNumber}`; +} + +/** Attempt start — `pullNumber` carries the ISSUE number (the same overload issue_watch_match uses). */ +export function buildAmsAttemptStartedEvent(input: { + recipientLogin: string; + repoFullName: string; + issueNumber: number; + attemptId: string; + detectedAt?: string; +}): DetectedNotificationEvent { + const recipientLogin = normalizeLogin(input.recipientLogin); + const detectedAt = input.detectedAt ?? nowIso(); + return { + eventType: "ams_attempt_started", + recipientLogin, + repoFullName: input.repoFullName, + pullNumber: input.issueNumber, + dedupKey: `ams_attempt_started:${input.repoFullName}#${input.issueNumber}:${input.attemptId}`, + deeplink: issueDeeplink(input.repoFullName, input.issueNumber), + actorLogin: recipientLogin, + detectedAt, + }; +} + +/** Attempt fail — same issue-number overload as start; `reason` folds into the dedupKey so distinct failure + * modes of one attempt notify distinctly while a retried identical failure stays deduped. */ +export function buildAmsAttemptFailedEvent(input: { + recipientLogin: string; + repoFullName: string; + issueNumber: number; + attemptId: string; + reason?: string | null | undefined; + detectedAt?: string; +}): DetectedNotificationEvent { + const recipientLogin = normalizeLogin(input.recipientLogin); + const detectedAt = input.detectedAt ?? nowIso(); + const reasonKey = input.reason?.trim() ? `:${input.reason.trim().slice(0, 80)}` : ""; + return { + eventType: "ams_attempt_failed", + recipientLogin, + repoFullName: input.repoFullName, + pullNumber: input.issueNumber, + dedupKey: `ams_attempt_failed:${input.repoFullName}#${input.issueNumber}:${input.attemptId}${reasonKey}`, + deeplink: issueDeeplink(input.repoFullName, input.issueNumber), + actorLogin: recipientLogin, + detectedAt, + }; +} + +/** Governor pause — not repo/PR-scoped, so `repoFullName` is a stable synthetic scope and `pullNumber` 0 + * (the same field-overload convention issue_watch_match set for non-PR events). */ +export function buildAmsGovernorPausedEvent(input: { + recipientLogin: string; + reason?: string | null | undefined; + pausedAt?: string | null | undefined; + detectedAt?: string; +}): DetectedNotificationEvent { + const recipientLogin = normalizeLogin(input.recipientLogin); + const detectedAt = input.detectedAt ?? nowIso(); + const pausedAt = input.pausedAt ?? detectedAt; + const reasonKey = input.reason?.trim() ? `:${input.reason.trim().slice(0, 80)}` : ""; + return { + eventType: "ams_governor_paused", + recipientLogin, + repoFullName: "ams/governor", + pullNumber: 0, + dedupKey: `ams_governor_paused:${recipientLogin}:${pausedAt}${reasonKey}`, + deeplink: "https://github.com/JSONbored/loopover", + actorLogin: recipientLogin, + detectedAt, + }; +} + +/** Miner-local PR outcome (merged or closed-without-merge). The decision sits right after the eventType in + * the dedupKey (ams_pr_outcome:{merged|closed}:{repo}#{n}:{closedAt}) so buildAmsPrOutcomeNotification can + * read it back without a payload column. */ +export function buildAmsPrOutcomeEvent(input: { + recipientLogin: string; + repoFullName: string; + pullNumber: number; + decision: "merged" | "closed"; + closedAt?: string | null | undefined; + detectedAt?: string; +}): DetectedNotificationEvent { + const recipientLogin = normalizeLogin(input.recipientLogin); + const detectedAt = input.detectedAt ?? nowIso(); + const closedAt = input.closedAt?.trim() || detectedAt; + return { + eventType: "ams_pr_outcome", + recipientLogin, + repoFullName: input.repoFullName, + pullNumber: input.pullNumber, + dedupKey: `ams_pr_outcome:${input.decision}:${input.repoFullName}#${input.pullNumber}:${closedAt}`, + deeplink: pullDeeplink(input.repoFullName, input.pullNumber), + actorLogin: recipientLogin, + detectedAt, + }; +} + +/** + * Validate one miner-posted AMS event and stamp the authenticated recipient onto it. Only AMS kinds pass — + * the ingest route must not let a client forge webhook-detected notification types — and both recipient and + * actor are forced to the authenticated login, never trusted from the payload. + */ +export function normalizeAmsNotificationEventInput(raw: unknown, recipientLogin: string): DetectedNotificationEvent | null { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const record = raw as Record; + if (!isAmsNotificationEventType(record.eventType)) return null; + if (typeof record.repoFullName !== "string" || !record.repoFullName.trim()) return null; + if (typeof record.dedupKey !== "string" || !record.dedupKey.trim()) return null; + if (typeof record.deeplink !== "string" || !record.deeplink.trim()) return null; + if (typeof record.detectedAt !== "string" || !record.detectedAt.trim()) return null; + if (typeof record.pullNumber !== "number" || !Number.isInteger(record.pullNumber) || record.pullNumber < 0) return null; + const login = normalizeLogin(recipientLogin); + return { + eventType: record.eventType, + recipientLogin: login, + repoFullName: record.repoFullName.trim(), + pullNumber: record.pullNumber, + dedupKey: record.dedupKey.trim(), + deeplink: record.deeplink.trim(), + actorLogin: login, + detectedAt: record.detectedAt.trim(), + }; +} diff --git a/src/notifications/service.ts b/src/notifications/service.ts index 826e2de800..0b5ce8874c 100644 --- a/src/notifications/service.ts +++ b/src/notifications/service.ts @@ -54,6 +54,52 @@ export function buildIssueWatchNotification(event: DetectedNotificationEvent): { }; } +// AMS attempt lifecycle (#7657) — `pullNumber` carries the ISSUE number (same overload issue_watch_match +// uses). Public-safe: outcome framing only, never raw figures. +export function buildAmsAttemptStartedNotification(event: DetectedNotificationEvent): { title: string; body: string } { + const ref = `${event.repoFullName}#${event.pullNumber}`; + return { + title: sanitizePublicComment(`Attempt started on ${ref}`), + body: sanitizePublicComment(`Your AMS miner started an attempt on ${ref}. Watch the attempt log for progress and the outcome.`), + }; +} + +export function buildAmsAttemptFailedNotification(event: DetectedNotificationEvent): { title: string; body: string } { + const ref = `${event.repoFullName}#${event.pullNumber}`; + return { + title: sanitizePublicComment(`Attempt failed on ${ref}`), + body: sanitizePublicComment(`Your AMS miner attempt on ${ref} did not complete. Check the attempt log, then retry or move to the next high-fit issue.`), + }; +} + +// Governor pause (#7657) is miner-global, not repo-scoped, so the content ignores the event's synthetic scope. +export function buildAmsGovernorPausedNotification(_event: DetectedNotificationEvent): { title: string; body: string } { + return { + title: sanitizePublicComment("AMS governor paused"), + body: sanitizePublicComment("Your AMS governor is paused. New attempt cycles wait until you run `loopover-miner governor resume`."), + }; +} + +export function buildAmsPrOutcomeNotification(event: DetectedNotificationEvent): { title: string; body: string } { + const ref = `${event.repoFullName}#${event.pullNumber}`; + // dedupKey layout from buildAmsPrOutcomeEvent: ams_pr_outcome:{merged|closed}:{repo}#{n}:{closedAt}. + const merged = event.dedupKey.startsWith("ams_pr_outcome:merged:"); + if (merged) { + return { + title: sanitizePublicComment(`AMS recorded merge: ${ref}`), + body: sanitizePublicComment( + `Your AMS miner recorded that ${ref} merged. Merged contributions strengthen your standing on ${event.repoFullName} — check your decision pack for the next high-fit issue.`, + ), + }; + } + return { + title: sanitizePublicComment(`AMS recorded close: ${ref}`), + body: sanitizePublicComment( + `Your AMS miner recorded that ${ref} closed without merging. Review the close feedback, then pick the next high-fit issue on ${event.repoFullName}.`, + ), + }; +} + // Maps a detected event to its public-safe notification content. export function buildNotificationContent(event: DetectedNotificationEvent): { title: string; body: string } { switch (event.eventType) { @@ -61,11 +107,44 @@ export function buildNotificationContent(event: DetectedNotificationEvent): { ti return buildMergedOutcomeNotification(event); case "issue_watch_match": return buildIssueWatchNotification(event); - default: + case "ams_attempt_started": + return buildAmsAttemptStartedNotification(event); + case "ams_attempt_failed": + return buildAmsAttemptFailedNotification(event); + case "ams_governor_paused": + return buildAmsGovernorPausedNotification(event); + case "ams_pr_outcome": + return buildAmsPrOutcomeNotification(event); + case "pull_request_changes_requested": return buildChangesRequestedNotification(event); } } +/** + * Mirrors job-dispatch.ts's notify-evaluate handoff (#7657): evaluate each event, then enqueue one + * `notify-deliver` job per freshly-created pending delivery. The AMS ingest route uses this so AMS kinds + * ride the exact same evaluate → deliver path as webhook-detected kinds, never a parallel one. + */ +export async function evaluateAndEnqueueNotificationDeliveries( + env: Env, + events: DetectedNotificationEvent[], +): Promise { + const pending: NotificationDeliveryRecord[] = []; + for (const event of events) { + pending.push(...(await evaluateNotificationEvent(env, event))); + } + await Promise.all( + pending.map((delivery) => + env.JOBS.send({ + type: "notify-deliver", + requestedBy: "notify-evaluate", + deliveryId: delivery.id, + }), + ), + ); + return pending; +} + /** * #699 path B: when a webhook opens a NEW grabbable, high-multiplier issue, fan out one notification event * per watching miner (matching their optional label filter), skipping the issue's own author. DB-backed diff --git a/src/openapi/schemas.ts b/src/openapi/schemas.ts index 7a1030d623..1b747c0d42 100644 --- a/src/openapi/schemas.ts +++ b/src/openapi/schemas.ts @@ -519,6 +519,14 @@ export const NotificationsMarkedSchema = z }) .openapi("NotificationsMarked"); +export const AmsNotificationsAcceptedSchema = z + .object({ + login: z.string(), + accepted: z.number(), + enqueued: z.number(), + }) + .openapi("AmsNotificationsAccepted"); + export const ContributorOpportunitySchema = z .object({ repoFullName: z.string(), diff --git a/src/openapi/spec.ts b/src/openapi/spec.ts index 53acd2ca76..5fc8efb22d 100644 --- a/src/openapi/spec.ts +++ b/src/openapi/spec.ts @@ -27,6 +27,7 @@ import { ContributorPrOutcomesSchema, NotificationFeedSchema, NotificationsMarkedSchema, + AmsNotificationsAcceptedSchema, ContributorRewardRiskStrategySchema, ContributorProfileSchema, ContributorScoringProfileSchema, @@ -841,6 +842,46 @@ export function buildOpenApiSpec() { 400: { description: "Invalid mark-read body" }, }, }); + registry.registerPath({ + method: "post", + path: "/v1/contributors/{login}/ams-notifications", + summary: "Ingest AMS notification events for a contributor", + request: { + params: z.object({ login: z.string() }), + body: { + content: { + "application/json": { + // Inline Zod (not a registered components.schemas $ref) — request bodies in this generator stay + // inline so Cloudflare schema pruning remains response-only (#write-cloudflare-schema). + schema: z.object({ + events: z + .array( + z.object({ + eventType: z.enum(["ams_attempt_started", "ams_attempt_failed", "ams_governor_paused", "ams_pr_outcome"]), + repoFullName: z.string(), + pullNumber: z.number().int().min(0), + dedupKey: z.string(), + deeplink: z.string(), + detectedAt: z.string(), + }), + ) + .min(1) + .max(20), + }), + }, + }, + }, + }, + responses: { + 200: { + description: + "Accepts AMS-relevant notification events (attempt start/fail, governor pause, PR outcome) posted by the contributor's own AMS miner and evaluates them through the existing notify-evaluate → notify-deliver path (#7657).", + content: { "application/json": { schema: AmsNotificationsAcceptedSchema } }, + }, + 400: { description: "Invalid AMS notification body" }, + 403: { description: "Forbidden when login does not match the authenticated session" }, + }, + }); registry.registerPath({ method: "get", path: "/v1/contributors/{login}/repos/{owner}/{repo}/decision", diff --git a/src/types.ts b/src/types.ts index bd8e3cc5fb..ee3a71e934 100644 --- a/src/types.ts +++ b/src/types.ts @@ -2429,7 +2429,17 @@ export type DigestSubscriptionRecord = { // unless a row is `paused`). export type NotificationChannel = "badge" | "email"; export type NotificationDeliveryStatus = "pending" | "delivered" | "read" | "suppressed"; -export type NotificationEventType = "pull_request_changes_requested" | "pull_request_merged" | "issue_watch_match"; +export type NotificationEventType = + | "pull_request_changes_requested" + | "pull_request_merged" + | "issue_watch_match" + // AMS-relevant kinds (#7657): attempt lifecycle, governor pause, and the miner's own PR-outcome change. + // Ingested via POST /v1/contributors/:login/ams-notifications and delivered through the same + // evaluateNotificationEvent → notify-deliver path as the webhook-detected kinds above. + | "ams_attempt_started" + | "ams_attempt_failed" + | "ams_governor_paused" + | "ams_pr_outcome"; /** #699 path B: a miner's standing watch on a repo for new grabbable issues. `labels` ([]=any) filters * which issues notify. The `pullNumber` field of the resulting notification event carries the ISSUE number. */ diff --git a/test/unit/miner-ams-notifications.test.ts b/test/unit/miner-ams-notifications.test.ts new file mode 100644 index 0000000000..0fd4448156 --- /dev/null +++ b/test/unit/miner-ams-notifications.test.ts @@ -0,0 +1,279 @@ +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + buildAmsAttemptFailedPayload, + buildAmsAttemptStartedPayload, + buildAmsGovernorPausedPayload, + buildAmsPrOutcomePayload, + DEFAULT_AMS_NOTIFICATION_TIMEOUT_MS, + publishAmsNotificationEvents, + scheduleAmsNotificationEvents, + type AmsNotificationEventPayload, + type AmsNotificationFetch, +} from "../../packages/loopover-miner/lib/ams-notifications.js"; + +// #7657: the miner-side AMS notification client. Payload builders mirror src/notifications/ams-events.ts's +// dedupKey/deeplink layouts by hand (this package cannot import src/) — the builder assertions here pin that +// lockstep. publishAmsNotificationEvents is fail-soft by contract: every failure mode collapses to a +// structured { sent: 0, error } result and never throws into the miner's real work. + +// Session posture mirrors miner-github-token-resolution.test.ts: a temp LOOPOVER_CONFIG_DIR (never this +// machine's real ~/.config) holding a loopover-mcp config.json with a session token. +const dirs: string[] = []; + +afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); +}); + +function sessionEnv(overrides: Record = {}): Record { + const dir = mkdtempSync(join(tmpdir(), "loopover-miner-ams-notifications-")); + dirs.push(dir); + writeFileSync(join(dir, "config.json"), JSON.stringify({ profiles: { default: { session: { token: "session-token-1" } } } }), { mode: 0o600 }); + return { LOOPOVER_CONFIG_DIR: dir, LOOPOVER_API_URL: "https://api.example.test", ...overrides }; +} + +function sessionlessEnv(): Record { + const dir = mkdtempSync(join(tmpdir(), "loopover-miner-ams-notifications-nosession-")); + dirs.push(dir); + return { LOOPOVER_CONFIG_DIR: dir }; +} + +function payload(overrides: Partial = {}): AmsNotificationEventPayload { + return { + eventType: "ams_attempt_started", + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 41, + dedupKey: "ams_attempt_started:acme/widgets#41:attempt-9", + deeplink: "https://github.com/acme/widgets/issues/41", + detectedAt: "2026-07-22T10:00:00.000Z", + ...overrides, + }; +} + +describe("AMS notification payload builders (#7657)", () => { + it("mirrors the hosted attempt-started dedupKey/deeplink layout with the issue number in pullNumber", () => { + expect( + buildAmsAttemptStartedPayload({ + recipientLogin: " Miner1 ", + repoFullName: "acme/widgets", + issueNumber: 41, + attemptId: "attempt-9", + detectedAt: "2026-07-22T10:00:00.000Z", + }), + ).toEqual(payload()); + }); + + it("folds a trimmed, 80-char-capped reason into the attempt-failed dedupKey and omits it when blank", () => { + const withReason = buildAmsAttemptFailedPayload({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + issueNumber: 41, + attemptId: "attempt-9", + reason: ` ${"r".repeat(120)} `, + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(withReason.dedupKey).toBe(`ams_attempt_failed:acme/widgets#41:attempt-9:${"r".repeat(80)}`); + for (const reason of [undefined, null, " "]) { + const bare = buildAmsAttemptFailedPayload({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + issueNumber: 41, + attemptId: "attempt-9", + reason, + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(bare.dedupKey).toBe("ams_attempt_failed:acme/widgets#41:attempt-9"); + } + }); + + it("scopes a governor pause to ams/governor with pullNumber 0, defaulting pausedAt to detectedAt", () => { + const explicit = buildAmsGovernorPausedPayload({ + recipientLogin: "Miner1", + reason: "manual stop", + pausedAt: "2026-07-22T09:00:00.000Z", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(explicit).toMatchObject({ + eventType: "ams_governor_paused", + recipientLogin: "miner1", + repoFullName: "ams/governor", + pullNumber: 0, + dedupKey: "ams_governor_paused:miner1:2026-07-22T09:00:00.000Z:manual stop", + }); + const defaulted = buildAmsGovernorPausedPayload({ recipientLogin: "miner1", detectedAt: "2026-07-22T10:00:00.000Z" }); + expect(defaulted.dedupKey).toBe("ams_governor_paused:miner1:2026-07-22T10:00:00.000Z"); + }); + + it("encodes the decision into the pr-outcome dedupKey and falls back to detectedAt when closedAt is blank", () => { + const merged = buildAmsPrOutcomePayload({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 9, + decision: "merged", + closedAt: "2026-07-22T08:00:00.000Z", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(merged.dedupKey).toBe("ams_pr_outcome:merged:acme/widgets#9:2026-07-22T08:00:00.000Z"); + expect(merged.deeplink).toBe("https://github.com/acme/widgets/pull/9"); + for (const closedAt of [undefined, null, " "]) { + const closed = buildAmsPrOutcomePayload({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 9, + decision: "closed", + closedAt, + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(closed.dedupKey).toBe("ams_pr_outcome:closed:acme/widgets#9:2026-07-22T10:00:00.000Z"); + } + }); + + it("defaults detectedAt to now in every builder when omitted", () => { + const built = [ + buildAmsAttemptStartedPayload({ recipientLogin: "miner1", repoFullName: "acme/widgets", issueNumber: 41, attemptId: "a" }), + buildAmsAttemptFailedPayload({ recipientLogin: "miner1", repoFullName: "acme/widgets", issueNumber: 41, attemptId: "a" }), + buildAmsGovernorPausedPayload({ recipientLogin: "miner1" }), + buildAmsPrOutcomePayload({ recipientLogin: "miner1", repoFullName: "acme/widgets", pullNumber: 9, decision: "merged" }), + ]; + for (const payloadBuilt of built) expect(Number.isNaN(Date.parse(payloadBuilt.detectedAt))).toBe(false); + }); +}); + +describe("publishAmsNotificationEvents (#7657)", () => { + it("POSTs the batch to the recipient's ams-notifications ingest with the session bearer token", async () => { + const calls: Array<{ url: string; init?: Parameters[1] }> = []; + const fetchFn: AmsNotificationFetch = async (url, init) => { + calls.push({ url, init }); + return new Response(JSON.stringify({ login: "miner1", accepted: 1, enqueued: 1 }), { status: 200 }); + }; + const result = await publishAmsNotificationEvents([payload()], { env: sessionEnv(), fetchFn }); + expect(result).toEqual({ sent: 1 }); + expect(calls).toHaveLength(1); + expect(calls[0]!.url).toBe("https://api.example.test/v1/contributors/miner1/ams-notifications"); + expect(calls[0]!.init?.method).toBe("POST"); + expect(calls[0]!.init?.headers?.authorization).toBe("Bearer session-token-1"); + const body = JSON.parse(calls[0]!.init?.body ?? "{}") as { events: Array> }; + expect(body.events).toHaveLength(1); + // recipientLogin rides the URL, never the wire payload — the server re-stamps recipient AND actor anyway. + expect(body.events[0]!).toEqual({ + eventType: "ams_attempt_started", + repoFullName: "acme/widgets", + pullNumber: 41, + dedupKey: "ams_attempt_started:acme/widgets#41:attempt-9", + deeplink: "https://github.com/acme/widgets/issues/41", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(calls[0]!.init?.signal).toBeInstanceOf(AbortSignal); + }); + + it("no-ops on an empty batch without touching the session or network", async () => { + const fetchFn = vi.fn(); + expect(await publishAmsNotificationEvents([], { env: sessionEnv(), fetchFn })).toEqual({ sent: 0 }); + expect(fetchFn).not.toHaveBeenCalled(); + }); + + it("skips silently when no loopover session is on disk — notifications are a nicety, not infrastructure", async () => { + const fetchFn = vi.fn(); + expect(await publishAmsNotificationEvents([payload()], { env: sessionlessEnv(), fetchFn })).toEqual({ sent: 0, error: "no_session" }); + expect(fetchFn).not.toHaveBeenCalled(); + }); + + it("refuses a batch whose first recipient normalizes to empty", async () => { + const fetchFn = vi.fn(); + expect(await publishAmsNotificationEvents([payload({ recipientLogin: " " })], { env: sessionEnv(), fetchFn })).toEqual({ + sent: 0, + error: "missing_recipient", + }); + expect(fetchFn).not.toHaveBeenCalled(); + }); + + it("refuses a mixed-recipient batch — the ingest is self-scoped per login", async () => { + const fetchFn = vi.fn(); + const result = await publishAmsNotificationEvents([payload(), payload({ recipientLogin: "other" })], { + env: sessionEnv(), + fetchFn, + }); + expect(result).toEqual({ sent: 0, error: "mixed_recipients" }); + expect(fetchFn).not.toHaveBeenCalled(); + }); + + it("treats matching recipients that differ only in case/whitespace as one recipient", async () => { + const fetchFn: AmsNotificationFetch = async () => new Response("{}", { status: 200 }); + const result = await publishAmsNotificationEvents([payload(), payload({ recipientLogin: " MINER1 " })], { + env: sessionEnv(), + fetchFn, + }); + expect(result).toEqual({ sent: 2 }); + }); + + it("reports a non-2xx response as http_ without throwing", async () => { + const fetchFn: AmsNotificationFetch = async () => new Response("nope", { status: 403 }); + expect(await publishAmsNotificationEvents([payload()], { env: sessionEnv(), fetchFn })).toEqual({ + sent: 0, + error: "http_403", + }); + }); + + it("collapses a thrown fetch (network blip / timeout) to a structured error without throwing", async () => { + const fetchFn: AmsNotificationFetch = async () => { + throw new Error(`boom ${"x".repeat(300)}`); + }; + const result = await publishAmsNotificationEvents([payload()], { env: sessionEnv(), fetchFn }); + expect(result.sent).toBe(0); + expect(result.error).toHaveLength(160); + const nonError: AmsNotificationFetch = async () => { + throw "string-throw"; + }; + expect(await publishAmsNotificationEvents([payload()], { env: sessionEnv(), fetchFn: nonError })).toEqual({ + sent: 0, + error: "network_failed", + }); + }); + + it("defaults to process.env when no env option is given (stubbed to a sessionless dir — never this box's real config)", async () => { + const dir = mkdtempSync(join(tmpdir(), "loopover-miner-ams-notifications-procenv-")); + dirs.push(dir); + vi.stubEnv("LOOPOVER_CONFIG_DIR", dir); + try { + await expect(publishAmsNotificationEvents([payload()])).resolves.toEqual({ sent: 0, error: "no_session" }); + } finally { + vi.unstubAllEnvs(); + } + }); + + it("uses the global fetch and the injected timeout when no fetchFn is given", async () => { + let capturedUrl: string | undefined; + let capturedSignal: AbortSignal | undefined; + vi.stubGlobal("fetch", async (url: string, init?: { signal?: AbortSignal }) => { + capturedUrl = url; + capturedSignal = init?.signal; + return new Response("{}", { status: 200 }); + }); + try { + const result = await publishAmsNotificationEvents([payload()], { env: sessionEnv(), timeoutMs: 5_000 }); + expect(result).toEqual({ sent: 1 }); + expect(capturedUrl).toBe("https://api.example.test/v1/contributors/miner1/ams-notifications"); + expect(capturedSignal).toBeInstanceOf(AbortSignal); + } finally { + vi.unstubAllGlobals(); + } + expect(DEFAULT_AMS_NOTIFICATION_TIMEOUT_MS).toBe(10_000); + }); +}); + +describe("scheduleAmsNotificationEvents (#7657)", () => { + it("fires publish without awaiting into the caller (fire-and-forget)", async () => { + let resolved = false; + const fetchFn: AmsNotificationFetch = async () => { + await new Promise((resolve) => setTimeout(resolve, 5)); + resolved = true; + return new Response("{}", { status: 200 }); + }; + scheduleAmsNotificationEvents([payload()], { env: sessionEnv(), fetchFn }); + expect(resolved).toBe(false); + await vi.waitFor(() => expect(resolved).toBe(true)); + }); +}); diff --git a/test/unit/miner-attempt-cli.test.ts b/test/unit/miner-attempt-cli.test.ts index ca3f725b24..009ca2e551 100644 --- a/test/unit/miner-attempt-cli.test.ts +++ b/test/unit/miner-attempt-cli.test.ts @@ -105,6 +105,9 @@ function readyPipelineOptions(overrides: Record = {}) { // Never touches the real (filesystem-backed) default governor-state store (#5655 follow-up) -- a test // that cares whether recordOwnSubmission was actually called overrides this explicitly. recordOwnSubmission: vi.fn(), + // Never lets the real fire-and-forget AMS badge notify (#7657) read process.env for a session -- a test + // that cares about the notify payloads overrides this explicitly. + scheduleAmsNotifications: vi.fn(), ...overrides, }; } @@ -2121,3 +2124,129 @@ describe("runAttempt: hosted soft-claim submission (#7168)", () => { expect(exitCode).toBe(7); }); }); + +describe("AMS badge notifications from the attempt lifecycle (#7657)", () => { + it("schedules exactly one attempt-started notification (and no failure) on a submitted outcome", async () => { + const { allocator, claimLedger, eventLedger, attemptLog, governorLedger } = tempLedgers(); + const scheduleAmsNotificationsSpy = vi.fn(); + const worktreeResult = fakeWorktreeResult(); + + const exitCode = await runAttempt(["acme/widgets", "7", "--miner-login", "Alice", "--json"], { + env: { MINER_CODING_AGENT_PROVIDER: "noop" }, + attemptId: "fixed-attempt-id", + openWorktreeAllocator: () => allocator, + openClaimLedger: () => claimLedger, + initEventLedger: () => eventLedger, + initAttemptLog: () => attemptLog, + initGovernorLedger: () => governorLedger, + ...readyPipelineOptions({ + scheduleAmsNotifications: scheduleAmsNotificationsSpy, + runMinerAttempt: async () => ({ + outcome: "submitted", + spec: { command: "gh pr create", cwd: worktreeResult.worktreePath, timeoutMs: 1000 }, + execResult: { code: 0 }, + loopResult: fakeLoopResult(), + }), + }), + }); + + expect(exitCode).toBe(0); + expect(scheduleAmsNotificationsSpy).toHaveBeenCalledTimes(1); + const [events, options] = scheduleAmsNotificationsSpy.mock.calls[0]!; + expect(events).toEqual([ + expect.objectContaining({ + eventType: "ams_attempt_started", + recipientLogin: "alice", + repoFullName: "acme/widgets", + pullNumber: 7, + dedupKey: "ams_attempt_started:acme/widgets#7:fixed-attempt-id", + }), + ]); + expect(options).toEqual({ env: { MINER_CODING_AGENT_PROVIDER: "noop" } }); + }); + + it("schedules an attempt-failed notification carrying the non-submitted outcome as its reason", async () => { + const { allocator, claimLedger, eventLedger, attemptLog, governorLedger } = tempLedgers(); + const scheduleAmsNotificationsSpy = vi.fn(); + + const exitCode = await runAttempt(["acme/widgets", "7", "--miner-login", "alice", "--json"], { + env: { MINER_CODING_AGENT_PROVIDER: "noop" }, + attemptId: "fixed-attempt-id", + openWorktreeAllocator: () => allocator, + openClaimLedger: () => claimLedger, + initEventLedger: () => eventLedger, + initAttemptLog: () => attemptLog, + initGovernorLedger: () => governorLedger, + ...readyPipelineOptions({ + scheduleAmsNotifications: scheduleAmsNotificationsSpy, + runMinerAttempt: async () => ({ outcome: "abandon", loopResult: fakeLoopResult() }), + }), + }); + + expect(exitCode).toBe(7); + expect(scheduleAmsNotificationsSpy).toHaveBeenCalledTimes(2); + const [failedEvents] = scheduleAmsNotificationsSpy.mock.calls[1]!; + expect(failedEvents).toEqual([ + expect.objectContaining({ + eventType: "ams_attempt_failed", + recipientLogin: "alice", + pullNumber: 7, + dedupKey: "ams_attempt_failed:acme/widgets#7:fixed-attempt-id:abandon", + }), + ]); + }); + + it("schedules an attempt_crashed failure notification when runMinerAttempt throws", async () => { + const { allocator, claimLedger, eventLedger, attemptLog, governorLedger } = tempLedgers(); + vi.spyOn(console, "error").mockImplementation(() => undefined); + const scheduleAmsNotificationsSpy = vi.fn(); + + const exitCode = await runAttempt(["acme/widgets", "7", "--miner-login", "alice"], { + env: { MINER_CODING_AGENT_PROVIDER: "noop" }, + attemptId: "fixed-attempt-id", + openWorktreeAllocator: () => allocator, + openClaimLedger: () => claimLedger, + initEventLedger: () => eventLedger, + initAttemptLog: () => attemptLog, + initGovernorLedger: () => governorLedger, + ...readyPipelineOptions({ + cleanupAttemptWorktree: vi.fn().mockResolvedValue({ ok: true, removed: false }), + scheduleAmsNotifications: scheduleAmsNotificationsSpy, + runMinerAttempt: async () => { + throw new Error("boom"); + }, + }), + }); + + expect(exitCode).toBe(2); + expect(scheduleAmsNotificationsSpy).toHaveBeenCalledTimes(2); + const [failedEvents] = scheduleAmsNotificationsSpy.mock.calls[1]!; + expect(failedEvents).toEqual([ + expect.objectContaining({ + eventType: "ams_attempt_failed", + dedupKey: "ams_attempt_failed:acme/widgets#7:fixed-attempt-id:attempt_crashed", + }), + ]); + }); + + it("falls back to the real fire-and-forget scheduler (a session-less no-op here) when none is injected", async () => { + const { allocator, claimLedger, eventLedger, attemptLog, governorLedger } = tempLedgers(); + // Mirrors the recordOwnSubmission omission pattern above: drop the injected mock so the `??` default + // branch runs. The env carries no loopover session, so the real scheduler resolves no_session silently. + const { scheduleAmsNotifications: _omitted, ...optionsWithoutScheduler } = readyPipelineOptions({ + runMinerAttempt: async () => ({ outcome: "abandon", loopResult: fakeLoopResult() }), + }); + + const exitCode = await runAttempt(["acme/widgets", "7", "--miner-login", "alice", "--json"], { + env: { MINER_CODING_AGENT_PROVIDER: "noop" }, + openWorktreeAllocator: () => allocator, + openClaimLedger: () => claimLedger, + initEventLedger: () => eventLedger, + initAttemptLog: () => attemptLog, + initGovernorLedger: () => governorLedger, + ...optionsWithoutScheduler, + }); + + expect(exitCode).toBe(7); + }); +}); diff --git a/test/unit/miner-governor-pause-cli.test.ts b/test/unit/miner-governor-pause-cli.test.ts index d481780282..c96a80fce5 100644 --- a/test/unit/miner-governor-pause-cli.test.ts +++ b/test/unit/miner-governor-pause-cli.test.ts @@ -1,4 +1,4 @@ -import { mkdtempSync, rmSync } from "node:fs"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; @@ -379,3 +379,183 @@ describe("governor pause/resume/status --json error contract (#5914)", () => { expect(String(log.mock.calls.at(-1)?.[0])).not.toContain("("); }); }); + +describe("AMS badge notify on governor pause (#7657)", () => { + it("publishes a governor-paused notification stamped with the persisted pause state", async () => { + const state = tempGovernorState(); + const log = vi.spyOn(console, "log").mockImplementation(() => undefined); + const publishSpy = vi.fn().mockResolvedValue({ sent: 1 }); + + const exitCode = await runGovernorPause(["--reason", "investigating", "--json"], { + openGovernorState: () => state, + env: { SOME_ENV: "x" }, + fetchSessionLogin: async () => "Miner1", + publishAmsNotifications: publishSpy, + }); + + expect(exitCode).toBe(0); + expect(publishSpy).toHaveBeenCalledTimes(1); + const [events, options] = publishSpy.mock.calls[0]!; + const persisted = state.loadPauseState(); + expect(events).toEqual([ + expect.objectContaining({ + eventType: "ams_governor_paused", + recipientLogin: "miner1", + repoFullName: "ams/governor", + pullNumber: 0, + dedupKey: `ams_governor_paused:miner1:${persisted.pausedAt}:investigating`, + }), + ]); + expect(options).toEqual({ env: { SOME_ENV: "x" } }); + // The pause itself still printed its normal JSON result. + expect(JSON.parse(String(log.mock.calls[0]?.[0]))).toMatchObject({ paused: true, reason: "investigating" }); + }); + + it("skips the notification when no session login resolves, without failing the pause", async () => { + const state = tempGovernorState(); + vi.spyOn(console, "log").mockImplementation(() => undefined); + const publishSpy = vi.fn(); + + const exitCode = await runGovernorPause([], { + openGovernorState: () => state, + fetchSessionLogin: async () => null, + publishAmsNotifications: publishSpy, + }); + + expect(exitCode).toBe(0); + expect(publishSpy).not.toHaveBeenCalled(); + expect(state.loadPauseState().paused).toBe(true); + }); + + it("swallows a thrown notify (rejecting fetchSessionLogin) — the persisted pause still succeeds", async () => { + const state = tempGovernorState(); + vi.spyOn(console, "log").mockImplementation(() => undefined); + + const exitCode = await runGovernorPause([], { + openGovernorState: () => state, + fetchSessionLogin: async () => { + throw new Error("session backend down"); + }, + }); + + expect(exitCode).toBe(0); + expect(state.loadPauseState().paused).toBe(true); + }); + + it("does not notify on --dry-run (nothing was persisted)", async () => { + vi.spyOn(console, "log").mockImplementation(() => undefined); + const publishSpy = vi.fn(); + const fetchSessionLoginSpy = vi.fn(); + + const exitCode = await runGovernorPause(["--dry-run"], { + fetchSessionLogin: fetchSessionLoginSpy, + publishAmsNotifications: publishSpy, + }); + + expect(exitCode).toBe(0); + expect(fetchSessionLoginSpy).not.toHaveBeenCalled(); + expect(publishSpy).not.toHaveBeenCalled(); + }); + + it("resolves the login from the on-disk session by default — no session dir means skip, publish untouched", async () => { + const state = tempGovernorState(); + vi.spyOn(console, "log").mockImplementation(() => undefined); + const publishSpy = vi.fn(); + const dir = mkdtempSync(join(tmpdir(), "loopover-miner-governor-pause-nosession-")); + roots.push(dir); + + const exitCode = await runGovernorPause([], { + openGovernorState: () => state, + env: { LOOPOVER_CONFIG_DIR: dir }, + publishAmsNotifications: publishSpy, + }); + + expect(exitCode).toBe(0); + expect(publishSpy).not.toHaveBeenCalled(); + }); + + it("publishes through the real client on the default path (session GET + ingest POST both stubbed)", async () => { + const state = tempGovernorState(); + vi.spyOn(console, "log").mockImplementation(() => undefined); + const dir = mkdtempSync(join(tmpdir(), "loopover-miner-governor-pause-realpublish-")); + roots.push(dir); + writeFileSync(join(dir, "config.json"), JSON.stringify({ profiles: { default: { session: { token: "session-token-1" } } } }), { mode: 0o600 }); + const fetchCalls: string[] = []; + vi.stubGlobal("fetch", async (url: string) => { + fetchCalls.push(url); + if (url.endsWith("/v1/auth/session")) return Response.json({ status: "authenticated", login: "miner1" }); + return Response.json({ login: "miner1", accepted: 1, enqueued: 1 }); + }); + + const exitCode = await runGovernorPause(["--reason", "maintenance"], { + openGovernorState: () => state, + env: { LOOPOVER_CONFIG_DIR: dir, LOOPOVER_API_URL: "https://api.example.test" }, + }); + + expect(exitCode).toBe(0); + expect(fetchCalls).toEqual([ + "https://api.example.test/v1/auth/session", + "https://api.example.test/v1/contributors/miner1/ams-notifications", + ]); + vi.unstubAllGlobals(); + }); + + it("fetches the session login from GET /v1/auth/session on the default path", async () => { + const state = tempGovernorState(); + vi.spyOn(console, "log").mockImplementation(() => undefined); + const publishSpy = vi.fn().mockResolvedValue({ sent: 1 }); + const dir = mkdtempSync(join(tmpdir(), "loopover-miner-governor-pause-session-")); + roots.push(dir); + writeFileSync(join(dir, "config.json"), JSON.stringify({ profiles: { default: { session: { token: "session-token-1" } } } }), { mode: 0o600 }); + const fetchCalls: string[] = []; + vi.stubGlobal("fetch", async (url: string, init?: { headers?: Record }) => { + fetchCalls.push(url); + expect(init?.headers?.authorization).toBe("Bearer session-token-1"); + return Response.json({ status: "authenticated", login: "Miner1" }); + }); + + const exitCode = await runGovernorPause([], { + openGovernorState: () => state, + env: { LOOPOVER_CONFIG_DIR: dir, LOOPOVER_API_URL: "https://api.example.test" }, + publishAmsNotifications: publishSpy, + }); + + expect(exitCode).toBe(0); + expect(fetchCalls).toEqual(["https://api.example.test/v1/auth/session"]); + expect(publishSpy).toHaveBeenCalledTimes(1); + expect(publishSpy.mock.calls[0]![0]).toEqual([expect.objectContaining({ recipientLogin: "miner1" })]); + vi.unstubAllGlobals(); + }); + + it.each([ + ["a non-OK session response", async () => new Response("nope", { status: 401 })], + ["a non-JSON session body", async () => new Response("not json", { status: 200 })], + ["a blank login", async () => Response.json({ login: " " })], + ["a non-string login", async () => Response.json({ login: 42 })], + [ + "a thrown fetch", + async () => { + throw new Error("network down"); + }, + ], + ])("skips the notification on %s from the default session lookup", async (_label, fetchImpl) => { + const state = tempGovernorState(); + vi.spyOn(console, "log").mockImplementation(() => undefined); + const publishSpy = vi.fn(); + const dir = mkdtempSync(join(tmpdir(), "loopover-miner-governor-pause-badsession-")); + roots.push(dir); + writeFileSync(join(dir, "config.json"), JSON.stringify({ profiles: { default: { session: { token: "session-token-1" } } } }), { mode: 0o600 }); + vi.stubGlobal("fetch", fetchImpl); + + const exitCode = await runGovernorPause([], { + openGovernorState: () => state, + env: { LOOPOVER_CONFIG_DIR: dir }, + publishAmsNotifications: publishSpy, + }); + + expect(exitCode).toBe(0); + expect(publishSpy).not.toHaveBeenCalled(); + expect(state.loadPauseState().paused).toBe(true); + vi.unstubAllGlobals(); + }); +}); diff --git a/test/unit/miner-pr-outcome.test.ts b/test/unit/miner-pr-outcome.test.ts index 5ae4cf293d..2fa4765a5d 100644 --- a/test/unit/miner-pr-outcome.test.ts +++ b/test/unit/miner-pr-outcome.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { MINER_PR_OUTCOME_DECISIONS, MINER_PR_OUTCOME_EVENT, @@ -117,3 +117,76 @@ describe("readPrOutcomes (#4274)", () => { expect([...readPrOutcomes(ledger).keys()]).toEqual(["acme/widgets:8"]); }); }); + +describe("AMS badge notify from recordPrOutcomeSnapshot (#7657)", () => { + it("schedules one pr-outcome notification after a successful ledger write when a recipient is known", () => { + const ledger = mockLedger(); + const scheduleSpy = vi.fn(); + const entry = recordPrOutcomeSnapshot( + { repoFullName: "acme/widgets", prNumber: 9, decision: "merged", closedAt: "2026-07-22T08:00:00Z" }, + { eventLedger: ledger, recipientLogin: " Miner1 ", env: { SOME_ENV: "x" }, scheduleAmsNotifications: scheduleSpy }, + ); + expect(entry).not.toBeNull(); + expect(scheduleSpy).toHaveBeenCalledTimes(1); + const [events, options] = scheduleSpy.mock.calls[0]!; + expect(events).toEqual([ + expect.objectContaining({ + eventType: "ams_pr_outcome", + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 9, + dedupKey: "ams_pr_outcome:merged:acme/widgets#9:2026-07-22T08:00:00Z", + }), + ]); + expect(options).toEqual({ env: { SOME_ENV: "x" } }); + }); + + it("skips the notification when no recipient (or a whitespace one) is given — the ledger write still happens", () => { + const ledger = mockLedger(); + const scheduleSpy = vi.fn(); + recordPrOutcomeSnapshot( + { repoFullName: "acme/widgets", prNumber: 9, decision: "closed", closedAt: "2026-07-22T08:00:00Z" }, + { eventLedger: ledger, scheduleAmsNotifications: scheduleSpy }, + ); + recordPrOutcomeSnapshot( + { repoFullName: "acme/widgets", prNumber: 10, decision: "closed", closedAt: "2026-07-22T08:00:00Z" }, + { eventLedger: ledger, recipientLogin: " ", scheduleAmsNotifications: scheduleSpy }, + ); + expect(ledger._events).toHaveLength(2); + expect(scheduleSpy).not.toHaveBeenCalled(); + }); + + it("never notifies for a snapshot the normalizer rejected (nothing was recorded)", () => { + const ledger = mockLedger(); + const scheduleSpy = vi.fn(); + const entry = recordPrOutcomeSnapshot( + { repoFullName: "acme/widgets", prNumber: 0, decision: "merged" }, + { eventLedger: ledger, recipientLogin: "miner1", scheduleAmsNotifications: scheduleSpy }, + ); + expect(entry).toBeNull(); + expect(scheduleSpy).not.toHaveBeenCalled(); + }); + + it("falls back to the real fire-and-forget scheduler (a session-less no-op here) when none is injected", () => { + const ledger = mockLedger(); + // No scheduleAmsNotifications injected → the real scheduleAmsNotificationEvents runs. The env points at + // an empty config dir, so the underlying publish resolves no_session and never touches the network. + const entry = recordPrOutcomeSnapshot( + { repoFullName: "acme/widgets", prNumber: 9, decision: "merged", closedAt: "2026-07-22T08:00:00Z" }, + { eventLedger: ledger, recipientLogin: "miner1", env: { LOOPOVER_CONFIG_DIR: "/nonexistent-loopover-config" } }, + ); + expect(entry).not.toBeNull(); + expect(ledger._events).toHaveLength(1); + }); + + it("defaults env to process.env in the scheduled options when none is injected", () => { + const ledger = mockLedger(); + const scheduleSpy = vi.fn(); + recordPrOutcomeSnapshot( + { repoFullName: "acme/widgets", prNumber: 9, decision: "merged", closedAt: "2026-07-22T08:00:00Z" }, + { eventLedger: ledger, recipientLogin: "miner1", scheduleAmsNotifications: scheduleSpy }, + ); + expect(scheduleSpy).toHaveBeenCalledTimes(1); + expect(scheduleSpy.mock.calls[0]![1]).toEqual({ env: process.env }); + }); +}); diff --git a/test/unit/notifications-ams-events.test.ts b/test/unit/notifications-ams-events.test.ts new file mode 100644 index 0000000000..21a1a67a69 --- /dev/null +++ b/test/unit/notifications-ams-events.test.ts @@ -0,0 +1,204 @@ +import { describe, expect, it } from "vitest"; +import { + AMS_NOTIFICATION_EVENT_TYPES, + buildAmsAttemptFailedEvent, + buildAmsAttemptStartedEvent, + buildAmsGovernorPausedEvent, + buildAmsPrOutcomeEvent, + isAmsNotificationEventType, + normalizeAmsNotificationEventInput, +} from "../../src/notifications/ams-events"; + +// #7657: the AMS event builders + the ingest-side validator. These pin the dedupKey layouts (mirrored by hand +// in packages/loopover-miner/lib/ams-notifications.ts — the miner cannot import src/) and the ingest rule that +// recipient AND actor are always re-stamped from the authenticated login, never trusted from the payload. + +describe("isAmsNotificationEventType (#7657)", () => { + it("accepts exactly the four AMS kinds", () => { + for (const eventType of AMS_NOTIFICATION_EVENT_TYPES) expect(isAmsNotificationEventType(eventType)).toBe(true); + }); + + it("rejects webhook kinds and non-strings — the ingest must not forge webhook notification types", () => { + expect(isAmsNotificationEventType("pull_request_merged")).toBe(false); + expect(isAmsNotificationEventType("pull_request_changes_requested")).toBe(false); + expect(isAmsNotificationEventType("issue_watch_match")).toBe(false); + expect(isAmsNotificationEventType(undefined)).toBe(false); + expect(isAmsNotificationEventType(7)).toBe(false); + }); +}); + +describe("AMS event builders (#7657)", () => { + it("builds an attempt-started event with the issue number in pullNumber and an issue deeplink", () => { + const event = buildAmsAttemptStartedEvent({ + recipientLogin: " Miner1 ", + repoFullName: "acme/widgets", + issueNumber: 41, + attemptId: "attempt-9", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(event).toEqual({ + eventType: "ams_attempt_started", + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 41, + dedupKey: "ams_attempt_started:acme/widgets#41:attempt-9", + deeplink: "https://github.com/acme/widgets/issues/41", + actorLogin: "miner1", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + }); + + it("defaults detectedAt to now in every builder when omitted", () => { + const built = [ + buildAmsAttemptStartedEvent({ recipientLogin: "miner1", repoFullName: "acme/widgets", issueNumber: 41, attemptId: "a" }), + buildAmsAttemptFailedEvent({ recipientLogin: "miner1", repoFullName: "acme/widgets", issueNumber: 41, attemptId: "a" }), + buildAmsGovernorPausedEvent({ recipientLogin: "miner1" }), + buildAmsPrOutcomeEvent({ recipientLogin: "miner1", repoFullName: "acme/widgets", pullNumber: 9, decision: "merged" }), + ]; + for (const event of built) expect(Number.isNaN(Date.parse(event.detectedAt))).toBe(false); + }); + + it("folds a failure reason into the attempt-failed dedupKey, truncated to 80 chars", () => { + const event = buildAmsAttemptFailedEvent({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + issueNumber: 41, + attemptId: "attempt-9", + reason: ` ${"r".repeat(120)} `, + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(event.eventType).toBe("ams_attempt_failed"); + expect(event.dedupKey).toBe(`ams_attempt_failed:acme/widgets#41:attempt-9:${"r".repeat(80)}`); + expect(event.deeplink).toBe("https://github.com/acme/widgets/issues/41"); + }); + + it("omits the reason segment when the reason is absent or blank", () => { + for (const reason of [undefined, null, " "]) { + const event = buildAmsAttemptFailedEvent({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + issueNumber: 41, + attemptId: "attempt-9", + reason, + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(event.dedupKey).toBe("ams_attempt_failed:acme/widgets#41:attempt-9"); + } + }); + + it("scopes a governor pause to the synthetic ams/governor repo with pullNumber 0", () => { + const event = buildAmsGovernorPausedEvent({ + recipientLogin: "Miner1", + reason: "manual stop", + pausedAt: "2026-07-22T09:00:00.000Z", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(event).toEqual({ + eventType: "ams_governor_paused", + recipientLogin: "miner1", + repoFullName: "ams/governor", + pullNumber: 0, + dedupKey: "ams_governor_paused:miner1:2026-07-22T09:00:00.000Z:manual stop", + deeplink: "https://github.com/JSONbored/loopover", + actorLogin: "miner1", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + }); + + it("defaults pausedAt to detectedAt and omits the reason segment when absent", () => { + const event = buildAmsGovernorPausedEvent({ recipientLogin: "miner1", detectedAt: "2026-07-22T10:00:00.000Z" }); + expect(event.dedupKey).toBe("ams_governor_paused:miner1:2026-07-22T10:00:00.000Z"); + }); + + it("puts the decision right after the eventType in the pr-outcome dedupKey (content builder reads it back)", () => { + const merged = buildAmsPrOutcomeEvent({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 9, + decision: "merged", + closedAt: "2026-07-22T08:00:00.000Z", + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(merged.dedupKey).toBe("ams_pr_outcome:merged:acme/widgets#9:2026-07-22T08:00:00.000Z"); + expect(merged.deeplink).toBe("https://github.com/acme/widgets/pull/9"); + expect(merged.pullNumber).toBe(9); + }); + + it("falls back to detectedAt when closedAt is absent or blank", () => { + for (const closedAt of [undefined, null, " "]) { + const event = buildAmsPrOutcomeEvent({ + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 9, + decision: "closed", + closedAt, + detectedAt: "2026-07-22T10:00:00.000Z", + }); + expect(event.dedupKey).toBe("ams_pr_outcome:closed:acme/widgets#9:2026-07-22T10:00:00.000Z"); + } + }); +}); + +describe("normalizeAmsNotificationEventInput (#7657)", () => { + const valid = { + eventType: "ams_attempt_started", + repoFullName: "acme/widgets", + pullNumber: 41, + dedupKey: "ams_attempt_started:acme/widgets#41:attempt-9", + deeplink: "https://github.com/acme/widgets/issues/41", + detectedAt: "2026-07-22T10:00:00.000Z", + }; + + it("stamps recipient AND actor from the authenticated login, never the payload", () => { + const event = normalizeAmsNotificationEventInput({ ...valid, actorLogin: "someone-else" }, " Miner1 "); + expect(event).toEqual({ + eventType: "ams_attempt_started", + recipientLogin: "miner1", + repoFullName: "acme/widgets", + pullNumber: 41, + dedupKey: valid.dedupKey, + deeplink: valid.deeplink, + actorLogin: "miner1", + detectedAt: valid.detectedAt, + }); + }); + + it("trims string fields", () => { + const event = normalizeAmsNotificationEventInput( + { ...valid, repoFullName: " acme/widgets ", dedupKey: ` ${valid.dedupKey} `, deeplink: ` ${valid.deeplink} `, detectedAt: ` ${valid.detectedAt} ` }, + "miner1", + ); + expect(event?.repoFullName).toBe("acme/widgets"); + expect(event?.dedupKey).toBe(valid.dedupKey); + expect(event?.deeplink).toBe(valid.deeplink); + expect(event?.detectedAt).toBe(valid.detectedAt); + }); + + it.each([ + ["a non-object", "nope"], + ["null", null], + ["an array", [valid]], + ["a webhook eventType", { ...valid, eventType: "pull_request_merged" }], + ["a missing repoFullName", { ...valid, repoFullName: undefined }], + ["a whitespace repoFullName", { ...valid, repoFullName: " " }], + ["a missing dedupKey", { ...valid, dedupKey: undefined }], + ["a whitespace dedupKey", { ...valid, dedupKey: " " }], + ["a missing deeplink", { ...valid, deeplink: undefined }], + ["a whitespace deeplink", { ...valid, deeplink: " " }], + ["a missing detectedAt", { ...valid, detectedAt: undefined }], + ["a whitespace detectedAt", { ...valid, detectedAt: " " }], + ["a non-integer pullNumber", { ...valid, pullNumber: 4.5 }], + ["a negative pullNumber", { ...valid, pullNumber: -1 }], + ["a string pullNumber", { ...valid, pullNumber: "41" }], + ])("rejects %s", (_label, raw) => { + expect(normalizeAmsNotificationEventInput(raw, "miner1")).toBeNull(); + }); + + it("accepts pullNumber 0 (the governor-pause synthetic scope)", () => { + const event = normalizeAmsNotificationEventInput( + { ...valid, eventType: "ams_governor_paused", repoFullName: "ams/governor", pullNumber: 0, dedupKey: "ams_governor_paused:miner1:t" }, + "miner1", + ); + expect(event?.pullNumber).toBe(0); + }); +}); diff --git a/test/unit/notifications-service.test.ts b/test/unit/notifications-service.test.ts index 9ac0ef9522..3e52fbf779 100644 --- a/test/unit/notifications-service.test.ts +++ b/test/unit/notifications-service.test.ts @@ -1,9 +1,14 @@ import { describe, expect, it } from "vitest"; import { + buildAmsAttemptFailedNotification, + buildAmsAttemptStartedNotification, + buildAmsGovernorPausedNotification, + buildAmsPrOutcomeNotification, buildChangesRequestedNotification, buildNotificationContent, buildNotificationFeed, deliverNotification, + evaluateAndEnqueueNotificationDeliveries, evaluateNotificationEvent, NOTIFICATION_RATE_LIMIT, resolveNotificationChannels, @@ -334,3 +339,78 @@ describe("notification repository helpers", () => { expect(await getNotificationDeliveryById(env, "missing")).toBeNull(); }); }); + +describe("AMS notification kinds (#7657)", () => { + it("routes each AMS event type to its own public-safe copy", () => { + const started = buildNotificationContent(event({ eventType: "ams_attempt_started", pullNumber: 41 })); + expect(started.title).toContain("Attempt started on owner/repo#41"); + const failed = buildNotificationContent(event({ eventType: "ams_attempt_failed", pullNumber: 41 })); + expect(failed.title).toContain("Attempt failed on owner/repo#41"); + const paused = buildNotificationContent(event({ eventType: "ams_governor_paused", repoFullName: "ams/governor", pullNumber: 0 })); + expect(paused.title).toBe("AMS governor paused"); + expect(paused.body).toContain("governor resume"); + const outcome = buildNotificationContent( + event({ eventType: "ams_pr_outcome", dedupKey: "ams_pr_outcome:merged:owner/repo#7:t" }), + ); + expect(outcome.title).toContain("AMS recorded merge: owner/repo#7"); + for (const content of [started, failed, paused, outcome]) { + expect(JSON.stringify(content)).not.toMatch(/reward|payout|trust score|wallet|hotkey|\$/i); + } + }); + + it("keeps attempt copy pointed at the attempt log for both lifecycle kinds", () => { + expect(buildAmsAttemptStartedNotification(event({ eventType: "ams_attempt_started" })).body).toContain("attempt log"); + expect(buildAmsAttemptFailedNotification(event({ eventType: "ams_attempt_failed" })).body).toContain("attempt log"); + }); + + it("ignores the synthetic governor scope in the pause copy", () => { + const content = buildAmsGovernorPausedNotification(event({ eventType: "ams_governor_paused", repoFullName: "ams/governor" })); + expect(content.title).toBe("AMS governor paused"); + expect(JSON.stringify(content)).not.toContain("ams/governor"); + }); + + it("reads the decision back out of the pr-outcome dedupKey for merged vs closed copy", () => { + const merged = buildAmsPrOutcomeNotification( + event({ eventType: "ams_pr_outcome", dedupKey: "ams_pr_outcome:merged:owner/repo#7:t" }), + ); + expect(merged.title).toContain("AMS recorded merge"); + expect(merged.body).toContain("merged"); + const closed = buildAmsPrOutcomeNotification( + event({ eventType: "ams_pr_outcome", dedupKey: "ams_pr_outcome:closed:owner/repo#7:t" }), + ); + expect(closed.title).toContain("AMS recorded close"); + expect(closed.body).toContain("without merging"); + }); + + it("evaluateAndEnqueueNotificationDeliveries creates deliveries and enqueues one notify-deliver per pending row", async () => { + const sent: Array> = []; + const env = createTestEnv({ + JOBS: { send: async (message: Record) => void sent.push(message) } as unknown as Queue, + }); + const pending = await evaluateAndEnqueueNotificationDeliveries(env, [ + event({ eventType: "ams_attempt_started", dedupKey: "ams_attempt_started:owner/repo#41:a1", pullNumber: 41 }), + event({ eventType: "ams_pr_outcome", dedupKey: "ams_pr_outcome:merged:owner/repo#7:t" }), + ]); + expect(pending).toHaveLength(2); + expect(sent).toHaveLength(2); + for (const [index, message] of sent.entries()) { + expect(message).toEqual({ type: "notify-deliver", requestedBy: "notify-evaluate", deliveryId: pending[index]!.id }); + } + }); + + it("evaluateAndEnqueueNotificationDeliveries enqueues nothing when every event dedupes to an existing row", async () => { + const sent: unknown[] = []; + const env = createTestEnv({ + JOBS: { send: async (message: unknown) => void sent.push(message) } as unknown as Queue, + }); + const first = await evaluateAndEnqueueNotificationDeliveries(env, [ + event({ eventType: "ams_attempt_started", dedupKey: "ams_attempt_started:owner/repo#41:a1", pullNumber: 41 }), + ]); + expect(first).toHaveLength(1); + const second = await evaluateAndEnqueueNotificationDeliveries(env, [ + event({ eventType: "ams_attempt_started", dedupKey: "ams_attempt_started:owner/repo#41:a1", pullNumber: 41 }), + ]); + expect(second).toHaveLength(0); + expect(sent).toHaveLength(1); + }); +}); diff --git a/test/unit/routes-ams-notifications.test.ts b/test/unit/routes-ams-notifications.test.ts new file mode 100644 index 0000000000..41c7370810 --- /dev/null +++ b/test/unit/routes-ams-notifications.test.ts @@ -0,0 +1,150 @@ +import { describe, expect, it } from "vitest"; +import { createApp } from "../../src/api/routes"; +import { listNotificationDeliveriesForRecipient } from "../../src/db/repositories"; +import { createTestEnv } from "../helpers/d1"; + +// #7657: POST /v1/contributors/:login/ams-notifications — the AMS miner's ingest for its own notification +// events. These pin the route contract: the requireContributorAccess guard, the zod body gate, the +// normalize re-stamp (recipient/actor forced to the path login), and that accepted events run through the +// SAME evaluateNotificationEvent → notify-deliver handoff as webhook kinds (deliveries + queued jobs). + +const jsonHeaders = (env: Env) => ({ authorization: `Bearer ${env.LOOPOVER_API_TOKEN}`, "content-type": "application/json" }); + +function amsEvent(overrides: Partial> = {}): Record { + return { + eventType: "ams_attempt_started", + repoFullName: "acme/widgets", + pullNumber: 41, + dedupKey: "ams_attempt_started:acme/widgets#41:attempt-9", + deeplink: "https://github.com/acme/widgets/issues/41", + detectedAt: "2026-07-22T10:00:00.000Z", + ...overrides, + }; +} + +function post(app: ReturnType, env: Env, body: unknown, login = "miner1") { + return app.request( + `/v1/contributors/${login}/ams-notifications`, + { method: "POST", headers: jsonHeaders(env), body: JSON.stringify(body) }, + env, + ); +} + +describe("POST /v1/contributors/:login/ams-notifications (#7657)", () => { + it("accepts AMS events, creates pending badge deliveries, and enqueues one notify-deliver per delivery", async () => { + const app = createApp(); + const sent: Array> = []; + const env = createTestEnv({ + JOBS: { send: async (message: Record) => void sent.push(message) } as unknown as Queue, + }); + + const response = await post(app, env, { + events: [amsEvent(), amsEvent({ eventType: "ams_pr_outcome", dedupKey: "ams_pr_outcome:merged:acme/widgets#9:t", pullNumber: 9 })], + }); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ login: "miner1", accepted: 2, enqueued: 2 }); + + const rows = await listNotificationDeliveriesForRecipient(env, "miner1"); + expect(rows).toHaveLength(2); + for (const row of rows) { + expect(row.recipientLogin).toBe("miner1"); + expect(row.status).toBe("pending"); + } + expect(sent).toHaveLength(2); + for (const message of sent) expect(message).toMatchObject({ type: "notify-deliver", requestedBy: "notify-evaluate" }); + }); + + it("re-stamps recipient and actor from the path login even when the payload claims someone else", async () => { + const app = createApp(); + const env = createTestEnv(); + const response = await post(app, env, { events: [amsEvent({ actorLogin: "mallory" })] }, "Miner1"); + expect(response.status).toBe(200); + const rows = await listNotificationDeliveriesForRecipient(env, "miner1"); + expect(rows).toHaveLength(1); + expect(rows[0]!.actorLogin).toBe("miner1"); + expect(rows[0]!.recipientLogin).toBe("miner1"); + }); + + it("is idempotent on a redelivered batch (dedupKey), reporting zero newly-enqueued deliveries", async () => { + const app = createApp(); + const env = createTestEnv(); + await post(app, env, { events: [amsEvent()] }); + const again = await post(app, env, { events: [amsEvent()] }); + expect(again.status).toBe(200); + await expect(again.json()).resolves.toEqual({ login: "miner1", accepted: 1, enqueued: 0 }); + expect(await listNotificationDeliveriesForRecipient(env, "miner1")).toHaveLength(1); + }); + + it("400s a malformed body: not JSON, empty batch, unknown eventType, oversized batch", async () => { + const app = createApp(); + const env = createTestEnv(); + for (const body of ["not json", { events: [] }, { events: [amsEvent({ eventType: "pull_request_merged" })] }]) { + const response = await app.request( + "/v1/contributors/miner1/ams-notifications", + { method: "POST", headers: jsonHeaders(env), body: typeof body === "string" ? body : JSON.stringify(body) }, + env, + ); + expect(response.status).toBe(400); + await expect(response.json()).resolves.toMatchObject({ error: "invalid_ams_notifications" }); + } + const oversized = await post(app, env, { events: Array.from({ length: 21 }, (_, index) => amsEvent({ pullNumber: index })) }); + expect(oversized.status).toBe(400); + }); + + it("400s no_valid_events when every event passes zod but fails normalize (whitespace-only repoFullName)", async () => { + const app = createApp(); + const env = createTestEnv(); + const response = await post(app, env, { events: [amsEvent({ repoFullName: " " })] }); + expect(response.status).toBe(400); + await expect(response.json()).resolves.toEqual({ error: "invalid_ams_notifications", detail: "no_valid_events" }); + expect(await listNotificationDeliveriesForRecipient(env, "miner1")).toHaveLength(0); + }); + + it("accepts the miner's OWN session — the real loopover-mcp session posture the miner-side client uses", async () => { + const app = createApp(); + const env = createTestEnv(); + const { createSessionForGitHubUser } = await import("../../src/auth/security"); + const session = await createSessionForGitHubUser(env, { login: "Miner1", id: 42 }); + const response = await app.request( + "/v1/contributors/miner1/ams-notifications", + { + method: "POST", + headers: { authorization: `Bearer ${session.token}`, "content-type": "application/json" }, + body: JSON.stringify({ events: [amsEvent()] }), + }, + env, + ); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ login: "miner1", accepted: 1, enqueued: 1 }); + }); + + it("403s a session for a different login — the ingest is strictly self-scoped", async () => { + const app = createApp(); + const env = createTestEnv(); + const { createSessionForGitHubUser } = await import("../../src/auth/security"); + const session = await createSessionForGitHubUser(env, { login: "someone-else", id: 77 }); + const response = await app.request( + "/v1/contributors/miner1/ams-notifications", + { + method: "POST", + headers: { authorization: `Bearer ${session.token}`, "content-type": "application/json" }, + body: JSON.stringify({ events: [amsEvent()] }), + }, + env, + ); + expect(response.status).toBe(403); + await expect(response.json()).resolves.toEqual({ error: "forbidden_contributor" }); + }); + + it("403s the shared mcp token unless fully unscoped (#2455 parity with the other contributor surfaces)", async () => { + const app = createApp(); + const env = createTestEnv({ MCP_READ_REPO_ALLOWLIST: "acme/widgets" }); + const response = await app.request( + "/v1/contributors/miner1/ams-notifications", + { method: "POST", headers: { authorization: `Bearer ${env.LOOPOVER_MCP_TOKEN}`, "content-type": "application/json" }, body: JSON.stringify({ events: [amsEvent()] }) }, + env, + ); + expect(response.status).toBe(403); + await expect(response.json()).resolves.toEqual({ error: "forbidden_contributor" }); + }); +});