Publish curated projectbrain source from 88fab8df1542019289e947f092c2b387e0825c60
This commit is contained in:
Public Source Publisher committed
1 parent
48e72d07e0
commit
eca2891b07
45 files changed
+2697
-328
No files matched your search
@@ -2,6 +2,7 @@ import { describe, expect, it, vi } from "vitest";
|
||||
import { loadEnv } from "../src/config/env.js";
|
||||
import type { GiteaActionJob } from "../src/integrations/gitea/types.js";
|
||||
import { compactGiteaRuntime, createGiteaActivityRuntimeLoader, repositoryFromActionJob } from "../src/services/activityRuntime.js";
|
||||
import { evaluateRunnerClaimHealth } from "../src/services/runnerClaimHealth.js";
|
||||
|
||||
const env = loadEnv({
|
||||
NODE_ENV: "test",
|
||||
@@ -44,6 +45,26 @@ describe("compact activity runtime", () => {
|
||||
expect(snapshot.counts).toEqual({ running: 2, runnable: 30, dependencyBlocked: 0 });
|
||||
expect(snapshot.jobs).toHaveLength(32);
|
||||
});
|
||||
it("shows queued jobs oldest first so the visible lane reflects waiting age", () => {
|
||||
const snapshot = compactGiteaRuntime([], [
|
||||
job({ id: 9, status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-08-28T19:09:00Z" }),
|
||||
job({ id: 7, status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-08-28T19:03:00Z" }),
|
||||
job({ id: 8, status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-08-28T19:06:00Z" }),
|
||||
], "2026-08-28T19:10:00Z");
|
||||
expect(snapshot.jobs.map((item) => item.id)).toEqual([7, 8, 9]);
|
||||
});
|
||||
|
||||
it("puts unknown queued timestamps last and breaks equal creation times by ascending job id", () => {
|
||||
const snapshot = compactGiteaRuntime([], [
|
||||
job({ id: 9, status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-08-28T19:03:00Z" }),
|
||||
job({ id: 4, status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-08-28T19:03:00Z" }),
|
||||
job({ id: 7, status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-08-28T19:06:00Z" }),
|
||||
job({ id: 2, status: "queued", runner_id: 0, runner_name: undefined, created_at: "not-a-timestamp" }),
|
||||
], "2026-08-28T19:10:00Z");
|
||||
|
||||
expect(snapshot.jobs.map((item) => item.id)).toEqual([4, 9, 7, 2]);
|
||||
});
|
||||
|
||||
it("extracts a repository safely from a Gitea job URL", () => {
|
||||
expect(repositoryFromActionJob(job({ html_url: "https://gitea.example.com/Space%20Owner/Repo/actions/runs/1/jobs/2" }))).toBe("Space Owner/Repo");
|
||||
expect(repositoryFromActionJob(job({ html_url: "not-a-url" }))).toBeUndefined();
|
||||
@@ -123,6 +144,49 @@ describe("compact activity runtime", () => {
|
||||
expect(snapshot.jobs[0].queueState).toBe("dependency-blocked");
|
||||
});
|
||||
|
||||
it("keeps runner health unknown when runner discovery fails but queued jobs remain visible", async () => {
|
||||
const queuedJob = job({ status: "queued", runner_id: 0, runner_name: undefined });
|
||||
const listUserActionRunners = vi.fn().mockRejectedValue(new Error("runner inventory unavailable"));
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => input.status === "queued"
|
||||
? { total_count: 1, jobs: [queuedJob] }
|
||||
: { total_count: 0, jobs: [] });
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs } });
|
||||
|
||||
const snapshot = await load();
|
||||
const health = evaluateRunnerClaimHealth(snapshot, { now: Date.parse("2026-09-30T00:05:00.000Z") });
|
||||
|
||||
expect(snapshot).toMatchObject({
|
||||
availability: "live",
|
||||
runnerInventoryComplete: false,
|
||||
warning: "Partial Gitea monitoring failure: runner inventory unavailable",
|
||||
runnableJobScan: { complete: true, observed: 1, total: 1 },
|
||||
});
|
||||
expect(health).toMatchObject({
|
||||
status: "incomplete",
|
||||
runnerInventoryComplete: false,
|
||||
coverage: { complete: true, observed: 1, total: 1 },
|
||||
counts: { queued: 1, unknown: 1, labelBlocked: 0, pollingStalled: 0 },
|
||||
jobs: [{ state: "unknown", claimableWaitSeconds: 0, requiredLabels: ["ubuntu-latest"], compatibleRunnerNames: [], compatibleIdleRunnerNames: [] }],
|
||||
});
|
||||
|
||||
const emptyInventory = createGiteaActivityRuntimeLoader(env, {
|
||||
client: {
|
||||
listUserActionRunners: vi.fn().mockResolvedValue({ total_count: 0, runners: [] }),
|
||||
listUserActionJobs,
|
||||
},
|
||||
});
|
||||
const knownEmptySnapshot = await emptyInventory();
|
||||
const knownEmptyHealth = evaluateRunnerClaimHealth(knownEmptySnapshot, { now: Date.parse("2026-09-30T00:05:00.000Z") });
|
||||
|
||||
expect(knownEmptySnapshot.runnerInventoryComplete).toBe(true);
|
||||
expect(knownEmptyHealth).toMatchObject({
|
||||
status: "degraded",
|
||||
runnerInventoryComplete: true,
|
||||
counts: { queued: 1, unknown: 0, labelBlocked: 1 },
|
||||
jobs: [{ state: "label-blocked" }],
|
||||
});
|
||||
});
|
||||
|
||||
it("uses Gitea's documented pending filter for dependency-blocked jobs", async () => {
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 0, runners: [] });
|
||||
const listUserActionJobs = vi.fn().mockResolvedValue({ total_count: 0, jobs: [] });
|
||||
@@ -133,4 +197,361 @@ describe("compact activity runtime", () => {
|
||||
expect(listUserActionJobs).toHaveBeenCalledWith({ status: "pending", limit: 50 });
|
||||
expect(listUserActionJobs).not.toHaveBeenCalledWith({ status: "waiting", limit: 50 });
|
||||
});
|
||||
|
||||
it("keeps Gitea requested jobs returned by the pending filter visible without a claim-health alarm", async () => {
|
||||
const requestedJob = job({
|
||||
id: 37_222,
|
||||
run_id: 21_559,
|
||||
name: "quality",
|
||||
status: "requested",
|
||||
runner_id: 0,
|
||||
runner_name: undefined,
|
||||
steps: [],
|
||||
created_at: "2026-10-03T02:20:00.000Z",
|
||||
});
|
||||
const recentCompleted = Array.from({ length: 30 }, (_, index) => job({
|
||||
id: 40_000 + index,
|
||||
run_id: 30_000 + index,
|
||||
status: "completed",
|
||||
conclusion: "success",
|
||||
runner_id: 0,
|
||||
runner_name: undefined,
|
||||
created_at: "2026-10-03T02:21:00.000Z",
|
||||
completed_at: "2026-10-03T02:22:00.000Z",
|
||||
}));
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: false, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => {
|
||||
if (input.status === "pending") return { total_count: 1, jobs: [requestedJob] };
|
||||
if (input.status === "queued") return { total_count: 0, jobs: [] };
|
||||
if (input.status === "in_progress") return { total_count: 0, jobs: [] };
|
||||
return { total_count: recentCompleted.length, jobs: recentCompleted };
|
||||
});
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs } });
|
||||
|
||||
const snapshot = await load();
|
||||
const requested = snapshot.jobs.find((entry) => entry.id === requestedJob.id);
|
||||
const health = evaluateRunnerClaimHealth(snapshot, { now: Date.parse("2026-10-03T02:24:00.000Z") });
|
||||
|
||||
expect(listUserActionJobs).toHaveBeenCalledWith({ status: "pending", limit: 50 });
|
||||
expect(requested).toMatchObject({
|
||||
id: 37_222,
|
||||
runId: 21_559,
|
||||
status: "requested",
|
||||
queueState: "dependency-blocked",
|
||||
});
|
||||
expect(snapshot.counts).toEqual({ running: 0, runnable: 0, dependencyBlocked: 1 });
|
||||
expect(health).toMatchObject({
|
||||
status: "healthy",
|
||||
counts: { queued: 0, pollingStalled: 0 },
|
||||
oldestWaitSeconds: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("paginates the queued-job health scan so old jobs beyond page one are diagnosed", async () => {
|
||||
let clock = new Date("2026-09-30T00:05:00.000Z");
|
||||
const queuedJobs = [
|
||||
...Array.from({ length: 50 }, (_, index) => job({
|
||||
id: 1_000 - index,
|
||||
status: "queued",
|
||||
runner_id: 0,
|
||||
runner_name: undefined,
|
||||
created_at: "2026-09-30T00:04:30.000Z",
|
||||
})),
|
||||
...Array.from({ length: 10 }, (_, index) => job({
|
||||
id: 900 - index,
|
||||
status: "queued",
|
||||
runner_id: 0,
|
||||
runner_name: undefined,
|
||||
created_at: "2026-09-30T00:00:00.000Z",
|
||||
})),
|
||||
];
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: false, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string; limit?: number; page?: number } = {}) => {
|
||||
if (input.status === "queued") {
|
||||
const start = ((input.page ?? 1) - 1) * (input.limit ?? 50);
|
||||
return { total_count: queuedJobs.length, jobs: queuedJobs.slice(start, start + (input.limit ?? 50)) };
|
||||
}
|
||||
return { total_count: 0, jobs: [] };
|
||||
});
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs }, now: () => clock });
|
||||
|
||||
let snapshot = await load();
|
||||
for (let sample = 0; sample < 4; sample += 1) {
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
snapshot = await load();
|
||||
}
|
||||
const health = evaluateRunnerClaimHealth(snapshot, { now: clock.getTime(), stallSeconds: 120 });
|
||||
|
||||
expect(listUserActionJobs).toHaveBeenCalledWith({ status: "queued", limit: 50, page: 1 });
|
||||
expect(listUserActionJobs).toHaveBeenCalledWith({ status: "queued", limit: 50, page: 2 });
|
||||
expect(snapshot.runnableJobScan).toEqual({ complete: true, observed: 60, total: 60 });
|
||||
expect(health).toMatchObject({
|
||||
status: "stalled",
|
||||
coverage: { complete: true, observed: 60, total: 60 },
|
||||
counts: { queued: 60, pollingStalled: 60 },
|
||||
oldestWaitSeconds: 120,
|
||||
});
|
||||
});
|
||||
|
||||
it("does not treat an old database-created job as a long-waiting queued job when first observed", async () => {
|
||||
const clock = new Date("2026-09-30T00:05:00.000Z");
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: false, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => input.status === "queued"
|
||||
? { total_count: 1, jobs: [job({ status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-09-29T00:00:00.000Z" })] }
|
||||
: { total_count: 0, jobs: [] });
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs }, now: () => clock });
|
||||
|
||||
const snapshot = await load();
|
||||
const health = evaluateRunnerClaimHealth(snapshot, { now: Date.parse("2026-09-30T04:00:00.000Z"), stallSeconds: 120 });
|
||||
|
||||
expect(snapshot.jobs[0]).toMatchObject({
|
||||
createdAt: "2026-09-29T00:00:00.000Z",
|
||||
runnableObservedSince: clock.toISOString(),
|
||||
claimableObservedSince: clock.toISOString(),
|
||||
});
|
||||
expect(health).toMatchObject({
|
||||
status: "healthy",
|
||||
counts: { queued: 1, claimable: 1, pollingStalled: 0 },
|
||||
oldestWaitSeconds: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("starts polling-stall timing only after compatible idle capacity is observed", async () => {
|
||||
let clock = new Date("2026-09-30T00:00:00.000Z");
|
||||
let runnerBusy = true;
|
||||
const listUserActionRunners = vi.fn().mockImplementation(async () => ({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: runnerBusy, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] }));
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => input.status === "queued"
|
||||
? { total_count: 1, jobs: [job({ status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-09-29T00:00:00.000Z" })] }
|
||||
: { total_count: 0, jobs: [] });
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs }, now: () => clock });
|
||||
|
||||
let snapshot = await load();
|
||||
expect(evaluateRunnerClaimHealth(snapshot, { now: clock.getTime(), stallSeconds: 120 }).jobs[0].state).toBe("capacity-wait");
|
||||
for (let sample = 0; sample < 3; sample += 1) {
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
await load();
|
||||
}
|
||||
|
||||
runnerBusy = false;
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
snapshot = await load();
|
||||
expect(snapshot.jobs[0]).toMatchObject({
|
||||
runnableObservedSince: "2026-09-30T00:00:00.000Z",
|
||||
claimableObservedSince: clock.toISOString(),
|
||||
});
|
||||
expect(evaluateRunnerClaimHealth(snapshot, { now: clock.getTime(), stallSeconds: 120 }).jobs[0].state).toBe("claimable");
|
||||
|
||||
for (let sample = 0; sample < 3; sample += 1) {
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
snapshot = await load();
|
||||
expect(evaluateRunnerClaimHealth(snapshot, { now: clock.getTime(), stallSeconds: 120 }).jobs[0].state).toBe("claimable");
|
||||
}
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
snapshot = await load();
|
||||
const finalHealth = evaluateRunnerClaimHealth(snapshot, { now: clock.getTime(), stallSeconds: 120 });
|
||||
expect(finalHealth.jobs[0]).toMatchObject({ state: "polling-stalled", waitSeconds: 240, claimableWaitSeconds: 120 });
|
||||
expect(finalHealth.oldestClaimableWaitSeconds).toBe(120);
|
||||
});
|
||||
|
||||
it("does not age a cached health sample with wall-clock time", async () => {
|
||||
let clock = new Date("2026-09-30T00:00:00.000Z");
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: false, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => input.status === "queued"
|
||||
? { total_count: 1, jobs: [job({ status: "queued", runner_id: 0, runner_name: undefined })] }
|
||||
: { total_count: 0, jobs: [] });
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs }, now: () => clock });
|
||||
|
||||
await load();
|
||||
clock = new Date(clock.getTime() + 14_000);
|
||||
const cached = await load();
|
||||
const health = evaluateRunnerClaimHealth(cached, { now: clock.getTime() + 3_600_000, stallSeconds: 30 });
|
||||
|
||||
expect(listUserActionRunners).toHaveBeenCalledTimes(1);
|
||||
expect(health).toMatchObject({ status: "healthy", oldestWaitSeconds: 0, counts: { pollingStalled: 0 } });
|
||||
});
|
||||
|
||||
it("resets observation timing after a long gap between complete samples", async () => {
|
||||
let clock = new Date("2026-09-30T00:00:00.000Z");
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: false, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => input.status === "queued"
|
||||
? { total_count: 1, jobs: [job({ status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-09-29T00:00:00.000Z" })] }
|
||||
: { total_count: 0, jobs: [] });
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs }, now: () => clock });
|
||||
|
||||
await load();
|
||||
clock = new Date(clock.getTime() + 46_000);
|
||||
const snapshot = await load();
|
||||
const health = evaluateRunnerClaimHealth(snapshot, { now: clock.getTime(), stallSeconds: 30 });
|
||||
|
||||
expect(snapshot.jobs[0].runnableObservedSince).toBe(clock.toISOString());
|
||||
expect(snapshot.jobs[0].claimableObservedSince).toBe(clock.toISOString());
|
||||
expect(health).toMatchObject({ status: "healthy", oldestWaitSeconds: 0, counts: { pollingStalled: 0 } });
|
||||
});
|
||||
|
||||
it("resets claimable timing after partial and stale samples", async () => {
|
||||
let clock = new Date("2026-09-30T00:00:00.000Z");
|
||||
let failSecondPage = false;
|
||||
let failAllJobQueries = false;
|
||||
const queuedJobs = Array.from({ length: 51 }, (_, index) => job({
|
||||
id: 1_000 - index,
|
||||
status: "queued",
|
||||
runner_id: 0,
|
||||
runner_name: undefined,
|
||||
created_at: "2026-09-29T00:00:00.000Z",
|
||||
}));
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: false, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string; limit?: number; page?: number } = {}) => {
|
||||
if (failAllJobQueries) throw new Error("job queries unavailable");
|
||||
if (input.status === "queued") {
|
||||
if (input.page === 2 && failSecondPage) throw new Error("queued page unavailable");
|
||||
const start = ((input.page ?? 1) - 1) * (input.limit ?? 50);
|
||||
return { total_count: queuedJobs.length, jobs: queuedJobs.slice(start, start + (input.limit ?? 50)) };
|
||||
}
|
||||
return { total_count: 0, jobs: [] };
|
||||
});
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs }, now: () => clock });
|
||||
|
||||
await load();
|
||||
failSecondPage = true;
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
const partial = await load();
|
||||
expect(partial.runnableJobScan.complete).toBe(false);
|
||||
expect(evaluateRunnerClaimHealth(partial, { now: clock.getTime() }).status).toBe("incomplete");
|
||||
|
||||
failSecondPage = false;
|
||||
clock = new Date(clock.getTime() + 15_000);
|
||||
let snapshot = await load();
|
||||
expect(snapshot.jobs.find((entry) => entry.id === queuedJobs[0].id)?.claimableObservedSince).toBe(clock.toISOString());
|
||||
|
||||
failAllJobQueries = true;
|
||||
clock = new Date(clock.getTime() + 15_000);
|
||||
const stale = await load();
|
||||
expect(stale.availability).toBe("stale");
|
||||
expect(evaluateRunnerClaimHealth(stale, { now: clock.getTime() }).jobs[0].state).toBe("unknown");
|
||||
|
||||
failAllJobQueries = false;
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
snapshot = await load();
|
||||
const health = evaluateRunnerClaimHealth(snapshot, { now: clock.getTime(), stallSeconds: 30 });
|
||||
expect(snapshot.jobs.find((entry) => entry.id === queuedJobs[0].id)?.claimableObservedSince).toBe(clock.toISOString());
|
||||
expect(health).toMatchObject({ status: "healthy", counts: { pollingStalled: 0 }, oldestWaitSeconds: 0 });
|
||||
});
|
||||
|
||||
it("restarts runnable timing after a known queued-to-pending-to-queued transition", async () => {
|
||||
let clock = new Date("2026-09-30T00:00:00.000Z");
|
||||
let queued = true;
|
||||
const queuedJob = job({ status: "queued", runner_id: 0, runner_name: undefined, created_at: "2026-09-29T00:00:00.000Z" });
|
||||
const pendingJob = { ...queuedJob, status: "pending" };
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 1, runners: [
|
||||
{ id: 9, name: "validation-1", status: "online", busy: false, disabled: false, ephemeral: false, labels: [
|
||||
{ id: 1, name: "ubuntu-latest", type: "custom" },
|
||||
] },
|
||||
] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => {
|
||||
if (input.status === "queued") return { total_count: queued ? 1 : 0, jobs: queued ? [queuedJob] : [] };
|
||||
if (input.status === "pending" || input.status === undefined) return { total_count: queued ? 0 : 1, jobs: queued ? [] : [pendingJob] };
|
||||
return { total_count: 0, jobs: [] };
|
||||
});
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs }, now: () => clock });
|
||||
|
||||
const first = await load();
|
||||
expect(first.jobs[0].runnableObservedSince).toBe(clock.toISOString());
|
||||
|
||||
queued = false;
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
const pending = await load();
|
||||
expect(pending.jobs[0].queueState).toBe("dependency-blocked");
|
||||
|
||||
queued = true;
|
||||
clock = new Date(clock.getTime() + 30_000);
|
||||
const requeued = await load();
|
||||
expect(requeued.jobs[0].runnableObservedSince).toBe(clock.toISOString());
|
||||
expect(requeued.jobs[0].claimableObservedSince).toBe(clock.toISOString());
|
||||
expect(evaluateRunnerClaimHealth(requeued, { now: clock.getTime(), stallSeconds: 30 }).jobs[0].state).toBe("claimable");
|
||||
});
|
||||
|
||||
it("keeps paginated health scans bounded and marks queues beyond the cap incomplete", async () => {
|
||||
const queuedJobs = Array.from({ length: 501 }, (_, index) => job({
|
||||
id: 2_000 - index,
|
||||
status: "queued",
|
||||
runner_id: 0,
|
||||
runner_name: undefined,
|
||||
}));
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 0, runners: [] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string; limit?: number; page?: number } = {}) => {
|
||||
if (input.status === "queued") {
|
||||
const start = ((input.page ?? 1) - 1) * (input.limit ?? 50);
|
||||
return { total_count: queuedJobs.length, jobs: queuedJobs.slice(start, start + (input.limit ?? 50)) };
|
||||
}
|
||||
return { total_count: 0, jobs: [] };
|
||||
});
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs } });
|
||||
|
||||
const snapshot = await load();
|
||||
|
||||
expect(listUserActionJobs.mock.calls.filter(([input]) => input?.status === "queued")).toHaveLength(10);
|
||||
expect(snapshot.runnableJobScan).toEqual({ complete: false, observed: 500, total: 501 });
|
||||
expect(evaluateRunnerClaimHealth(snapshot).status).toBe("incomplete");
|
||||
});
|
||||
|
||||
it("does not treat a missing queued total as an empty complete scan", async () => {
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 0, runners: [] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string } = {}) => input.status === "queued"
|
||||
? { jobs: [] } as unknown as { total_count: number; jobs: GiteaActionJob[] }
|
||||
: { total_count: 0, jobs: [] });
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs } });
|
||||
|
||||
const snapshot = await load();
|
||||
|
||||
expect(snapshot.runnableJobScan).toEqual({ complete: false, observed: 0, total: null });
|
||||
expect(snapshot.warning).toContain("invalid queued-job total");
|
||||
expect(evaluateRunnerClaimHealth(snapshot).status).toBe("incomplete");
|
||||
});
|
||||
|
||||
it("marks the queued health scan incomplete when a later page fails", async () => {
|
||||
const queuedJobs = Array.from({ length: 60 }, (_, index) => job({ id: 1_000 - index, status: "queued" }));
|
||||
const listUserActionRunners = vi.fn().mockResolvedValue({ total_count: 0, runners: [] });
|
||||
const listUserActionJobs = vi.fn().mockImplementation(async (input: { status?: string; limit?: number; page?: number } = {}) => {
|
||||
if (input.status === "queued") {
|
||||
if (input.page === 2) throw new Error("page two unavailable");
|
||||
return { total_count: queuedJobs.length, jobs: queuedJobs.slice(0, input.limit ?? 50) };
|
||||
}
|
||||
return { total_count: 0, jobs: [] };
|
||||
});
|
||||
const load = createGiteaActivityRuntimeLoader(env, { client: { listUserActionRunners, listUserActionJobs } });
|
||||
|
||||
const snapshot = await load();
|
||||
|
||||
expect(snapshot.runnableJobScan).toEqual({ complete: false, observed: 50, total: 60 });
|
||||
expect(snapshot.warning).toContain("page two unavailable");
|
||||
expect(evaluateRunnerClaimHealth(snapshot).status).toBe("incomplete");
|
||||
});
|
||||
});
|
||||
@@ -3,7 +3,7 @@ import { Client } from "@modelcontextprotocol/sdk/client/index.js";
|
||||
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
|
||||
import { loadEnv } from "../src/config/env.js";
|
||||
import { compareAutoModeProjectRows, createProjectBrainMcpServer } from "../src/mcp/autonomyServer.js";
|
||||
import { autonomyRetryDelayMinutes, canResumeAutonomyLease, classifyAutonomyBlocker, enqueueAutonomyTaskInState, evaluateAutonomyRunBudget, isAutonomyLeaseOwner, normalizeAutonomyState, parseRoadmapMarkdown, releaseActiveAutonomyLease, retryAutonomyBlockerIfDue } from "../src/repositories/autonomyRepository.js";
|
||||
import { autonomyRetryDelayMinutes, canResumeAutonomyLease, classifyAutonomyBlocker, enqueueAutonomyTaskInState, evaluateAutonomyRunBudget, isAutonomyLeaseOwner, normalizeAutonomyState, parseRoadmapMarkdown, releaseActiveAutonomyLease, reorderAutonomyQueueInState, resetAutonomyQueueOrderInState, retryAutonomyBlockerIfDue } from "../src/repositories/autonomyRepository.js";
|
||||
import { autonomyHeartbeatSlotId, compactAutonomyState, resolveAutonomyHeartbeat, summarizeAutonomyProject } from "../src/services/autonomyRuntime.js";
|
||||
|
||||
const env = loadEnv({
|
||||
@@ -99,6 +99,25 @@ describe("Auto Mode state safety", () => {
|
||||
expect(queued.state.roadmap?.phases[1].items[1].title).toBe("Lopend werk");
|
||||
});
|
||||
|
||||
it("persists an exact owner-defined order for pending executable work only", () => {
|
||||
const roadmap = parseRoadmapMarkdown("## Fase\n- [ ] A\n- [ ] B\n- [ ] C");
|
||||
roadmap.phases[0].items[2].scope = "future";
|
||||
const current = normalizeAutonomyState({ enabled: true, roadmap }, "owner");
|
||||
const [a, b] = current.roadmap!.phases[0].items;
|
||||
|
||||
const reordered = reorderAutonomyQueueInState(current, [b.id, a.id], "owner");
|
||||
|
||||
expect(reordered.roadmap?.phases[0].items[0].manualOrder).toBe(1);
|
||||
expect(reordered.roadmap?.phases[0].items[1].manualOrder).toBe(0);
|
||||
expect(reordered.roadmap?.phases[0].items[2].manualOrder).toBeUndefined();
|
||||
expect(reordered.events.at(-1)?.type).toBe("autonomy.queue.reordered");
|
||||
expect(() => reorderAutonomyQueueInState(current, [a.id], "owner")).toThrow("AUTO_MODE_QUEUE_CHANGED");
|
||||
|
||||
const automatic = resetAutonomyQueueOrderInState(reordered, "owner");
|
||||
expect(automatic.roadmap?.phases.flatMap((phase) => phase.items).every((item) => item.manualOrder === undefined)).toBe(true);
|
||||
expect(automatic.events.at(-1)?.type).toBe("autonomy.queue.order_reset");
|
||||
});
|
||||
|
||||
it("derives the public status from durable facts instead of stale stored copy", () => {
|
||||
expect(normalizeAutonomyState({ enabled: true, status: "running" }).status).toBe("waiting_roadmap");
|
||||
expect(normalizeAutonomyState({}).profile).toBe("normal");
|
||||
@@ -305,8 +324,11 @@ describe("Auto Mode MCP surface", () => {
|
||||
"list_auto_mode_projects",
|
||||
"heartbeat_auto_executor",
|
||||
"get_auto_mode_state",
|
||||
"get_runner_claim_health",
|
||||
"configure_auto_mode",
|
||||
"import_auto_roadmap",
|
||||
"reorder_auto_queue",
|
||||
"reset_auto_queue_order",
|
||||
"resume_auto_mode",
|
||||
"claim_auto_work",
|
||||
"report_auto_work",
|
||||
@@ -314,6 +336,9 @@ describe("Auto Mode MCP surface", () => {
|
||||
expect(advertisedTools.find((tool) => tool.name === "claim_auto_work")?.inputSchema).toMatchObject({
|
||||
required: expect.arrayContaining(["project_id", "run_id"]),
|
||||
});
|
||||
expect(advertisedTools.find((tool) => tool.name === "reorder_auto_queue")?.inputSchema).toMatchObject({
|
||||
required: expect.arrayContaining(["project_id", "item_ids"]),
|
||||
});
|
||||
expect(advertisedTools.find((tool) => tool.name === "configure_auto_mode")?.inputSchema).toMatchObject({
|
||||
properties: expect.objectContaining({ max_work_items_per_run: expect.any(Object), max_run_minutes: expect.any(Object), max_recovery_minutes: expect.any(Object), pause_after_current: expect.any(Object), scheduling_policy: expect.any(Object) }),
|
||||
});
|
||||
|
||||
@@ -36,6 +36,27 @@ describe("smart autonomy scheduler", () => {
|
||||
expect(chooseNextAutonomyWork(roadmap.phases, 0, "downstream-impact")?.item.id).toBe(a.id);
|
||||
});
|
||||
|
||||
it("honors explicit manual queue order before automatic scoring", () => {
|
||||
const roadmap = parseRoadmapMarkdown("## Delivery\n- [ ] High score\n- [ ] Owner first");
|
||||
const [highScore, ownerFirst] = roadmap.phases[0].items;
|
||||
highScore.priority = 100;
|
||||
ownerFirst.priority = 0;
|
||||
highScore.manualOrder = 1;
|
||||
ownerFirst.manualOrder = 0;
|
||||
|
||||
expect(chooseNextAutonomyWork(roadmap.phases, 0)?.item.id).toBe(ownerFirst.id);
|
||||
});
|
||||
|
||||
it("never lets manual ordering bypass an unmet dependency", () => {
|
||||
const roadmap = parseRoadmapMarkdown("## Delivery\n- [ ] Prerequisite\n- [ ] Manually first");
|
||||
const [prerequisite, manuallyFirst] = roadmap.phases[0].items;
|
||||
manuallyFirst.dependsOn = [prerequisite.id];
|
||||
manuallyFirst.manualOrder = 0;
|
||||
prerequisite.manualOrder = 1;
|
||||
|
||||
expect(chooseNextAutonomyWork(roadmap.phases, 0)?.item.id).toBe(prerequisite.id);
|
||||
});
|
||||
|
||||
it("does not claim work while its external dependency is unavailable", () => {
|
||||
const roadmap = parseRoadmapMarkdown("## Delivery\n- [ ] A\n- [ ] B");
|
||||
roadmap.phases[0].items[0].waitingExternal = true;
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import { afterAll, afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { Prisma } from "@prisma/client";
|
||||
import { prisma } from "../src/db/prisma.js";
|
||||
import { configureAutonomy, getAutonomyState, importAutonomyRoadmap, reportAutonomyWork } from "../src/repositories/autonomyRepository.js";
|
||||
import { claimAutonomyWork } from "../src/services/autonomyRuntimeControl.js";
|
||||
@@ -11,12 +13,204 @@ import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
|
||||
import { createProjectBrainMcpServer } from "../src/mcp/server.js";
|
||||
import { loadEnv } from "../src/config/env.js";
|
||||
import { GiteaApiError, GiteaClient } from "../src/integrations/gitea/client.js";
|
||||
import { claimReadyWork, getProjectWorkGraph, reclaimExpiredWork, recordWorkCompletion, syncProjectWorkGraph, updateWorkDependencies } from "../src/services/workGraph.js";
|
||||
|
||||
const databaseUrl = process.env.TEST_DATABASE_URL;
|
||||
if (databaseUrl) process.env.DATABASE_URL = databaseUrl;
|
||||
const databaseDescribe = databaseUrl ? describe : describe.skip;
|
||||
const slugs: string[] = [];
|
||||
|
||||
const createWorkGraphProject = async (markdown: string) => {
|
||||
const slug = `workgraph-${randomUUID()}`;
|
||||
slugs.push(slug);
|
||||
await createProject({ id: slug, name: "Work graph concurrency proof", description: "Isolated database regression" });
|
||||
await configureAutonomy(slug, { enabled: true, autoBrainstorm: false }, "database-test");
|
||||
await importAutonomyRoadmap(slug, { markdown }, "database-test");
|
||||
const project = await prisma.project.findUniqueOrThrow({ where: { slug } });
|
||||
const graph = await getProjectWorkGraph(slug);
|
||||
return { slug, project, graph };
|
||||
};
|
||||
|
||||
const nodeByTitle = (graph: Awaited<ReturnType<typeof getProjectWorkGraph>>, title: string) => {
|
||||
const node = graph.nodes.find((candidate) => candidate.title === title);
|
||||
if (!node) throw new Error(`Missing work graph fixture node '${title}'.`);
|
||||
return node;
|
||||
};
|
||||
|
||||
const setAutonomyDependencies = async (projectId: string, slug: string, title: string, dependsOnTitles: string[]) => {
|
||||
const state = await getAutonomyState(slug);
|
||||
const items = state.roadmap?.phases.flatMap((phase) => phase.items) ?? [];
|
||||
const target = items.find((item) => item.title === title);
|
||||
if (!target) throw new Error(`Missing autonomy fixture item '${title}'.`);
|
||||
target.dependsOn = dependsOnTitles.map((dependencyTitle) => {
|
||||
const dependency = items.find((item) => item.title === dependencyTitle);
|
||||
if (!dependency) throw new Error(`Missing autonomy fixture dependency '${dependencyTitle}'.`);
|
||||
return dependency.id;
|
||||
});
|
||||
const workspace = await prisma.workspaceState.findUniqueOrThrow({ where: { projectId }, select: { notesJson: true } });
|
||||
const notes = typeof workspace.notesJson === "object" && workspace.notesJson !== null && !Array.isArray(workspace.notesJson) ? workspace.notesJson : {};
|
||||
await prisma.workspaceState.update({ where: { projectId }, data: { notesJson: { ...notes, autonomy: state } as Prisma.InputJsonValue } });
|
||||
};
|
||||
|
||||
const setAutonomyPriority = async (projectId: string, slug: string, title: string, priority: number) => {
|
||||
const state = await getAutonomyState(slug);
|
||||
const target = state.roadmap?.phases.flatMap((phase) => phase.items).find((item) => item.title === title);
|
||||
if (!target) throw new Error(`Missing autonomy fixture item '${title}'.`);
|
||||
target.priority = priority;
|
||||
const workspace = await prisma.workspaceState.findUniqueOrThrow({ where: { projectId }, select: { notesJson: true } });
|
||||
const notes = typeof workspace.notesJson === "object" && workspace.notesJson !== null && !Array.isArray(workspace.notesJson) ? workspace.notesJson : {};
|
||||
await prisma.workspaceState.update({ where: { projectId }, data: { notesJson: { ...notes, autonomy: state } as Prisma.InputJsonValue } });
|
||||
};
|
||||
|
||||
const setAutonomyRoadmapOrder = async (projectId: string, slug: string, titles: string[]) => {
|
||||
const state = await getAutonomyState(slug);
|
||||
const itemOrder = new Map(titles.map((title, index) => [title, index]));
|
||||
const items = state.roadmap?.phases.flatMap((phase) => phase.items) ?? [];
|
||||
if (items.length !== titles.length || titles.some((title) => !items.some((item) => item.title === title))) {
|
||||
throw new Error("The requested roadmap order does not match the source fixture.");
|
||||
}
|
||||
for (const phase of state.roadmap?.phases ?? []) {
|
||||
phase.items.sort((left, right) => itemOrder.get(left.title)! - itemOrder.get(right.title)!);
|
||||
}
|
||||
const workspace = await prisma.workspaceState.findUniqueOrThrow({ where: { projectId }, select: { notesJson: true } });
|
||||
const notes = typeof workspace.notesJson === "object" && workspace.notesJson !== null && !Array.isArray(workspace.notesJson) ? workspace.notesJson : {};
|
||||
await prisma.workspaceState.update({ where: { projectId }, data: { notesJson: { ...notes, autonomy: state } as Prisma.InputJsonValue } });
|
||||
};
|
||||
|
||||
type DatabaseLockWaiter = { pid: number; query: string; blockingPids: number[]; hasProjectRowShareLock: boolean };
|
||||
const readDatabaseLockWaiters = () => prisma.$queryRaw<DatabaseLockWaiter[]>`
|
||||
SELECT activity.pid, activity.query, pg_blocking_pids(activity.pid) AS "blockingPids",
|
||||
EXISTS (
|
||||
SELECT 1 FROM pg_locks
|
||||
WHERE pg_locks.pid = activity.pid
|
||||
AND pg_locks.relation = '"Project"'::regclass
|
||||
AND pg_locks.mode = 'RowShareLock'
|
||||
AND pg_locks.granted
|
||||
) AS "hasProjectRowShareLock"
|
||||
FROM pg_stat_activity AS activity
|
||||
WHERE activity.datname = current_database()
|
||||
AND activity.pid <> pg_backend_pid()
|
||||
AND activity.state = 'active'
|
||||
AND activity.wait_event_type = 'Lock'
|
||||
`;
|
||||
|
||||
const isProjectCoordinationLockQuery = (query: string) => {
|
||||
const normalized = query.toLowerCase();
|
||||
return normalized.includes("for update") || normalized.includes("for no key update");
|
||||
};
|
||||
|
||||
const waitForDatabaseLockWaiter = async (predicate: (waiter: DatabaseLockWaiter) => boolean, timeoutMs = 2000) => {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
const waiter = (await readDatabaseLockWaiters()).find(predicate);
|
||||
if (waiter) return waiter;
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
return undefined;
|
||||
};
|
||||
|
||||
const createAuditInsertBarrier = async (projectId: string, eventType: "work.completed" | "work.reclaimed") => {
|
||||
const suffix = randomUUID().replaceAll("-", "");
|
||||
const triggerName = `test_audit_${suffix}`;
|
||||
const functionName = `${triggerName}_fn`;
|
||||
const advisoryKey = [Number.parseInt(suffix.slice(0, 7), 16), Number.parseInt(suffix.slice(7, 14), 16)] as const;
|
||||
let blockerPid = 0;
|
||||
let releaseAdvisoryLock: () => void = () => undefined;
|
||||
let acquiredAdvisoryLock!: () => void;
|
||||
let functionCreated = false;
|
||||
let triggerCreated = false;
|
||||
let lockHolder: Promise<void> | undefined;
|
||||
const locked = new Promise<void>((resolve) => { acquiredAdvisoryLock = resolve; });
|
||||
const hold = new Promise<void>((resolve) => { releaseAdvisoryLock = resolve; });
|
||||
const cleanup = async () => {
|
||||
releaseAdvisoryLock();
|
||||
if (lockHolder) await lockHolder.catch(() => undefined);
|
||||
if (triggerCreated) await prisma.$executeRawUnsafe(`DROP TRIGGER IF EXISTS ${triggerName} ON "WorkGraphAuditEvent"`);
|
||||
if (functionCreated) await prisma.$executeRawUnsafe(`DROP FUNCTION IF EXISTS ${functionName}()`);
|
||||
};
|
||||
try {
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE FUNCTION ${functionName}() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
BEGIN
|
||||
PERFORM pg_advisory_xact_lock(${advisoryKey[0]}, ${advisoryKey[1]});
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$
|
||||
`);
|
||||
functionCreated = true;
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE TRIGGER ${triggerName}
|
||||
BEFORE INSERT ON "WorkGraphAuditEvent"
|
||||
FOR EACH ROW WHEN (NEW."projectId" = '${projectId}' AND NEW."eventType" = '${eventType}')
|
||||
EXECUTE FUNCTION ${functionName}();
|
||||
`);
|
||||
triggerCreated = true;
|
||||
lockHolder = prisma.$transaction(async (transaction) => {
|
||||
const [backend] = await transaction.$queryRaw<Array<{ pid: number }>>`SELECT pg_backend_pid() AS pid`;
|
||||
blockerPid = backend!.pid;
|
||||
await transaction.$queryRaw`SELECT pg_advisory_xact_lock(${advisoryKey[0]}::integer, ${advisoryKey[1]}::integer) IS NULL`;
|
||||
acquiredAdvisoryLock();
|
||||
await hold;
|
||||
});
|
||||
await Promise.race([locked, lockHolder.then(() => { throw new Error("The advisory lock holder ended before acquiring its lock."); })]);
|
||||
return { blockerPid, release: releaseAdvisoryLock, cleanup };
|
||||
} catch (error) {
|
||||
await cleanup();
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
const createWorkGraphNodeUpdateBarrier = async (nodeId: string) => {
|
||||
const suffix = randomUUID().replaceAll("-", "");
|
||||
const triggerName = `test_node_${suffix}`;
|
||||
const functionName = `${triggerName}_fn`;
|
||||
const advisoryKey = [Number.parseInt(suffix.slice(0, 7), 16), Number.parseInt(suffix.slice(7, 14), 16)] as const;
|
||||
let blockerPid = 0;
|
||||
let releaseAdvisoryLock: () => void = () => undefined;
|
||||
let acquiredAdvisoryLock!: () => void;
|
||||
let functionCreated = false;
|
||||
let triggerCreated = false;
|
||||
let lockHolder: Promise<void> | undefined;
|
||||
const locked = new Promise<void>((resolve) => { acquiredAdvisoryLock = resolve; });
|
||||
const hold = new Promise<void>((resolve) => { releaseAdvisoryLock = resolve; });
|
||||
const cleanup = async () => {
|
||||
releaseAdvisoryLock();
|
||||
if (lockHolder) await lockHolder.catch(() => undefined);
|
||||
if (triggerCreated) await prisma.$executeRawUnsafe(`DROP TRIGGER IF EXISTS ${triggerName} ON "WorkGraphNode"`);
|
||||
if (functionCreated) await prisma.$executeRawUnsafe(`DROP FUNCTION IF EXISTS ${functionName}()`);
|
||||
};
|
||||
try {
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE FUNCTION ${functionName}() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
BEGIN
|
||||
PERFORM pg_advisory_xact_lock(${advisoryKey[0]}, ${advisoryKey[1]});
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$
|
||||
`);
|
||||
functionCreated = true;
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE TRIGGER ${triggerName}
|
||||
BEFORE UPDATE ON "WorkGraphNode"
|
||||
FOR EACH ROW WHEN (NEW."id" = '${nodeId}')
|
||||
EXECUTE FUNCTION ${functionName}();
|
||||
`);
|
||||
triggerCreated = true;
|
||||
lockHolder = prisma.$transaction(async (transaction) => {
|
||||
const [backend] = await transaction.$queryRaw<Array<{ pid: number }>>`SELECT pg_backend_pid() AS pid`;
|
||||
blockerPid = backend!.pid;
|
||||
await transaction.$queryRaw`SELECT pg_advisory_xact_lock(${advisoryKey[0]}::integer, ${advisoryKey[1]}::integer) IS NULL`;
|
||||
acquiredAdvisoryLock();
|
||||
await hold;
|
||||
});
|
||||
await Promise.race([locked, lockHolder.then(() => { throw new Error("The advisory lock holder ended before acquiring its lock."); })]);
|
||||
return { blockerPid, release: releaseAdvisoryLock, cleanup };
|
||||
} catch (error) {
|
||||
await cleanup();
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
databaseDescribe("autonomy database write isolation", () => {
|
||||
it("preserves progress and the active lease when the same roadmap is imported again", async () => {
|
||||
const slug = `autonomy-import-${Date.now()}`;
|
||||
@@ -107,7 +301,10 @@ databaseDescribe("autonomy database write isolation", () => {
|
||||
vi.spyOn(GiteaClient.prototype, "getBranch").mockResolvedValue({ name: "chatgpt/existing", commit: { id: head, message: "Repair" }, protected: false });
|
||||
vi.spyOn(GiteaClient.prototype, "getContents").mockResolvedValue({ type: "file", path: "src/feature.ts", name: "feature.ts", sha: "file-sha", size: 4, encoding: "base64", content: "dGVzdA==" });
|
||||
const write = vi.spyOn(GiteaClient.prototype, "updateFile").mockResolvedValue({});
|
||||
vi.spyOn(GiteaClient.prototype, "listActionRuns").mockResolvedValue({ total_count: 1, workflow_runs: [{ id: 2, run_number: 2, display_title: "CI", path: ".gitea/workflows/ci.yml", event: "pull_request", head_sha: head, head_branch: "chatgpt/existing", status: "completed", conclusion: "success", html_url: "https://gitea.example.com/runs/2" }] });
|
||||
vi.spyOn(GiteaClient.prototype, "listActionRuns").mockResolvedValue({ total_count: 2, workflow_runs: [
|
||||
{ id: 2, run_number: 2, display_title: "Managed validation / full", path: ".gitea/workflows/managed-validation.yml", event: "pull_request", head_sha: head, head_branch: "chatgpt/existing", status: "completed", conclusion: "success", html_url: "https://gitea.example.com/runs/2" },
|
||||
{ id: 3, run_number: 3, display_title: "CI", path: ".gitea/workflows/ci.yml", event: "pull_request", head_sha: head, head_branch: "chatgpt/existing", status: "completed", conclusion: "success", html_url: "https://gitea.example.com/runs/3" },
|
||||
] });
|
||||
vi.spyOn(GiteaClient.prototype, "getBranchProtection").mockResolvedValue({ enable_status_check: true, status_check_contexts: ["CI / test"] });
|
||||
vi.spyOn(GiteaClient.prototype, "getCombinedCommitStatus").mockResolvedValue({ state: "success", statuses: [{ context: "CI / test", status: "success" }] });
|
||||
const merge = vi.spyOn(GiteaClient.prototype, "mergePullRequest").mockImplementation(async () => { remoteMerged = true; });
|
||||
@@ -162,4 +359,379 @@ databaseDescribe("autonomy database write isolation", () => {
|
||||
expect(integrated.spec.validationEvidence).toBeUndefined();
|
||||
expect(compare).toHaveBeenLastCalledWith("AutonomyProof", slug, "b".repeat(40), "a".repeat(40));
|
||||
});
|
||||
|
||||
it("serializes concurrent dependency edits before checking for cycles", async () => {
|
||||
const { slug, project, graph } = await createWorkGraphProject("## Work\n- Task A\n- Task B");
|
||||
const taskA = nodeByTitle(graph, "Task A");
|
||||
const taskB = nodeByTitle(graph, "Task B");
|
||||
const suffix = randomUUID().replaceAll("-", "");
|
||||
const triggerName = `test_cycle_${suffix}`;
|
||||
const functionName = `${triggerName}_fn`;
|
||||
const advisoryKey = [Number.parseInt(suffix.slice(0, 7), 16), Number.parseInt(suffix.slice(7, 14), 16)] as const;
|
||||
let blockerPid = 0;
|
||||
let releaseAdvisoryLock: () => void = () => undefined;
|
||||
let acquiredAdvisoryLock!: () => void;
|
||||
let functionCreated = false;
|
||||
let triggerCreated = false;
|
||||
const locked = new Promise<void>((resolve) => { acquiredAdvisoryLock = resolve; });
|
||||
const hold = new Promise<void>((resolve) => { releaseAdvisoryLock = resolve; });
|
||||
let lockHolder: Promise<void> | undefined;
|
||||
let edits: Array<Promise<Awaited<ReturnType<typeof updateWorkDependencies>>>> | undefined;
|
||||
try {
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE FUNCTION ${functionName}() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
BEGIN
|
||||
PERFORM pg_advisory_xact_lock(${advisoryKey[0]}, ${advisoryKey[1]});
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$
|
||||
`);
|
||||
functionCreated = true;
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE TRIGGER ${triggerName}
|
||||
BEFORE INSERT ON "WorkGraphDependency"
|
||||
FOR EACH ROW WHEN (NEW."projectId" = '${project.id}')
|
||||
EXECUTE FUNCTION ${functionName}();
|
||||
`);
|
||||
triggerCreated = true;
|
||||
lockHolder = prisma.$transaction(async (transaction) => {
|
||||
const [backend] = await transaction.$queryRaw<Array<{ pid: number }>>`SELECT pg_backend_pid() AS pid`;
|
||||
blockerPid = backend!.pid;
|
||||
await transaction.$queryRaw`SELECT pg_advisory_xact_lock(${advisoryKey[0]}::integer, ${advisoryKey[1]}::integer) IS NULL`;
|
||||
acquiredAdvisoryLock();
|
||||
await hold;
|
||||
});
|
||||
await Promise.race([locked, lockHolder.then(() => { throw new Error("The advisory lock holder ended before acquiring its lock."); })]);
|
||||
edits = [
|
||||
updateWorkDependencies(slug, taskA.id, [taskB.id], "database-test"),
|
||||
updateWorkDependencies(slug, taskB.id, [taskA.id], "database-test"),
|
||||
];
|
||||
const waitDeadline = Date.now() + 2000;
|
||||
let bothAtSerializedBoundary = false;
|
||||
while (!bothAtSerializedBoundary && Date.now() < waitDeadline) {
|
||||
const waiters = await readDatabaseLockWaiters();
|
||||
const blockedInserts = waiters.filter((row) => row.query.toLowerCase().includes("workgraphdependency") && row.blockingPids.includes(blockerPid));
|
||||
const projectLockWaiters = waiters.filter((row) => isProjectCoordinationLockQuery(row.query));
|
||||
const baselinePair = blockedInserts.length >= 2;
|
||||
const serializedPair = blockedInserts.length === 1 && projectLockWaiters.some((row) => row.blockingPids.includes(blockedInserts[0]!.pid));
|
||||
bothAtSerializedBoundary = baselinePair || serializedPair;
|
||||
if (!bothAtSerializedBoundary) await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
expect(bothAtSerializedBoundary).toBe(true);
|
||||
releaseAdvisoryLock();
|
||||
const results = await Promise.allSettled(edits);
|
||||
expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(1);
|
||||
const rejected = results.filter((result): result is PromiseRejectedResult => result.status === "rejected");
|
||||
expect(rejected).toHaveLength(1);
|
||||
expect(rejected[0]?.reason).toMatchObject({ message: "WORK_GRAPH_DEPENDENCY_CYCLE" });
|
||||
const finalGraph = await getProjectWorkGraph(slug);
|
||||
const finalA = nodeByTitle(finalGraph, "Task A");
|
||||
const finalB = nodeByTitle(finalGraph, "Task B");
|
||||
expect(finalA.dependencies.some((edge) => edge.dependsOnNodeId === finalB.id)).not.toBe(finalB.dependencies.some((edge) => edge.dependsOnNodeId === finalA.id));
|
||||
} finally {
|
||||
releaseAdvisoryLock();
|
||||
if (lockHolder) await lockHolder.catch(() => undefined);
|
||||
if (edits) await Promise.allSettled(edits);
|
||||
if (triggerCreated) await prisma.$executeRawUnsafe(`DROP TRIGGER IF EXISTS ${triggerName} ON "WorkGraphDependency"`);
|
||||
if (functionCreated) await prisma.$executeRawUnsafe(`DROP FUNCTION IF EXISTS ${functionName}()`);
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it("does not attach a new unresolved prerequisite to work with an active lease", async () => {
|
||||
const { slug, project } = await createWorkGraphProject("## Work\n- Runnable task\n- Prerequisite task");
|
||||
await setAutonomyPriority(project.id, slug, "Runnable task", 100);
|
||||
const currentGraph = await getProjectWorkGraph(slug);
|
||||
const task = nodeByTitle(currentGraph, "Runnable task");
|
||||
const prerequisite = nodeByTitle(currentGraph, "Prerequisite task");
|
||||
const beforeDependencies = await prisma.workGraphDependency.findMany({ where: { projectId: project.id, nodeId: task.id }, select: { dependsOnNodeId: true, dependencyType: true, createdBy: true } });
|
||||
expect(beforeDependencies).toHaveLength(0);
|
||||
const claim = await claimReadyWork(slug, "claimant");
|
||||
expect(claim?.node.id).toBe(task.id);
|
||||
expect(claim?.node.dependencies.map((edge) => edge.dependsOnNodeId)).toEqual(beforeDependencies.map((edge) => edge.dependsOnNodeId));
|
||||
await expect(updateWorkDependencies(slug, task.id, [prerequisite.id], "database-test")).rejects.toThrow("WORK_GRAPH_ACTIVE_LEASE");
|
||||
const persisted = await prisma.workGraphDependency.findMany({ where: { projectId: project.id, nodeId: task.id }, select: { dependsOnNodeId: true, dependencyType: true, createdBy: true } });
|
||||
expect(persisted).toEqual(beforeDependencies);
|
||||
});
|
||||
|
||||
it("serializes a dependency edit behind an in-flight claim", async () => {
|
||||
const { slug, project } = await createWorkGraphProject("## Work\n- Runnable task\n- Prerequisite task");
|
||||
await setAutonomyPriority(project.id, slug, "Runnable task", 100);
|
||||
const currentGraph = await getProjectWorkGraph(slug);
|
||||
const task = nodeByTitle(currentGraph, "Runnable task");
|
||||
const prerequisite = nodeByTitle(currentGraph, "Prerequisite task");
|
||||
const suffix = randomUUID().replaceAll("-", "");
|
||||
const triggerName = `test_claim_${suffix}`;
|
||||
const functionName = `${triggerName}_fn`;
|
||||
const advisoryKey = [Number.parseInt(suffix.slice(0, 7), 16), Number.parseInt(suffix.slice(7, 14), 16)] as const;
|
||||
|
||||
let releaseAdvisoryLock: () => void = () => undefined;
|
||||
let acquiredAdvisoryLock!: () => void;
|
||||
let blockerPid = 0;
|
||||
let functionCreated = false;
|
||||
let triggerCreated = false;
|
||||
const beforeDependencies = await prisma.workGraphDependency.findMany({ where: { projectId: project.id, nodeId: task.id }, select: { dependsOnNodeId: true, dependencyType: true, createdBy: true } });
|
||||
expect(beforeDependencies).toHaveLength(0);
|
||||
const locked = new Promise<void>((resolve) => { acquiredAdvisoryLock = resolve; });
|
||||
const hold = new Promise<void>((resolve) => { releaseAdvisoryLock = resolve; });
|
||||
let claim: ReturnType<typeof claimReadyWork> | undefined;
|
||||
let edit: Promise<{ status: "fulfilled" } | { status: "rejected"; error: unknown }> | undefined;
|
||||
let editSettled = false;
|
||||
let lockHolder: Promise<void> | undefined;
|
||||
try {
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE FUNCTION ${functionName}() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
BEGIN
|
||||
PERFORM pg_advisory_xact_lock(${advisoryKey[0]}, ${advisoryKey[1]});
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$
|
||||
`);
|
||||
functionCreated = true;
|
||||
await prisma.$executeRawUnsafe(`
|
||||
CREATE TRIGGER ${triggerName}
|
||||
BEFORE UPDATE ON "WorkGraphNode"
|
||||
FOR EACH ROW WHEN (NEW."projectId" = '${project.id}' AND NEW."owner" IS NOT NULL AND OLD."owner" IS DISTINCT FROM NEW."owner")
|
||||
EXECUTE FUNCTION ${functionName}();
|
||||
`);
|
||||
triggerCreated = true;
|
||||
lockHolder = prisma.$transaction(async (transaction) => {
|
||||
const [backend] = await transaction.$queryRaw<Array<{ pid: number }>>`SELECT pg_backend_pid() AS pid`;
|
||||
blockerPid = backend!.pid;
|
||||
await transaction.$queryRaw`SELECT pg_advisory_xact_lock(${advisoryKey[0]}::integer, ${advisoryKey[1]}::integer) IS NULL`;
|
||||
acquiredAdvisoryLock();
|
||||
await hold;
|
||||
});
|
||||
await Promise.race([locked, lockHolder!.then(() => { throw new Error("The advisory lock holder ended before acquiring its lock."); })]);
|
||||
claim = claimReadyWork(slug, "claimant");
|
||||
const waitDeadline = Date.now() + 2000;
|
||||
let claimWaiter: DatabaseLockWaiter | undefined;
|
||||
while (!claimWaiter && Date.now() < waitDeadline) {
|
||||
claimWaiter = (await readDatabaseLockWaiters()).find((row) => row.query.toLowerCase().includes("update") && row.query.toLowerCase().includes("workgraphnode") && row.blockingPids.includes(blockerPid));
|
||||
if (!claimWaiter) await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
expect(claimWaiter).toBeDefined();
|
||||
|
||||
edit = updateWorkDependencies(slug, task.id, [prerequisite.id], "database-test").then(
|
||||
() => ({ status: "fulfilled" as const }),
|
||||
(error: unknown) => ({ status: "rejected" as const, error }),
|
||||
).finally(() => { editSettled = true; });
|
||||
const editWaitDeadline = Date.now() + 2000;
|
||||
let dependencyEditWaiter: DatabaseLockWaiter | undefined;
|
||||
while (!dependencyEditWaiter && !editSettled && Date.now() < editWaitDeadline) {
|
||||
dependencyEditWaiter = (await readDatabaseLockWaiters()).find((row) => isProjectCoordinationLockQuery(row.query) && row.blockingPids.includes(claimWaiter!.pid));
|
||||
if (!dependencyEditWaiter && !editSettled) await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
expect(editSettled).toBe(false);
|
||||
expect(dependencyEditWaiter).toBeDefined();
|
||||
} finally {
|
||||
releaseAdvisoryLock?.();
|
||||
if (lockHolder) await lockHolder.catch(() => undefined);
|
||||
if (claim) await claim.catch(() => undefined);
|
||||
if (edit) await edit.catch(() => undefined);
|
||||
if (triggerCreated) await prisma.$executeRawUnsafe(`DROP TRIGGER IF EXISTS ${triggerName} ON "WorkGraphNode"`);
|
||||
if (functionCreated) await prisma.$executeRawUnsafe(`DROP FUNCTION IF EXISTS ${functionName}()`);
|
||||
}
|
||||
|
||||
const claimResult = await claim!;
|
||||
expect(claimResult?.node.id).toBe(task.id);
|
||||
expect(claimResult?.node.dependencies.map((edge) => edge.dependsOnNodeId)).toEqual(beforeDependencies.map((edge) => edge.dependsOnNodeId));
|
||||
const editResult = await edit!;
|
||||
expect(editResult.status).toBe("rejected");
|
||||
if (editResult.status === "rejected") expect(editResult.error).toMatchObject({ message: "WORK_GRAPH_ACTIVE_LEASE" });
|
||||
const persisted = await prisma.workGraphDependency.findMany({ where: { projectId: project.id, nodeId: task.id }, select: { dependsOnNodeId: true, dependencyType: true, createdBy: true } });
|
||||
expect(persisted).toEqual(beforeDependencies);
|
||||
}, 15_000);
|
||||
|
||||
it("keeps completion audit foreign-key checks compatible with a waiting graph sync", async () => {
|
||||
const { slug, project, graph } = await createWorkGraphProject("## Work\n- Completing task\n- Other task");
|
||||
await setAutonomyPriority(project.id, slug, "Completing task", 100);
|
||||
const task = nodeByTitle(graph, "Completing task");
|
||||
const claim = await claimReadyWork(slug, "completion-worker");
|
||||
expect(claim?.node.id).toBe(task.id);
|
||||
const barrier = await createAuditInsertBarrier(project.id, "work.completed");
|
||||
let completion: ReturnType<typeof recordWorkCompletion> | undefined;
|
||||
let sync: ReturnType<typeof syncProjectWorkGraph> | undefined;
|
||||
try {
|
||||
completion = recordWorkCompletion(slug, task.id, "completion-worker", claim!.leaseToken, { summary: "Completion lock-order proof" });
|
||||
const completionWaiter = await waitForDatabaseLockWaiter((waiter) => waiter.query.toLowerCase().includes("workgraphauditevent") && waiter.blockingPids.includes(barrier.blockerPid));
|
||||
expect(completionWaiter).toBeDefined();
|
||||
|
||||
sync = syncProjectWorkGraph(slug);
|
||||
const syncWaiter = await waitForDatabaseLockWaiter((waiter) => waiter.query.toLowerCase().includes("workgraphnode") && waiter.blockingPids.includes(completionWaiter!.pid) && waiter.hasProjectRowShareLock);
|
||||
expect(syncWaiter).toBeDefined();
|
||||
|
||||
barrier.release();
|
||||
const results = await Promise.allSettled([completion, sync]);
|
||||
expect(results.map((result) => result.status)).toEqual(["fulfilled", "fulfilled"]);
|
||||
} finally {
|
||||
barrier.release();
|
||||
await Promise.allSettled([...(completion ? [completion] : []), ...(sync ? [sync] : [])]);
|
||||
await barrier.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it("keeps expired-work audit foreign-key checks compatible with a waiting claim", async () => {
|
||||
const { slug, project, graph } = await createWorkGraphProject("## Work\n- Expired task\n- Other task");
|
||||
await setAutonomyPriority(project.id, slug, "Expired task", 100);
|
||||
const task = nodeByTitle(graph, "Expired task");
|
||||
const initialClaim = await claimReadyWork(slug, "expired-worker");
|
||||
expect(initialClaim?.node.id).toBe(task.id);
|
||||
await prisma.workGraphNode.update({ where: { id: task.id }, data: { leaseExpiresAt: new Date(Date.now() - 60_000) } });
|
||||
const barrier = await createAuditInsertBarrier(project.id, "work.reclaimed");
|
||||
let reclaim: ReturnType<typeof reclaimExpiredWork> | undefined;
|
||||
let claimAfterExpiry: ReturnType<typeof claimReadyWork> | undefined;
|
||||
try {
|
||||
reclaim = reclaimExpiredWork(slug);
|
||||
const reclaimWaiter = await waitForDatabaseLockWaiter((waiter) => waiter.query.toLowerCase().includes("workgraphauditevent") && waiter.blockingPids.includes(barrier.blockerPid));
|
||||
expect(reclaimWaiter).toBeDefined();
|
||||
|
||||
claimAfterExpiry = claimReadyWork(slug, "reclaim-race-claimant");
|
||||
const claimWaiter = await waitForDatabaseLockWaiter((waiter) => isProjectCoordinationLockQuery(waiter.query) && waiter.blockingPids.includes(reclaimWaiter!.pid));
|
||||
expect(claimWaiter).toBeDefined();
|
||||
|
||||
barrier.release();
|
||||
const [reclaimResult, claimResult] = await Promise.all([reclaim, claimAfterExpiry]);
|
||||
expect(reclaimResult.reclaimedCount).toBe(1);
|
||||
expect(claimResult?.node.id).toBe(task.id);
|
||||
} finally {
|
||||
barrier.release();
|
||||
await Promise.allSettled([...(reclaim ? [reclaim] : []), ...(claimAfterExpiry ? [claimAfterExpiry] : [])]);
|
||||
await barrier.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it("serializes multi-node reclaim before it can invert locks with graph reconciliation", async () => {
|
||||
const { slug, project, graph } = await createWorkGraphProject("## Work\n- Task B\n- Task A");
|
||||
const taskA = nodeByTitle(graph, "Task A");
|
||||
const taskB = nodeByTitle(graph, "Task B");
|
||||
const now = Date.now();
|
||||
await prisma.workGraphNode.update({ where: { id: taskB.id }, data: { owner: "expired-worker", leaseToken: randomUUID(), leaseExpiresAt: new Date(now - 120_000), executionEligibility: "running", canonicalLifecycle: "running" } });
|
||||
await prisma.workGraphNode.update({ where: { id: taskA.id }, data: { owner: "expired-worker", leaseToken: randomUUID(), leaseExpiresAt: new Date(now - 60_000), executionEligibility: "running", canonicalLifecycle: "running" } });
|
||||
const reclaimWhere = { projectId: project.id, leaseExpiresAt: { lt: new Date() }, owner: { not: null }, executionEligibility: "running" as const };
|
||||
const reclaimOrder = await prisma.workGraphNode.findMany({ where: reclaimWhere, select: { id: true, title: true } });
|
||||
expect(reclaimOrder).toHaveLength(2);
|
||||
const graphOrder = reclaimOrder.map((node) => node.title).reverse();
|
||||
await setAutonomyRoadmapOrder(project.id, slug, graphOrder);
|
||||
|
||||
const graphFirstNode = graphOrder[0] === "Task A" ? taskA : taskB;
|
||||
const barrier = await createWorkGraphNodeUpdateBarrier(graphFirstNode.id);
|
||||
let sync: ReturnType<typeof syncProjectWorkGraph> | undefined;
|
||||
let reclaim: ReturnType<typeof reclaimExpiredWork> | undefined;
|
||||
try {
|
||||
sync = syncProjectWorkGraph(slug);
|
||||
const syncWaiter = await waitForDatabaseLockWaiter((waiter) => waiter.query.toLowerCase().includes("workgraphnode") && waiter.blockingPids.includes(barrier.blockerPid) && waiter.hasProjectRowShareLock);
|
||||
expect(syncWaiter).toBeDefined();
|
||||
|
||||
reclaim = reclaimExpiredWork(slug);
|
||||
const reclaimProjectWaiter = await waitForDatabaseLockWaiter((waiter) => isProjectCoordinationLockQuery(waiter.query) && waiter.blockingPids.includes(syncWaiter!.pid));
|
||||
const reclaimNodeWaiter = reclaimProjectWaiter ? undefined : await waitForDatabaseLockWaiter((waiter) => waiter.query.toLowerCase().includes("workgraphnode") && waiter.blockingPids.includes(syncWaiter!.pid));
|
||||
expect(reclaimProjectWaiter ?? reclaimNodeWaiter).toBeDefined();
|
||||
|
||||
barrier.release();
|
||||
if (reclaimNodeWaiter) {
|
||||
const graphReverseWaiter = await waitForDatabaseLockWaiter((waiter) => waiter.query.toLowerCase().includes("workgraphnode") && waiter.blockingPids.includes(reclaimNodeWaiter.pid) && waiter.hasProjectRowShareLock);
|
||||
expect(graphReverseWaiter).toBeDefined();
|
||||
}
|
||||
const results = await Promise.allSettled([sync, reclaim]);
|
||||
const failures = results.filter((result): result is PromiseRejectedResult => result.status === "rejected");
|
||||
expect(results.map((result) => result.status), failures.map((failure) => String(failure.reason)).join("\n")).toEqual(["fulfilled", "fulfilled"]);
|
||||
} finally {
|
||||
barrier.release();
|
||||
await Promise.allSettled([...(sync ? [sync] : []), ...(reclaim ? [reclaim] : [])]);
|
||||
await barrier.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it("keeps a live lease's derived dependencies until expiry, then applies the latest source edges", async () => {
|
||||
const { slug, project } = await createWorkGraphProject("## Work\n- Runnable task\n- Prerequisite task");
|
||||
await setAutonomyPriority(project.id, slug, "Runnable task", 100);
|
||||
const graph = await getProjectWorkGraph(slug);
|
||||
const task = nodeByTitle(graph, "Runnable task");
|
||||
const prerequisite = nodeByTitle(graph, "Prerequisite task");
|
||||
const claim = await claimReadyWork(slug, "claimant");
|
||||
expect(claim?.node.id).toBe(task.id);
|
||||
expect(claim?.node.dependencies).toHaveLength(0);
|
||||
|
||||
await setAutonomyDependencies(project.id, slug, "Runnable task", ["Prerequisite task"]);
|
||||
const duringLease = await getProjectWorkGraph(slug);
|
||||
const leasedTask = nodeByTitle(duringLease, "Runnable task");
|
||||
expect(leasedTask.executionEligibility).toBe("running");
|
||||
expect(leasedTask.dependencies).toHaveLength(0);
|
||||
|
||||
await prisma.workGraphNode.update({ where: { id: task.id }, data: { leaseExpiresAt: new Date(Date.now() - 60_000) } });
|
||||
const afterExpiry = await getProjectWorkGraph(slug);
|
||||
const blockedTask = nodeByTitle(afterExpiry, "Runnable task");
|
||||
expect(blockedTask.executionEligibility).toBe("waiting");
|
||||
expect(blockedTask.dependencies.map((edge) => edge.dependsOnNodeId)).toEqual([prerequisite.id]);
|
||||
const nextClaim = await claimReadyWork(slug, "next-claimant");
|
||||
expect(nextClaim?.node.id).toBe(prerequisite.id);
|
||||
expect(nextClaim?.node.id).not.toBe(task.id);
|
||||
}, 15_000);
|
||||
|
||||
it("keeps the old complete derived edge set visible until reconciliation commits", async () => {
|
||||
const { slug, project } = await createWorkGraphProject("## Work\n- Task A\n- Task B\n- Task C");
|
||||
await setAutonomyDependencies(project.id, slug, "Task A", ["Task B"]);
|
||||
const initialGraph = await getProjectWorkGraph(slug);
|
||||
const taskA = nodeByTitle(initialGraph, "Task A");
|
||||
const taskB = nodeByTitle(initialGraph, "Task B");
|
||||
const taskC = nodeByTitle(initialGraph, "Task C");
|
||||
await setAutonomyDependencies(project.id, slug, "Task A", ["Task C"]);
|
||||
|
||||
let releaseProjectLock!: () => void;
|
||||
let acquiredProjectLock!: () => void;
|
||||
let blockerPid = 0;
|
||||
const projectLocked = new Promise<void>((resolve) => { acquiredProjectLock = resolve; });
|
||||
const hold = new Promise<void>((resolve) => { releaseProjectLock = resolve; });
|
||||
const lockHolder = prisma.$transaction(async (transaction) => {
|
||||
const [backend] = await transaction.$queryRaw<Array<{ pid: number }>>`SELECT pg_backend_pid() AS pid`;
|
||||
blockerPid = backend!.pid;
|
||||
await transaction.$queryRaw`SELECT "id" FROM "Project" WHERE "id" = ${project.id} FOR UPDATE`;
|
||||
acquiredProjectLock();
|
||||
await hold;
|
||||
});
|
||||
let sync: Promise<string> | undefined;
|
||||
try {
|
||||
await Promise.race([projectLocked, lockHolder.then(() => { throw new Error("The project lock holder ended before acquiring its lock."); })]);
|
||||
let syncFinished = false;
|
||||
sync = syncProjectWorkGraph(slug).then((result) => { syncFinished = true; return result; });
|
||||
const waitDeadline = Date.now() + 5000;
|
||||
let waitingTransactions: DatabaseLockWaiter[] = [];
|
||||
while (!waitingTransactions.length && Date.now() < waitDeadline) {
|
||||
waitingTransactions = (await readDatabaseLockWaiters()).filter((row) => row.blockingPids.includes(blockerPid));
|
||||
if (!waitingTransactions.length) await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
expect(waitingTransactions.length).toBeGreaterThan(0);
|
||||
expect(syncFinished).toBe(false);
|
||||
const duringSync = await prisma.workGraphDependency.findMany({ where: { projectId: project.id, createdBy: "autonomy-sync" } });
|
||||
expect(duringSync.map((edge) => [edge.nodeId, edge.dependsOnNodeId])).toEqual([[taskA.id, taskB.id]]);
|
||||
} finally {
|
||||
releaseProjectLock();
|
||||
await lockHolder.catch(() => undefined);
|
||||
if (sync) await sync;
|
||||
}
|
||||
|
||||
const afterSync = await prisma.workGraphDependency.findMany({ where: { projectId: project.id, createdBy: "autonomy-sync" } });
|
||||
expect(afterSync.map((edge) => [edge.nodeId, edge.dependsOnNodeId])).toEqual([[taskA.id, taskC.id]]);
|
||||
}, 15_000);
|
||||
|
||||
it("rechecks prerequisites before treating an expired lease as reclaimable", async () => {
|
||||
const { slug, project, graph } = await createWorkGraphProject("## Work\n- Dependent task\n- Prerequisite task");
|
||||
const task = nodeByTitle(graph, "Dependent task");
|
||||
const prerequisite = nodeByTitle(graph, "Prerequisite task");
|
||||
await updateWorkDependencies(slug, task.id, [prerequisite.id], "database-test");
|
||||
await prisma.workGraphNode.update({ where: { id: task.id }, data: {
|
||||
owner: "expired-worker",
|
||||
leaseToken: randomUUID(),
|
||||
leaseExpiresAt: new Date(Date.now() - 60_000),
|
||||
executionEligibility: "running",
|
||||
canonicalLifecycle: "running",
|
||||
} });
|
||||
|
||||
const snapshot = await getProjectWorkGraph(slug);
|
||||
expect(nodeByTitle(snapshot, "Dependent task").executionEligibility).toBe("waiting");
|
||||
expect(nodeByTitle(snapshot, "Dependent task").dependencies.some((edge) => edge.dependsOnNodeId === prerequisite.id)).toBe(true);
|
||||
const claim = await claimReadyWork(slug, "claimant");
|
||||
expect(claim?.node.id).toBe(prerequisite.id);
|
||||
expect(claim?.node.id).not.toBe(task.id);
|
||||
});
|
||||
});
|
||||
@@ -107,13 +107,36 @@ describe("Gitea webhook verification", () => {
|
||||
});
|
||||
|
||||
describe("Gitea API client", () => {
|
||||
it("accepts Gitea's empty HTTP 200 merge response without inventing a failed merge", async () => {
|
||||
const fetcher = vi.fn(async (input: string | URL | Request) => String(input).endsWith("/version")
|
||||
? new Response(JSON.stringify({ version: "1.27.3" }), { status: 200 })
|
||||
: new Response(null, { status: 200 }));
|
||||
it.each([
|
||||
["1.14.0", "Do"],
|
||||
["1.14.7", "Do"],
|
||||
["1.25.5", "Do"],
|
||||
["1.26.4", "do"],
|
||||
["1.27.3", "do"],
|
||||
["28.0.0", "do"],
|
||||
["28.1.3", "do"],
|
||||
] as const)("uses the supported Gitea merge field for %s", async (version, expectedKey) => {
|
||||
const bodies: Record<string, unknown>[] = [];
|
||||
const fetcher = vi.fn(async (input: string | URL | Request, init?: RequestInit) => {
|
||||
if (String(input).endsWith("/version")) return new Response(JSON.stringify({ version }), { status: 200 });
|
||||
bodies.push(JSON.parse(String(init?.body ?? "{}")) as Record<string, unknown>);
|
||||
return new Response(null, { status: 200 });
|
||||
});
|
||||
const client = new GiteaClient("https://gitea.example.com", "test-token", fetcher as typeof fetch);
|
||||
await expect(client.mergePullRequest("Owner", "Repo", 7, { method: "merge", headCommitId: "a".repeat(40) })).resolves.toBeUndefined();
|
||||
expect(fetcher).toHaveBeenCalledTimes(2);
|
||||
expect(bodies[0]?.[expectedKey]).toBe("merge");
|
||||
expect(bodies[0]?.[expectedKey === "Do" ? "do" : "Do"]).toBeUndefined();
|
||||
});
|
||||
|
||||
it.each(["1.24.6", "29.0.0", "not-a-version", ""])("fails closed for unsupported Gitea merge version %s", async (version) => {
|
||||
const fetcher = vi.fn(async (input: string | URL | Request) => String(input).endsWith("/version")
|
||||
? new Response(JSON.stringify({ version }), { status: 200 })
|
||||
: new Response(null, { status: 200 }));
|
||||
const client = new GiteaClient("https://gitea.example.com", "test-token", fetcher as typeof fetch);
|
||||
await expect(client.mergePullRequest("Owner", "Repo", 7, { method: "merge", headCommitId: "a".repeat(40) }))
|
||||
.rejects.toMatchObject({ code: "GITEA_VERSION_UNSUPPORTED" });
|
||||
expect(fetcher).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
it("creates a private initialized repository through the bounded user endpoint", async () => {
|
||||
const fetcher = vi.fn(async () => new Response(JSON.stringify({
|
||||
@@ -509,11 +532,11 @@ describe("ChatGPT feature delivery approval", () => {
|
||||
|
||||
it("requires every Gitea Actions workflow for the exact branch head to succeed", () => {
|
||||
const run = (id: number, conclusion: string, headSha = "current-sha"): GiteaActionRun => ({
|
||||
id, run_number: id, display_title: "CI", path: ".gitea/workflows/ci.yml", event: "push", head_sha: headSha,
|
||||
id, run_number: id, display_title: "Managed validation / full", path: ".gitea/workflows/managed-validation.yml", event: "workflow_dispatch", head_sha: headSha,
|
||||
head_branch: "chatgpt/feature", status: "completed", conclusion, html_url: `https://gitea.example.com/runs/${id}`,
|
||||
});
|
||||
expect(assertSuccessfulActionRuns([run(1, "success"), run(2, "success")], "chatgpt/feature", "current-sha")).toHaveLength(2);
|
||||
expect(() => assertSuccessfulActionRuns([run(1, "success"), run(2, "failure")], "chatgpt/feature", "current-sha")).toThrow("FEATURE_DELIVERY_CI_NOT_SUCCESSFUL");
|
||||
expect(() => assertSuccessfulActionRuns([run(1, "success"), run(2, "failure")], "chatgpt/feature", "current-sha")).toThrow("FEATURE_DELIVERY_FULL_VALIDATION_REQUIRED");
|
||||
expect(() => assertSuccessfulActionRuns([run(1, "success", "stale-sha")], "chatgpt/feature", "current-sha")).toThrow("FEATURE_DELIVERY_CI_NOT_SUCCESSFUL");
|
||||
});
|
||||
});
|
||||
@@ -7,7 +7,7 @@ const branch = "codex/native-ci";
|
||||
const sha = "a".repeat(40);
|
||||
const context = "CI / build-test (pull_request)";
|
||||
const run = (id: number, overrides: Partial<GiteaActionRun> = {}): GiteaActionRun => ({
|
||||
id, run_number: id, display_title: "CI", path: ".gitea/workflows/ci.yml",
|
||||
id, run_number: id, display_title: "Managed validation / full", path: ".gitea/workflows/managed-validation.yml",
|
||||
event: "pull_request", head_sha: sha, head_branch: "", status: "completed",
|
||||
conclusion: "success", html_url: `https://gitea.example/runs/${id}`, ...overrides,
|
||||
});
|
||||
@@ -25,11 +25,11 @@ describe("native PR validation discovery", () => {
|
||||
});
|
||||
|
||||
it.each(["pending", "queued", "waiting", "in_progress"])("does not authorize %s", (status) => {
|
||||
expect(() => assertSuccessfulActionRuns([run(1, { status, conclusion: undefined })], branch, sha)).toThrow("FEATURE_DELIVERY_CI_NOT_SUCCESSFUL");
|
||||
expect(() => assertSuccessfulActionRuns([run(1, { status, conclusion: undefined })], branch, sha)).toThrow("FEATURE_DELIVERY_FULL_VALIDATION_REQUIRED");
|
||||
});
|
||||
|
||||
it.each(["cancelled", "canceled", "failure"])("does not authorize a %s job that never started", (conclusion) => {
|
||||
expect(() => assertSuccessfulActionRuns([run(1, { conclusion })], branch, sha)).toThrow("FEATURE_DELIVERY_CI_NOT_SUCCESSFUL");
|
||||
expect(() => assertSuccessfulActionRuns([run(1, { conclusion })], branch, sha)).toThrow("FEATURE_DELIVERY_FULL_VALIDATION_REQUIRED");
|
||||
});
|
||||
|
||||
it("uses the newest attempt of each workflow and preserves all evidence", () => {
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import Fastify from "fastify";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { loadEnv } from "../src/config/env.js";
|
||||
import { registerOperatingSystemRoutes } from "../src/routes/operatingSystem.js";
|
||||
|
||||
const mocks = vi.hoisted(() => ({ updateWorkDependencies: vi.fn() }));
|
||||
|
||||
vi.mock("../src/services/workGraph.js", () => ({
|
||||
claimReadyWork: vi.fn(),
|
||||
getProjectWorkGraph: vi.fn(),
|
||||
listWorkGraphHistory: vi.fn(),
|
||||
recordWorkCompletion: vi.fn(),
|
||||
reclaimExpiredWork: vi.fn(),
|
||||
releaseWork: vi.fn(),
|
||||
updateWorkDependencies: mocks.updateWorkDependencies,
|
||||
}));
|
||||
|
||||
const env = loadEnv({
|
||||
NODE_ENV: "test",
|
||||
PROJECTBRAIN_DATA_SOURCE: "database",
|
||||
DATABASE_URL: "postgresql://fixture:fixture@localhost:5432/projectbrain_fixture",
|
||||
PROJECTBRAIN_OWNER_USERNAME: "fixture-owner",
|
||||
PROJECTBRAIN_OWNER_PASSWORD: "fixture-only-owner-password",
|
||||
PROJECTBRAIN_AUTH_SECRET: "fixture-only-auth-secret-with-at-least-32-characters",
|
||||
});
|
||||
|
||||
describe("operating-system work-graph routes", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
it("maps an active work-graph lease conflict to 409 with the authenticated actor", async () => {
|
||||
mocks.updateWorkDependencies.mockRejectedValue(new Error("WORK_GRAPH_ACTIVE_LEASE"));
|
||||
const app = Fastify();
|
||||
app.decorateRequest("authUser", null);
|
||||
app.addHook("onRequest", async (request) => {
|
||||
request.authUser = {
|
||||
id: "member-id",
|
||||
username: "authenticated-member",
|
||||
role: "member",
|
||||
credentialVersion: 1,
|
||||
};
|
||||
});
|
||||
await registerOperatingSystemRoutes(app, { env });
|
||||
|
||||
try {
|
||||
const response = await app.inject({
|
||||
method: "PUT",
|
||||
url: "/api/projects/project-1/work-graph/node-1/dependencies",
|
||||
payload: { dependencyNodeIds: ["node-2"], actor: "forged-caller" },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(409);
|
||||
expect(response.json().error.code).toBe("WORK_GRAPH_ACTIVE_LEASE");
|
||||
expect(mocks.updateWorkDependencies).toHaveBeenCalledWith(
|
||||
"project-1",
|
||||
"node-1",
|
||||
["node-2"],
|
||||
"authenticated-member",
|
||||
);
|
||||
} finally {
|
||||
await app.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -15,9 +15,9 @@ const actionRun = (input: {
|
||||
}): GiteaActionRun => ({
|
||||
id: input.id,
|
||||
run_number: input.id,
|
||||
display_title: input.workflow ?? "CI",
|
||||
path: `.gitea/workflows/${input.workflow ?? "ci.yml"}`,
|
||||
event: "push",
|
||||
display_title: input.workflow ?? "Managed validation / full",
|
||||
path: `.gitea/workflows/${input.workflow ?? "managed-validation.yml"}`,
|
||||
event: input.workflow ? "pull_request" : "workflow_dispatch",
|
||||
head_sha: input.headSha ?? "current-sha",
|
||||
head_branch: "chatgpt/feature",
|
||||
status: input.status ?? "completed",
|
||||
@@ -46,6 +46,7 @@ describe("review remediation regressions", () => {
|
||||
|
||||
it("still fails closed when the latest run of any workflow is not successful", () => {
|
||||
expect(() => assertSuccessfulActionRuns([
|
||||
actionRun({ id: 9, conclusion: "success" }),
|
||||
actionRun({ id: 10, conclusion: "success", workflow: "quality.yml" }),
|
||||
actionRun({ id: 11, conclusion: "failure", workflow: "security.yml" }),
|
||||
], "chatgpt/feature", "current-sha")).toThrow("FEATURE_DELIVERY_CI_NOT_SUCCESSFUL");
|
||||
@@ -53,7 +54,7 @@ describe("review remediation regressions", () => {
|
||||
expect(() => assertSuccessfulActionRuns([
|
||||
actionRun({ id: 12, conclusion: "success" }),
|
||||
actionRun({ id: 13, conclusion: "", status: "in_progress" }),
|
||||
], "chatgpt/feature", "current-sha")).toThrow("FEATURE_DELIVERY_CI_NOT_SUCCESSFUL");
|
||||
], "chatgpt/feature", "current-sha")).toThrow("FEATURE_DELIVERY_FULL_VALIDATION_REQUIRED");
|
||||
});
|
||||
|
||||
it("rejects shipped placeholder credentials in production", () => {
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { CompactRunner, CompactRunnerJob } from "../src/services/activityRuntime.js";
|
||||
import { evaluateRunnerClaimHealth } from "../src/services/runnerClaimHealth.js";
|
||||
|
||||
const runner = (input: Partial<CompactRunner> = {}): CompactRunner => ({
|
||||
id: 1,
|
||||
name: "validation-1",
|
||||
status: "online",
|
||||
busy: false,
|
||||
disabled: false,
|
||||
labels: ["linux-validation"],
|
||||
...input,
|
||||
});
|
||||
|
||||
const job = (input: Partial<CompactRunnerJob> = {}): CompactRunnerJob => ({
|
||||
id: 99,
|
||||
runId: 9,
|
||||
name: "full",
|
||||
repository: "Jens/projectbrain",
|
||||
branch: "feature",
|
||||
status: "queued",
|
||||
queueState: "runnable",
|
||||
labels: ["linux-validation"],
|
||||
completedSteps: 0,
|
||||
totalSteps: 1,
|
||||
url: "https://gitea.example/jobs/99",
|
||||
createdAt: "2026-09-30T00:00:00.000Z",
|
||||
runnableObservedSince: "2026-09-30T00:00:00.000Z",
|
||||
claimableObservedSince: "2026-09-30T00:00:00.000Z",
|
||||
...input,
|
||||
});
|
||||
|
||||
describe("runner claim health", () => {
|
||||
const now = Date.parse("2026-09-30T00:05:00.000Z");
|
||||
|
||||
it("marks an old queued job with a compatible idle runner as polling-stalled without inventing a restart action", () => {
|
||||
const health = evaluateRunnerClaimHealth({ runners: [runner()], jobs: [job()], generatedAt: "2026-09-30T00:05:00.000Z" }, { now, stallSeconds: 120 });
|
||||
expect(health).toMatchObject({
|
||||
status: "stalled",
|
||||
recoveryOwner: "appops-runner-fleet",
|
||||
recoveryPolicy: "controller-owned",
|
||||
counts: { pollingStalled: 1 },
|
||||
oldestWaitSeconds: 300,
|
||||
oldestClaimableWaitSeconds: 300,
|
||||
});
|
||||
expect(health.jobs[0]).toMatchObject({ state: "polling-stalled", compatibleIdleRunnerNames: ["validation-1"] });
|
||||
expect(JSON.stringify(health)).not.toContain("container.restart");
|
||||
});
|
||||
|
||||
it("classifies a busy compatible runner as capacity-wait", () => {
|
||||
const health = evaluateRunnerClaimHealth({ runners: [runner({ busy: true })], jobs: [job()], generatedAt: "2026-09-30T00:05:00.000Z" }, { now });
|
||||
expect(health.jobs[0]).toMatchObject({ state: "capacity-wait", waitSeconds: 300, claimableWaitSeconds: 0 });
|
||||
expect(health.oldestClaimableWaitSeconds).toBe(0);
|
||||
});
|
||||
|
||||
it("reports a label block when no online runner can satisfy the job", () => {
|
||||
const health = evaluateRunnerClaimHealth({ runners: [runner({ labels: ["security"] })], jobs: [job()], generatedAt: "2026-09-30T00:05:00.000Z" }, { now });
|
||||
expect(health.status).toBe("degraded");
|
||||
expect(health.counts.labelBlocked).toBe(1);
|
||||
});
|
||||
|
||||
it("reports a polling stall independently of how many compatible idle runners exist", () => {
|
||||
const health = evaluateRunnerClaimHealth({
|
||||
runners: [runner(), runner({ id: 2, name: "validation-2" })],
|
||||
jobs: [job()],
|
||||
generatedAt: "2026-09-30T00:05:00.000Z",
|
||||
}, { now });
|
||||
expect(health.counts.pollingStalled).toBe(1);
|
||||
expect(health.jobs[0].compatibleIdleRunnerNames).toEqual(["validation-1", "validation-2"]);
|
||||
});
|
||||
|
||||
it("keeps a newly queued compatible job inside the normal claim grace window", () => {
|
||||
const health = evaluateRunnerClaimHealth({
|
||||
runners: [runner()],
|
||||
jobs: [job({
|
||||
createdAt: "2026-09-30T00:04:30.000Z",
|
||||
runnableObservedSince: "2026-09-30T00:04:30.000Z",
|
||||
claimableObservedSince: "2026-09-30T00:04:30.000Z",
|
||||
})],
|
||||
generatedAt: "2026-09-30T00:05:00.000Z",
|
||||
}, { now, stallSeconds: 120 });
|
||||
expect(health.status).toBe("healthy");
|
||||
expect(health.jobs[0].state).toBe("claimable");
|
||||
});
|
||||
|
||||
it("reports incomplete coverage rather than healthy when queued jobs were omitted from the scan", () => {
|
||||
const health = evaluateRunnerClaimHealth({
|
||||
availability: "live",
|
||||
runners: [runner()],
|
||||
jobs: Array.from({ length: 50 }, (_, index) => job({ id: index + 1, createdAt: "2026-09-30T00:04:30.000Z" })),
|
||||
runnableJobScan: { complete: false, observed: 50, total: 60 },
|
||||
generatedAt: "2026-09-30T00:05:00.000Z",
|
||||
}, { now });
|
||||
|
||||
expect(health).toMatchObject({
|
||||
status: "incomplete",
|
||||
coverage: { complete: false, observed: 50, total: 60 },
|
||||
counts: { queued: 50, pollingStalled: 0 },
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user