Merge pull request #4265 from cardstack/worktree-cs-10419

Implement test realm management for AI-generated tests

Author: habdelra

.gitignore

@@ -13,3 +13,4 @@ test-results/.last-run.json
 traefik/dynamic/*.yml
 .mise.local.toml
 packages/openrouter-realm/OpenRouterModel/
+packages/host/dist

packages/matrix/docker/synapse/index.ts

@@ -102,7 +102,7 @@ export async function cfgDirFromTemplate(
   }
   const configDir = dataDir
     ? dataDir
-    : await fse.mkdtemp(path.join(os.tmpdir(), 'synapsedocker-'));
+    : await fse.mkdtemp(path.join(os.tmpdir(), 'sf-test-synapse-'));
 
   // copy the contents of the template dir, omitting homeserver.yaml as we'll template that
   console.log(`Copy ${templateDir} -> ${configDir}`);

packages/realm-server/tests/realm-endpoints/cancel-indexing-job-test.ts

@@ -189,6 +189,142 @@ module(`realm-endpoints/${basename(__filename)}`, function () {
         );
       });
 
+      test('cancels both running and pending jobs when cancelPending is true', async function (assert) {
+        let concurrencyGroup = `indexing:${testRealm.url}`;
+
+        // Create a running job (with active reservation)
+        let [{ id: runningJobId }] = (await dbAdapter.execute(`INSERT INTO jobs
+        (args, job_type, concurrency_group, timeout, priority)
+        VALUES
+        (
+          '{"realmURL": "${testRealm.url}", "realmUsername":"node-test_realm"}',
+          'from-scratch-index',
+          '${concurrencyGroup}',
+          180,
+          0
+        ) RETURNING id`)) as { id: string }[];
+        await dbAdapter.execute(`INSERT INTO job_reservations
+        (job_id, locked_until ) VALUES (${runningJobId}, NOW() + INTERVAL '3 minutes')`);
+
+        // Create a pending job (no reservation)
+        let [{ id: pendingJobId }] = (await dbAdapter.execute(`INSERT INTO jobs
+        (args, job_type, concurrency_group, timeout, priority)
+        VALUES
+        (
+          '{"realmURL": "${testRealm.url}", "realmUsername":"node-test_realm"}',
+          'incremental-index',
+          '${concurrencyGroup}',
+          180,
+          0
+        ) RETURNING id`)) as { id: string }[];
+
+        let response = await request
+          .post('/_cancel-indexing-job')
+          .set('Accept', 'application/json')
+          .set(
+            'Authorization',
+            `Bearer ${createJWT(testRealm, 'writer', ['read', 'write'])}`,
+          )
+          .send({ cancelPending: true });
+
+        assert.strictEqual(response.status, 204, 'HTTP 204 response');
+
+        // Running job should be cancelled
+        let [runningJob] = await dbAdapter.execute(
+          `SELECT status, result, finished_at FROM jobs WHERE id = ${runningJobId}`,
+        );
+        assert.strictEqual(
+          runningJob.status,
+          'rejected',
+          'running job was canceled',
+        );
+        assert.deepEqual(
+          runningJob.result,
+          {
+            status: 418,
+            message: 'User initiated job cancellation',
+          },
+          'running job result is cancellation payload',
+        );
+        assert.ok(runningJob.finished_at, 'running job has finish time');
+
+        // Pending job should ALSO be cancelled
+        let [pendingJob] = await dbAdapter.execute(
+          `SELECT status, result, finished_at FROM jobs WHERE id = ${pendingJobId}`,
+        );
+        assert.strictEqual(
+          pendingJob.status,
+          'rejected',
+          'pending job was also canceled when cancelPending is true',
+        );
+        assert.deepEqual(
+          pendingJob.result,
+          {
+            status: 418,
+            message: 'User initiated job cancellation',
+          },
+          'pending job result is cancellation payload',
+        );
+        assert.ok(pendingJob.finished_at, 'pending job has finish time');
+      });
+
+      test('default behavior (no body) only cancels running jobs, not pending', async function (assert) {
+        let concurrencyGroup = `indexing:${testRealm.url}`;
+
+        let [{ id: runningJobId }] = (await dbAdapter.execute(`INSERT INTO jobs
+        (args, job_type, concurrency_group, timeout, priority)
+        VALUES
+        (
+          '{"realmURL": "${testRealm.url}", "realmUsername":"node-test_realm"}',
+          'from-scratch-index',
+          '${concurrencyGroup}',
+          180,
+          0
+        ) RETURNING id`)) as { id: string }[];
+        await dbAdapter.execute(`INSERT INTO job_reservations
+        (job_id, locked_until ) VALUES (${runningJobId}, NOW() + INTERVAL '3 minutes')`);
+
+        let [{ id: pendingJobId }] = (await dbAdapter.execute(`INSERT INTO jobs
+        (args, job_type, concurrency_group, timeout, priority)
+        VALUES
+        (
+          '{"realmURL": "${testRealm.url}", "realmUsername":"node-test_realm"}',
+          'incremental-index',
+          '${concurrencyGroup}',
+          180,
+          0
+        ) RETURNING id`)) as { id: string }[];
+
+        // No body — default behavior
+        let response = await request
+          .post('/_cancel-indexing-job')
+          .set('Accept', 'application/json')
+          .set(
+            'Authorization',
+            `Bearer ${createJWT(testRealm, 'writer', ['read', 'write'])}`,
+          );
+
+        assert.strictEqual(response.status, 204, 'HTTP 204 response');
+
+        let [runningJob] = await dbAdapter.execute(
+          `SELECT status FROM jobs WHERE id = ${runningJobId}`,
+        );
+        assert.strictEqual(
+          runningJob.status,
+          'rejected',
+          'running job canceled',
+        );
+
+        let [pendingJob] = await dbAdapter.execute(
+          `SELECT status FROM jobs WHERE id = ${pendingJobId}`,
+        );
+        assert.strictEqual(
+          pendingJob.status,
+          'unfulfilled',
+          'pending job NOT canceled when cancelPending is not set',
+        );
+      });
+
       test('does not treat expired reservations as running jobs', async function (assert) {
         let concurrencyGroup = `indexing:${testRealm.url}`;
         let [{ id: jobId }] = (await dbAdapter.execute(`INSERT INTO jobs

packages/realm-server/tests/scripts/boot_preseeded.sh

@@ -10,7 +10,7 @@ exec docker-entrypoint.sh postgres \
   -c full_page_writes=off \
   -c synchronous_commit=off \
   -c shared_buffers=16MB \
-  -c max_connections=20 \
+  -c max_connections=50 \
   -c wal_level=minimal \
   -c max_wal_senders=0 \
   -c max_replication_slots=0 \

packages/runtime-common/job-utils.ts

@@ -96,3 +96,31 @@ export async function cancelRunningJobsInConcurrencyGroup(
   }
   return runningJobIds;
 }
+
+/**
+ * Cancel ALL jobs in a concurrency group — both running (active reservations)
+ * and pending (unfulfilled, no active reservation).
+ */
+export async function cancelAllJobsInConcurrencyGroup(
+  dbAdapter: DBAdapter,
+  concurrencyGroup: string,
+): Promise<{ cancelledRunning: string[]; cancelledPending: string[] }> {
+  let cancelledRunning = await cancelRunningJobsInConcurrencyGroup(
+    dbAdapter,
+    concurrencyGroup,
+  );
+
+  let pendingRows = (await query(dbAdapter, [
+    `SELECT id FROM jobs WHERE concurrency_group =`,
+    param(concurrencyGroup),
+    `AND status = 'unfulfilled'`,
+  ] as Expression)) as { id: string }[];
+
+  let cancelledPending: string[] = [];
+  for (let row of pendingRows) {
+    await forceCancelJobById(dbAdapter, row.id);
+    cancelledPending.push(row.id);
+  }
+
+  return { cancelledRunning, cancelledPending };
+}

