feat(github): redeliver failed webhooks

apps/docs/docs/guides/runners.md

@@ -44,7 +44,7 @@ A runner operation that succeeded means the VM ran and was cleaned up. It does n
 - **App:** open a connection in **Repositories** to see its recent jobs.
 - **CLI:** `vectis job list <binding-id> --json` returns the latest 100 jobs with their GitHub status and conclusion.
 
-Job states arrive through GitHub webhooks. If one seems missing or stale:
+Job states arrive through GitHub webhooks. Vectis checks for failed deliveries every two minutes and requests one automatic redelivery per webhook, usually recovering missed updates within a few minutes without a connected Mac. If one still seems missing or stale:
 
 ```sh
 vectis job scan <binding-id> --wait             # find queued and running jobs Vectis missed

convex/_generated/api.d.ts

@@ -28,6 +28,8 @@ import type * as githubIdentity from "../githubIdentity.js";
 import type * as githubJobs from "../githubJobs.js";
 import type * as githubMigrations from "../githubMigrations.js";
 import type * as githubOAuth from "../githubOAuth.js";
+import type * as githubRedelivery from "../githubRedelivery.js";
+import type * as githubRedeliveryState from "../githubRedeliveryState.js";
 import type * as githubRepositories from "../githubRepositories.js";
 import type * as githubRunnerProbe from "../githubRunnerProbe.js";
 import type * as githubRunners from "../githubRunners.js";
@@ -78,6 +80,8 @@ declare const fullApi: ApiFromModules<{
   githubJobs: typeof githubJobs;
   githubMigrations: typeof githubMigrations;
   githubOAuth: typeof githubOAuth;
+  githubRedelivery: typeof githubRedelivery;
+  githubRedeliveryState: typeof githubRedeliveryState;
   githubRepositories: typeof githubRepositories;
   githubRunnerProbe: typeof githubRunnerProbe;
   githubRunners: typeof githubRunners;

convex/crons.ts

@@ -2,5 +2,6 @@ import { cronJobs } from "convex/server";
 import { internal } from "./_generated/api.js";
 
 const crons = cronJobs();
+crons.interval("GitHub webhook redelivery", { minutes: 2 }, internal.githubRedelivery.scan, {});
 crons.daily("retention sweep", { hourUTC: 3, minuteUTC: 0 }, internal.retention.sweep, {});
 export default crons;

convex/githubRedelivery.ts

@@ -0,0 +1,144 @@
+"use node";
+import { Schema } from "effect";
+import { createGitHubAppJwt, GitHubApiError } from "../packages/github/src/app-client.js";
+import { internal } from "./_generated/api.js";
+import { internalAction } from "./_generated/server.js";
+
+const minute = 60_000;
+const maxAge = 3 * 24 * 60 * minute;
+const events = new Set(["workflow_job", "installation", "installation_repositories", "repository"]);
+const deliveries = Schema.Array(
+  Schema.Struct({
+    id: Schema.String.check(Schema.isPattern(/^\d{1,30}$/)),
+    guid: Schema.NonEmptyString,
+    delivered_at: Schema.String,
+    status_code: Schema.NullOr(Schema.Int),
+    event: Schema.String,
+  }),
+);
+
+function nextCursor(link: string | null) {
+  if (!link) return null;
+  for (const part of link.split(",")) {
+    const match = part.match(/<([^>]+)>;\s*rel="next"/);
+    const url = match?.[1];
+    if (url) {
+      const cursor = new URL(url).searchParams.get("cursor");
+      if (!cursor) throw new Error("GitHub delivery pagination is missing its cursor.");
+      return cursor;
+    }
+  }
+  return null;
+}
+
+export const scan = internalAction({
+  args: {},
+  handler: async (ctx): Promise<void> => {
+    const state = await ctx.runQuery(internal.githubRedeliveryState.load, {});
+    if (!state) return;
+    const now = Date.now();
+    const since = Math.max(state.checkpoint?.deliveredAt ?? now - 60 * minute, now - maxAge);
+    const through = now - minute;
+    const deadline = now + 90_000;
+    const signal = AbortSignal.timeout(90_000);
+    const headers = {
+      Authorization: `Bearer ${createGitHubAppJwt(state.app)}`,
+      Accept: "application/vnd.github+json",
+      "X-GitHub-Api-Version": "2026-03-10",
+    };
+    const request = async (path: string, method: "GET" | "POST") => {
+      const response = await fetch(`https://api.github.com${path}`, {
+        method,
+        headers,
+        redirect: "error",
+        signal: AbortSignal.any([signal, AbortSignal.timeout(10_000)]),
+      });
+      if (!response.ok) {
+        await response.body?.cancel();
+        throw new GitHubApiError(response.status);
+      }
+      return response;
+    };
+
+    const attempts = new Map<string, { id: string; deliveredAt: number; successful: boolean }>();
+    let complete = false;
+    let cursor: string | null = null;
+    try {
+      for (let page = 0; page < 10; page++) {
+        const params = new URLSearchParams({ per_page: "100" });
+        if (cursor) params.set("cursor", cursor);
+        const response = await request(`/app/hook/deliveries?${params}`, "GET");
+        // Delivery IDs exceed Number.MAX_SAFE_INTEGER, so they must stay strings.
+        const text = (await response.text()).replace(/"id":\s*(\d+)/g, '"id":"$1"');
+        const rows = Schema.decodeUnknownSync(deliveries)(JSON.parse(text));
+        if (rows.length > 100) throw new Error("GitHub exceeded the delivery page limit.");
+        for (const row of rows) {
+          const deliveredAt = Date.parse(row.delivered_at);
+          if (!Number.isFinite(deliveredAt)) throw new Error("Invalid GitHub delivery timestamp.");
+          if (deliveredAt < since) {
+            complete = true;
+            break;
+          }
+          if (!events.has(row.event)) continue;
+          const successful =
+            row.status_code !== null && row.status_code >= 200 && row.status_code < 300;
+          const existing = attempts.get(row.guid);
+          if (!existing || deliveredAt > existing.deliveredAt)
+            attempts.set(row.guid, {
+              id: row.id,
+              deliveredAt,
+              successful: successful || existing?.successful === true,
+            });
+          else existing.successful ||= successful;
+        }
+        cursor = nextCursor(response.headers.get("link"));
+        if (complete || !cursor || rows.length === 0) {
+          complete = true;
+          break;
+        }
+      }
+    } catch (error) {
+      console.warn("GitHub webhook delivery listing failed; checkpoint unchanged.", String(error));
+      return;
+    }
+    if (!complete) {
+      console.warn("GitHub webhook delivery page limit reached; checkpoint unchanged.");
+      return;
+    }
+
+    const candidates = [...attempts.entries()]
+      .filter(([, attempt]) => !attempt.successful && attempt.deliveredAt <= through)
+      .sort((a, b) => a[1].deliveredAt - b[1].deliveredAt);
+    let requested = 0;
+    for (const [guid, attempt] of candidates) {
+      if (requested >= 20 || Date.now() >= deadline || signal.aborted) {
+        complete = false;
+        break;
+      }
+      if (!(await ctx.runMutation(internal.githubRedeliveryState.claim, { guid }))) continue;
+      requested++;
+      try {
+        const response = await request(`/app/hook/deliveries/${attempt.id}/attempts`, "POST");
+        await response.body?.cancel();
+      } catch (error) {
+        console.warn("GitHub webhook redelivery request failed; it will not be repeated.", {
+          guid,
+          deliveryId: attempt.id,
+          error,
+        });
+        complete = false;
+        break;
+      }
+    }
+    if (!complete) {
+      console.warn("GitHub webhook redelivery scan incomplete; checkpoint unchanged.");
+      return;
+    }
+    // Fresh attempts remain in the next scan, even when this run saw newer successful deliveries.
+    await ctx.runMutation(internal.githubRedeliveryState.advance, {
+      id: state.checkpoint?._id ?? null,
+      deliveredAt: through,
+      scannedAt: now,
+    });
+  },
+});

