From 3a54e05e48740cb29c752381649bd6ec6359b5bf Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:27 +0800
Subject: [PATCH 1/8] feat(api): typed GitHubRateLimitError for 403/429 signals
 (#1176)

resolveRefHeadSha now classifies a GitHub rate-limit response (429, or 403 with X-RateLimit-Remaining:0 / a Retry-After header) into a typed GitHubRateLimitError carrying the recommended wait (from Retry-After or X-RateLimit-Reset). Other non-OK statuses stay generic 'transient' errors. This lets the batch drift scheduler back off cleanly instead of hammering a limited API.

Part of #1176
---
 .../skills/crud/utils/githubPull.test.ts      | 45 ++++++++++++++++
 .../domains/skills/crud/utils/githubPull.ts   | 53 +++++++++++++++++--
 2 files changed, 95 insertions(+), 3 deletions(-)

diff --git a/ornn-api/src/domains/skills/crud/utils/githubPull.test.ts b/ornn-api/src/domains/skills/crud/utils/githubPull.test.ts
index 21c1624b..92c11d74 100644
--- a/ornn-api/src/domains/skills/crud/utils/githubPull.test.ts
+++ b/ornn-api/src/domains/skills/crud/utils/githubPull.test.ts
@@ -8,6 +8,7 @@ import {
   authHeaders,
   resolveRefHeadSha,
   GitHubSourceNotFoundError,
+  GitHubRateLimitError,
 } from "./githubPull";
 
 /** Records each request's URL + Authorization header, returns caller-chosen responses. */
@@ -444,4 +445,48 @@ describe("resolveRefHeadSha", () => {
     await resolveRefHeadSha("acme/x", "feature/foo", {}, impl);
     expect(calls[0]!.url).toContain("/git/ref/heads/feature/foo");
   });
+
+  test("429 → GitHubRateLimitError with Retry-After", async () => {
+    const { impl } = recordingFetch(
+      () => new Response("slow down", { status: 429, headers: { "retry-after": "30" } }),
+    );
+    let err: unknown;
+    try {
+      await resolveRefHeadSha("acme/x", "main", {}, impl);
+    } catch (e) {
+      err = e;
+    }
+    expect(err).toBeInstanceOf(GitHubRateLimitError);
+    expect((err as GitHubRateLimitError).retryAfterMs).toBe(30_000);
+  });
+
+  test("403 with X-RateLimit-Remaining: 0 → GitHubRateLimitError", async () => {
+    const { impl } = recordingFetch(
+      () =>
+        new Response("forbidden", {
+          status: 403,
+          headers: { "x-ratelimit-remaining": "0" },
+        }),
+    );
+    await expect(resolveRefHeadSha("acme/x", "main", {}, impl)).rejects.toBeInstanceOf(
+      GitHubRateLimitError,
+    );
+  });
+
+  test("403 WITHOUT rate-limit signals → generic error (not rate-limit)", async () => {
+    const { impl } = recordingFetch(() => new Response("nope", { status: 403 }));
+    let err: unknown;
+    try {
+      await resolveRefHeadSha("acme/x", "main", {}, impl);
+    } catch (e) {
+      err = e;
+    }
+    expect(err).toBeInstanceOf(Error);
+    expect(err).not.toBeInstanceOf(GitHubRateLimitError);
+  });
+
+  test("500 → generic transient error", async () => {
+    const { impl } = recordingFetch(() => new Response("boom", { status: 500 }));
+    await expect(resolveRefHeadSha("acme/x", "main", {}, impl)).rejects.toThrow(/500/);
+  });
 });
diff --git a/ornn-api/src/domains/skills/crud/utils/githubPull.ts b/ornn-api/src/domains/skills/crud/utils/githubPull.ts
index 30bcae2e..17aa2934 100644
--- a/ornn-api/src/domains/skills/crud/utils/githubPull.ts
+++ b/ornn-api/src/domains/skills/crud/utils/githubPull.ts
@@ -61,6 +61,53 @@ export class GitHubSourceNotFoundError extends Error {
   }
 }
 
