fix(integration): serialize adopted request replay

TASK-f40126ee-2485-45c8-a67b-4279e4e8704a

docs/TEST_INTEGRITY.md

@@ -3176,8 +3176,12 @@ test_integrity_note:
       then: "The transition refuses with its structured boundary code and no integration queue record"
     - id: SCN-integration-adoption-request-replay
       given: "A finalized adoption that has created its one ordinary private-ref integration request"
-      when: "A concurrent exact-row insert or later request replay is attempted"
-      then: "Real SQLite rejects the duplicate exact source row, the journal replay refuses, and exactly one queue row remains"
+      when: "Two independently opened authenticated API callers request it concurrently"
+      then: "The existing transactional adoption journal serializes them, one caller refuses after the completed transition, and exactly one queue row remains"
+    - id: SCN-integration-adoption-legacy-retries
+      given: "An existing project database containing two historical ordinary-request retry rows with the same source branch and commit"
+      when: "IntegrationService starts"
+      then: "Both records remain readable; startup adds no global uniqueness constraint and does not reject historical evidence"
   observable_outcomes:
     - "Git ref values, structured refusal codes, durable journal state, and native integration-request fields"
   determinism_controls:
@@ -3189,7 +3193,8 @@ test_integrity_note:
       - "Claiming crash recovery without preserving actual ref effects"
       - "Treating a private candidate as an ordinary specialist branch"
       - "Using an unrelated, source-only, stale, or mismatched approval as adoption authority"
-      - "Duplicating a private-ref queue record through a concurrent row or replay"
+      - "Duplicating a private-ref queue record through concurrent authenticated API requests or replay"
+      - "Rejecting historical ordinary-request retry evidence with an unqualified schema constraint"
   relaxation:
     did_relax_any_assertion: false
     if_true_explain_spec_basis: ""

src/integration/adoption.mjs

@@ -236,23 +236,26 @@ export class AdoptionService {
   }
 
   #append(adoptionId, expectedRevision, record, operation) {
-    const now = this.clock().toISOString();
     this.controlPlane.database.exec('BEGIN IMMEDIATE');
     try {
-      const current = this.#find(adoptionId);
-      if (expectedRevision === null ? current : !current || current.revision !== expectedRevision) {
-        throw new TorchError('Adoption journal changed before this transition could be recorded', {
-          code: 'ADOPTION_RECORD_REVISION_STALE', details: { adoptionId, expectedRevision, actual: current?.revision ?? null },
-        });
-      }
-      this.controlPlane.database.prepare(`INSERT INTO audit_events
-        (id, project_id, actor_id, operation, entity_type, entity_id, details, created_at)
-        VALUES (?, ?, 'owner', ?, 'integration-adoption', ?, ?, ?)`)
-        .run(this.idFactory(), this.controlPlane.projectId, operation, adoptionId, JSON.stringify(record), now);
+      this.#appendInTransaction(adoptionId, expectedRevision, record, operation);
       this.controlPlane.database.exec('COMMIT');
     } catch (error) { this.controlPlane.database.exec('ROLLBACK'); throw error; }
   }
 