packages/runtime-common/realm.ts

@@ -169,7 +169,10 @@ import {
   type PublishabilityWarningType,
   type ResourceIndexEntry,
 } from './publishability';
-import { cancelRunningJobsInConcurrencyGroup } from './job-utils';
+import {
+  cancelAllJobsInConcurrencyGroup,
+  cancelRunningJobsInConcurrencyGroup,
+} from './job-utils';
 
 export const REALM_ROOM_RETENTION_POLICY_MAX_LIFETIME = 60 * 60 * 1000;
 
@@ -825,13 +828,31 @@ export class Realm {
   }
 
   private async cancelIndexingJob(
-    _request: Request,
+    request: Request,
     requestContext: RequestContext,
   ) {
-    await cancelRunningJobsInConcurrencyGroup(
-      this.#dbAdapter,
-      `indexing:${this.url}`,
-    );
+    let cancelPending = false;
+    try {
+      let body = await request.text();
+      if (body) {
+        let parsed = JSON.parse(body) as { cancelPending?: boolean };
+        cancelPending = parsed.cancelPending === true;
+      }
+    } catch {
+      // No body or invalid JSON — use default (running only).
+    }
+
+    if (cancelPending) {
+      await cancelAllJobsInConcurrencyGroup(
+        this.#dbAdapter,
+        `indexing:${this.url}`,
+      );
+    } else {
+      await cancelRunningJobsInConcurrencyGroup(
+        this.#dbAdapter,
+        `indexing:${this.url}`,
+      );
+    }
 
     return createResponse({
       body: null,