Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 213 additions & 0 deletions packages/loopover-miner/lib/ams-notifications.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,213 @@
// AMS → hosted badge notifications (#7657). Builds DetectedNotificationEvent-shaped AMS kinds and POSTs them
// to the contributor ams-notifications ingest, which evaluates through evaluateNotificationEvent →
// notify-deliver (same handoff as src/queue/job-dispatch.ts). Fail-soft: a missing session or network blip
// never breaks the miner's real work. No parallel local notification store.

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;
actorLogin: string;
detectedAt: string;
};

export type AmsNotificationPublishResult = { sent: number; error?: string };

export type AmsNotificationFetch = (
url: string,
init?: { method?: string; headers?: Record<string, string>; body?: string; signal?: AbortSignal },
) => Promise<Response>;

export type PublishAmsNotificationEventsOptions = {
env?: Record<string, string | undefined>;
fetchFn?: AmsNotificationFetch;
timeoutMs?: number;
/** Test/self-host inject: mirrors job-dispatch evaluate → notify-deliver without HTTP. */
dispatch?: (events: AmsNotificationEventPayload[]) => Promise<void>;
};

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 githubIssueDeeplink(repoFullName: string, issueNumber: number): string {
return `https://github.com/${repoFullName}/issues/${issueNumber}`;
}

function githubPullDeeplink(repoFullName: string, pullNumber: number): string {
return `https://github.com/${repoFullName}/pull/${pullNumber}`;
}

export function buildAmsAttemptStartedPayload(input: {
recipientLogin: string;
repoFullName: string;
issueNumber: number;
attemptId: string;
detectedAt?: string;
}): AmsNotificationEventPayload {
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: githubIssueDeeplink(input.repoFullName, input.issueNumber),
actorLogin: recipientLogin,
detectedAt,
};
}

export function buildAmsAttemptFailedPayload(input: {
recipientLogin: string;
repoFullName: string;
issueNumber: number;
attemptId: string;
reason?: string | null;
detectedAt?: string;
}): AmsNotificationEventPayload {
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: githubIssueDeeplink(input.repoFullName, input.issueNumber),
actorLogin: recipientLogin,
detectedAt,
};
}

export function buildAmsGovernorPausedPayload(input: {
recipientLogin: string;
reason?: string | null;
pausedAt?: string;
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",
actorLogin: recipientLogin,
detectedAt,
};
}

export function buildAmsPrOutcomePayload(input: {
recipientLogin: string;
repoFullName: string;
pullNumber: number;
decision: "merged" | "closed";
closedAt?: string | null;
detectedAt?: string;
}): AmsNotificationEventPayload {
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.repoFullName}#${input.pullNumber}:${input.decision}:${closedAt}`,
deeplink: githubPullDeeplink(input.repoFullName, input.pullNumber),
actorLogin: recipientLogin,
detectedAt,
};
}

