admin/clawhub
1
0
Fork 0
mirror of https://github.com/openclaw/clawhub.git synced 2026-08-14 08:52:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: dispatch ClawScan workers from Convex (#3029)

Browse Source
This commit is contained in:
Patrick Erichsen
2026-07-08 21:57:14 -07:00
committed by GitHub
parent dfca6f5d9d
commit 683a1bf5dd
14 changed files with 1035 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/security-scan-codex.yml
+1 -4
View File
@@ -19,9 +19,6 @@ on:
description: "Stop claiming new batches after this many minutes"
required: true
default: "8"
schedule:
- cron: "*/5 * * * *"
permissions:
contents: read
@@ -37,7 +34,7 @@ jobs:
environment: Production
strategy:
fail-fast: false
max-parallel: 2
max-parallel: 4
matrix:
shard: [0, 1, 2, 3]
env:
convex/_generated/api.d.ts
Vendored
+2
View File
@@ -150,6 +150,7 @@ import type * as search from "../search.js";
import type * as securityDataset from "../securityDataset.js";
import type * as securityDatasetNode from "../securityDatasetNode.js";
import type * as securityScan from "../securityScan.js";
import type * as securityScanDispatch from "../securityScanDispatch.js";
import type * as skillCards from "../skillCards.js";
import type * as skillStatEvents from "../skillStatEvents.js";
import type * as skillTransfers from "../skillTransfers.js";
@@ -312,6 +313,7 @@ declare const fullApi: ApiFromModules<{
securityDataset: typeof securityDataset;
securityDatasetNode: typeof securityDatasetNode;
securityScan: typeof securityScan;
securityScanDispatch: typeof securityScanDispatch;
skillCards: typeof skillCards;
skillStatEvents: typeof skillStatEvents;
skillTransfers: typeof skillTransfers;
convex/crons.test.ts
+16
View File
@@ -16,6 +16,7 @@ const mocks = vi.hoisted(() => {
const authRefreshTokensPruneRef = Symbol("auth-refresh-tokens-prune");
const publisherInvitesPruneRef = Symbol("publisher-invites-prune");
const promotionsFeedPublishRef = Symbol("promotions-feed-publish");
const securityScanDispatchWatchdogRef = Symbol("security-scan-dispatch-watchdog");
return {
interval,
githubSkillSyncRef,
@@ -31,6 +32,7 @@ const mocks = vi.hoisted(() => {
authRefreshTokensPruneRef,
publisherInvitesPruneRef,
promotionsFeedPublishRef,
securityScanDispatchWatchdogRef,
};
});
@@ -78,6 +80,9 @@ vi.mock("./_generated/api", () => ({
securityScan: {
pruneExpiredSkillScanRequestsInternal: Symbol("skill-scan-request-prune"),
},
securityScanDispatch: {
requestSecurityScanDispatchInternal: mocks.securityScanDispatchWatchdogRef,
},
downloadMetrics: {
pruneDownloadMetricDedupesInternal: Symbol("download-metric-dedupe-prune"),
},
@@ -157,6 +162,17 @@ describe("crons", () => {
);
});
it("runs the security scan dispatch recovery watchdog every five minutes", async () => {
await import("./crons");
expect(mocks.interval).toHaveBeenCalledWith(
"codex-scan-dispatch-watchdog",
{ minutes: 5 },
mocks.securityScanDispatchWatchdogRef,
{},
);
});
it("prunes install telemetry dedupe rows daily", async () => {
await import("./crons");
convex/crons.ts
+7
View File
@@ -169,6 +169,13 @@ if (process.env.CLAWHUB_DISABLE_CRONS !== "1" && process.env.CLAWHUB_PREVIEW !==
{},
);
crons.interval(
"codex-scan-dispatch-watchdog",
{ minutes: 5 },
internal.securityScanDispatch.requestSecurityScanDispatchInternal,
{},
);
crons.interval(
"download-metric-dedupe-prune",
{ hours: 24 },
convex/lib/githubAuth.test.ts
+2
View File
@@ -27,6 +27,7 @@ describe("githubAuth", () => {
Response.json({
token: "ghs_app_token",
expires_at: "2026-02-02T13:00:00Z",
permissions: { actions: "write", contents: "read" },
}),
);
@@ -35,6 +36,7 @@ describe("githubAuth", () => {
).resolves.toEqual({
token: "ghs_app_token",
expiresAt: Date.parse("2026-02-02T13:00:00Z"),
permissions: { actions: "write", contents: "read" },
});
expect(fetchMock).toHaveBeenCalledWith(
convex/lib/githubAuth.ts
+7 -2
View File
@@ -14,6 +14,7 @@ type GitHubAppConfig = {
type InstallationToken = {
token: string;
expiresAt: number;
permissions: Record<string, string>;
};
type CachedInstallationToken = InstallationToken & {
@@ -106,12 +107,16 @@ export async function createGitHubAppInstallationToken(
throw new Error(`GitHub App token failed: ${message}`);
}
const payload = (await response.json()) as { token?: string; expires_at?: string };
const payload = (await response.json()) as {
token?: string;
expires_at?: string;
permissions?: Record<string, string>;
};
const token = payload.token?.trim();
if (!token) throw new Error("GitHub App token missing");
const expiresAt = payload.expires_at ? Date.parse(payload.expires_at) : Number.NaN;
if (!Number.isFinite(expiresAt)) throw new Error("GitHub App token expiry missing");
return { token, expiresAt };
return { token, expiresAt, permissions: payload.permissions ?? {} };
}
async function getCachedGitHubAppInstallationToken(options: {
convex/lib/retentionPolicy.ts
+1
View File
@@ -129,6 +129,7 @@ export const RETENTION_POLICIES = {
),
packageInspectorScanCursors: permanent("Package inspector scan progress cursor."),
securityScanJobs: permanent("Security scan job history and current processing state."),
securityScanDispatchState: permanent("Security scan worker dispatch coordination state."),
skillScanRequests: ephemeral(
"Uploaded or GitHub scan requests expire and delete temporary blobs.",
{
convex/schema.ts
+15
View File
@@ -1828,6 +1828,20 @@ const securityScanJobs = defineTable({
.index("by_package_release", ["packageReleaseId"])
.index("by_skill_scan_request", ["skillScanRequestId"]);
const securityScanDispatchState = defineTable({
key: v.string(),
scheduledToken: v.optional(v.string()),
scheduledAt: v.optional(v.number()),
leaseToken: v.optional(v.string()),
leaseExpiresAt: v.optional(v.number()),
lastDispatchAt: v.optional(v.number()),
lastDispatchStatus: v.optional(
v.union(v.literal("succeeded"), v.literal("failed"), v.literal("unknown")),
),
lastError: v.optional(v.string()),
updatedAt: v.number(),
}).index("by_key", ["key"]);
const skillScanRequests = defineTable({
actorUserId: v.id("users"),
sourceKind: skillScanRequestSourceKindValidator,
@@ -3189,6 +3203,7 @@ export default defineSchema({
packageInspectorFindingNotifications,
packageInspectorScanCursors,
securityScanJobs,
securityScanDispatchState,
skillScanRequests,
skillScanRequestFileChunks,
skillCardGenerationJobs,
convex/securityScan.test.ts
+152 -4
View File
@@ -8,6 +8,7 @@ import {
claimQueuedJobsInternal,
completeCodexScanJob,
enqueueBulkSkillRescanBatchForAdminInternal,
enqueuePackageReleaseScanInternal,
enqueueSkillVersionScanInternal,
failCodexScanJob,
failJobInternal,
@@ -362,7 +363,7 @@ const enqueueSkillVersionScanInternalHandler = (
enqueueSkillVersionScanInternal as unknown as WrappedHandler<
{
versionId: string;
source: "publish";
source: "publish" | "vt-update" | "backfill" | "bulk-rescan" | "manual";
priority?: number;
waitForVtMs?: number;
preserveActiveJob?: boolean;
@@ -372,6 +373,18 @@ const enqueueSkillVersionScanInternalHandler = (
>
)._handler;
const enqueuePackageReleaseScanInternalHandler = (
enqueuePackageReleaseScanInternal as unknown as WrappedHandler<
{
releaseId: string;
source: "publish" | "vt-update" | "backfill" | "bulk-rescan" | "manual";
priority?: number;
waitForVtMs?: number;
},
{ ok: true; skipped?: string; jobId?: string; alreadyQueued?: boolean }
>
)._handler;
const enqueueBulkSkillRescanBatchForAdminInternalHandler = (
enqueueBulkSkillRescanBatchForAdminInternal as unknown as WrappedHandler<
{
@@ -560,6 +573,7 @@ function makeRescanCtx(options: {
role: options.actorRole ?? "user",
},
...options.docs,
...Object.fromEntries((options.activeJobs ?? []).map((job) => [String(job._id), job])),
}),
);
const inserts: Array<{ table: string; doc: Record<string, unknown> }> = [];
@@ -568,6 +582,7 @@ function makeRescanCtx(options: {
const insert = vi.fn(async (table: string, doc: Record<string, unknown>) => {
const id = `${table}:${inserts.length + 1}`;
inserts.push({ table, doc });
docs.set(id, { _id: id, _creationTime: Date.now(), ...doc });
return id;
});
const patch = vi.fn(async (id: string, doc: Record<string, unknown>) => {
@@ -583,9 +598,31 @@ function makeRescanCtx(options: {
buildRange({ eq });
return {
collect: vi.fn(async () => {
if (table === "securityScanJobs") return options.activeJobs ?? [];
if (table === "securityScanJobs") {
return Array.from(docs.values()).filter((doc) =>
String(doc._id).startsWith("securityScanJobs:"),
);
}
return [];
}),
order: vi.fn(() => ({
first: vi.fn(async () => {
if (table !== "securityScanJobs") return null;
return (
Array.from(docs.values())
.filter(
(doc) =>
String(doc._id).startsWith("securityScanJobs:") &&
(!equals.has("status") || doc.status === equals.get("status")),
)
.sort(
(a, b) =>
Number(a.nextRunAt ?? 0) - Number(b.nextRunAt ?? 0) ||
Number(a._creationTime ?? 0) - Number(b._creationTime ?? 0),
)[0] ?? null
);
}),
})),
take: vi.fn(async () => {
if (table === "skills") {
return Array.from(docs.values()).filter((doc) => {
@@ -663,7 +700,10 @@ function makeRescanCtx(options: {
};
}),
}));
const scheduler = { runAfter: vi.fn(async () => undefined) };
const scheduler = {
runAfter: vi.fn(async () => undefined),
runAt: vi.fn(async () => "_scheduled_functions:1"),
};
return {
ctx: {
@@ -1196,6 +1236,113 @@ describe("securityScan", () => {
expect(patches).toEqual([]);
});
it("keeps an unscanned skill publish at publish priority when VirusTotal finishes", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_000_000);
const existingJob = makeScanJob({
_id: "securityScanJobs:skill-publish",
source: "publish",
skillVersionId: "skillVersions:skill-publish",
waitForVtUntil: 1_500_000,
nextRunAt: 1_500_000,
});
const { ctx, patches } = makeRescanCtx({
actorId: "users:owner",
docs: {
"skillVersions:skill-publish": {
_id: "skillVersions:skill-publish",
skillId: "skills:skill-publish",
version: "1.0.0",
vtAnalysis: { status: "clean", checkedAt: 1_000_000 },
},
},
activeJobs: [existingJob],
});
await enqueueSkillVersionScanInternalHandler(ctx, {
versionId: "skillVersions:skill-publish",
source: "vt-update",
waitForVtMs: 0,
});
expect(patches).toContainEqual({
id: "securityScanJobs:skill-publish",
patch: expect.objectContaining({
source: "publish",
nextRunAt: 1_000_000,
}),
});
});
it("keeps an unscanned package publish at publish priority when VirusTotal finishes", async () => {
vi.useFakeTimers();
vi.setSystemTime(2_000_000);
const existingJob = makeScanJob({
_id: "securityScanJobs:package-publish",
targetKind: "packageRelease",
skillVersionId: undefined,
packageReleaseId: "packageReleases:package-publish",
source: "publish",
waitForVtUntil: 2_500_000,
nextRunAt: 2_500_000,
});
const { ctx, patches } = makeRescanCtx({
actorId: "users:owner",
docs: {
"packageReleases:package-publish": {
_id: "packageReleases:package-publish",
packageId: "packages:package-publish",
version: "1.0.0",
vtAnalysis: { status: "clean", checkedAt: 2_000_000 },
},
},
activeJobs: [existingJob],
});
await enqueuePackageReleaseScanInternalHandler(ctx, {
releaseId: "packageReleases:package-publish",
source: "vt-update",
waitForVtMs: 0,
});
expect(patches).toContainEqual({
id: "securityScanJobs:package-publish",
patch: expect.objectContaining({
source: "publish",
nextRunAt: 2_000_000,
}),
});
});
it("requests an immediate worker dispatch when a publish scan becomes claimable", async () => {
vi.useFakeTimers();
vi.setSystemTime(3_000_000);
vi.stubEnv("SECURITY_SCAN_EVENT_DISPATCH_ENABLED", "1");
vi.stubEnv("GITHUB_APP_ID", "configured");
vi.stubEnv("GITHUB_APP_INSTALLATION_ID", "configured");
vi.stubEnv("GITHUB_APP_PRIVATE_KEY", "configured");
const { ctx, scheduler } = makeRescanCtx({
actorId: "users:owner",
docs: {
"skillVersions:dispatch": {
_id: "skillVersions:dispatch",
skillId: "skills:dispatch",
version: "1.0.0",
vtAnalysis: { status: "clean", checkedAt: 3_000_000 },
},
},
});
await enqueueSkillVersionScanInternalHandler(ctx, {
versionId: "skillVersions:dispatch",
source: "publish",
});
expect(scheduler.runAt).toHaveBeenCalledWith(3_000_000, expect.anything(), {
scheduleToken: expect.any(String),
});
});
it("lets platform moderators request skill rescans", async () => {
const { ctx, inserts } = makeRescanCtx({
actorId: "users:moderator",
@@ -3277,7 +3424,7 @@ describe("securityScan", () => {
},
);
expect(runMutation).toHaveBeenCalledTimes(2);
expect(runMutation).toHaveBeenCalledTimes(3);
expect(runMutation).toHaveBeenNthCalledWith(
1,
expect.anything(),
@@ -3294,6 +3441,7 @@ describe("securityScan", () => {
leaseToken: "lease-token",
}),
);
expect(runMutation).toHaveBeenNthCalledWith(3, expect.anything(), {});
});
it.each([
convex/securityScan.ts
+40 -7
View File
@@ -22,6 +22,7 @@ import {
serializedSkillScanRequestFilesBytes,
} from "./lib/skillScanRequestFiles";
import { redactWorkerPublicText } from "./lib/workerTextRedaction";
import { requestSecurityScanDispatch } from "./securityScanDispatch";
const DEFAULT_VT_WAIT_MS = 10 * 60 * 1000;
const DEFAULT_LEASE_MS = 60 * 60 * 1000;
@@ -155,6 +156,21 @@ const CLAIM_SOURCE_ORDER: SecurityScanJobSource[] = [
"bulk-rescan",
];
const SOURCE_PRIORITY: Record<SecurityScanJobSource, number> = {
manual: 5,
backfill: 4,
publish: 3,
"vt-update": 2,
"bulk-rescan": 1,
};
function higherPrioritySource(
current: SecurityScanJobSource,
requested: SecurityScanJobSource,
): SecurityScanJobSource {
return SOURCE_PRIORITY[requested] > SOURCE_PRIORITY[current] ? requested : current;
}
type EnqueueSkillVersionScanArgs = {
versionId: Id<"skillVersions">;
source: SecurityScanJobSource;
@@ -313,6 +329,9 @@ const internalRefs = internal as unknown as {
recordSkillScanRequestSucceededInternal: unknown;
succeedJobInternal: unknown;
};
securityScanDispatch: {
requestSecurityScanDispatchInternal: unknown;
};
skills: {
getSkillByIdInternal: unknown;
getVersionByIdInternal: unknown;
@@ -2089,13 +2108,14 @@ async function enqueueSkillVersionScan(ctx: MutationCtx, args: EnqueueSkillVersi
return { ok: true as const, jobId: active._id, alreadyQueued: true as const };
}
await ctx.db.patch(active._id, {
source: args.source,
source: higherPrioritySource(active.source, args.source),
priority: Math.max(active.priority, args.priority ?? 0),
hasMaliciousSignal,
waitForVtUntil: Math.min(active.waitForVtUntil, waitForVtUntil),
nextRunAt: Math.min(active.nextRunAt, nextRunAt),
updatedAt: now,
});
await requestSecurityScanDispatch(ctx);
return { ok: true as const, jobId: active._id, alreadyQueued: true as const };
}
const preservedExisting = args.preserveExistingJob
@@ -2120,6 +2140,7 @@ async function enqueueSkillVersionScan(ctx: MutationCtx, args: EnqueueSkillVersi
createdAt: now,
updatedAt: now,
});
await requestSecurityScanDispatch(ctx);
return { ok: true as const, jobId, alreadyQueued: false as const };
}
@@ -2150,13 +2171,14 @@ async function enqueuePackageReleaseScan(ctx: MutationCtx, args: EnqueuePackageR
const active = existing.find((job) => job.status === "queued" || job.status === "running");
if (active) {
await ctx.db.patch(active._id, {
source: args.source,
source: higherPrioritySource(active.source, args.source),
priority: Math.max(active.priority, args.priority ?? 0),
hasMaliciousSignal,
waitForVtUntil: Math.min(active.waitForVtUntil, waitForVtUntil),
nextRunAt: Math.min(active.nextRunAt, nextRunAt),
updatedAt: now,
});
await requestSecurityScanDispatch(ctx);
return { ok: true as const, jobId: active._id, alreadyQueued: true as const };
}
@@ -2173,6 +2195,7 @@ async function enqueuePackageReleaseScan(ctx: MutationCtx, args: EnqueuePackageR
createdAt: now,
updatedAt: now,
});
await requestSecurityScanDispatch(ctx);
return { ok: true as const, jobId, alreadyQueued: false as const };
}
@@ -2740,11 +2763,21 @@ export const completeCodexScanJob = action({
throw new ConvexError("Unsupported security scan target");
}
return await runMutationRef(ctx, internalRefs.securityScan.succeedJobInternal, {
jobId: args.jobId,
leaseToken: args.leaseToken,
runId: args.runId,
});
const result = await runMutationRef<{ ok: true }>(
ctx,
internalRefs.securityScan.succeedJobInternal,
{
jobId: args.jobId,
leaseToken: args.leaseToken,
runId: args.runId,
},
);
await runMutationRef(
ctx,
internalRefs.securityScanDispatch.requestSecurityScanDispatchInternal,
{},
);
return result;
},
});
convex/securityScanDispatch.test.ts
+480
View File
@@ -0,0 +1,480 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import {
beginSecurityScanDispatchInternal,
dispatchSecurityScanWorkflow,
finishSecurityScanDispatchInternal,
requestSecurityScanDispatchInternal,
} from "./securityScanDispatch";
type WrappedHandler<TArgs, TResult = unknown> = {
_handler: (ctx: unknown, args: TArgs) => Promise<TResult>;
};
const requestSecurityScanDispatchInternalHandler = (
requestSecurityScanDispatchInternal as unknown as WrappedHandler<
Record<string, never>,
{ scheduled: boolean; scheduledAt?: number }
>
)._handler;
const beginSecurityScanDispatchInternalHandler = (
beginSecurityScanDispatchInternal as unknown as WrappedHandler<
{ scheduleToken: string },
{ shouldDispatch: boolean; leaseToken?: string }
>
)._handler;
const finishSecurityScanDispatchInternalHandler = (
finishSecurityScanDispatchInternal as unknown as WrappedHandler<
{
leaseToken: string;
outcome: "succeeded" | "failed" | "unknown";
error?: string;
},
{ ok: boolean; stale?: boolean }
>
)._handler;
describe("securityScanDispatch", () => {
afterEach(() => {
vi.useRealTimers();
vi.unstubAllEnvs();
});
it("schedules an immediate worker dispatch for claimable queue work", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_000_000);
vi.stubEnv("SECURITY_SCAN_EVENT_DISPATCH_ENABLED", "1");
vi.stubEnv("GITHUB_APP_ID", "configured");
vi.stubEnv("GITHUB_APP_INSTALLATION_ID", "configured");
vi.stubEnv("GITHUB_APP_PRIVATE_KEY", "configured");
const insert = vi.fn(async () => "securityScanDispatchState:1");
const runAt = vi.fn(async () => "_scheduled_functions:1");
const query = vi.fn((table: string) => {
if (table === "securityScanJobs") {
return {
withIndex: vi.fn(() => ({
order: vi.fn(() => ({
first: vi.fn(async () => ({
_id: "securityScanJobs:1",
status: "queued",
nextRunAt: 900_000,
})),
})),
})),
};
}
return {
withIndex: vi.fn(() => ({
unique: vi.fn(async () => null),
})),
};
});
const result = await requestSecurityScanDispatchInternalHandler(
{
db: {
get: vi.fn(),
insert,
patch: vi.fn(),
query,
replace: vi.fn(),
delete: vi.fn(),
normalizeId: vi.fn(),
system: {},
},
scheduler: { runAt },
},
{},
);
expect(result).toEqual({ scheduled: true, scheduledAt: 1_000_000 });
expect(runAt).toHaveBeenCalledWith(1_000_000, expect.anything(), {
scheduleToken: expect.any(String),
});
expect(insert).toHaveBeenCalledWith(
"securityScanDispatchState",
expect.objectContaining({
key: "codex-worker",
scheduledAt: 1_000_000,
scheduledToken: expect.any(String),
}),
);
});
it("coalesces simultaneous queue requests behind the existing scheduled dispatch", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_000_000);
vi.stubEnv("SECURITY_SCAN_EVENT_DISPATCH_ENABLED", "1");
vi.stubEnv("GITHUB_APP_ID", "configured");
vi.stubEnv("GITHUB_APP_INSTALLATION_ID", "configured");
vi.stubEnv("GITHUB_APP_PRIVATE_KEY", "configured");
const runAt = vi.fn();
const query = vi.fn((table: string) => {
if (table === "securityScanJobs") {
return {
withIndex: vi.fn(() => ({
order: vi.fn(() => ({
first: vi.fn(async () => ({
_id: "securityScanJobs:1",
status: "queued",
nextRunAt: 900_000,
})),
})),
})),
};
}
return {
withIndex: vi.fn(() => ({
unique: vi.fn(async () => ({
_id: "securityScanDispatchState:1",
key: "codex-worker",
scheduledToken: "existing-token",
scheduledAt: 1_000_000,
updatedAt: 999_000,
})),
})),
};
});
const result = await requestSecurityScanDispatchInternalHandler(
{
db: {
get: vi.fn(),
insert: vi.fn(),
patch: vi.fn(),
query,
replace: vi.fn(),
delete: vi.fn(),
normalizeId: vi.fn(),
system: {},
},
scheduler: { runAt },
},
{},
);
expect(result).toEqual({ scheduled: false, scheduledAt: 1_000_000 });
expect(runAt).not.toHaveBeenCalled();
});
it("defers the next drain wave until the active dispatch lease expires", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_000_000);
vi.stubEnv("SECURITY_SCAN_EVENT_DISPATCH_ENABLED", "1");
vi.stubEnv("GITHUB_APP_ID", "configured");
vi.stubEnv("GITHUB_APP_INSTALLATION_ID", "configured");
vi.stubEnv("GITHUB_APP_PRIVATE_KEY", "configured");
const patch = vi.fn();
const runAt = vi.fn(async () => "_scheduled_functions:next");
const query = vi.fn((table: string) => {
if (table === "securityScanJobs") {
return {
withIndex: vi.fn(() => ({
order: vi.fn(() => ({
first: vi.fn(async () => ({
_id: "securityScanJobs:2",
status: "queued",
nextRunAt: 900_000,
})),
})),
})),
};
}
return {
withIndex: vi.fn(() => ({
unique: vi.fn(async () => ({
_id: "securityScanDispatchState:1",
key: "codex-worker",
leaseToken: "active-lease",
leaseExpiresAt: 1_300_000,
updatedAt: 999_000,
})),
})),
};
});
const result = await requestSecurityScanDispatchInternalHandler(
{
db: {
get: vi.fn(),
insert: vi.fn(),
patch,
query,
replace: vi.fn(),
delete: vi.fn(),
normalizeId: vi.fn(),
system: {},
},
scheduler: { runAt },
},
{},
);
expect(result).toEqual({ scheduled: true, scheduledAt: 1_300_000 });
expect(runAt).toHaveBeenCalledWith(1_300_000, expect.anything(), {
scheduleToken: expect.any(String),
});
expect(patch).toHaveBeenCalledWith(
"securityScanDispatchState:1",
expect.objectContaining({ scheduledAt: 1_300_000 }),
);
});
it("replaces a stale scheduled token when the watchdog finds claimable work", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_000_000);
vi.stubEnv("SECURITY_SCAN_EVENT_DISPATCH_ENABLED", "1");
vi.stubEnv("GITHUB_APP_ID", "configured");
vi.stubEnv("GITHUB_APP_INSTALLATION_ID", "configured");
vi.stubEnv("GITHUB_APP_PRIVATE_KEY", "configured");
const patch = vi.fn();
const runAt = vi.fn(async () => "_scheduled_functions:replacement");
const query = vi.fn((table: string) => {
if (table === "securityScanJobs") {
return {
withIndex: vi.fn(() => ({
order: vi.fn(() => ({
first: vi.fn(async () => ({
_id: "securityScanJobs:stuck",
status: "queued",
nextRunAt: 800_000,
})),
})),
})),
};
}
return {
withIndex: vi.fn(() => ({
unique: vi.fn(async () => ({
_id: "securityScanDispatchState:1",
key: "codex-worker",
scheduledToken: "lost-token",
scheduledAt: 900_000,
updatedAt: 900_000,
})),
})),
};
});
const result = await requestSecurityScanDispatchInternalHandler(
{
db: {
get: vi.fn(),
insert: vi.fn(),
patch,
query,
replace: vi.fn(),
delete: vi.fn(),
normalizeId: vi.fn(),
system: {},
},
scheduler: { runAt },
},
{},
);
expect(result).toEqual({ scheduled: true, scheduledAt: 1_000_000 });
expect(runAt).toHaveBeenCalledWith(1_000_000, expect.anything(), {
scheduleToken: expect.not.stringMatching(/^lost-token$/),
});
});
it("atomically acquires a dispatch lease only for the current scheduled token", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_000_000);
vi.stubEnv("SECURITY_SCAN_EVENT_DISPATCH_ENABLED", "1");
vi.stubEnv("GITHUB_APP_ID", "configured");
vi.stubEnv("GITHUB_APP_INSTALLATION_ID", "configured");
vi.stubEnv("GITHUB_APP_PRIVATE_KEY", "configured");
const patch = vi.fn();
const query = vi.fn((table: string) => {
if (table === "securityScanJobs") {
return {
withIndex: vi.fn(() => ({
order: vi.fn(() => ({
first: vi.fn(async () => ({
_id: "securityScanJobs:claimable",
status: "queued",
nextRunAt: 900_000,
})),
})),
})),
};
}
return {
withIndex: vi.fn(() => ({
unique: vi.fn(async () => ({
_id: "securityScanDispatchState:1",
key: "codex-worker",
scheduledToken: "current-token",
scheduledAt: 1_000_000,
updatedAt: 999_000,
})),
})),
};
});
const result = await beginSecurityScanDispatchInternalHandler(
{
db: {
get: vi.fn(),
insert: vi.fn(),
patch,
query,
replace: vi.fn(),
delete: vi.fn(),
normalizeId: vi.fn(),
system: {},
},
scheduler: { runAt: vi.fn() },
},
{ scheduleToken: "current-token" },
);
expect(result).toEqual({
shouldDispatch: true,
leaseToken: expect.any(String),
});
expect(patch).toHaveBeenCalledWith(
"securityScanDispatchState:1",
expect.objectContaining({
scheduledToken: undefined,
scheduledAt: undefined,
leaseToken: result.leaseToken,
leaseExpiresAt: 1_300_000,
}),
);
});
it("refuses to dispatch when the GitHub App lacks Actions write permission", async () => {
const fetchImpl = vi.fn();
await expect(
dispatchSecurityScanWorkflow(
{
token: "installation-token",
permissions: { actions: "read", contents: "read" },
},
fetchImpl,
),
).resolves.toEqual({
ok: false,
reason: "actions-write-required",
});
expect(fetchImpl).not.toHaveBeenCalled();
});
it("dispatches the production workflow on main with bounded worker inputs", async () => {
const fetchImpl = vi.fn(async () => new Response(null, { status: 204 }));
await expect(
dispatchSecurityScanWorkflow(
{
token: "installation-token",
permissions: { actions: "write", contents: "read" },
},
fetchImpl,
),
).resolves.toEqual({ ok: true });
expect(fetchImpl).toHaveBeenCalledWith(
"https://api.github.com/repos/openclaw/clawhub/actions/workflows/security-scan-codex.yml/dispatches",
expect.objectContaining({
method: "POST",
headers: expect.objectContaining({
Authorization: "Bearer installation-token",
}),
body: JSON.stringify({
ref: "main",
inputs: {
"batch-limit": "4",
"max-runtime-minutes": "8",
},
}),
}),
);
});
it("releases a rejected dispatch lease and schedules a bounded retry", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_000_000);
vi.stubEnv("SECURITY_SCAN_EVENT_DISPATCH_ENABLED", "1");
vi.stubEnv("GITHUB_APP_ID", "configured");
vi.stubEnv("GITHUB_APP_INSTALLATION_ID", "configured");
vi.stubEnv("GITHUB_APP_PRIVATE_KEY", "configured");
const state: Record<string, unknown> = {
_id: "securityScanDispatchState:1",
key: "codex-worker",
leaseToken: "dispatch-lease",
leaseExpiresAt: 1_300_000,
updatedAt: 999_000,
};
const patch = vi.fn(async (_id: string, next: Record<string, unknown>) => {
for (const [key, value] of Object.entries(next)) {
if (value === undefined) delete state[key];
else state[key] = value;
}
});
const runAt = vi.fn(async () => "_scheduled_functions:retry");
const query = vi.fn((table: string) => {
if (table === "securityScanJobs") {
return {
withIndex: vi.fn(() => ({
order: vi.fn(() => ({
first: vi.fn(async () => ({
_id: "securityScanJobs:retry",
status: "queued",
nextRunAt: 900_000,
})),
})),
})),
};
}
return {
withIndex: vi.fn(() => ({
unique: vi.fn(async () => state),
})),
};
});
const result = await finishSecurityScanDispatchInternalHandler(
{
db: {
get: vi.fn(),
insert: vi.fn(),
patch,
query,
replace: vi.fn(),
delete: vi.fn(),
normalizeId: vi.fn(),
system: {},
},
scheduler: { runAt },
},
{
leaseToken: "dispatch-lease",
outcome: "failed",
error: "GitHub rejected the dispatch",
},
);
expect(result).toEqual({ ok: true });
expect(runAt).toHaveBeenCalledWith(1_060_000, expect.anything(), {
scheduleToken: expect.any(String),
});
expect(state).toMatchObject({
lastDispatchStatus: "failed",
lastError: "GitHub rejected the dispatch",
scheduledAt: 1_060_000,
});
expect(state).not.toHaveProperty("leaseToken");
expect(state).not.toHaveProperty("leaseExpiresAt");
});
});
convex/securityScanDispatch.ts
+293
View File
@@ -0,0 +1,293 @@
import { v } from "convex/values";
import { internal } from "./_generated/api";
import type { MutationCtx } from "./_generated/server";
import { internalAction, internalMutation } from "./functions";
import { createGitHubAppInstallationToken, isGitHubAppConfigured } from "./lib/githubAuth";
const DISPATCH_STATE_KEY = "codex-worker";
const DISPATCH_LEASE_MS = 5 * 60 * 1000;
const SCHEDULE_STALE_MS = 60 * 1000;
const GITHUB_WORKFLOW_DISPATCH_URL =
"https://api.github.com/repos/openclaw/clawhub/actions/workflows/security-scan-codex.yml/dispatches";
const internalRefs = internal as unknown as {
securityScanDispatch: {
beginSecurityScanDispatchInternal: unknown;
dispatchSecurityScanWorkerInternal: unknown;
finishSecurityScanDispatchInternal: unknown;
};
};
async function runMutationRef<T>(
ctx: { runMutation: (ref: never, args: never) => Promise<unknown> },
ref: unknown,
args: unknown,
): Promise<T> {
return (await ctx.runMutation(ref as never, args as never)) as T;
}
export function isSecurityScanEventDispatchEnabled(env: NodeJS.ProcessEnv = process.env) {
return (
env.SECURITY_SCAN_EVENT_DISPATCH_ENABLED === "1" &&
env.CLAWHUB_PREVIEW !== "1" &&
Boolean(
env.GITHUB_APP_ID?.trim() &&
env.GITHUB_APP_INSTALLATION_ID?.trim() &&
env.GITHUB_APP_PRIVATE_KEY?.trim(),
)
);
}
export async function dispatchSecurityScanWorkflow(
installationToken: {
token: string;
permissions: Record<string, string>;
},
fetchImpl: typeof fetch = fetch,
) {
if (installationToken.permissions.actions !== "write") {
return { ok: false as const, reason: "actions-write-required" as const };
}
const response = await fetchImpl(GITHUB_WORKFLOW_DISPATCH_URL, {
method: "POST",
headers: {
Accept: "application/vnd.github+json",
Authorization: `Bearer ${installationToken.token}`,
"Content-Type": "application/json",
"User-Agent": "clawhub/security-scan-dispatch",
"X-GitHub-Api-Version": "2022-11-28",
},
body: JSON.stringify({
ref: "main",
inputs: {
"batch-limit": "4",
"max-runtime-minutes": "8",
},
}),
});
if (!response.ok) {
return {
ok: false as const,
reason: "github-rejected" as const,
status: response.status,
};
}
return { ok: true as const };
}
export async function requestSecurityScanDispatch(ctx: MutationCtx, notBefore = 0) {
if (!isSecurityScanEventDispatchEnabled()) return { scheduled: false as const };
const earliestQueued = await ctx.db
.query("securityScanJobs")
.withIndex("by_status_and_next_run_at", (q) => q.eq("status", "queued"))
.order("asc")
.first();
if (!earliestQueued) return { scheduled: false as const };
const now = Date.now();
const state = await ctx.db
.query("securityScanDispatchState")
.withIndex("by_key", (q) => q.eq("key", DISPATCH_STATE_KEY))
.unique();
const activeUntil =
state?.leaseExpiresAt !== undefined && state.leaseExpiresAt > now ? state.leaseExpiresAt : now;
const scheduledAt = Math.max(now, earliestQueued.nextRunAt, activeUntil, notBefore);
if (
state?.scheduledAt !== undefined &&
state.scheduledAt >= now - SCHEDULE_STALE_MS &&
state.scheduledAt <= scheduledAt
) {
return { scheduled: false as const, scheduledAt: state.scheduledAt };
}
const scheduleToken = crypto.randomUUID();
await ctx.scheduler.runAt(
scheduledAt,
internalRefs.securityScanDispatch.dispatchSecurityScanWorkerInternal as never,
{ scheduleToken } as never,
);
const patch = {
scheduledToken: scheduleToken,
scheduledAt,
updatedAt: now,
};
if (state) {
await ctx.db.patch(state._id, patch);
} else {
await ctx.db.insert("securityScanDispatchState", {
key: DISPATCH_STATE_KEY,
...patch,
});
}
return { scheduled: true as const, scheduledAt };
}
export const requestSecurityScanDispatchInternal = internalMutation({
args: {},
handler: async (ctx) => {
return requestSecurityScanDispatch(ctx);
},
});
export const beginSecurityScanDispatchInternal = internalMutation({
args: {
scheduleToken: v.string(),
},
handler: async (ctx, args) => {
if (!isSecurityScanEventDispatchEnabled()) return { shouldDispatch: false as const };
const state = await ctx.db
.query("securityScanDispatchState")
.withIndex("by_key", (q) => q.eq("key", DISPATCH_STATE_KEY))
.unique();
if (!state || state.scheduledToken !== args.scheduleToken) {
return { shouldDispatch: false as const };
}
const now = Date.now();
const claimable = await ctx.db
.query("securityScanJobs")
.withIndex("by_status_and_next_run_at", (q) => q.eq("status", "queued").lte("nextRunAt", now))
.order("asc")
.first();
if (!claimable) {
await ctx.db.patch(state._id, {
scheduledToken: undefined,
scheduledAt: undefined,
updatedAt: now,
});
await requestSecurityScanDispatch(ctx);
return { shouldDispatch: false as const };
}
const leaseToken = crypto.randomUUID();
await ctx.db.patch(state._id, {
scheduledToken: undefined,
scheduledAt: undefined,
leaseToken,
leaseExpiresAt: now + DISPATCH_LEASE_MS,
updatedAt: now,
});
return { shouldDispatch: true as const, leaseToken };
},
});
export const finishSecurityScanDispatchInternal = internalMutation({
args: {
leaseToken: v.string(),
outcome: v.union(v.literal("succeeded"), v.literal("failed"), v.literal("unknown")),
error: v.optional(v.string()),
},
handler: async (ctx, args) => {
const state = await ctx.db
.query("securityScanDispatchState")
.withIndex("by_key", (q) => q.eq("key", DISPATCH_STATE_KEY))
.unique();
if (!state || state.leaseToken !== args.leaseToken) {
return { ok: false as const, stale: true as const };
}
const now = Date.now();
await ctx.db.patch(state._id, {
...(args.outcome === "failed"
? {
leaseToken: undefined,
leaseExpiresAt: undefined,
}
: {}),
lastDispatchAt: now,
lastDispatchStatus: args.outcome,
lastError: args.error?.slice(0, 500),
updatedAt: now,
});
if (args.outcome === "failed") {
await requestSecurityScanDispatch(ctx, now + 60_000);
}
return { ok: true as const };
},
});
export const checkGitHubActionsPermissionInternal = internalAction({
args: {},
handler: async () => {
if (!isGitHubAppConfigured()) {
return {
configured: false as const,
actionsPermission: null,
canDispatch: false,
};
}
const installationToken = await createGitHubAppInstallationToken({
userAgent: "clawhub/security-scan-dispatch-preflight",
});
const actionsPermission = installationToken.permissions.actions ?? "none";
return {
configured: true as const,
actionsPermission,
canDispatch: actionsPermission === "write",
};
},
});
export const dispatchSecurityScanWorkerInternal = internalAction({
args: {
scheduleToken: v.string(),
},
handler: async (ctx, args) => {
const begin = await runMutationRef<{ shouldDispatch: boolean; leaseToken?: string }>(
ctx,
internalRefs.securityScanDispatch.beginSecurityScanDispatchInternal,
args,
);
if (!begin.shouldDispatch || !begin.leaseToken) {
return { dispatched: false as const, reason: "coalesced-or-empty" as const };
}
try {
const installationToken = await createGitHubAppInstallationToken({
userAgent: "clawhub/security-scan-dispatch",
});
const result = await dispatchSecurityScanWorkflow(installationToken);
if (result.ok) {
await runMutationRef(
ctx,
internalRefs.securityScanDispatch.finishSecurityScanDispatchInternal,
{
leaseToken: begin.leaseToken,
outcome: "succeeded",
},
);
return { dispatched: true as const };
}
const error =
result.reason === "actions-write-required"
? "GitHub App Actions write permission is required"
: `GitHub workflow dispatch rejected with HTTP ${result.status}`;
await runMutationRef(
ctx,
internalRefs.securityScanDispatch.finishSecurityScanDispatchInternal,
{
leaseToken: begin.leaseToken,
outcome: "failed",
error,
},
);
return { dispatched: false as const, reason: result.reason };
} catch {
await runMutationRef(
ctx,
internalRefs.securityScanDispatch.finishSecurityScanDispatchInternal,
{
leaseToken: begin.leaseToken,
outcome: "unknown",
error: "GitHub workflow dispatch outcome could not be confirmed",
},
);
return { dispatched: false as const, reason: "unknown" as const };
}
},
});
scripts/security/security-scan-worker-workflow.test.ts
+9
View File
@@ -36,9 +36,14 @@ describe("security-scan-codex workflow", () => {
"codex-security-scan": {
env?: Record<string, unknown>;
steps: WorkflowStep[];
strategy?: { "max-parallel"?: number; matrix?: { shard?: number[] } };
"timeout-minutes"?: number;
};
};
on?: {
schedule?: Array<{ cron?: string }>;
workflow_dispatch?: unknown;
};
};
const steps = workflow.jobs["codex-security-scan"].steps;
const jobEnv = workflow.jobs["codex-security-scan"].env ?? {};
@@ -63,6 +68,10 @@ describe("security-scan-codex workflow", () => {
);
expect(uploadStep?.with?.path).toBe("${{ env.CODEX_SECURITY_SCAN_DIAGNOSTICS_DIR }}");
expect(workflow.jobs["codex-security-scan"]["timeout-minutes"]).toBe(20);
expect(workflow.on?.workflow_dispatch).toBeDefined();
expect(workflow.on?.schedule).toBeUndefined();
expect(workflow.jobs["codex-security-scan"].strategy?.["max-parallel"]).toBe(4);
expect(workflow.jobs["codex-security-scan"].strategy?.matrix?.shard).toEqual([0, 1, 2, 3]);
expect(jobEnv.CODEX_SECURITY_SCAN_MAX_RUNTIME_MINUTES).toBe(
"${{ inputs['max-runtime-minutes'] || '8' }}",
);
specs/security-moderation.md
+10
View File
@@ -244,6 +244,16 @@ See also: [acceptable-usage.md](./acceptable-usage.md) for the marketplace polic
workspace with static and VT signals as context.
- Current skill and plugin scans are queued through `securityScanJobs` and
completed by the external Codex worker.
- Claimable queue work edge-triggers a coalesced GitHub Actions worker dispatch.
Successful completion requests another dispatch while queued work remains; a
five-minute Convex cron is only a recovery watchdog for lost dispatch signals.
Event-driven dispatch stays disabled until the production GitHub App is
verified to have Actions write permission.
- Queue source priority is `manual`, `backfill`, `publish`, `vt-update`, then
`bulk-rescan`. A later VirusTotal update may make a waiting publish job
claimable immediately, but it must not demote that job from publish priority.
- Bulk rescans stay lowest priority and use the bounded operator campaign flow,
which enqueues one page at a time and waits for that page before continuing.
- ClawScan worker concurrency is an operator-controlled compute concern. The
backend claim path must cap only a single worker claim size and must not impose
a global active-scan ceiling; horizontal capacity is controlled by worker