convex/githubRedeliveryState.ts

@@ -0,0 +1,47 @@
+import { v } from "convex/values";
+import { internalMutation, internalQuery } from "./_generated/server.js";
+
+export const load = internalQuery({
+  args: {},
+  handler: async (ctx) => {
+    const app = await ctx.db.query("githubApps").first();
+    if (!app) return null;
+    return {
+      app: { clientId: app.clientId, privateKey: app.privateKey },
+      checkpoint: await ctx.db.query("githubRedeliveryCheckpoint").first(),
+    };
+  },
+});
+
+export const advance = internalMutation({
+  args: {
+    id: v.union(v.id("githubRedeliveryCheckpoint"), v.null()),
+    deliveredAt: v.number(),
+    scannedAt: v.number(),
+  },
+  handler: async (ctx, { id, ...checkpoint }) => {
+    if (id) {
+      await ctx.db.patch("githubRedeliveryCheckpoint", id, checkpoint);
+      return;
+    }
+    const existing = await ctx.db.query("githubRedeliveryCheckpoint").first();
+    if (existing) {
+      if (existing.deliveredAt < checkpoint.deliveredAt)
+        await ctx.db.patch("githubRedeliveryCheckpoint", existing._id, checkpoint);
+    } else await ctx.db.insert("githubRedeliveryCheckpoint", checkpoint);
+  },
+});
+
+export const claim = internalMutation({
+  args: { guid: v.string() },
+  handler: async (ctx, { guid }) => {
+    const existing = await ctx.db
+      .query("githubRedeliveryRequests")
+      .withIndex("by_guid", (q) => q.eq("guid", guid))
+      .unique();
+    if (existing) return false;
+    // Reserve before POST: a timeout may still mean GitHub accepted the request.
+    await ctx.db.insert("githubRedeliveryRequests", { guid, requestedAt: Date.now() });
+    return true;
+  },
+});