/**
* Publish AMS notification events through the hosted evaluate → notify-deliver path. Prefer an injected
* `dispatch` (tests / in-process self-host). Otherwise POST to `/v1/contributors/:login/ams-notifications`
* when a loopover-mcp session is on disk. Never throws.
*/
export async function publishAmsNotificationEvents(
events: AmsNotificationEventPayload[],
options: PublishAmsNotificationEventsOptions = {},
): Promise<AmsNotificationPublishResult> {
if (!Array.isArray(events) || events.length === 0) return { sent: 0 };
if (options.dispatch) {
try {
await options.dispatch(events);
return { sent: events.length };
} catch (error) {
return { sent: 0, error: error instanceof Error ? error.message.slice(0, 160) : "dispatch_failed" };
}
}

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" };
if (events.some((event) => normalizeLogin(event.recipientLogin) !== recipientLogin)) {
return { sent: 0, error: "mixed_recipients" };
}

const fetchFn = options.fetchFn ?? (fetch as AmsNotificationFetch);
const timeoutMs = options.timeoutMs ?? DEFAULT_AMS_NOTIFICATION_TIMEOUT_MS;
const url = `${session.apiUrl}/v1/contributors/${encodeURIComponent(recipientLogin)}/ams-notifications`;
const body = JSON.stringify({
events: events.map(({ eventType, repoFullName, pullNumber, dedupKey, deeplink, actorLogin, detectedAt }) => ({
eventType,
repoFullName,
pullNumber,
dedupKey,
deeplink,
actorLogin,
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(timeoutMs),
});
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 sync call sites (never awaits into the caller's critical path). */
export function scheduleAmsNotificationEvents(
events: AmsNotificationEventPayload[],
options: PublishAmsNotificationEventsOptions = {},
): void {
void publishAmsNotificationEvents(events, options).catch(() => {
// publishAmsNotificationEvents is already fail-soft; this only guards a rejected promise from an inject.
});
}
50 changes: 50 additions & 0 deletions packages/loopover-miner/lib/attempt-cli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,12 @@ 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,
type PublishAmsNotificationEventsOptions,
} 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";
Expand Down Expand Up @@ -141,6 +147,11 @@ 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 (session POST / inject). */
scheduleAmsNotifications?: (
events: Parameters<typeof scheduleAmsNotificationEvents>[0],
options?: PublishAmsNotificationEventsOptions,
) => void;
/** 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;
Expand Down Expand Up @@ -666,6 +677,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 through the hosted evaluate → deliver path.
scheduleAmsNotifications(
[
buildAmsAttemptStartedPayload({
recipientLogin: parsed.minerLogin,
repoFullName: parsed.repoFullName,
issueNumber: parsed.issueNumber,
attemptId,
}),
],
{ env: env as NodeJS.ProcessEnv },
);
let result;
try {
result = await runAttemptPipeline(
Expand All @@ -691,10 +715,36 @@ 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: env as NodeJS.ProcessEnv },
);
throw error;
}

worktreeResult.attemptOk = result.outcome === "submitted";
if (result.outcome !== "submitted") {
scheduleAmsNotifications(
[
buildAmsAttemptFailedPayload({
recipientLogin: parsed.minerLogin,
repoFullName: parsed.repoFullName,
issueNumber: parsed.issueNumber,
attemptId,
reason: result.outcome,
}),
],
{ env: env as NodeJS.ProcessEnv },
);
}

// 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
Expand Down
74 changes: 65 additions & 9 deletions packages/loopover-miner/lib/governor-pause-cli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,12 @@
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,
type PublishAmsNotificationEventsOptions,
} from "./ams-notifications.js";
import { resolveLoopoverBackendSession } from "./github-token-resolution.js";

const GOVERNOR_PAUSE_USAGE = "Usage: loopover-miner governor pause [--reason <text>] [--dry-run] [--json]";
const GOVERNOR_RESUME_USAGE = "Usage: loopover-miner governor resume [--dry-run] [--json]";
Expand All @@ -24,6 +30,10 @@ export type ParsedGovernorNoArgsSubcommand = { json: boolean } | { error: string

export type GovernorPauseCliOptions = {
openGovernorState?: () => GovernorState;
env?: Record<string, string | undefined>;
/** Override AMS badge notify (#7657). Defaults to publishAmsNotificationEvents. */
publishAmsNotifications?: typeof publishAmsNotificationEvents;
fetchSessionLogin?: (session: { apiUrl: string; sessionToken: string }) => Promise<string | null>;
};

export function parseGovernorPauseArgs(args: string[]): ParsedGovernorPauseArgs {
Expand Down Expand Up @@ -101,6 +111,50 @@ function renderPauseState(pauseState: GovernorPauseState): string {
return `governor is PAUSED since ${pauseState.pausedAt}${reason}`;
}

async function resolveSessionLogin(env: NodeJS.ProcessEnv): Promise<string | null> {
const session = resolveLoopoverBackendSession(env);
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;
}
}

async function notifyGovernorPaused(
pauseState: GovernorPauseState,
options: GovernorPauseCliOptions,
): Promise<void> {
const env = options.env ?? process.env;
// Injected fetchSessionLogin (tests) may resolve a login without a disk session; only require a real
// session when falling back to GET /v1/auth/session.
const processEnv = env as NodeJS.ProcessEnv;
const login = options.fetchSessionLogin
? await options.fetchSessionLogin(
resolveLoopoverBackendSession(processEnv) ?? { apiUrl: "https://api.loopover.ai", sessionToken: "" },
)
: await resolveSessionLogin(processEnv);
if (!login) return;
const publish = options.publishAmsNotifications ?? publishAmsNotificationEvents;
const publishOptions: PublishAmsNotificationEventsOptions = { env };
await publish(
[
buildAmsGovernorPausedPayload({
recipientLogin: login,
reason: pauseState.reason,
...(pauseState.pausedAt ? { pausedAt: pauseState.pausedAt } : {}),
}),
],
publishOptions,
);
}

export async function runGovernorPause(args: string[], options: GovernorPauseCliOptions = {}): Promise<number> {
const parsed = parseGovernorPauseArgs(args);
if ("error" in parsed) {
Expand All @@ -119,15 +173,17 @@ 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): best-effort; a notify miss must not fail the pause itself.
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));
}
Expand Down
Loading