+  #appendInTransaction(adoptionId, expectedRevision, record, operation) {
+    const current = this.#find(adoptionId);
+    if (expectedRevision === null ? current : !current || current.revision !== expectedRevision) {
+      throw new TorchError('Adoption journal changed before this transition could be recorded', {
+        code: 'ADOPTION_RECORD_REVISION_STALE', details: { adoptionId, expectedRevision, actual: current?.revision ?? null },
+      });
+    }
+    this.controlPlane.database.prepare(`INSERT INTO audit_events
+      (id, project_id, actor_id, operation, entity_type, entity_id, details, created_at)
+      VALUES (?, ?, 'owner', ?, 'integration-adoption', ?, ?, ?)`)
+      .run(this.idFactory(), this.controlPlane.projectId, operation, adoptionId, JSON.stringify(record), this.clock().toISOString());
+  }
+
   #fromRecord(record, input, action) {
     exactInput(record, input);
     return this.plan({ action, actorId: input.actorId, approvalId: input.approvalId, approvalRevision: input.approvalRevision,
@@ -326,20 +329,37 @@ export class AdoptionService {
   }
 
   requestAdoptedCandidate(input = {}, { requestIntegration, findIntegrationRequest } = {}) {
-    const record = this.getPrepared(text(input.adoptionId, 'adoptionId'));
-    if (!['finalized', 'requesting'].includes(record.state)) throw new TorchError('Only a finalized adoption can request native qualification', { code: 'ADOPTION_REQUEST_STATE_INVALID', details: { state: record.state } });
-    const plan = this.#fromRecord(record, input, 'request');
-    if (!plan.canProceed) throw new TorchError('Adopted candidate request is blocked', { code: 'ADOPTION_REQUEST_BLOCKED', details: plan.blockers });
     if (typeof requestIntegration !== 'function') throw new TorchError('Native integration request adapter is unavailable', { code: 'ADOPTION_NATIVE_REQUEST_UNAVAILABLE' });
-    const requesting = record.state === 'requesting' ? record : { ...record, revision: record.revision + 1, state: 'requesting',
-      requestApprovalId: text(input.approvalId, 'approvalId'), requestApprovalRevision: input.approvalRevision, requestStartedAt: this.clock().toISOString() };
-    if (record.state !== 'requesting') this.#append(record.id, record.revision, requesting, 'integration.adoption.request-started');
-    const request = findIntegrationRequest?.({ candidateRef: requesting.candidateRef, commit: requesting.candidateSha })
-      ?? requestIntegration({ adoptionId: requesting.id, areaId: requesting.contributorArea,
+    const adoptionId = text(input.adoptionId, 'adoptionId');
+    this.controlPlane.database.exec('BEGIN IMMEDIATE');
+    try {
+      const record = this.getPrepared(adoptionId);
+      if (!['finalized', 'requesting'].includes(record.state)) {
+        throw new TorchError('Only a finalized adoption can request native qualification', {
+          code: 'ADOPTION_REQUEST_STATE_INVALID', details: { state: record.state },
+        });
+      }
+      const plan = this.#fromRecord(record, input, 'request');
+      if (!plan.canProceed) throw new TorchError('Adopted candidate request is blocked', { code: 'ADOPTION_REQUEST_BLOCKED', details: plan.blockers });
+      const requesting = record.state === 'requesting' ? record : { ...record, revision: record.revision + 1, state: 'requesting',
+        requestApprovalId: text(input.approvalId, 'approvalId'), requestApprovalRevision: input.approvalRevision, requestStartedAt: this.clock().toISOString() };
+      if (record.state !== 'requesting') this.#appendInTransaction(record.id, record.revision, requesting, 'integration.adoption.request-started');
+      const existing = findIntegrationRequest?.({ candidateRef: requesting.candidateRef, commit: requesting.candidateSha });
+      if (existing && (existing.sourceArea !== requesting.contributorArea || existing.targetBranch !== requesting.targetBranch)) {
+        throw new TorchError('Existing private-ref integration request does not match this adoption', {
+          code: 'ADOPTION_INTEGRATION_REQUEST_CONFLICT', details: { adoptionId: requesting.id, integrationRequestId: existing.id },
+        });
+      }
+      const request = existing ?? requestIntegration({ adoptionId: requesting.id, areaId: requesting.contributorArea,
         commit: requesting.candidateSha, candidateRef: requesting.candidateRef });
-    const requested = { ...requesting, revision: requesting.revision + 1, state: 'requested', integrationRequestId: request.id, requestedAt: this.clock().toISOString() };
-    this.#append(requesting.id, requesting.revision, requested, 'integration.adoption.request');
-    return { ...requested, integrationRequest: request, mutationPerformed: true };
+      const requested = { ...requesting, revision: requesting.revision + 1, state: 'requested', integrationRequestId: request.id, requestedAt: this.clock().toISOString() };
+      this.#appendInTransaction(requesting.id, requesting.revision, requested, 'integration.adoption.request');
+      this.controlPlane.database.exec('COMMIT');
+      return { ...requested, integrationRequest: request, mutationPerformed: true };
+    } catch (error) {
+      this.controlPlane.database.exec('ROLLBACK');
+      throw error;
+    }
   }
 
   #find(adoptionId) {

src/integration/service.mjs

@@ -80,8 +80,6 @@ export class IntegrationService {
       );
       CREATE INDEX IF NOT EXISTS integration_state_created
         ON integration_requests(state, created_at, id);
-      CREATE UNIQUE INDEX IF NOT EXISTS integration_request_exact_source
-        ON integration_requests(project_id, source_branch, source_commit);
     `);
   }
 

test/fleet/integration-adoption.test.mjs

@@ -1,6 +1,6 @@
 import assert from 'node:assert/strict';
 import { createHash } from 'node:crypto';
-import { execFileSync } from 'node:child_process';
+import { execFileSync, spawn } from 'node:child_process';
 import { mkdtempSync, writeFileSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { basename, join } from 'node:path';
@@ -47,7 +47,43 @@ function fixture(name) {
   const control = openControlPlane({ repositoryRoot: root, env });
   const service = new IntegrationService({ repositoryRoot: root, controlPlane: control, checkService: { exactPasses: () => false } });
   const contributor = proposal.domains[0].id;
-  return { root, control, service, contributor, destination: 'review-area', base, candidate };
+  return { root, env, control, service, contributor, destination: 'review-area', base, candidate };
+}
+
+function concurrentAdoptedRequest(context, requestInput) {
+  const childSource = `
+    import { pathToFileURL } from 'node:url';
+    import { resolve } from 'node:path';
+    const { repositoryRoot, stateRoot, requestInput } = JSON.parse(process.argv[1]);
+    const sourceRoot = process.cwd();
+    const { openControlPlane } = await import(pathToFileURL(resolve(sourceRoot, 'src/control-plane/service.mjs')).href);
+    const { IntegrationService } = await import(pathToFileURL(resolve(sourceRoot, 'src/integration/service.mjs')).href);
+    const control = openControlPlane({ repositoryRoot, env: { ...process.env, XDG_DATA_HOME: stateRoot } });
+    const service = new IntegrationService({ repositoryRoot, controlPlane: control, checkService: { exactPasses: () => false } });
+    try {
+      const result = service.requestAdoptedCandidate(requestInput);
+      console.log(JSON.stringify({ outcome: 'requested', integrationRequestId: result.integrationRequest.id }));
+    } catch (error) {
+      console.log(JSON.stringify({ outcome: 'refused', code: error.code ?? null }));
+    } finally {
+      control.close();
+    }
+  `;
+  const encoded = JSON.stringify({ repositoryRoot: context.root, stateRoot: context.env.XDG_DATA_HOME, requestInput });
+  return new Promise((resolve, reject) => {
+    const child = spawn(process.execPath, ['--input-type=module', '-e', childSource, encoded], {
+      cwd: process.cwd(), env: process.env, stdio: ['ignore', 'pipe', 'pipe'],
+    });
+    let stdout = ''; let stderr = '';
+    child.stdout.on('data', chunk => { stdout += chunk; });
+    child.stderr.on('data', chunk => { stderr += chunk; });
+    child.on('error', reject);
+    child.on('close', (code) => {
+      if (code !== 0) return reject(new Error(`concurrent adopted request child failed: ${stderr}`));
+      try { resolve(JSON.parse(stdout.trim().split('\n').at(-1))); }
+      catch (error) { reject(new Error(`invalid concurrent adopted request output: ${stdout}\n${stderr}\n${error.message}`)); }
+    });
+  });
 }
 
 function authority(context, { action, adoptionId }) {
@@ -266,22 +302,37 @@ test('SCN-integration-adoption-approval-boundaries: unrelated or source-only app
   wrongProvenance.control.close();
 });
 
-test('SCN-integration-adoption-request-replay: real SQLite rejects a duplicate exact private-ref queue row, and a completed adoption request cannot replay into a second row', () => {
+test('SCN-integration-adoption-request-replay: concurrent authenticated API callers serialize on the adoption journal and create exactly one private-ref queue request', async () => {
   const context = fixture('torch-adoption-request-replay');
   const adoptionId = 'fixture-adoption';
   prepare(context, adoptionId);
-  const finalized = context.service.finalizeAdoption(lifecycleInput(context, 'finalize', adoptionId));
-  const requested = context.service.requestAdoptedCandidate(lifecycleInput(context, 'request', adoptionId));
-  const duplicate = context.control.database.prepare(`INSERT INTO integration_requests (
+  context.service.finalizeAdoption(lifecycleInput(context, 'finalize', adoptionId));
+  const requestInput = lifecycleInput(context, 'request', adoptionId);
+  const outcomes = await Promise.all([
+    concurrentAdoptedRequest(context, requestInput),
+    concurrentAdoptedRequest(context, requestInput),
+  ]);
+  assert.deepEqual(outcomes.map((entry) => entry.outcome).sort(), ['refused', 'requested']);
+  assert.equal(outcomes.find((entry) => entry.outcome === 'refused').code, 'ADOPTION_REQUEST_STATE_INVALID');
+  const requests = context.service.list();
+  assert.equal(requests.length, 1);
+  assert.equal(context.service.getPreparedAdoption(adoptionId).state, 'requested');
+  context.control.close();
+});
+
+test('SCN-integration-adoption-legacy-retries: historical ordinary-request retries remain readable when integration service starts', () => {
+  const context = fixture('torch-adoption-legacy-retries');
+  const insert = context.control.database.prepare(`INSERT INTO integration_requests (
     id, project_id, source_area, source_branch, source_commit, target_branch,
     base_target_commit, state, reason, required_checks, created_at, updated_at
   ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`);
-  assert.throws(() => duplicate.run(
-    'duplicate-race-row', context.control.projectId, context.contributor, finalized.candidateRef, context.candidate,
-    'main', context.base, 'testing', null, '[]', requested.requestedAt, requested.requestedAt,
-  ), /UNIQUE constraint failed: integration_requests.project_id, integration_requests.source_branch, integration_requests.source_commit/);
-  assert.throws(() => context.service.requestAdoptedCandidate(lifecycleInput(context, 'request', adoptionId)),
-    (error) => error.code === 'ADOPTION_REQUEST_STATE_INVALID');
-  assert.deepEqual(context.service.list().map((record) => record.id), [requested.integrationRequest.id]);
+  for (const [id, state] of [['historical-failed', 'blocked'], ['historical-retry', 'superseded']]) {
+    insert.run(id, context.control.projectId, context.contributor, 'refs/heads/candidate', context.candidate,
+      'main', context.base, state, 'historical-retry', '[]', '2026-10-03T00:00:00.000Z', '2026-10-03T00:00:00.000Z');
+  }
+  assert.doesNotThrow(() => new IntegrationService({
+    repositoryRoot: context.root, controlPlane: context.control, checkService: { exactPasses: () => false },
+  }));
+  assert.equal(context.service.list().length, 2);
   context.control.close();
 });