convex/retention.ts

@@ -6,6 +6,7 @@ import { internalMutation, type MutationCtx } from "./_generated/server.js";
 const day = 24 * 60 * 60 * 1000;
 export const windows = {
   deliveries: 30 * day,
+  redeliveryRequests: 3 * day,
   jobs: 90 * day,
   operations: 90 * day,
   leases: 30 * day,
@@ -32,6 +33,14 @@ async function sweepTable<T extends { _id: { __tableName: string } }>(
 
 export async function sweepOnce(ctx: MutationCtx, now: number) {
   let deleted = 0;
+  deleted += await sweepTable(
+    await ctx.db
+      .query("githubRedeliveryRequests")
+      .withIndex("by_requested", (q) => q.lt("requestedAt", now - windows.redeliveryRequests))
+      .take(limit),
+    () => false,
+    async (row) => ctx.db.delete("githubRedeliveryRequests", row._id),
+  );
   deleted += await sweepTable(
     await ctx.db
       .query("githubDeliveries")

convex/schema.ts

@@ -229,6 +229,16 @@ export default defineSchema({
   })
     .index("by_delivery", ["deliveryId"])
     .index("by_received", ["receivedAt"]),
+  githubRedeliveryCheckpoint: defineTable({
+    deliveredAt: v.number(),
+    scannedAt: v.number(),
+  }),
+  githubRedeliveryRequests: defineTable({
+    guid: v.string(),
+    requestedAt: v.number(),
+  })
+    .index("by_guid", ["guid"])
+    .index("by_requested", ["requestedAt"]),
   githubAppSetups: defineTable({
     stateDigest: v.string(),
     ownerId: v.number(),

packages/github/src/app-client.ts

@@ -92,13 +92,16 @@ export class GitHubPermissionError extends GitHubApiError {
     this.message = "The Vectis GitHub App needs updated Actions write permission.";
   }
 }
+export function createGitHubAppJwt(app: { clientId: string; privateKey: string }) {
+  const now = Math.floor(Date.now() / 1000);
+  const encode = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url");
+  const data = `${encode({ alg: "RS256", typ: "JWT" })}.${encode({ iss: app.clientId, iat: now - 60, exp: now + 300 })}`;
+  return `${data}.${sign("RSA-SHA256", Buffer.from(data), createPrivateKey(app.privateKey)).toString("base64url")}`;
+}
 export class GitHubAppClient {
   constructor(private readonly app: { appId: number; clientId: string; privateKey: string }) {}
   private jwt() {
-    const now = Math.floor(Date.now() / 1000);
-    const encode = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url");
-    const data = `${encode({ alg: "RS256", typ: "JWT" })}.${encode({ iss: this.app.clientId, iat: now - 60, exp: now + 300 })}`;
-    return `${data}.${sign("RSA-SHA256", Buffer.from(data), createPrivateKey(this.app.privateKey)).toString("base64url")}`;
+    return createGitHubAppJwt(this.app);
   }
   async request(
     token: string,