feat(notifications): active issue-watch monitor — gittensory_watch_issues (#699 path B)

migrations/0036_issue_watch_subscriptions.sql

@@ -0,0 +1,13 @@
+-- #699 path B: miners subscribe to watch a repo for NEW grabbable, high-multiplier issues. When such an
+-- issue opens, the watchers are notified through the #535 notification pipeline. `labels_json` is an
+-- optional label filter ([] = any label); UNIQUE(login, repo_full_name) makes subscribe idempotent.
+CREATE TABLE IF NOT EXISTS issue_watch_subscriptions (
+  id TEXT PRIMARY KEY,
+  login TEXT NOT NULL,
+  repo_full_name TEXT NOT NULL,
+  labels_json TEXT NOT NULL DEFAULT '[]',
+  created_at TEXT NOT NULL,
+  updated_at TEXT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS issue_watch_subscriptions_login_repo_unique ON issue_watch_subscriptions (login, repo_full_name);
+CREATE INDEX IF NOT EXISTS issue_watch_subscriptions_repo_idx ON issue_watch_subscriptions (repo_full_name);

src/db/repositories.ts

@@ -27,6 +27,7 @@ import {
   issues,
   githubRateLimitObservations,
   notificationDeliveries,
+  issueWatchSubscriptions,
   notificationSubscriptions,
   officialMinerDetections,
   pullRequestFiles,
@@ -100,6 +101,7 @@ import type {
   NotificationChannel,
   NotificationDeliveryRecord,
   NotificationDeliveryStatus,
+  IssueWatchSubscription,
   NotificationSubscriptionRecord,
   ProductUsageActivationFunnel,
   ProductUsageDailyRollupRecord,
@@ -1303,6 +1305,54 @@ export async function listNotificationSubscriptionsForLogin(env: Env, login: str
   return rows.map(toNotificationSubscriptionRecord);
 }
 
+// ─── Issue-watch subscriptions (#699 path B) ─────────────────────────────────────────────────────────
+
+function toIssueWatchSubscription(row: typeof issueWatchSubscriptions.$inferSelect): IssueWatchSubscription {
+  return { login: row.login, repoFullName: row.repoFullName, labels: parseJson<string[]>(row.labelsJson, []), createdAt: row.createdAt, updatedAt: row.updatedAt };
+}
+
+/** Subscribe a miner to a repo's new grabbable issues; idempotent on (login, repo) — re-subscribing just
+ *  updates the label filter. `labels` ([]=any) are lowercased for case-insensitive matching at delivery. */
+export async function upsertIssueWatchSubscription(env: Env, input: { login: string; repoFullName: string; labels?: string[] | undefined }): Promise<IssueWatchSubscription> {
+  const db = getDb(env.DB);
+  const login = input.login.toLowerCase();
+  const labels = [...new Set((input.labels ?? []).map((label) => label.toLowerCase().trim()).filter(Boolean))];
+  await db
+    .insert(issueWatchSubscriptions)
+    .values({ id: crypto.randomUUID(), login, repoFullName: input.repoFullName, labelsJson: jsonString(labels), updatedAt: nowIso() })
+    .onConflictDoUpdate({ target: [issueWatchSubscriptions.login, issueWatchSubscriptions.repoFullName], set: { labelsJson: jsonString(labels), updatedAt: nowIso() } });
+  const [row] = await db
+    .select()
+    .from(issueWatchSubscriptions)
+    .where(and(eq(issueWatchSubscriptions.login, login), eq(issueWatchSubscriptions.repoFullName, input.repoFullName)));
+  return row ? toIssueWatchSubscription(row) : { login, repoFullName: input.repoFullName, labels };
+}
+
+export async function listIssueWatchSubscriptionsForLogin(env: Env, login: string): Promise<IssueWatchSubscription[]> {
+  const db = getDb(env.DB);
+  const rows = await db.select().from(issueWatchSubscriptions).where(eq(issueWatchSubscriptions.login, login.toLowerCase())).limit(200);
+  return rows.map(toIssueWatchSubscription);
+}
+
+/** Returns whether a watch existed and was removed (so the caller can report it accurately). */
+export async function deleteIssueWatchSubscription(env: Env, login: string, repoFullName: string): Promise<boolean> {
+  const db = getDb(env.DB);
+  const existing = await db
+    .select({ id: issueWatchSubscriptions.id })
+    .from(issueWatchSubscriptions)
+    .where(and(eq(issueWatchSubscriptions.login, login.toLowerCase()), eq(issueWatchSubscriptions.repoFullName, repoFullName)));
+  if (existing.length === 0) return false;
+  await db.delete(issueWatchSubscriptions).where(and(eq(issueWatchSubscriptions.login, login.toLowerCase()), eq(issueWatchSubscriptions.repoFullName, repoFullName)));
+  return true;
+}
+
+/** All miners watching a repo — the candidate recipients when a new grabbable issue opens there. */
+export async function listIssueWatchersForRepo(env: Env, repoFullName: string): Promise<IssueWatchSubscription[]> {
+  const db = getDb(env.DB);
+  const rows = await db.select().from(issueWatchSubscriptions).where(eq(issueWatchSubscriptions.repoFullName, repoFullName)).limit(5000);
+  return rows.map(toIssueWatchSubscription);
+}
+
 // Idempotency guard: UNIQUE(dedup_key, channel) means a duplicate webhook / queue retry inserts nothing
 // and returns the existing row. Returns whether THIS call created the row (so only the first enqueues delivery).
 export async function insertNotificationDeliveryIfAbsent(

src/db/schema.ts

@@ -859,6 +859,24 @@ export const notificationDeliveries = sqliteTable(
   }),
 );
 
+// #699 path B: a miner's standing watch on a repo for NEW grabbable, high-multiplier issues. `labelsJson`
+// is an optional label filter ([] = any). UNIQUE(login, repoFullName) makes subscribe idempotent.
+export const issueWatchSubscriptions = sqliteTable(
+  "issue_watch_subscriptions",
+  {
+    id: text("id").primaryKey(),
+    login: text("login").notNull(),
+    repoFullName: text("repo_full_name").notNull(),
+    labelsJson: text("labels_json").notNull().default("[]"),
+    createdAt: text("created_at").notNull().$defaultFn(() => nowIso()),
+    updatedAt: text("updated_at").notNull().$defaultFn(() => nowIso()),
+  },
+  (table) => ({
+    loginRepo: uniqueIndex("issue_watch_subscriptions_login_repo_unique").on(table.login, table.repoFullName),
+    repo: index("issue_watch_subscriptions_repo_idx").on(table.repoFullName),
+  }),
+);
+
 export const githubAgentCommandAnswers = sqliteTable(
   "github_agent_command_answers",
   {

src/mcp/server.ts

@@ -22,7 +22,10 @@ import {
   listContributorPullRequests,
   listIssueSignalSample,
   listIssues,
+  deleteIssueWatchSubscription,
+  listIssueWatchSubscriptionsForLogin,
   listNotificationDeliveriesForRecipient,
+  upsertIssueWatchSubscription,
   listOpenPullRequests,
   listPullRequests,
   listRecentMergedPullRequests,
@@ -451,6 +454,20 @@ const markNotificationsReadShape = {
     .optional(),
 };
 
+// #699 path B: a miner's self-scoped issue-watch subscriptions. `action` defaults to `list`; `watch`/`unwatch`
+// require repoFullName. `labels` ([]/omitted = any) filters which new issues notify.
+const watchIssuesShape = {
+  login: z.string().min(1),
+  action: z.enum(["watch", "unwatch", "list"]).default("list"),
+  repoFullName: z.string().min(3).max(200).optional(),
+  labels: z.array(z.string().min(1).max(100)).max(50).optional(),
+};
+
+const watchIssuesOutputSchema = {
+  watching: z.array(z.object({ repoFullName: z.string(), labels: z.array(z.string()) })).optional(),
+  changed: z.string().optional(),
+};
+
 const explainRepoDecisionOutputSchema = {
   status: z.string().optional(),
   login: z.string().optional(),
@@ -722,6 +739,17 @@ export class GittensoryMcp {
       async (input) => this.toolResult(await this.markNotificationsRead(input.login, input.ids)),
     );
 
+    server.registerTool(
+      "gittensory_watch_issues",
+      {
+        description:
+          "Watch repos for NEW grabbable, high-multiplier issues (maintainer-created, not WIP). action=watch subscribes a repo (optional label filter), unwatch removes it, list (default) returns your watches. When a matching issue opens you're notified via gittensory_list_notifications. Self-scoped to the authenticated login.",
+        inputSchema: watchIssuesShape,
+        outputSchema: watchIssuesOutputSchema,
+      },
+      async (input) => this.toolResult(await this.watchIssues(input)),
+    );
+
     server.registerTool(
       "gittensory_explain_repo_decision",
       {
@@ -1393,6 +1421,27 @@ export class GittensoryMcp {
     };
   }
 
+  // #699 path B: manage a miner's issue-watch subscriptions. Self-scoped; watch/unwatch need repoFullName.
+  private async watchIssues(input: z.infer<z.ZodObject<typeof watchIssuesShape>>): Promise<ToolPayload> {
+    this.requireContributorAccess(input.login);
+    let changed: string | undefined;
+    if (input.action === "watch" || input.action === "unwatch") {
+      if (!input.repoFullName) return { summary: `${input.action} requires repoFullName.`, data: {} };
+      if (input.action === "watch") {
+        await upsertIssueWatchSubscription(this.env, { login: input.login, repoFullName: input.repoFullName, labels: input.labels });
+        changed = `watching ${input.repoFullName}${input.labels && input.labels.length > 0 ? ` (labels: ${input.labels.join(", ")})` : ""}`;
+      } else {
+        const removed = await deleteIssueWatchSubscription(this.env, input.login, input.repoFullName);
+        changed = removed ? `unwatched ${input.repoFullName}` : `was not watching ${input.repoFullName}`;
+      }
+    }
+    const watching = (await listIssueWatchSubscriptionsForLogin(this.env, input.login)).map((sub) => ({ repoFullName: sub.repoFullName, labels: sub.labels }));
+    return {
+      summary: `Watching ${watching.length} repo(s) for new grabbable issues${changed ? ` (${changed})` : ""}.`,
+      data: { watching, ...(changed ? { changed } : {}) } as unknown as Record<string, unknown>,
+    };
+  }
+
   private async markNotificationsRead(login: string, ids?: string[]): Promise<ToolPayload> {
     this.requireContributorAccess(login);
     const marked = await markNotificationDeliveriesRead(this.env, login, ids);

src/notifications/service.ts

@@ -3,10 +3,12 @@ import {
   countRecentNotificationDeliveries,
   getNotificationDeliveryById,
   insertNotificationDeliveryIfAbsent,
+  listIssueWatchersForRepo,
   listNotificationSubscriptionsForLogin,
   markNotificationDeliveryDelivered,
 } from "../db/repositories";
-import type { DetectedNotificationEvent, NotificationChannel, NotificationDeliveryRecord, NotificationSubscriptionRecord } from "../types";
+import { isGrabbableHighMultiplierIssue } from "../signals/engine";
+import type { DetectedNotificationEvent, IssueRecord, NotificationChannel, NotificationDeliveryRecord, NotificationSubscriptionRecord } from "../types";
 import { nowIso } from "../utils/json";
 
 // Per-recipient, per-channel safety cap. The killer event (changes_requested) delivers immediately, but a
@@ -40,9 +42,55 @@ export function buildMergedOutcomeNotification(event: DetectedNotificationEvent)
   };
 }
 
+// #699 path B: a repo a miner watches opened a NEW grabbable, high-multiplier issue. For this eventType the
+// `pullNumber` field carries the ISSUE number. Public-safe — "open to grab" framing, never raw reward/score.
+export function buildIssueWatchNotification(event: DetectedNotificationEvent): { title: string; body: string } {
+  const ref = `${event.repoFullName}#${event.pullNumber}`;
+  return {
+    title: sanitizePublicComment(`New issue to grab on ${ref}`),
+    body: sanitizePublicComment(`A new maintainer-created issue opened on ${ref} that is open for you to grab. Maintainer-created issues are strong early targets on ${event.repoFullName} — claim it to line up your next contribution.`),
+  };
+}
+
 // Maps a detected event to its public-safe notification content.
 export function buildNotificationContent(event: DetectedNotificationEvent): { title: string; body: string } {
-  return event.eventType === "pull_request_merged" ? buildMergedOutcomeNotification(event) : buildChangesRequestedNotification(event);
+  switch (event.eventType) {
+    case "pull_request_merged":
+      return buildMergedOutcomeNotification(event);
+    case "issue_watch_match":
+      return buildIssueWatchNotification(event);
+    default:
+      return buildChangesRequestedNotification(event);
+  }
+}
+
+/**
+ * #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
+ * (reads the repo's watchers), so it lives here rather than in the pure payload-only detectNotificationEvents.
+ */
+export async function detectIssueWatchEvents(env: Env, repoFullName: string, issue: IssueRecord): Promise<DetectedNotificationEvent[]> {
+  if (!isGrabbableHighMultiplierIssue(issue)) return [];
+  const watchers = await listIssueWatchersForRepo(env, repoFullName);
+  if (watchers.length === 0) return [];
+  const detectedAt = nowIso();
+  const issueLabels = new Set(issue.labels.map((label) => label.toLowerCase().trim()));
+  const authorLogin = issue.authorLogin?.toLowerCase();
+  return watchers
+    // An empty label filter matches any issue; otherwise at least one watched label must be present.
+    .filter((watcher) => watcher.labels.length === 0 || watcher.labels.some((label) => issueLabels.has(label)))
+    // Don't ping the maintainer who opened the issue about their own issue.
+    .filter((watcher) => watcher.login.toLowerCase() !== authorLogin)
+    .map((watcher) => ({
+      eventType: "issue_watch_match" as const,
+      recipientLogin: watcher.login,
+      repoFullName,
+      pullNumber: issue.number, // carries the ISSUE number for this eventType
+      dedupKey: `issue_watch_match:${repoFullName}#${issue.number}:${watcher.login.toLowerCase()}`,
+      deeplink: `https://github.com/${repoFullName}/issues/${issue.number}`,
+      actorLogin: issue.authorLogin ?? "unknown",
+      detectedAt,
+    }));
 }
 
 function rateLimitWindowStart(now: string): string {

src/queue/processors.ts

@@ -82,7 +82,7 @@ import { fetchPublicContributorProfile } from "../github/public";
 import { refreshRegistry } from "../registry/sync";
 import { buildIssueAdvisory, buildPullRequestAdvisory, evaluateGateCheck } from "../rules/advisory";
 import { detectNotificationEvents } from "../notifications/events";
-import { deliverNotification, evaluateNotificationEvent } from "../notifications/service";
+import { deliverNotification, detectIssueWatchEvents, evaluateNotificationEvent } from "../notifications/service";
 import { getOrCreateScoringModelSnapshot, refreshScoringModelSnapshot } from "../scoring/model";
 import { buildAndPersistContributorDecisionPack, loadDecisionPackSharedInputs } from "../services/decision-pack";
 import {
@@ -139,7 +139,7 @@ import { loadRepoFocusManifest } from "../signals/focus-manifest-loader";
 import { resolveEffectiveSettings } from "../signals/focus-manifest";
 import type { LocalBranchAnalysisInput } from "../signals/local-branch";
 import { runGittensoryAiReview } from "../services/ai-review";
-import type { AdvisoryFinding, ContributorEvidenceRecord, GitHubWebhookPayload, JobMessage, JsonValue, PullRequestRecord, RepositorySettings } from "../types";
+import type { AdvisoryFinding, ContributorEvidenceRecord, DetectedNotificationEvent, GitHubWebhookPayload, JobMessage, JsonValue, PullRequestRecord, RepositorySettings } from "../types";
 import { sha256Hex } from "../utils/crypto";
 import { errorMessage, nowIso } from "../utils/json";
 
@@ -759,6 +759,7 @@ async function processGitHubWebhook(env: Env, deliveryId: string, eventName: str
       }
     }
 
+    let issueWatchEvents: DetectedNotificationEvent[] = [];
     if (payload.repository?.full_name && payload.issue && !payload.issue.pull_request) {
       const issue = await upsertIssueFromGitHub(env, payload.repository.full_name, payload.issue);
       const repo = await getRepository(env, payload.repository.full_name);
@@ -770,9 +771,12 @@ async function processGitHubWebhook(env: Env, deliveryId: string, eventName: str
         advisory.findings.push(...buildIssueSlopAssessment({ title: issue.title, body: issue.body }).findings);
       }
       await persistAdvisory(env, advisory);
+      // #699 path B: a newly opened grabbable, high-multiplier issue notifies the miners watching this repo
+      // (fanned out through the same #535 pipeline below).
+      if (payload.action === "opened") issueWatchEvents = await detectIssueWatchEvents(env, payload.repository.full_name, issue);
     }
 
-    for (const notificationEvent of detectNotificationEvents(eventName, payload)) {
+    for (const notificationEvent of [...detectNotificationEvents(eventName, payload), ...issueWatchEvents]) {
       await recordAuditEvent(env, {
         eventType: "notification.event_detected",
         actor: notificationEvent.actorLogin,

src/signals/engine.ts

@@ -214,6 +214,15 @@ function isMaintainerWipIssue(issue: IssueRecord): boolean {
   return isMaintainerAssociation(issue.authorAssociation) && issue.labels.some((label) => MAINTAINER_WIP_LABELS.has(label.toLowerCase().trim()));
 }
 
+/**
+ * True iff an issue is the highest-multiplier, immediately-grabbable target (#699): open, maintainer-created
+ * (the biggest reward multiplier), and NOT flagged as the maintainer's own WIP/internal work. This is the
+ * exact condition the issue-watch monitor (#699 path B) notifies subscribers about.
+ */
+export function isGrabbableHighMultiplierIssue(issue: IssueRecord): boolean {
+  return issue.state === "open" && isMaintainerAssociation(issue.authorAssociation) && !isMaintainerWipIssue(issue);
+}
+
 export type ContributorFit = {
   login: string;
   generatedAt: string;

src/types.ts

@@ -1120,7 +1120,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";
+export type NotificationEventType = "pull_request_changes_requested" | "pull_request_merged" | "issue_watch_match";
+
+/** #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. */
+export type IssueWatchSubscription = {
+  login: string;
+  repoFullName: string;
+  labels: string[];
+  createdAt?: string | null | undefined;
+  updatedAt?: string | null | undefined;
+};
 
 // A notification-worthy event extracted from a webhook payload (src/notifications/events.ts).
 export type DetectedNotificationEvent = {