+/**
+ * Thrown when GitHub signals a primary or secondary rate limit (`429`, or
+ * `403` with `X-RateLimit-Remaining: 0` / a `Retry-After` header). Carries
+ * the recommended wait so a batch caller (the drift scheduler, #1176) can
+ * short-circuit the rest of its tick and let the next fire retry.
+ */
+export class GitHubRateLimitError extends Error {
+  constructor(
+    public readonly status: number,
+    public readonly retryAfterMs: number | undefined,
+  ) {
+    super(`GitHub rate limit hit (status ${status})`);
+    this.name = "GitHubRateLimitError";
+  }
+}
+
+/**
+ * Classify a non-OK GitHub response: a rate-limit signal becomes a typed
+ * {@link GitHubRateLimitError} (with the wait derived from `Retry-After` or
+ * `X-RateLimit-Reset`); everything else becomes a generic error the caller
+ * treats as transient. `nowMs` is injected for deterministic tests.
+ */
+function rateLimitOrGeneric(
+  res: Response,
+  context: string,
+  nowMs: number,
+): Error {
+  const retryAfter = res.headers.get("retry-after");
+  const remaining = res.headers.get("x-ratelimit-remaining");
+  const reset = res.headers.get("x-ratelimit-reset");
+  const isRateLimit =
+    res.status === 429 ||
+    (res.status === 403 && (remaining === "0" || retryAfter !== null));
+  if (!isRateLimit) {
+    return new Error(`GitHub API returned ${res.status} for ${context}`);
+  }
+  let retryAfterMs: number | undefined;
+  if (retryAfter !== null) {
+    const secs = Number(retryAfter);
+    if (Number.isFinite(secs)) retryAfterMs = Math.max(0, secs * 1000);
+  } else if (reset !== null) {
+    const resetMs = Number(reset) * 1000;
+    if (Number.isFinite(resetMs)) retryAfterMs = Math.max(0, resetMs - nowMs);
+  }
+  return new GitHubRateLimitError(res.status, retryAfterMs);
+}
+
 export interface GitHubPullInput {
   /** `owner/name`. */
   readonly repo: string;
@@ -359,9 +406,9 @@ export async function resolveRefHeadSha(
     if (res.status === 304) return { notModified: true };
     if (res.status === 404) return "not-found";
     if (!res.ok) {
-      throw new Error(
-        `GitHub git/ref API returned ${res.status} for ${normalizedRepo}@${trimmedRef}`,
-      );
+      // A 403/429 rate-limit signal becomes a typed error so the batch
+      // caller can back off; anything else is treated as transient.
+      throw rateLimitOrGeneric(res, `${normalizedRepo}@${trimmedRef}`, Date.now());
     }
     const body = (await res.json()) as { object?: { sha?: string } };
     const sha = body.object?.sha;

From 17c8ff7e10be947ea3dc219625e83ccd8454b896 Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:27 +0800
Subject: [PATCH 2/8] refactor(api): share classifyProbeResult +
 pickSourceSyncToken (#1176)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Extract the probe-result-to-verdict mapping (classifyProbeResult) and the settings>env>anonymous token resolution (pickSourceSyncToken) out of the single-skill drift path so the upcoming batch job can reuse them verbatim — the batch probes once per (repo,ref) group and classifies each member against its own lastSyncedCommit. No behavior change: runSourceDriftCheck and SkillService.resolveSourceSyncToken now delegate to the shared helpers.

Part of #1176
---
 ornn-api/src/domains/skills/crud/service.ts   | 11 ++-
 .../domains/skills/crud/sourceDrift.test.ts   | 56 ++++++++++++-
 .../src/domains/skills/crud/sourceDrift.ts    | 80 ++++++++++++++-----
 3 files changed, 120 insertions(+), 27 deletions(-)

diff --git a/ornn-api/src/domains/skills/crud/service.ts b/ornn-api/src/domains/skills/crud/service.ts
index 61aa42ec..6cfb8fbf 100644
--- a/ornn-api/src/domains/skills/crud/service.ts
+++ b/ornn-api/src/domains/skills/crud/service.ts
@@ -12,7 +12,11 @@ import type { IStorageClient } from "../../../clients/storageClient";
 import type { SkillDocument, SkillMetadata, SkillDetailResponse, SkillVersionDocument, SkillSource } from "../../../shared/types/index";
 import { AppError } from "../../../shared/types/index";
 import { fetchSkillFromGitHub, parseGithubUrl, type GitHubPullInput } from "./utils/githubPull";
-import { runSourceDriftCheck, type SourceDriftResult } from "./sourceDrift";
+import {
+  runSourceDriftCheck,
+  pickSourceSyncToken,
+  type SourceDriftResult,
+} from "./sourceDrift";
 import type { SourceSyncSection } from "../../settings/sections/sourceSync";
 import { computeVersionDiff, type VersionDiffResult } from "./utils/versionDiff";
 import { isReservedVerb } from "../../../shared/reservedVerbs";
@@ -212,9 +216,8 @@ export class SkillService {
    * bogus `Authorization` header. NEVER logged.
    */
   private async resolveSourceSyncToken(): Promise<string> {
-    const configured = (await this.sourceSyncSettings?.getSourceSync())?.githubToken?.trim();
-    if (configured) return configured;
-    return this.sourceSyncGithubTokenFallback?.trim() ?? "";
+    const settingsToken = (await this.sourceSyncSettings?.getSourceSync())?.githubToken;
+    return pickSourceSyncToken(settingsToken, this.sourceSyncGithubTokenFallback);
   }
 
   /**
diff --git a/ornn-api/src/domains/skills/crud/sourceDrift.test.ts b/ornn-api/src/domains/skills/crud/sourceDrift.test.ts
index 0041d403..67c8a726 100644
--- a/ornn-api/src/domains/skills/crud/sourceDrift.test.ts
+++ b/ornn-api/src/domains/skills/crud/sourceDrift.test.ts
@@ -1,5 +1,10 @@
 import { describe, expect, test } from "bun:test";
-import { runSourceDriftCheck, type SourceDriftDeps } from "./sourceDrift";
+import {
+  runSourceDriftCheck,
+  classifyProbeResult,
+  pickSourceSyncToken,
+  type SourceDriftDeps,
+} from "./sourceDrift";
 import {
   GitHubSourceNotFoundError,
   type RefHeadProbeResult,
@@ -135,3 +140,52 @@ describe("runSourceDriftCheck", () => {
     expect(res.applicable).toBe(false);
   });
 });
+
+describe("classifyProbeResult", () => {
+  test("304 notModified → in_sync, no upstreamHeadSha", () => {
+    const { driftState, patch } = classifyProbeResult(
+      { lastSyncedCommit: "x" },
+      { notModified: true },
+    );
+    expect(driftState).toBe("in_sync");
+    expect(patch).toEqual({ driftState: "in_sync" });
+  });
+
+  test("HEAD == lastSyncedCommit → in_sync with sha + etag", () => {
+    const { driftState, patch } = classifyProbeResult(
+      { lastSyncedCommit: "same" },
+      { sha: "same", etag: 'W/"e"', notModified: false },
+    );
+    expect(driftState).toBe("in_sync");
+    expect(patch).toEqual({ driftState: "in_sync", upstreamHeadSha: "same", etag: 'W/"e"' });
+  });
+
+  test("HEAD != lastSyncedCommit → drifted", () => {
+    const { driftState, patch } = classifyProbeResult(
+      { lastSyncedCommit: "old" },
+      { sha: "new", notModified: false },
+    );
+    expect(driftState).toBe("drifted");
+    expect(patch.upstreamHeadSha).toBe("new");
+    expect(patch.etag).toBeUndefined();
+  });
+});
+
+describe("pickSourceSyncToken", () => {
+  test("settings token wins", () => {
+    expect(pickSourceSyncToken("settings", "env")).toBe("settings");
+  });
+  test("empty/whitespace settings → env fallback", () => {
+    expect(pickSourceSyncToken("", "env")).toBe("env");
+    expect(pickSourceSyncToken("   ", "env")).toBe("env");
+    expect(pickSourceSyncToken(undefined, "env")).toBe("env");
+  });
+  test("neither → empty string (anonymous)", () => {
+    expect(pickSourceSyncToken(undefined, undefined)).toBe("");
+    expect(pickSourceSyncToken("", "")).toBe("");
+  });
+  test("trims the chosen token", () => {
+    expect(pickSourceSyncToken("  tok  ", undefined)).toBe("tok");
+    expect(pickSourceSyncToken("", "  env  ")).toBe("env");
+  });
+});
diff --git a/ornn-api/src/domains/skills/crud/sourceDrift.ts b/ornn-api/src/domains/skills/crud/sourceDrift.ts
index 1c2d23f7..20da60a2 100644
--- a/ornn-api/src/domains/skills/crud/sourceDrift.ts
+++ b/ornn-api/src/domains/skills/crud/sourceDrift.ts
@@ -32,6 +32,56 @@ export interface SourceDriftResult {
   readonly upstreamHeadSha?: string;
 }
 
+/** Persist patch (sans `lastCheckedAt`, which the caller stamps). */
+export interface DriftPatch {
+  readonly driftState: SkillSourceDriftState;
+  readonly upstreamHeadSha?: string;
+  readonly etag?: string;
+}
+
+/**
+ * Turn a probe result + a skill's last-synced commit into a drift verdict
+ * and the fields to persist. Pure — shared by the single-skill check and the
+ * batch scheduler (#1176), which probes once per `(repo, ref)` group and
+ * classifies each member against its own `lastSyncedCommit`.
+ *
+ * A `304` (nothing changed since the stored ETag) is `in_sync`; a live HEAD
+ * equal to `lastSyncedCommit` is `in_sync`; anything else is `drifted`.
+ */
+export function classifyProbeResult(
+  source: { lastSyncedCommit?: string | undefined },
+  result: RefHeadProbeResult,
+): { driftState: SkillSourceDriftState; patch: DriftPatch } {
+  if (result.notModified) {
+    return { driftState: "in_sync", patch: { driftState: "in_sync" } };
+  }
+  const sha = result.sha!;
+  const driftState: SkillSourceDriftState =
+    sha === source.lastSyncedCommit ? "in_sync" : "drifted";
+  return {
+    driftState,
+    patch: {
+      driftState,
+      upstreamHeadSha: sha,
+      ...(result.etag ? { etag: result.etag } : {}),
+    },
+  };
+}
+
+/**
+ * Resolve the effective GitHub token: an admin-set settings value wins, then
+ * the env fallback, else `""` (anonymous). Trimmed so stray whitespace never
+ * becomes a bogus bearer. Pure — shared by SkillService and the scheduler.
+ */
+export function pickSourceSyncToken(
+  settingsToken: string | undefined,
+  envFallback: string | undefined,
+): string {
+  const configured = settingsToken?.trim();
+  if (configured) return configured;
+  return envFallback?.trim() ?? "";
+}
+
 export interface SourceDriftDeps {
   readonly skillRepo: Pick<
     SkillRepository,
@@ -80,31 +130,17 @@ export async function runSourceDriftCheck(
       etag: source.etag,
     });
 
-    // 304 — nothing changed since the stored ETag. Free on authenticated
-    // requests; just record that we looked.
-    if (result.notModified) {
-      await deps.skillRepo.updateSourceDriftState(guid, {
-        driftState: "in_sync",
-        lastCheckedAt: now,
-      });
-      logger.debug({ guid, repo: source.repo }, "source drift check: 304 not modified");
-      return { applicable: true, driftState: "in_sync" };
-    }
-
-    const sha = result.sha!;
-    const drifted = sha !== source.lastSyncedCommit;
-    const driftState: SkillSourceDriftState = drifted ? "drifted" : "in_sync";
-    await deps.skillRepo.updateSourceDriftState(guid, {
-      driftState,
-      upstreamHeadSha: sha,
-      ...(result.etag ? { etag: result.etag } : {}),
-      lastCheckedAt: now,
-    });
+    const { driftState, patch } = classifyProbeResult(source, result);
+    await deps.skillRepo.updateSourceDriftState(guid, { ...patch, lastCheckedAt: now });
     logger.info(
-      { guid, repo: source.repo, ref: source.ref, driftState, upstreamHeadSha: sha },
+      { guid, repo: source.repo, ref: source.ref, driftState, upstreamHeadSha: patch.upstreamHeadSha },
       "source drift check complete",
     );
-    return { applicable: true, driftState, upstreamHeadSha: sha };
+    return {
+      applicable: true,
+      driftState,
+      ...(patch.upstreamHeadSha ? { upstreamHeadSha: patch.upstreamHeadSha } : {}),
+    };
   } catch (err) {
     // A genuinely missing repo/ref is a terminal, per-skill state — record
     // it and return. Transient failures (network, 5xx) are re-thrown so the

From e9f6d08c4e19501d2c6bfb82ce9528692e6c1b94 Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:28 +0800
Subject: [PATCH 3/8] feat(api): source-drift repository query, setter, and
 index (#1176)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add findGithubSourcedSkills({ notCheckedSince }) — enumerates github-sourced skills due for a check (excluding pinned 40-hex refs via a $not-regex, and skills checked more recently than the cutoff), returning light { guid, source, ownerId } rows. Back it with a partial index over github sources ordered by lastCheckedAt. Extract the source-coercion out of mapDoc into a shared coerceSkillSource helper so both read paths stay consistent.

Part of #1176
---
 .../crud/repository.sourceDrift.test.ts       | 133 ++++++++++++++++++
 .../src/domains/skills/crud/repository.ts     | 122 +++++++++++-----
 2 files changed, 218 insertions(+), 37 deletions(-)
 create mode 100644 ornn-api/src/domains/skills/crud/repository.sourceDrift.test.ts

diff --git a/ornn-api/src/domains/skills/crud/repository.sourceDrift.test.ts b/ornn-api/src/domains/skills/crud/repository.sourceDrift.test.ts
new file mode 100644
index 00000000..56a8b983
--- /dev/null
+++ b/ornn-api/src/domains/skills/crud/repository.sourceDrift.test.ts
@@ -0,0 +1,133 @@
+/**
+ * Repository tests for the source-drift query + setter (#1176), against a
+ * real in-memory MongoDB so the `$not`-regex query, the partial index, and
+ * the dot-path `$set` are exercised for real.
+ *
+ * @module domains/skills/crud/repository.sourceDrift.test
+ */
+import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test";
+import { MongoClient, type Db } from "mongodb";
+import { MongoMemoryServer } from "mongodb-memory-server";
+import { SkillRepository } from "./repository";
+
+let mongo: MongoMemoryServer;
+let client: MongoClient;
+let db: Db;
+let repo: SkillRepository;
+
+beforeAll(async () => {
+  mongo = await MongoMemoryServer.create();
+  client = new MongoClient(mongo.getUri());
+  await client.connect();
+  db = client.db("source_drift_test");
+  repo = new SkillRepository(db);
+  await repo.ensureIndexes();
+});
+
+afterAll(async () => {
+  await client.close();
+  await mongo.stop();
+});
+
+beforeEach(async () => {
+  await db.collection("skills").deleteMany({});
+});
+
+function skillDoc(overrides: Record<string, unknown>): Record<string, unknown> {
+  const now = new Date();
+  return {
+    name: `s-${overrides._id}`,
+    description: "d",
+    metadata: { category: "plain" },
+    skillHash: "h",
+    storageKey: "k",
+    createdBy: "owner",
+    createdOn: now,
+    updatedBy: "owner",
+    updatedOn: now,
+    isPrivate: true,
+    sharedWithUsers: [],
+    sharedWithOrgs: [],
+    latestVersion: "1.0",
+    ...overrides,
+  };
+}
+
+async function seed(...docs: Array<Record<string, unknown>>): Promise<void> {
+  await db.collection("skills").insertMany(docs.map(skillDoc) as never);
+}
+
+describe("findGithubSourcedSkills", () => {
+  test("selects due github skills only; excludes fresh, pinned, and source-less", async () => {
+    const threeHoursAgo = new Date(Date.now() - 3 * 60 * 60 * 1000);
+    await seed(
+      // github, never checked → included
+      { _id: "g1", createdBy: "owner-1", source: { type: "github", repo: "a/x", ref: "main", path: "" } },
+      // github, checked long ago → included
+      {
+        _id: "g2",
+        createdBy: "owner-2",
+        source: { type: "github", repo: "a/x", ref: "main", path: "", lastCheckedAt: threeHoursAgo },
+      },
+      // github, checked just now → excluded (fresh)
+      {
+        _id: "g3",
+        createdBy: "owner-3",
+        source: { type: "github", repo: "b/y", ref: "main", path: "", lastCheckedAt: new Date() },
+      },
+      // pinned 40-hex ref → excluded (never drifts)
+      { _id: "g4", createdBy: "owner-4", source: { type: "github", repo: "c/z", ref: "a".repeat(40), path: "" } },
+      // hand-uploaded (no source) → excluded
+      { _id: "g5", createdBy: "owner-5" },
+    );
+
+    const cutoff = new Date(Date.now() - 60 * 60 * 1000); // 1h ago
+    const due = await repo.findGithubSourcedSkills({ notCheckedSince: cutoff });
+
+    expect(due.map((d) => d.guid).sort()).toEqual(["g1", "g2"]);
+    const g1 = due.find((d) => d.guid === "g1")!;
+    expect(g1.ownerId).toBe("owner-1");
+    expect(g1.source.repo).toBe("a/x");
+    expect(g1.source.ref).toBe("main");
+  });
+});
+
+describe("updateSourceDriftState", () => {
+  test("sets only drift dot-paths; preserves lastSyncedCommit and does NOT bump updatedOn", async () => {
+    const created = new Date("2026-06-01T00:00:00.000Z");
+    await seed({
+      _id: "g1",
+      updatedOn: created,
+      updatedBy: "orig",
+      source: { type: "github", repo: "a/x", ref: "main", path: "", lastSyncedCommit: "keepme" },
+    });
+
+    const checkedAt = new Date("2026-07-01T08:00:00.000Z");
+    await repo.updateSourceDriftState("g1", {
+      driftState: "drifted",
+      upstreamHeadSha: "newsha",
+      etag: 'W/"e1"',
+      lastCheckedAt: checkedAt,
+    });
+
+    const doc = await repo.findByGuid("g1");
+    expect(doc).not.toBeNull();
+    const src = doc!.source!;
+    expect(src.driftState).toBe("drifted");
+    expect(src.upstreamHeadSha).toBe("newsha");
+    expect(src.etag).toBe('W/"e1"');
+    expect(src.lastCheckedAt?.toISOString()).toBe(checkedAt.toISOString());
+    // Load-bearing: the refresh-owned field is untouched, and a background
+    // drift check must not disturb sort-by-updated ordering.
+    expect(src.lastSyncedCommit).toBe("keepme");
+    expect(doc!.updatedOn.toISOString()).toBe(created.toISOString());
+    expect(doc!.updatedBy).toBe("orig");
+  });
+
+  test("is a no-op on a skill with no source", async () => {
+    await seed({ _id: "g1" }); // no source
+    await repo.updateSourceDriftState("g1", { driftState: "broken", lastCheckedAt: new Date() });
+    const doc = await repo.findByGuid("g1");
+    expect(doc!.source).toBeUndefined();
+  });
+});
diff --git a/ornn-api/src/domains/skills/crud/repository.ts b/ornn-api/src/domains/skills/crud/repository.ts
index f611c43c..fa457e05 100644
--- a/ornn-api/src/domains/skills/crud/repository.ts
+++ b/ornn-api/src/domains/skills/crud/repository.ts
@@ -8,6 +8,7 @@ import type {
   SkillDocument,
   SkillGrant,
   SkillMetadata,
+  SkillSource,
   SkillSourceDriftState,
 } from "../../../shared/types/index";
 import { AppError } from "../../../shared/types/index";
@@ -170,6 +171,13 @@ export class SkillRepository {
       this.collection.createIndex({ createdBy: 1, createdOn: -1 }),
       this.collection.createIndex({ createdOn: -1 }),
       this.collection.createIndex({ isPrivate: 1, createdOn: -1 }),
+      // Source-drift scan (#1176): partial index over github-sourced skills
+      // only, ordered by last-checked so the scheduler's "due for a check"
+      // query stays cheap as the catalogue grows.
+      this.collection.createIndex(
+        { "source.lastCheckedAt": 1 },
+        { partialFilterExpression: { "source.type": "github" } },
+      ),
     ]);
   }
 
@@ -389,6 +397,44 @@ export class SkillRepository {
     );
   }
 
+  /**
+   * Enumerate GitHub-sourced skills due for a drift check (#1176). Selects
+   * skills whose source is github, whose `ref` is NOT a pinned 40-hex SHA
+   * (those can never drift), and which were either never checked or last
+   * checked before `notCheckedSince`. Returns light `{ guid, source }`
+   * projections — the scheduler coalesces them by `(repo, ref)`.
+   */
+  async findGithubSourcedSkills(opts: {
+    notCheckedSince: Date;
+  }): Promise<Array<{ guid: string; source: SkillSource; ownerId: string }>> {
+    const docs = await this.collection
+      .find(
+        {
+          "source.type": "github",
+          // Exclude pinned commit SHAs — a 40-hex ref never moves.
+          "source.ref": { $not: /^[0-9a-f]{40}$/i },
+          $or: [
+            { "source.lastCheckedAt": { $exists: false } },
+            { "source.lastCheckedAt": { $lt: opts.notCheckedSince } },
+          ],
+        },
+        { projection: { source: 1, createdBy: 1 } },
+      )
+      .toArray();
+    const out: Array<{ guid: string; source: SkillSource; ownerId: string }> = [];
+    for (const doc of docs) {
+      const source = coerceSkillSource(doc.source);
+      if (source) {
+        out.push({
+          guid: String(doc._id),
+          source,
+          ownerId: typeof doc.createdBy === "string" ? doc.createdBy : "",
+        });
+      }
+    }
+    return out;
+  }
+
   /**
    * Set or clear a NyxID-service tie. When `data.nyxidServiceId` is `null`
    * we wipe all four cached fields. Caller must have already validated
@@ -930,6 +976,44 @@ export class SkillRepository {
   }
 }
 
+/**
+ * Coerce a stored `source` sub-document into the typed `SkillSource`,
+ * omitting absent optional fields so we never fabricate an Invalid Date.
+ * Shared by `mapDoc` and `findGithubSourcedSkills` (#1176). Returns
+ * `undefined` for hand-uploaded skills (no `source`).
+ */
+function coerceSkillSource(raw: unknown): SkillSource | undefined {
+  if (!raw || typeof raw !== "object") return undefined;
+  const s = raw as Record<string, unknown>;
+  return {
+    type: "github",
+    repo: String(s.repo ?? ""),
+    ref: String(s.ref ?? ""),
+    path: String(s.path ?? ""),
+    ...(s.lastSyncedAt instanceof Date
+      ? { lastSyncedAt: s.lastSyncedAt }
+      : s.lastSyncedAt != null
+        ? { lastSyncedAt: new Date(s.lastSyncedAt as string | number) }
+        : {}),
+    ...(typeof s.lastSyncedCommit === "string" && s.lastSyncedCommit
+      ? { lastSyncedCommit: s.lastSyncedCommit }
+      : {}),
+    // Drift-detection fields (#1175). Absent until the first drift check.
+    ...(typeof s.upstreamHeadSha === "string" && s.upstreamHeadSha
+      ? { upstreamHeadSha: s.upstreamHeadSha }
+      : {}),
+    ...(typeof s.etag === "string" && s.etag ? { etag: s.etag } : {}),
+    ...(s.lastCheckedAt instanceof Date
+      ? { lastCheckedAt: s.lastCheckedAt }
+      : s.lastCheckedAt != null
+        ? { lastCheckedAt: new Date(s.lastCheckedAt as string | number) }
+        : {}),
+    ...(typeof s.driftState === "string"
+      ? { driftState: s.driftState as SkillSourceDriftState }
+      : {}),
+  };
+}
+
 function mapDoc(doc: Document | null): SkillDocument | null {
   if (!doc) return null;
   return {
@@ -959,43 +1043,7 @@ function mapDoc(doc: Document | null): SkillDocument | null {
     // to fall back to read-grants derived from the legacy lists.
     grants: coerceStoredGrants(doc.grants),
     latestVersion: doc.latestVersion ?? "0.1",
-    source: doc.source
-      ? {
-          type: "github",
-          repo: String(doc.source.repo ?? ""),
-          ref: String(doc.source.ref ?? ""),
-          path: String(doc.source.path ?? ""),
-          // Both fields are optional — when the user attached a GitHub
-          // link via PUT /skills/:id/source without an immediate sync,
-          // they're absent from the doc. Don't fabricate an Invalid
-          // Date by feeding `undefined` to the Date constructor.
-          ...(doc.source.lastSyncedAt instanceof Date
-            ? { lastSyncedAt: doc.source.lastSyncedAt }
-            : doc.source.lastSyncedAt != null
-              ? { lastSyncedAt: new Date(doc.source.lastSyncedAt) }
-              : {}),
-          ...(typeof doc.source.lastSyncedCommit === "string" && doc.source.lastSyncedCommit
-            ? { lastSyncedCommit: doc.source.lastSyncedCommit }
-            : {}),
-          // Drift-detection fields (#1175). All optional — absent until the
-          // first drift check runs. Same absent-field care as above so we
-          // never fabricate an Invalid Date.
-          ...(typeof doc.source.upstreamHeadSha === "string" && doc.source.upstreamHeadSha
-            ? { upstreamHeadSha: doc.source.upstreamHeadSha }
-            : {}),
-          ...(typeof doc.source.etag === "string" && doc.source.etag
-            ? { etag: doc.source.etag }
-            : {}),
-          ...(doc.source.lastCheckedAt instanceof Date
-            ? { lastCheckedAt: doc.source.lastCheckedAt }
-            : doc.source.lastCheckedAt != null
-              ? { lastCheckedAt: new Date(doc.source.lastCheckedAt) }
-              : {}),
-          ...(typeof doc.source.driftState === "string"
-            ? { driftState: doc.source.driftState as SkillSourceDriftState }
-            : {}),
-        }
-      : undefined,
+    source: coerceSkillSource(doc.source),
     nyxidServiceId: typeof doc.nyxidServiceId === "string" ? doc.nyxidServiceId : null,
     nyxidServiceSlug: typeof doc.nyxidServiceSlug === "string" ? doc.nyxidServiceSlug : null,
     nyxidServiceLabel: typeof doc.nyxidServiceLabel === "string" ? doc.nyxidServiceLabel : null,

From 441145a419f28205d6bc2ff474babee872341b28 Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:28 +0800
Subject: [PATCH 4/8] feat(api): skill.source_broken owner notification (#1176)

Add the skill.source_broken category to the canonical NOTIFICATION_CATEGORIES vocabulary (so the boot migration's allow-list can't wipe it) and a notifySourceBroken() method that tells a skill's owner their linked GitHub source could not be resolved, deep-linking to the skill so they can re-link it.

Part of #1176
---
 ornn-api/src/domains/notifications/service.ts | 26 +++++++++++++++++++
 ornn-api/src/domains/notifications/types.ts   |  3 +++
 2 files changed, 29 insertions(+)

diff --git a/ornn-api/src/domains/notifications/service.ts b/ornn-api/src/domains/notifications/service.ts
index ebc3ca4d..de31c4d6 100644
--- a/ornn-api/src/domains/notifications/service.ts
+++ b/ornn-api/src/domains/notifications/service.ts
@@ -442,6 +442,32 @@ export class NotificationService {
     });
   }
 
+  /**
+   * Owner-side notification (#1176) fired when an automatic drift check finds
+   * a GitHub-sourced skill's upstream repo/ref can no longer be resolved
+   * (deleted, made private, or the branch/tag was removed). Lets the owner
+   * re-link or fix the source. Deep-links to the skill detail page.
+   */
+  async notifySourceBroken(params: {
+    ownerId: string;
+    skillGuid: string;
+    repo: string;
+    ref: string;
+  }): Promise<void> {
+    const title = `The GitHub source for one of your skills is unavailable`;
+    const body =
+      `Ornn could not reach the upstream source (${params.repo}@${params.ref}) during an ` +
+      `automatic sync check — it may have been deleted, made private, or the branch/tag ` +
+      `was removed. Re-link the skill to a valid GitHub folder to resume automatic syncing.`;
+    await this.emit(params.ownerId, {
+      category: "skill.source_broken",
+      title,
+      body,
+      link: `/skills/${encodeURIComponent(params.skillGuid)}`,
+      data: { skillGuid: params.skillGuid, repo: params.repo, ref: params.ref },
+    });
+  }
+
   private async emit(
     userId: string,
     payload: {
diff --git a/ornn-api/src/domains/notifications/types.ts b/ornn-api/src/domains/notifications/types.ts
index 17189e2f..0b7a1d1a 100644
--- a/ornn-api/src/domains/notifications/types.ts
+++ b/ornn-api/src/domains/notifications/types.ts
@@ -35,6 +35,9 @@ export const NOTIFICATION_CATEGORIES = [
   "launchPromo.codeDelivered",
   // A member skill became unreadable to the skillset owner (#1136).
   "skillset.member_unreadable",
+  // A GitHub-sourced skill's upstream repo/ref could not be resolved during
+  // an automatic drift check — the source is broken (404/private/deleted) (#1176).
+  "skill.source_broken",
 ] as const;
 
 export type NotificationCategory = (typeof NOTIFICATION_CATEGORIES)[number];

From 9352cc6b147a07bccd4e89b68f0089f94e64afb0 Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:48 +0800
Subject: [PATCH 5/8] feat(api): source-drift batch job (coalesce + rate-limit
 backoff) (#1176)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

runSourceDriftJob enumerates due github-sourced skills, coalesces them by (repo,ref) so a shared upstream costs exactly one probe, and persists each skill's drift verdict. A broken source marks the skill broken and notifies its owner; a rate-limit signal short-circuits the rest of the tick (the next fire retries); pinned refs are skipped; probes run under a bounded concurrency with optional jitter. No single skill's persist/notify/probe failure is allowed to abort the run. Records state only — no publish (that is #1177).

Part of #1176
---
 .../skills/crud/sourceDriftJob.test.ts        | 254 ++++++++++++++++++
 .../src/domains/skills/crud/sourceDriftJob.ts | Bin 0 -> 7285 bytes
 2 files changed, 254 insertions(+)
 create mode 100644 ornn-api/src/domains/skills/crud/sourceDriftJob.test.ts
 create mode 100644 ornn-api/src/domains/skills/crud/sourceDriftJob.ts

diff --git a/ornn-api/src/domains/skills/crud/sourceDriftJob.test.ts b/ornn-api/src/domains/skills/crud/sourceDriftJob.test.ts
new file mode 100644
index 00000000..fbff2a26
--- /dev/null
+++ b/ornn-api/src/domains/skills/crud/sourceDriftJob.test.ts
@@ -0,0 +1,254 @@
+import { describe, expect, test } from "bun:test";
+import pino from "pino";
+import { runSourceDriftJob, type SourceDriftJobDeps } from "./sourceDriftJob";
+import {
+  GitHubSourceNotFoundError,
+  GitHubRateLimitError,
+  type RefHeadProbeInput,
+  type RefHeadProbeResult,
+} from "./utils/githubPull";
+import type { SkillSource } from "../../../shared/types/index";
+
+const logger = pino({ level: "silent" });
+
+type Candidate = { guid: string; source: SkillSource; ownerId: string };
+type Probe = (
+  repo: string,
+  ref: string,
+  opts?: RefHeadProbeInput,
+) => Promise<RefHeadProbeResult>;
+
+function cand(
+  guid: string,
+  repo: string,
+  ref: string,
+  sourceExtra: Partial<Extract<SkillSource, { type: "github" }>> = {},
+): Candidate {
+  return {
+    guid,
+    ownerId: `owner-${guid}`,
+    source: { type: "github", repo, ref, path: "", ...sourceExtra },
+  };
+}
+
+function makeDeps(opts: {
+  enabled?: boolean;
+  candidates: Candidate[];
+  probe: Probe;
+  concurrency?: number;
+}): {
+  deps: SourceDriftJobDeps;
+  persisted: Array<{ guid: string; patch: Record<string, unknown> }>;
+  notified: Array<{ ownerId: string; skillGuid: string; repo: string; ref: string }>;
+} {
+  const persisted: Array<{ guid: string; patch: Record<string, unknown> }> = [];
+  const notified: Array<{ ownerId: string; skillGuid: string; repo: string; ref: string }> = [];
+  const deps: SourceDriftJobDeps = {
+    skillRepo: {
+      findGithubSourcedSkills: async () => opts.candidates,
+      updateSourceDriftState: async (guid, patch) => {
+        persisted.push({ guid, patch: patch as Record<string, unknown> });
+      },
+    },
+    settingsService: {
+      getSourceSync: async () => ({
+        enabled: opts.enabled ?? true,
+        githubToken: "tok",
+        pollSchedule: "*/15 * * * *",
+        minCheckIntervalMinutes: 60,
+        autoPublish: false,
+      }),
+    },
+    notifier: {
+      notifySourceBroken: async (p) => {
+        notified.push(p);
+      },
+    },
+    logger,
+    probeRefHead: opts.probe,
+    concurrency: opts.concurrency ?? 5,
+    jitterMs: 0,
+  };
+  return { deps, persisted, notified };
+}
+
+describe("runSourceDriftJob", () => {
+  test("disabled settings → no-op (no probe, no persist)", async () => {
+    let probed = 0;
+    const { deps, persisted } = makeDeps({
+      enabled: false,
+      candidates: [cand("g1", "acme/x", "main")],
+      probe: async () => {
+        probed++;
+        return { sha: "s", notModified: false };
+      },
+    });
+    const res = await runSourceDriftJob(deps);
+    expect(res.enabled).toBe(false);
+    expect(probed).toBe(0);
+    expect(persisted.length).toBe(0);
+  });
+
+  test("coalesces by (repo,ref): N skills sharing an upstream → ONE probe", async () => {
+    let probes = 0;
+    const { deps, persisted, notified } = makeDeps({
+      candidates: [
+        cand("g1", "acme/x", "main", { lastSyncedCommit: "c1" }),
+        cand("g2", "acme/x", "main", { lastSyncedCommit: "c1" }),
+        cand("g3", "acme/x", "main", { lastSyncedCommit: "cOLD" }),
+      ],
+      probe: async () => {
+        probes++;
+        return { sha: "c1", notModified: false };
+      },
+    });
+    const res = await runSourceDriftJob(deps);
+    expect(probes).toBe(1); // one probe for the shared (repo,ref)
+    expect(res.groups).toBe(1);
+    expect(res.checked).toBe(3); // fanned out to all three
+    expect(res.drifted).toBe(1); // only g3's lastSyncedCommit differs
+    expect(persisted.length).toBe(3);
+    expect(notified.length).toBe(0);
+  });
+
+  test("pinned 40-hex ref is skipped, never probed", async () => {
+    let probes = 0;
+    const { deps, persisted } = makeDeps({
+      candidates: [cand("g1", "acme/x", "a".repeat(40))],
+      probe: async () => {
+        probes++;
+        return { sha: "x", notModified: false };
+      },
+    });
+    const res = await runSourceDriftJob(deps);
+    expect(probes).toBe(0);
+    expect(res.skipped).toBe(1);
+    expect(persisted.length).toBe(0);
+  });
+
+  test("broken source → each skill marked broken + notified; loop continues", async () => {
+    const { deps, persisted, notified } = makeDeps({
+      concurrency: 1,
+      candidates: [
+        cand("g1", "gone/x", "main"),
+        cand("g2", "gone/x", "main"),
+        cand("g3", "live/y", "main", { lastSyncedCommit: "c1" }),
+      ],
+      probe: async (repo) => {
+        if (repo === "gone/x") throw new GitHubSourceNotFoundError(repo, "main");
+        return { sha: "c1", notModified: false };
+      },
+    });
+    const res = await runSourceDriftJob(deps);
+    expect(res.broken).toBe(2);
+    expect(notified.map((n) => n.skillGuid).sort()).toEqual(["g1", "g2"]);
+    expect(persisted.filter((p) => p.patch.driftState === "broken").length).toBe(2);
+    // The live group after the broken one still got processed.
+    expect(res.checked).toBe(1);
+    expect(persisted.some((p) => p.guid === "g3" && p.patch.driftState === "in_sync")).toBe(true);
+  });
+
+  test("rate-limit → short-circuits remaining groups; counts them skipped", async () => {
+    let calls = 0;
+    const { deps, persisted } = makeDeps({
+      concurrency: 1, // deterministic order
+      candidates: [
+        cand("g1", "a/x", "main"),
+        cand("g2", "b/y", "main"),
+        cand("g3", "c/z", "main"),
+      ],
+      probe: async () => {
+        calls++;
+        if (calls === 1) throw new GitHubRateLimitError(403, 1000);
+        return { sha: "s", notModified: false };
+      },
+    });
+    const res = await runSourceDriftJob(deps);
+    expect(calls).toBe(1); // no probing after the rate-limit signal
+    expect(res.skipped).toBe(3); // the rate-limited group + the two never attempted
+    expect(res.checked).toBe(0);
+    expect(persisted.length).toBe(0);
+  });
+
+  test("drift persists upstreamHeadSha + etag + lastCheckedAt", async () => {
+    const { deps, persisted } = makeDeps({
+      candidates: [cand("g1", "a/x", "main", { lastSyncedCommit: "old" })],
+      probe: async () => ({ sha: "new", etag: 'W/"e"', notModified: false }),
+    });
+    await runSourceDriftJob(deps);
+    const patch = persisted.find((p) => p.guid === "g1")!.patch;
+    expect(patch.driftState).toBe("drifted");
+    expect(patch.upstreamHeadSha).toBe("new");
+    expect(patch.etag).toBe('W/"e"');
+    expect(patch.lastCheckedAt).toBeInstanceOf(Date);
+  });
+
+  test("304 notModified → in_sync; the stored etag is sent", async () => {
+    let seenEtag: string | undefined;
+    const { deps, persisted } = makeDeps({
+      candidates: [cand("g1", "a/x", "main", { etag: 'W/"prev"' })],
+      probe: async (_r, _ref, o) => {
+        seenEtag = o?.etag;
+        return { notModified: true };
+      },
+    });
+    const res = await runSourceDriftJob(deps);
+    expect(seenEtag).toBe('W/"prev"');
+    expect(res.checked).toBe(1);
+    expect(res.drifted).toBe(0);
+    expect(persisted[0]!.patch.driftState).toBe("in_sync");
+  });
+
+  test("transient probe error → group skipped, other groups still processed", async () => {
+    const { deps, persisted } = makeDeps({
+      concurrency: 1,
+      candidates: [
+        cand("g1", "flaky/x", "main"),
+        cand("g2", "ok/y", "main", { lastSyncedCommit: "c1" }),
+      ],
+      probe: async (repo) => {
+        if (repo === "flaky/x") throw new Error("ECONNRESET");
+        return { sha: "c1", notModified: false };
+      },
+    });
+    const res = await runSourceDriftJob(deps);
+    expect(res.skipped).toBe(1);
+    expect(res.checked).toBe(1);
+    expect(persisted.some((p) => p.guid === "g2")).toBe(true);
+    expect(persisted.some((p) => p.guid === "g1")).toBe(false);
+  });
+
+  test("a persist failure for one skill does not abort the tick", async () => {
+    const persisted: string[] = [];
+    const deps: SourceDriftJobDeps = {
+      skillRepo: {
+        findGithubSourcedSkills: async () => [
+          cand("g1", "a/x", "main", { lastSyncedCommit: "c1" }),
+          cand("g2", "b/y", "main", { lastSyncedCommit: "c1" }),
+        ],
+        updateSourceDriftState: async (guid) => {
+          if (guid === "g1") throw new Error("mongo write failed");
+          persisted.push(guid);
+        },
+      },
+      settingsService: {
+        getSourceSync: async () => ({
+          enabled: true,
+          githubToken: "",
+          pollSchedule: "*/15 * * * *",
+          minCheckIntervalMinutes: 60,
+          autoPublish: false,
+        }),
+      },
+      notifier: { notifySourceBroken: async () => {} },
+      logger,
+      probeRefHead: async () => ({ sha: "c1", notModified: false }),
+      concurrency: 1,
+      jitterMs: 0,
+    };
+    const res = await runSourceDriftJob(deps);
+    // Both counted as checked; g2 still persisted despite g1's write throwing.
+    expect(res.checked).toBe(2);
+    expect(persisted).toContain("g2");
+  });
+});
diff --git a/ornn-api/src/domains/skills/crud/sourceDriftJob.ts b/ornn-api/src/domains/skills/crud/sourceDriftJob.ts
new file mode 100644
index 0000000000000000000000000000000000000000..e889bf3ed39c369b744a8c5deabf7705eb3258f0
GIT binary patch
literal 7285
zcmcIp>u%h}74C06#jzToTzSP^19pP0EICe`i;KuLEcch7)?9L^WyB?!8B)?#5zt?K
zfTB;BC+T<2nOSnzYdc6=!;q=DoH>{8Ts=8B*rNkFt2>*T!^{?WOY@{n7xZI2r|9>O
z9)0ooh<^FUzo}gqn%CJ%K3uvmSyviMd0~x<L4(PBQ*}#Ylh!!;u4upS=7&xQXXLJm
zvUHSnhVt4{LLiV{jVY~@(zp~CKz~W|6_a@vS<}>GvL+wBBUj_NTO`(G)HSZPCRyU1
zYaN-}ByGzT)s>;f)^jtaq{^r<))m-iS|))k^7<d}yvTQLY*wUgOwVa<>nl^?2^OEC
z$J57?#unF*V3L{Av?d$TGO<@eiZvjV=$+nsC$Cj?TjT{KD8NyDQyI&YReoY+9G1nh
zXvr0qRZ=ofZc*FzFfDA_@g1gX%`z#fqPnD;+FoI)z<QBhQN+8Ixot($IM-UdHHzs~
zEu?Uz^lU-EJSob~8p3S@F)1m_`UZMyYml_{4b@#sbuLeqb=?HRz2*1V;#mzrUj6wk
zHQl@{+`^kaX_+LW;ft7Ff<y(a!w{LHhxj+GExud}VFSf4mo)<f1X(aQQ2<O*+hr3~
z_X{@0pXxcpnCunH20Ed3)xeTPRa0*3CGGR){?W$n9i1_j-T)!4XluKodw)2dFl=;j
z_a%jSe<7a4@>kI{;n&@ys4{c={(XVS*{Vv<OxhN8wME65wuLPqsBYt9eWNFKw?|Ny
ztIKQi#^m2afoF?kj75cu^7pE4zpc9}`^MVZ_M2~@-5(f9>vhqVr}LWO|FUYjmT$bb
z^TxQY#8$7Xt}RM8xh&d6H-Fug<$y^_%fz`NUkOov6=q4VltI?QTXCG8oJuP;1*JV<
z;A=Qzk%4VGraQXqifjguSg7g9yWC8Tsbr7TdtR<j(LE?l4i4y50#)3Q_cTkcvv~rC
z<+zjqeO*%m<N_IXmmK22KiDnPgl%<Z>~)cvn2}~~1!+aq8YQBSe<hJXFTruxQ1aYr
zxE(b!*`aMtk8qqcID9u0Q<fQ(Yr*|WK3upT%?PSq7S24qu8ZvCXzzaSLyQKud0`qS
z%!p^PJAepG^i*Eh@)5=a(S8nW18K0hN|nhmw@*LQe%G+2bY0#!V|U)iMX&?eJifgu
zp$ED&tsZAQL<1T6Rqrb&Y)K+E<u}zeqEuPV69l(bKy<S}2$taL-|*`k#FkCF60;GR
z>RLeet-t*&s1V%4TMUFV89x(X0&s~7Zkm=)l=bDMu``J@@elG#;P0(*2-TcoI8QNG
zl30#_vpj(dAuMw&<%pe>boXUhR9El2^u6a<?s^XezY*T%0MO$>4TD4%gwch%X&uh*
z^z9KHpRD0BJfV&%)?!VcCbvkFRoYo=s<uy^!jp^Xg$hJZJg@7s=?YIHlh2I}TKP+Z
zoQm`hJyW!FBP0OMvWJ%zzK0}NA<WZ{MGK!|{kVp61Js;k{X0|tjvhN@`NAWgM!U);
zcxTd*e@vCk;rBD}RhK5InA!EFaOV0a23t~k@V)T-U@zmO^npE1`a7?|2ce+>Nl27;
zl_m;T8q+{);pV_o#M`}8ZD%WF`kP?e2vv||3NbkR?`^R($lqakOfbYqLPg@>fe3ry
z!6Y}YwFY9uWr!5P7lzS;p*9T4{CmT-9|B2YlO@U@>z-~!lJgG*ii?x;M;GJ0ox-)q
z(s!Uvl-wjJg_xrlI@2^B?+!f?^EOtFASNN<BATO`i;T4+PfsYRJI6PPt<=NFr<*b1
z=!}05YLGUmQiI}0@+Upx<#=hFOD+vTKa!Y~_Pu%`BBs;IP+-9Y5@LobR!S{SI*1Zn
zeu^}bkl;tC1Uh87eI=vJb4cG|C&@S~4N;#mRyw0~N@dd9fn+0$>@-A?Q}iH>rAmE?
z+7{&`P<KR=38E%lRoKwf#$K%X5AW<WUx>1?9vg<*NNqjHlItgfVntn=D$`gDE$~p8
z!Xe0`pZCAf{xkB#{|`9OXy6+hud4tGu1h`w17Ct{p}}^QDlz!z$Kf!r@8&h12E;+Y
z1k%}^Pk8X4WAQ!qCfy2Y9UDW}+)VcIR;K6cf~0Ue_Mtd<rFY1p%In~eJ3J{XoCI)+
z)^ya}xSj|kk5Mc*qb(>v)1V{R@GC})M9w=D2tFUkl`(}BFh!fm-6qFpe6K@1zC%dI
zrx2GTTCBibgOapNIo?$P_pECrEfosh-~hK>un34%?P=H6c@Cv>ynlf@E8-t;rJHC(
zhwEZ1UKZ8Y%=$~NR<D!t6oG}i>jV0H`nTzHIwTp2ZMS91b(sA?_2<16ZY)ok?5seX
zgzE^8?11ck_y<AaYu~P;QKvzZj;P4(hc??VE8<@|s5nrBpfg9&RKY={)KhB6Z>XrG
zCq=m7hIQrRXuCki+0-b~9dYm3qOJ6ZRjYc{ohHpw?I^~5Xgt3-i9|6Rn<KAq?t0U;
ziULq|^UAD1@7-s2X{-Q@IkMjWZet<8f5*5FY?xaS^AZ=8BX*Jql;magk8oy$L4}@<
zo6aqw)SodxcDM}};M-$5PcLM9fIn5Px^6ryDe2iWz<;Zq?429ZY4w+~e5Xmfh?mK2
z^vL6LxD7&2<XP27sL^mupuGMa=cdC4I<^pZcfY~IFkg}Rh%~iGrN`VDI3rk7Hbk>O
zKy}zVkK@>D8jjEz7mY5M6Q%+w9c+CJ-Wa7^4Pn|*)t%#3wph|H|NNJNlcy`FQDG|p
z3vEO36DCqZ78L^ee#ZG$T-GbY{-&?mNb-MlbQAy;;IUl2%x=Mua`y0{LPuetdnX<^
zEM8oi>aty|^ReJ8kW%gjr_Y7J#e*SDo-=tp_N*`kHV~IuT%OWHPsP6%2*^uZvq1NZ
zaCrO%2X|6w$UII`h;Vd1y?Ef=MJ&eGp!=^(*&f2r4qfLO1C?609v?#2@bIMMB7Fs)
zNXsrmFhfX3=SGjGhYNFyr*fVr${$WLlfRx%5C51P<`;L5r}v*tiWmWqi{O0pzme3x
zC1{j@wApuQIt_~HOCe^^2e6Pk#Wnk|IQBKF$|6Z*=3CpJ2Q6!;cKDcSlgpXnaiHIr
zm-}N{4e{|^6)~@~-nNQf9WZUqkcRq$Fg=6-+}8GJV3NNUNujyksK%K9*W?^l1xEY>
zs(g-K%bub+;Zl6KLhA}#ukNTGRG`6t6%gXOYh$>IVgWCO(W-_IBqB@}c)058bA7_c
zWc>x<XTu+QkNAjkwG4YToZf9}Sc4jpJoLLx{2T9XXDs1(lSt^H+@Q(h7>y!F3u)+C
zN*tNr%xGV0oc%Fjs25WF^A^o0_~@aea%&C9q#*_dfW9yB^x0}*E*~5s=u6^^KGjD%
z-VC9Hzdxo{#E!QZB)Ecy6;Wkc8H7*h$?a`l{i*FlD@QVgk;%9W${9YGjEB(i>k+%7
zdt_cW&<cG1H*F--+W}$C>pZgucB3{(-;2w|uFp=pKs$iWwq~Vz;AcOMxliIvuMZLB
zJ90?`JLbeusG%4vT;2_s?9mmb6WhQ=j+Y6<C&B8&k*_+2h~wDT<DZU?k7?hR{QIL2
z#{5EmZ>=xBzOHsoXgczJkq&XBhRX3jCpMB#oqVBYKk(D9QyIyy;u-fsRi%Ed^W7N3
zOKpjsTbrz6o>@oe&m{lPe2rFAT(ySM%{MX|b;A9a@F?bhp*R_FdVs3&Q%&Z&CH~`T
zwLSOIY=-zVXqPzD_uY~P$=DTt$0F>Zn%16Z*w^_7#_t0g94s-A$M9He@B1@<p66Bn
E0p>qOQvd(}

literal 0
HcmV?d00001


From 3f1860448cb75e24bd8614b88af0d3dd5232674c Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:48 +0800
Subject: [PATCH 6/8] feat(api): source-sync Agenda scheduler (#1176)

createSourceSyncScheduler clones the mirror scheduler's multi-pod-safe Agenda pattern (per-fire row lock on agendaJobs) for the inbound drift job: a source-drift-check job whose cron cadence is driven by settings.sourceSync.pollSchedule (Asia/Singapore), plus a 1-minute sync tick that re-registers it cluster-wide when an admin changes the cadence. Exposes getScheduledRunStatus() for parity with the mirror scheduler.

Part of #1176
---
 .../skills/crud/sourceSyncScheduler.test.ts   | 252 ++++++++++++++++++
 .../skills/crud/sourceSyncScheduler.ts        | 242 +++++++++++++++++
 2 files changed, 494 insertions(+)
 create mode 100644 ornn-api/src/domains/skills/crud/sourceSyncScheduler.test.ts
 create mode 100644 ornn-api/src/domains/skills/crud/sourceSyncScheduler.ts

diff --git a/ornn-api/src/domains/skills/crud/sourceSyncScheduler.test.ts b/ornn-api/src/domains/skills/crud/sourceSyncScheduler.test.ts
new file mode 100644
index 00000000..864f6195
--- /dev/null
+++ b/ornn-api/src/domains/skills/crud/sourceSyncScheduler.test.ts
@@ -0,0 +1,252 @@
+/**
+ * Source-sync scheduler unit tests. Mocks the Agenda surface (same approach
+ * as the mirror scheduler test) so assertions stay deterministic — the
+ * multi-pod row-lock guarantee is Agenda's own to test, not ours.
+ *
+ * @module domains/skills/crud/sourceSyncScheduler.test
+ */
+import { beforeEach, describe, expect, mock, test } from "bun:test";
+import pino from "pino";
+import type { SourceSyncSettingsReader } from "./sourceSyncScheduler";
+import type { SourceSyncSection } from "../../settings/sections/sourceSync";
+import type { SourceDriftJobResult } from "./sourceDriftJob";
+
+const logger = pino({ level: "silent" });
+
+let agendaCalls: {
+  define: Array<{ name: string }>;
+  every: Array<{ interval: string | number; name: string; options?: { timezone?: string } }>;
+  cancel: Array<{ name?: string }>;
+  now: string[];
+  started: boolean;
+  stopped: boolean;
+};
+const jobHandlers = new Map<string, () => Promise<void>>();
+let queryJobsResult: { jobs: Array<Record<string, unknown>> } = { jobs: [] };
+let queryJobsThrows: Error | null = null;
+
+mock.module("agenda", () => ({
+  Agenda: class FakeAgenda {
+    on() {}
+    define(name: string, fn: () => Promise<void>) {
+      agendaCalls.define.push({ name });
+      jobHandlers.set(name, fn);
+    }
+    async every(
+      interval: string | number,
+      name: string,
+      _data: unknown,
+      options?: { timezone?: string },
+    ) {
+      agendaCalls.every.push({ interval, name, ...(options !== undefined ? { options } : {}) });
+    }
+    async cancel(opts: { name?: string }) {
+      agendaCalls.cancel.push(opts);
+      return 1;
+    }
+    async now(name: string) {
+      agendaCalls.now.push(name);
+      const fn = jobHandlers.get(name);
+      if (fn) await fn();
+    }
+    async start() {
+      agendaCalls.started = true;
+    }
+    async stop() {
+      agendaCalls.stopped = true;
+    }
+    async queryJobs(_opts: { name: string }) {
+      if (queryJobsThrows) throw queryJobsThrows;
+      return queryJobsResult;
+    }
+  },
+}));
+mock.module("@agendajs/mongo-backend", () => ({
+  MongoBackend: class FakeBackend {
+    constructor(_: unknown) {}
+  },
+}));
+
+const { createSourceSyncScheduler } = await import("./sourceSyncScheduler");
+
+function resetAgendaCalls() {
+  agendaCalls = { define: [], every: [], cancel: [], now: [], started: false, stopped: false };
+  jobHandlers.clear();
+  queryJobsResult = { jobs: [] };
+  queryJobsThrows = null;
+}
+
+function makeSettings(initial: string): SourceSyncSettingsReader & { set(s: string): void } {
+  let cur: SourceSyncSection = {
+    enabled: true,
+    githubToken: "",
+    pollSchedule: initial,
+    minCheckIntervalMinutes: 60,
+    autoPublish: false,
+  };
+  return {
+    getSourceSync: mock(async () => cur),
+    set(next: string) {
+      cur = { ...cur, pollSchedule: next };
+    },
+  } as unknown as SourceSyncSettingsReader & { set(s: string): void };
+}
+
+const okResult: SourceDriftJobResult = {
+  enabled: true,
+  groups: 0,
+  checked: 0,
+  drifted: 0,
+  broken: 0,
+  skipped: 0,
+};
+const FAKE_DB = {} as Parameters<typeof createSourceSyncScheduler>[0]["db"];
+
+beforeEach(() => {
+  resetAgendaCalls();
+});
+
+describe("createSourceSyncScheduler", () => {
+  test("start registers both jobs + eager-syncs schedule from settings (SGT)", async () => {
+    const settings = makeSettings("*/15 * * * *");
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: settings,
+      runDriftJob: async () => okResult,
+    });
+    await sched.start();
+
+    expect(agendaCalls.define.map((d) => d.name).sort()).toEqual([
+      "source-drift-check",
+      "source-sync-schedule",
+    ]);
+    const everyJob = agendaCalls.every.find((e) => e.name === "source-drift-check");
+    expect(everyJob).toBeDefined();
+    expect(everyJob!.interval).toBe("*/15 * * * *");
+    expect(everyJob!.options?.timezone).toBe("Asia/Singapore");
+    const everySync = agendaCalls.every.find((e) => e.name === "source-sync-schedule");
+    expect(everySync!.interval).toBe("1 minute");
+
+    await sched.stop();
+    expect(agendaCalls.stopped).toBe(true);
+  });
+
+  test("settings change → next tick re-registers with new cron", async () => {
+    const settings = makeSettings("*/15 * * * *");
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: settings,
+      runDriftJob: async () => okResult,
+    });
+    await sched.start();
+    const before = agendaCalls.every.filter((e) => e.name === "source-drift-check").length;
+
+    settings.set("0 * * * *");
+    await sched.runSyncNow();
+
+    const after = agendaCalls.every.filter((e) => e.name === "source-drift-check");
+    expect(after.length).toBe(before + 1);
+    expect(after.at(-1)!.interval).toBe("0 * * * *");
+  });
+
+  test("unchanged schedule → no second every() for the job", async () => {
+    const settings = makeSettings("*/15 * * * *");
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: settings,
+      runDriftJob: async () => okResult,
+    });
+    await sched.start();
+    const baseline = agendaCalls.every.filter((e) => e.name === "source-drift-check").length;
+    await sched.runSyncNow();
+    await sched.runSyncNow();
+    expect(agendaCalls.every.filter((e) => e.name === "source-drift-check").length).toBe(baseline);
+  });
+
+  test("empty schedule → cancels the recurring job; re-enabling re-registers", async () => {
+    const settings = makeSettings("*/15 * * * *");
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: settings,
+      runDriftJob: async () => okResult,
+    });
+    await sched.start();
+
+    settings.set("");
+    await sched.runSyncNow();
+    expect(agendaCalls.cancel.filter((c) => c.name === "source-drift-check").length).toBe(1);
+
+    settings.set("0 3 * * *");
+    await sched.runSyncNow();
+    expect(
+      agendaCalls.every.filter((e) => e.name === "source-drift-check").at(-1)!.interval,
+    ).toBe("0 3 * * *");
+  });
+
+  test("drift-check handler delegates to runDriftJob", async () => {
+    let ran = 0;
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: makeSettings("*/15 * * * *"),
+      runDriftJob: async () => {
+        ran++;
+        return okResult;
+      },
+    });
+    await sched.start();
+    const fn = jobHandlers.get("source-drift-check");
+    expect(fn).toBeDefined();
+    await fn!();
+    expect(ran).toBe(1);
+  });
+
+  test("settings read failure on a tick is swallowed (no crash)", async () => {
+    const broken = {
+      getSourceSync: mock(async () => {
+        throw new Error("db down");
+      }),
+    } as unknown as SourceSyncSettingsReader;
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: broken,
+      runDriftJob: async () => okResult,
+    });
+    await sched.start(); // eager sync hits the broken read but must not throw
+    await sched.runSyncNow();
+    await sched.stop();
+  });
+
+  test("getScheduledRunStatus: succeeded derivation", async () => {
+    const lastRunAt = new Date("2026-07-01T02:00:00.000Z");
+    const lastFinishedAt = new Date("2026-07-01T02:00:03.500Z");
+    queryJobsResult = { jobs: [{ lastRunAt, lastFinishedAt }] };
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: makeSettings("*/15 * * * *"),
+      runDriftJob: async () => okResult,
+    });
+    await sched.start();
+    const s = await sched.getScheduledRunStatus();
+    expect(s.status).toBe("succeeded");
+    expect(s.lastDurationMs).toBe(3500);
+  });
+
+  test("getScheduledRunStatus: queryJobs throw → never_run", async () => {
+    queryJobsThrows = new Error("mongo unreachable");
+    const sched = createSourceSyncScheduler({
+      db: FAKE_DB,
+      logger,
+      settingsService: makeSettings("*/15 * * * *"),
+      runDriftJob: async () => okResult,
+    });
+    await sched.start();
+    expect((await sched.getScheduledRunStatus()).status).toBe("never_run");
+  });
+});
diff --git a/ornn-api/src/domains/skills/crud/sourceSyncScheduler.ts b/ornn-api/src/domains/skills/crud/sourceSyncScheduler.ts
new file mode 100644
index 00000000..747e2839
--- /dev/null
+++ b/ornn-api/src/domains/skills/crud/sourceSyncScheduler.ts
@@ -0,0 +1,242 @@
+/**
+ * In-process source-sync (drift-check) scheduler (#1176).
+ *
+ * Clones the mirror scheduler's multi-pod-safe Agenda pattern
+ * (`domains/skills/mirror/scheduler.ts`): exactly one pod claims each
+ * scheduled fire via Agenda's per-fire row lock on the `agendaJobs`
+ * collection; the rest skip. Two recurring jobs:
+ *
+ *   1. `source-drift-check` — runs the drift batch job. Cadence (cron) is
+ *      driven by `settings.sourceSync.pollSchedule`, interpreted in
+ *      `Asia/Singapore`. Empty schedule = unregistered.
+ *   2. `source-sync-schedule` — every minute on every pod, re-reads the
+ *      cron from settings and (re-)registers the drift job via
+ *      `agenda.every()`, so an admin cadence change converges cluster-wide
+ *      within ~65s without cross-pod messaging.
+ *
+ * Crash recovery: `defaultLockLifetime` (10 min) lets another pod re-claim
+ * a fire if the holder dies. The job itself is idempotent (it only records
+ * drift state), so a rare double-run is harmless.
+ *
+ * @module domains/skills/crud/sourceSyncScheduler
+ */
+import { Agenda } from "agenda";
+import { MongoBackend } from "@agendajs/mongo-backend";
+import type { Db } from "mongodb";
+import type pino from "pino";
+import type { SourceSyncSection } from "../../settings/sections/sourceSync";
+import type { SourceDriftJobResult } from "./sourceDriftJob";
+
+const JOB_DRIFT_CHECK = "source-drift-check";
+const JOB_SYNC_SCHEDULE = "source-sync-schedule";
+
+const DEFAULT_TIMEZONE = "Asia/Singapore";
+const DEFAULT_SYNC_INTERVAL = "1 minute";
+const DEFAULT_LOCK_LIFETIME_MS = 10 * 60 * 1000;
+const PROCESS_EVERY = "5 seconds";
+
+/** Narrow settings surface — just the source-sync section read. */
+export interface SourceSyncSettingsReader {
+  getSourceSync(): Promise<SourceSyncSection>;
+}
+
+export interface SourceSyncSchedulerDeps {
+  db: Db;
+  logger: pino.Logger;
+  settingsService: SourceSyncSettingsReader;
+  /** The actual work — constructed in bootstrap with the full job deps. */
+  runDriftJob: () => Promise<SourceDriftJobResult>;
+  lockLifetimeMs?: number;
+  syncInterval?: string | number;
+  timezone?: string;
+  processEvery?: string | number;
+}
+
+export interface ScheduledRunStatus {
+  status: "succeeded" | "failed" | "running" | "never_run";
+  lastRunAt: Date | null;
+  lastFinishedAt: Date | null;
+  lastDurationMs: number | null;
+  lastError: string | null;
+  nextRunAt: Date | null;
+}
+
+export interface SourceSyncScheduler {
+  start(): Promise<void>;
+  stop(): Promise<void>;
+  /** Test hook — force the sync tick immediately. */
+  runSyncNow(): Promise<void>;
+  getScheduledRunStatus(): Promise<ScheduledRunStatus>;
+}
+
+export function createSourceSyncScheduler(
+  deps: SourceSyncSchedulerDeps,
+): SourceSyncScheduler {
+  const { db, settingsService, runDriftJob, logger } = deps;
+  const lockLifetime = deps.lockLifetimeMs ?? DEFAULT_LOCK_LIFETIME_MS;
+  const syncInterval = deps.syncInterval ?? DEFAULT_SYNC_INTERVAL;
+  const timezone = deps.timezone ?? DEFAULT_TIMEZONE;
+
+  const agenda = new Agenda({
+    backend: new MongoBackend({ mongo: db }),
+    processEvery: deps.processEvery ?? PROCESS_EVERY,
+    defaultLockLifetime: lockLifetime,
+  });
+
+  // Per-process memo of the last-registered cron — lets the sync tick
+  // early-return when nothing changed. Convergence across pods still
+  // happens via the shared `agendaJobs` doc.
+  let currentSchedule: string | null = null;
+
+  agenda.define(JOB_DRIFT_CHECK, async () => {
+    const t0 = Date.now();
+    try {
+      const result = await runDriftJob();
+      logger.info(
+        { ...result, durationMs: Date.now() - t0 },
+        "scheduled source drift check completed",
+      );
+    } catch (err) {
+      logger.error(
+        {
+          err: err instanceof Error ? err.message : String(err),
+          durationMs: Date.now() - t0,
+        },
+        "scheduled source drift check failed",
+      );
+      throw err;
+    }
+  });
+
+  agenda.define(JOB_SYNC_SCHEDULE, async () => {
+    let section: SourceSyncSection;
+    try {
+      section = await settingsService.getSourceSync();
+    } catch (err) {
+      logger.error(
+        { err: err instanceof Error ? err.message : String(err) },
+        "source-sync-schedule: failed to read settings — skipping tick",
+      );
+      return;
+    }
+    const next = section.pollSchedule;
+    if (next === currentSchedule) return;
+
+    if (next === "") {
+      await agenda.cancel({ name: JOB_DRIFT_CHECK });
+      currentSchedule = "";
+      logger.info(
+        "source-sync schedule: cancelled (settings.sourceSync.pollSchedule is empty)",
+      );
+      return;
+    }
+
+    try {
+      await agenda.every(next, JOB_DRIFT_CHECK, undefined, { timezone });
+      currentSchedule = next;
+      logger.info({ cron: next, timezone }, "source-sync schedule: registered");
+    } catch (err) {
+      logger.error(
+        { err: err instanceof Error ? err.message : String(err), cron: next },
+        "source-sync schedule: failed to register — will retry on next sync tick",
+      );
+    }
+  });
+
+  agenda.on("error", (err: unknown) => {
+    logger.error(
+      { err: err instanceof Error ? err.message : String(err) },
+      "agenda error event (source-sync)",
+    );
+  });
+
+  return {
+    async start() {
+      await agenda.start();
+      try {
+        await agenda.now(JOB_SYNC_SCHEDULE);
+      } catch (err) {
+        logger.warn(
+          { err: err instanceof Error ? err.message : String(err) },
+          "source-sync scheduler: initial sync enqueue failed — recurring sync will catch up",
+        );
+      }
+      await agenda.every(syncInterval, JOB_SYNC_SCHEDULE);
+      logger.info(
+        { syncInterval, timezone, lockLifetimeMs: lockLifetime },
+        "source-sync scheduler started",
+      );
+    },
+    async stop() {
+      try {
+        await agenda.stop(false); // false = don't close the Mongo client (we own it)
+      } catch (err) {
+        logger.warn(
+          { err: err instanceof Error ? err.message : String(err) },
+          "source-sync scheduler: agenda.stop failed — continuing",
+        );
+      }
+    },
+    async runSyncNow() {
+      await agenda.now(JOB_SYNC_SCHEDULE);
+    },
+    async getScheduledRunStatus(): Promise<ScheduledRunStatus> {
+      let result;
+      try {
+        result = await agenda.queryJobs({ name: JOB_DRIFT_CHECK });
+      } catch (err) {
+        logger.warn(
+          { err: err instanceof Error ? err.message : String(err) },
+          "getScheduledRunStatus (source-sync): queryJobs failed — returning never_run",
+        );
+        return emptyStatus();
+      }
+      const job = result.jobs[0];
+      if (!job) return emptyStatus();
+
+      const lastRunAt = job.lastRunAt ?? null;
+      const lastFinishedAt = job.lastFinishedAt ?? null;
+      const failedAt = job.failedAt ?? null;
+      const lockedAt = job.lockedAt ?? null;
+
+      let status: ScheduledRunStatus["status"];
+      if (lockedAt) {
+        status = "running";
+      } else if (
+        failedAt &&
+        (!lastFinishedAt || failedAt.getTime() >= lastFinishedAt.getTime())
+      ) {
+        status = "failed";
+      } else if (lastFinishedAt) {
+        status = "succeeded";
+      } else {
+        status = "never_run";
+      }
+
+      const lastDurationMs =
+        lastFinishedAt && lastRunAt
+          ? lastFinishedAt.getTime() - lastRunAt.getTime()
+          : null;
+
+      return {
+        status,
+        lastRunAt,
+        lastFinishedAt,
+        lastDurationMs,
+        lastError: status === "failed" ? (job.failReason ?? null) : null,
+        nextRunAt: job.nextRunAt ?? null,
+      };
+    },
+  };
+}
+
+function emptyStatus(): ScheduledRunStatus {
+  return {
+    status: "never_run",
+    lastRunAt: null,
+    lastFinishedAt: null,
+    lastDurationMs: null,
+    lastError: null,
+    nextRunAt: null,
+  };
+}

From c23d8bad4f87d1e7696ec93fad5b1ee33f94280d Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:48 +0800
Subject: [PATCH 7/8] feat(api): wire source-sync scheduler into bootstrap
 (#1176)

Construct + start the source-sync scheduler next to the mirror scheduler, injecting the drift job with the skill repo, settings, env token fallback, and the notification service. A start failure logs and leaves this pod without the scan rather than crashing boot; shutdown stops it before the mirror scheduler.

Part of #1176
---
 ornn-api/src/bootstrap.ts | 51 ++++++++++++++++++++++++++++++++++++---
 1 file changed, 48 insertions(+), 3 deletions(-)

diff --git a/ornn-api/src/bootstrap.ts b/ornn-api/src/bootstrap.ts
index c88f25e6..e2e1e8fa 100644
--- a/ornn-api/src/bootstrap.ts
+++ b/ornn-api/src/bootstrap.ts
@@ -109,6 +109,11 @@ import {
   createMirrorScheduler,
   type MirrorScheduler,
 } from "./domains/skills/mirror/scheduler";
+import {
+  createSourceSyncScheduler,
+  type SourceSyncScheduler,
+} from "./domains/skills/crud/sourceSyncScheduler";
+import { runSourceDriftJob } from "./domains/skills/crud/sourceDriftJob";
 
 // Domain: Me (caller-scoped endpoints)
 import { createMeRoutes } from "./domains/me/routes";
@@ -784,6 +789,39 @@ export async function bootstrap(
     mirrorScheduler,
   });
 
+  // In-process source-drift scheduler (#1176). Same multi-pod-safe Agenda
+  // pattern as the mirror scheduler; cadence driven by
+  // `settings.sourceSync.pollSchedule`. Detects when a GitHub-sourced skill's
+  // upstream moved and records drift state (no publish — that is #1177). A
+  // start failure logs and leaves this pod without the scheduled scan rather
+  // than crashing boot.
+  let sourceSyncScheduler: SourceSyncScheduler | null = null;
+  try {
+    sourceSyncScheduler = createSourceSyncScheduler({
+      db,
+      logger,
+      settingsService,
+      runDriftJob: () =>
+        runSourceDriftJob({
+          skillRepo,
+          settingsService,
+          envTokenFallback: config.sourceSyncGithubToken,
+          notifier: notificationService,
+          logger,
+          // Small per-group jitter so a large catalogue spreads its probes
+          // across the tick instead of bursting one egress IP.
+          jitterMs: 250,
+        }),
+    });
+    await sourceSyncScheduler.start();
+  } catch (err) {
+    logger.error(
+      { err: err instanceof Error ? err.message : String(err) },
+      "source-sync scheduler failed to start — scheduled drift checks will not run on this pod",
+    );
+    sourceSyncScheduler = null;
+  }
+
   // Skill routes — sharing is now a direct PUT /permissions write; the
   // audit signal is surfaced as a per-version label, not a gate.
   // #1136 — forward reference: the skill routes fire a reactive skillset
@@ -1134,9 +1172,16 @@ export async function bootstrap(
   // ---- Shutdown ----
   async function shutdown(): Promise<void> {
     logger.info("Shutting down ornn-api...");
-    // Stop the scheduler first so no new mirror reconciles start while
-    // we're tearing the Mongo connection down. `stop()` is idempotent +
-    // already swallows its own errors.
+    // Stop the schedulers first so no new reconciles / drift checks start
+    // while we're tearing the Mongo connection down. `stop()` is idempotent +
+    // already swallows its own errors. Source-sync first, then mirror.
+    if (sourceSyncScheduler) {
+      try {
+        await sourceSyncScheduler.stop();
+      } catch (err) {
+        logger.warn({ err }, "Source-sync scheduler stop failed — continuing");
+      }
+    }
     if (mirrorScheduler) {
       try {
         await mirrorScheduler.stop();

From f4d7f09112db82e33e3548ee7814c5c013493742 Mon Sep 17 00:00:00 2001
From: Shining <250120269+chronoai-shining@users.noreply.github.com>
Date: Wed, 1 Jul 2026 15:54:48 +0800
Subject: [PATCH 8/8] docs: changeset for #1176 (scheduled source drift-check
 job)

---
 .changeset/gh-source-drift-scheduler.md | 5 +++++
 1 file changed, 5 insertions(+)
 create mode 100644 .changeset/gh-source-drift-scheduler.md

diff --git a/.changeset/gh-source-drift-scheduler.md b/.changeset/gh-source-drift-scheduler.md
new file mode 100644
index 00000000..86187116
--- /dev/null
+++ b/.changeset/gh-source-drift-scheduler.md
@@ -0,0 +1,5 @@
+---
+"ornn-api": minor
+---
+
+Add the scheduled GitHub source drift-check job (#1176): a multi-pod-safe Agenda scheduler periodically probes every GitHub-sourced skill's upstream (coalescing skills that share a repo/ref into one request), records whether it has drifted, and notifies the owner when a source repo becomes unreachable. Cadence is driven by the `sourceSync.pollSchedule` setting; the job honors GitHub rate limits and never lets one skill's failure abort the run. It only records drift state — automatic re-publish is a later change.