diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 3095569e7..44467e4a9 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -95,7 +95,7 @@ jobs: - name: Build public API exports before native acceptance consumer regression run: npm run build:dist - name: Reproduce native lease transition history before complete verification - run: node node_modules/vitest/vitest.mjs run --config ./vitest.control-plane.config.ts tests/unit/control-plane/capacity/execution/admission/postgres.test.ts tests/unit/control-plane/release/provider-prerequisites.test.ts tests/unit/control-plane/providers/assignments/lifecycle/architecture/terminal-observation-postgres.test.ts tests/unit/control-plane/capacity/execution/graph/persistence/concurrent-results-postgres.test.ts tests/unit/control-plane/providers/assignments/architecture/recovery/operator-recovery-postgres.test.ts tests/unit/control-plane/providers/assignments/architecture/recovery/recovery-observation-postgres.test.ts tests/unit/control-plane/capacity/workdays/scheduling/architecture/workday-start-native.test.ts tests/unit/control-plane/capacity/workdays/lifecycle/concurrent-runs-postgres.test.ts tests/unit/control-plane/capacity/execution/graph/architecture/workday-closeout-persistence.test.ts tests/unit/control-plane/capacity/execution/architecture/initial/admission/session-authority-unit.test.ts tests/unit/control-plane/capacity/planning/assignment-estimate-integration.test.ts tests/unit/control-plane/capacity/planning/estimate-source-native.test.ts tests/unit/control-plane/providers/assignments/architecture/cancellation-persistence.test.ts tests/unit/control-plane/capacity/execution/architecture/initial/admission/resource-closure-unit.test.ts tests/unit/control-plane/capacity/execution/architecture/monitoring/node-status-history-postgres.test.ts tests/unit/control-plane/capacity/accounting/architecture/accounting-history-postgres.test.ts tests/unit/control-plane/capacity/execution/architecture/monitoring/node-review-exhaustion-postgres.test.ts tests/unit/control-plane/capacity/workdays/scheduling/architecture/workday-node-history-native.test.ts tests/unit/control-plane/capacity/execution/admission/discussion-binding.test.ts tests/unit/control-plane/capacity/execution/architecture/admission-replay-authority.test.ts tests/unit/control-plane/capacity/execution/architecture/admission-replay-native.test.ts tests/unit/control-plane/capacity/execution/admission/timeout-postgres.test.ts tests/unit/control-plane/capacity/execution/scheduling/planning/planning-window-evidence.test.ts tests/unit/control-plane/capacity/execution/scheduling/planning/planning-phase-native.test.ts tests/unit/control-plane/capacity/execution/admission/closeout/closeout-admission.test.ts tests/unit/control-plane/capacity/workdays/lifecycle/workday-lifecycle-observations.test.ts tests/unit/control-plane/capacity/execution/reporting/report-context-scope.test.ts tests/unit/control-plane/capacity/execution/reporting/report-context-lifecycle.test.ts tests/unit/control-plane/capacity/execution/architecture/monitoring/node-status-history.test.ts tests/unit/control-plane/capacity/execution/graph/architecture/workday-observation-persistence.test.ts + run: node node_modules/vitest/vitest.mjs run --config ./vitest.control-plane.config.ts tests/unit/control-plane/capacity/execution/admission/postgres.test.ts tests/unit/control-plane/release/provider-prerequisites.test.ts tests/unit/control-plane/providers/assignments/lifecycle/architecture/terminal-observation-postgres.test.ts tests/unit/control-plane/capacity/execution/graph/persistence/concurrent-results-postgres.test.ts tests/unit/control-plane/providers/assignments/architecture/recovery/operator-recovery-postgres.test.ts tests/unit/control-plane/providers/assignments/architecture/recovery/recovery-observation-postgres.test.ts tests/unit/control-plane/capacity/workdays/scheduling/architecture/workday-start-native.test.ts tests/unit/control-plane/capacity/workdays/lifecycle/concurrent-runs-postgres.test.ts tests/unit/control-plane/capacity/execution/graph/architecture/workday-closeout-persistence.test.ts tests/unit/control-plane/capacity/execution/architecture/initial/admission/session-authority-unit.test.ts tests/unit/control-plane/capacity/planning/assignment-estimate-integration.test.ts tests/unit/control-plane/capacity/planning/estimate-source-native.test.ts tests/unit/control-plane/providers/assignments/architecture/cancellation-persistence.test.ts tests/unit/control-plane/capacity/execution/architecture/initial/admission/resource-closure-unit.test.ts tests/unit/control-plane/capacity/execution/architecture/monitoring/node-status-history-postgres.test.ts tests/unit/control-plane/capacity/accounting/architecture/accounting-history-postgres.test.ts tests/unit/control-plane/capacity/execution/architecture/monitoring/node-review-exhaustion-postgres.test.ts tests/unit/control-plane/capacity/workdays/scheduling/architecture/workday-node-history-native.test.ts tests/unit/control-plane/capacity/execution/admission/discussion-binding.test.ts tests/unit/control-plane/capacity/execution/architecture/admission-replay-authority.test.ts tests/unit/control-plane/capacity/execution/architecture/admission-replay-native.test.ts tests/unit/control-plane/capacity/execution/admission/timeout-postgres.test.ts tests/unit/control-plane/capacity/execution/scheduling/planning/planning-window-evidence.test.ts tests/unit/control-plane/capacity/execution/scheduling/planning/planning-phase-native.test.ts tests/unit/control-plane/capacity/execution/admission/closeout/closeout-admission.test.ts tests/unit/control-plane/capacity/workdays/lifecycle/workday-lifecycle-observations.test.ts tests/unit/control-plane/capacity/execution/reporting/report-context-scope.test.ts tests/unit/control-plane/capacity/execution/reporting/report-context-lifecycle.test.ts tests/unit/control-plane/capacity/execution/architecture/monitoring/node-status-history.test.ts tests/unit/control-plane/capacity/execution/graph/architecture/workday-observation-persistence.test.ts tests/unit/control-plane/capacity/ontology/availability-history.test.ts tests/unit/control-plane/capacity/ontology/availability-postgres.test.ts - run: npm run verify:direct - name: Verify real PostgreSQL backup-writer migration closure run: npx vitest run --config vitest.control-plane.config.ts tests/unit/control-plane/services/live/replication-migration-postgres.test.ts diff --git a/guarantees/agent/golden/scenes/component-boundaries.scene.yaml b/guarantees/agent/golden/scenes/component-boundaries.scene.yaml index 470f4f85d..27c11f97a 100644 --- a/guarantees/agent/golden/scenes/component-boundaries.scene.yaml +++ b/guarantees/agent/golden/scenes/component-boundaries.scene.yaml @@ -2834,3 +2834,27 @@ workflow: - id: execution-retired-participation-2 action: { verifier: api.golden.execution-retired-participation-2 } expect: { status: passed } + - id: provider-availability-history-unit + action: { verifier: api.golden.provider-availability-history-unit } + expect: { status: passed } + - id: capacity-audit-pagination-unit + action: { verifier: api.golden.capacity-audit-pagination-unit } + expect: { status: passed } + - id: provider-availability-fault-sql-unit + action: { verifier: api.golden.provider-availability-fault-sql-unit } + expect: { status: passed } + - id: provider-availability-expiry-unit + action: + verifier: api.golden.provider-availability-expiry-unit + expect: + status: passed + - id: provider-availability-offer-guard-unit + action: + verifier: api.golden.provider-availability-offer-guard-unit + expect: + status: passed + - id: provider-availability-current-claim-unit + action: + verifier: api.golden.provider-availability-current-claim-unit + expect: + status: passed diff --git a/guarantees/verifiers/golden.verifiers.yaml b/guarantees/verifiers/golden.verifiers.yaml index 4761cc6dd..dbbef52d5 100644 --- a/guarantees/verifiers/golden.verifiers.yaml +++ b/guarantees/verifiers/golden.verifiers.yaml @@ -4804,3 +4804,33 @@ verifiers: ownerPackage: "@treeseed/api" testFile: tests/unit/control-plane/capacity/execution/architecture/native-inventory.test.ts testName: "requires native absence of every retired table including source candidates without accepting ambiguous catalog or resolution results" + api.golden.provider-availability-history-unit: + kind: vitestCase + ownerPackage: "@treeseed/api" + testFile: tests/unit/control-plane/capacity/ontology/availability-history.test.ts + testName: "publishes the committed provider availability transition into existing scoped capacity audit history without private adapter observations" + api.golden.capacity-audit-pagination-unit: + kind: vitestCase + ownerPackage: "@treeseed/api" + testFile: tests/unit/control-plane/capacity/ontology/availability-history.test.ts + testName: "public capacity audit preserves raw records and complete cursor pagination including equal-clock identities" + api.golden.provider-availability-fault-sql-unit: + kind: vitestCase + ownerPackage: "@treeseed/api" + testFile: tests/unit/control-plane/capacity/ontology/availability-history.test.ts + testName: "executes the exact owning native availability fault SQL with missing interrupted and altered audit writes before original source retry" + api.golden.provider-availability-expiry-unit: + kind: vitestCase + ownerPackage: "@treeseed/api" + testFile: tests/unit/control-plane/capacity/ontology/availability-history.test.ts + testName: "retains each actual runtime expiry in the original availability audit clock and reports only newly expired sessions through unchanged replay" + api.golden.provider-availability-offer-guard-unit: + kind: vitestCase + ownerPackage: "@treeseed/api" + testFile: tests/unit/control-plane/capacity/ontology/availability-history.test.ts + testName: "retains exact capability offer bytes on stale foreign or terminal availability publication before original guarded retry" + api.golden.provider-availability-current-claim-unit: + kind: vitestCase + ownerPackage: "@treeseed/api" + testFile: tests/unit/control-plane/capacity/ontology/availability-history.test.ts + testName: "denies a missing current availability claim before every offer lane or seed grant mutation without changing original requests" diff --git a/src/api/capacity/repositories/accounts/availability-observation.ts b/src/api/capacity/repositories/accounts/availability-observation.ts new file mode 100644 index 000000000..58118ad04 --- /dev/null +++ b/src/api/capacity/repositories/accounts/availability-observation.ts @@ -0,0 +1,41 @@ +import {canonicalJson,sha256} from '../../security.ts'; +import {isDeepStrictEqual} from 'node:util'; +import type {ProviderAvailabilitySession} from '@treeseed/sdk/capacity-provider/contracts'; +import type {CapacityGovernanceDatabase} from '../../database.ts'; +import {CapacityGovernanceError} from '../../database.ts'; +import {CapacityAuditRepository} from '../support/audit.ts'; +type Row=Record; +const object=(value:unknown):Row=>value&&typeof value==='object'&&!Array.isArray(value)?value as Row:{}; +const healthy=(value:unknown):boolean|null=>{if(value===undefined||value===null)return null;if(typeof value!=='boolean')throw new CapacityGovernanceError('provider_availability_observation_invalid','Availability health observation must remain an original boolean.',500);return value;}; + +/** Existing audit is observation history; the session remains the supply authority. */ +export async function observeAvailabilitySession(database:CapacityGovernanceDatabase,session:ProviderAvailabilitySession):Promise{ + const now=(session.status==='closed'||session.status==='expired')?session.closedAt:session.refreshedAt; + const clocks=[session.openedAt,session.refreshedAt,session.expiresAt,session.snapshot.availableFrom,...(session.closedAt?[session.closedAt]:[]),...(session.snapshot.availableUntil?[session.snapshot.availableUntil]:[])]; + if(!clocks.every(value=>typeof value==='string'&&Number.isFinite(Date.parse(value))&&new Date(value).toISOString()===value)||Date.parse(session.openedAt)>Date.parse(session.refreshedAt)||Date.parse(session.expiresAt)<=Date.parse(session.refreshedAt)||session.closedAt&&Date.parse(session.closedAt)Number.isSafeInteger(value)&&value>=0)) + throw new CapacityGovernanceError('provider_availability_observation_invalid','Original lifecycle clocks and measured worker quantities required.',500); + if(!now||!Number.isFinite(Date.parse(now))||!Number.isSafeInteger(session.sequence)||session.sequence<1) + throw new CapacityGovernanceError('provider_availability_observation_invalid','Original availability identity sequence and clock required.',500); + const adapters=session.snapshot.adapters.map(adapter=>{ + const accounting=object(adapter.accountingObservation),model=object(accounting.modelUsage),capabilities=object(accounting.capabilityUsage); + const policy=object(adapter.observations).modelPolicy; + return{id:adapter.id,status:adapter.status,runtimeBuild:adapter.runtimeBuild??null,capabilities:adapter.capabilities,laneIds:adapter.laneIds, + nativeLimitsDigest:'sha256:'+sha256(canonicalJson(adapter.nativeLimits)),modelPolicyDigest:policy===undefined?null:'sha256:'+sha256(canonicalJson(policy)), + offers:(adapter.offers??[]).map(offer=>({offerId:offer.offerId,offerDigest:offer.offerDigest})), + modelHealthy:healthy(model.healthy),capabilityHealth:Object.fromEntries(adapter.capabilities.map(id=>[id,healthy(object(capabilities[id]).healthy)]))}; + }); + if(new Set(adapters.map(adapter=>adapter.id)).size!==adapters.length) + throw new CapacityGovernanceError('provider_availability_observation_invalid','Ambiguous adapter inventory cannot become history.',500); + const metadata={providerId:session.providerId,membershipId:session.membershipId,availabilitySessionId:session.id,sequence:session.sequence,status:session.status, + openedAt:session.openedAt,refreshedAt:session.refreshedAt,expiresAt:session.expiresAt,closedAt:session.closedAt, + pressure:session.snapshot.pressure,availableFrom:session.snapshot.availableFrom,availableUntil:session.snapshot.availableUntil, + maxConcurrentWorkers:session.snapshot.maxConcurrentWorkers,availableWorkers:session.snapshot.availableWorkers,reservedWorkers:session.snapshot.reservedWorkers,borrowedWorkers:session.snapshot.borrowedWorkers, + lanes:session.snapshot.lanes,constraintsDigest:'sha256:'+sha256(canonicalJson(session.snapshot.constraints)),adapters}; + const action='provider.availability.updated',key=session.sequence+':'+session.status; + await new CapacityAuditRepository(database).recordOnce({teamId:session.teamId,providerId:session.providerId,membershipId:session.membershipId, + actorType:'system',actorId:null,action,resourceType:'provider-availability-session',resourceId:session.id,idempotencyKey:key,metadata,now},session.id+':'+key); + const rows=await database.all('SELECT * FROM capacity_audit_events WHERE team_id=? AND action=? AND resource_id=? AND idempotency_key=?',[session.teamId,action,session.id,key]); + const row=rows[0];let retained:unknown;try{retained=JSON.parse(String(row?.metadata_json));}catch{/* Original malformed bytes remain retained. */} + if(rows.length!==1||!row||row.capacity_provider_id!==session.providerId||row.membership_id!==session.membershipId||row.created_at!==now||!isDeepStrictEqual(retained,metadata)) + throw new CapacityGovernanceError('provider_availability_observation_conflict','Availability requires its exact original durable public audit observation.',500); +} diff --git a/src/api/capacity/repositories/accounts/availability-session.ts b/src/api/capacity/repositories/accounts/availability-session.ts index 136f6309a..1fe564945 100644 --- a/src/api/capacity/repositories/accounts/availability-session.ts +++ b/src/api/capacity/repositories/accounts/availability-session.ts @@ -1,3 +1,4 @@ +import {observeAvailabilitySession} from './availability-observation.ts'; import { encodeCapacityPageCursor,type CapacityPage,type CapacityPageCursor } from '@treeseed/sdk/capacity-pagination'; import type { ProviderAvailabilitySession,ProviderAvailabilitySessionStatus } from '@treeseed/sdk/capacity-provider/contracts'; import type { CapacityDatabaseOperation,CapacityGovernanceDatabase } from '../../database.ts'; @@ -72,8 +73,9 @@ export interface AvailabilitySessionWrite { export class AvailabilitySessionRepository { constructor(private readonly database: CapacityGovernanceDatabase) {} - async open(input: AvailabilitySessionWrite, providerOperations: CapacityDatabaseOperation[]) { + async open(input: AvailabilitySessionWrite, providerOperations: CapacityDatabaseOperation[]):Promise { await this.database.ensureInitialized(); + const previous=await this.database.all("SELECT id FROM capacity_provider_availability_sessions WHERE membership_id=? AND status IN ('open','draining') ORDER BY id",[input.membershipId]); // These rows serialize publication without changing their referenced keys. // KEY SHARE from in-flight assignment/reservation FKs must remain compatible. await this.database.batch([ @@ -82,26 +84,28 @@ export class AvailabilitySessionRepository { { query: `UPDATE capacity_provider_availability_sessions SET status = 'closed', closed_at = COALESCE(closed_at, ?), updated_at = ? WHERE membership_id = ? AND status IN ('open','draining')`, params: [input.openedAt, input.openedAt, input.membershipId] }, { query: `INSERT INTO capacity_provider_availability_sessions (id, membership_id, team_id, capacity_provider_id, environment, status, sequence, opened_at, refreshed_at, expires_at, available_from, available_until, execution_providers_json, capabilities_json, native_limits_json, runner_pressure_json, constraints_json, metadata_json, created_at, updated_at) VALUES (?, ?, ?, ?, ?, 'open', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, params: [input.id, input.membershipId, input.teamId, input.providerId, input.environment, input.sequence, input.openedAt, input.refreshedAt, input.expiresAt, input.availableFrom, input.availableUntil, JSON.stringify(input.executionProviders), JSON.stringify(input.capabilities), JSON.stringify(input.nativeLimits), JSON.stringify(input.runnerPressure), JSON.stringify(input.constraints), JSON.stringify(input.metadata), input.openedAt, input.openedAt] }, ]); - return this.get(input.teamId, input.id); + for(const value of previous){const closed=await this.get(input.teamId,String(value.id));if(!closed||closed.status!=='closed')throw new CapacityGovernanceError('provider_availability_observation_conflict','Replaced availability closure is missing.',500);await observeAvailabilitySession(this.database,closed);} + const current=await this.get(input.teamId,input.id);if(!current)throw new CapacityGovernanceError('provider_availability_observation_conflict','Committed availability is missing.',500);await observeAvailabilitySession(this.database,current);return current; } async refresh(input: AvailabilitySessionWrite, expectedSequence: number, providerOperations: CapacityDatabaseOperation[]) { await this.database.ensureInitialized(); + const currentClaim=await this.database.first(`SELECT id FROM capacity_provider_availability_sessions WHERE id = ? AND membership_id = ? AND team_id = ? AND capacity_provider_id = ? AND sequence = ? AND status = 'open' FOR NO KEY UPDATE`,[input.id,input.membershipId,input.teamId,input.providerId,expectedSequence]); + if(!currentClaim)return null; const results = await this.database.batch([ - { query: `SELECT id FROM capacity_provider_availability_sessions WHERE id = ? AND membership_id = ? AND team_id = ? FOR NO KEY UPDATE`, params: [input.id, input.membershipId, input.teamId] }, ...providerOperations, { query: `UPDATE capacity_provider_availability_sessions SET sequence = sequence + 1, refreshed_at = ?, expires_at = ?, available_from = ?, available_until = ?, execution_providers_json = ?, capabilities_json = ?, native_limits_json = ?, runner_pressure_json = ?, constraints_json = ?, metadata_json = ?, updated_at = ? WHERE id = ? AND membership_id = ? AND team_id = ? AND capacity_provider_id = ? AND status = 'open' AND sequence = ? RETURNING id, sequence`, params: [input.refreshedAt, input.expiresAt, input.availableFrom, input.availableUntil, JSON.stringify(input.executionProviders), JSON.stringify(input.capabilities), JSON.stringify(input.nativeLimits), JSON.stringify(input.runnerPressure), JSON.stringify(input.constraints), JSON.stringify(input.metadata), input.refreshedAt, input.id, input.membershipId, input.teamId, input.providerId, expectedSequence] }, ]); const updateResult = (results as Array<{ results?: Row[] }>).at(-1)?.results?.[0] ?? null; if (!updateResult || Number(updateResult.sequence) !== expectedSequence + 1) return null; - return this.get(input.teamId, input.id); + const current=await this.get(input.teamId,input.id);if(!current)throw new CapacityGovernanceError('provider_availability_observation_conflict','Committed refresh is missing.',500);await observeAvailabilitySession(this.database,current);return current; } async close(teamId: string, membershipId: string, sessionId: string) { await this.database.ensureInitialized(); const now = new Date().toISOString(); await this.database.run(`UPDATE capacity_provider_availability_sessions SET status = 'closed', closed_at = COALESCE(closed_at, ?), updated_at = ? WHERE id = ? AND membership_id = ? AND team_id = ? AND status IN ('open','draining')`, [now, now, sessionId, membershipId, teamId]); - return this.get(teamId, sessionId); + const current=await this.get(teamId,sessionId);if(current?.status==='closed')await observeAvailabilitySession(this.database,current);return current; } /** Retire only sessions referenced by this workday, after every consumer and reservation has closed. */ @@ -114,16 +118,19 @@ export class AvailabilitySessionRepository { AND owned.team_id=session.team_id)`; // The same row lock is acquired by final admission. A separate statement // after that lock observes claims committed while closure was waiting. - await this.database.batch([ + const results=await this.database.batch([ { query: `SELECT session.id FROM capacity_provider_availability_sessions session WHERE ${owned} ORDER BY session.id FOR NO KEY UPDATE`, params: [teamId,workdayId] }, { query: `UPDATE capacity_provider_availability_sessions session SET status='closed',closed_at=COALESCE(closed_at,?),updated_at=? WHERE session.status IN ('open','draining') AND ${owned} AND NOT EXISTS (SELECT 1 FROM capacity_provider_assignments active WHERE active.provider_session_id=session.id AND (active.status IN ('pending','leased','running') OR EXISTS (SELECT 1 FROM capacity_reservations reservation WHERE reservation.assignment_id=active.id - AND reservation.team_id=active.team_id AND reservation.state NOT IN ('consumed','released','expired','failed'))))`, + AND reservation.team_id=active.team_id AND reservation.state NOT IN ('consumed','released','expired','failed')))) RETURNING session.id`, params: [now,now,teamId,workdayId] }, ]); + const changed=Array.isArray(results)?Reflect.get(Object(results.at(-1)),'results'):undefined; + if(!Array.isArray(changed))throw new CapacityGovernanceError('provider_availability_observation_conflict','Native close inventory is unavailable.',500); + for(const value of changed){if(!value||typeof value.id!=='string')throw new CapacityGovernanceError('provider_availability_observation_conflict','Native closed identity is unavailable.',500);const closed=await this.get(teamId,value.id);if(!closed||closed.status!=='closed')throw new CapacityGovernanceError('provider_availability_observation_conflict','Closed native session is missing.',500);await observeAvailabilitySession(this.database,closed);} } async get(teamId: string, sessionId: string) { diff --git a/src/api/capacity/repositories/capacity/providers/execution-provider.ts b/src/api/capacity/repositories/capacity/providers/execution-provider.ts index 8255f8117..62db15a6f 100644 --- a/src/api/capacity/repositories/capacity/providers/execution-provider.ts +++ b/src/api/capacity/repositories/capacity/providers/execution-provider.ts @@ -156,7 +156,7 @@ export function upsertCapacityExecutionProviderOperations(input: { }) : []; const offerOperations: CapacityDatabaseOperation[] = Array.isArray(entry.offers) ? entry.offers.map((offerValue) => { const offer = record(offerValue); - return { query: `INSERT INTO execution_capability_offers (capacity_provider_id,execution_provider_id,offer_id,offer_digest,offer_json,status,last_seen_at) VALUES (?,?,?,?,?,'active',?) ON CONFLICT (capacity_provider_id,offer_id) DO UPDATE SET execution_provider_id=EXCLUDED.execution_provider_id,offer_digest=EXCLUDED.offer_digest,offer_json=EXCLUDED.offer_json,status=CASE WHEN execution_capability_offers.status='context_overflow' AND execution_capability_offers.offer_digest=EXCLUDED.offer_digest THEN 'context_overflow' ELSE 'active' END,last_seen_at=EXCLUDED.last_seen_at`, params: [input.providerId,id,String(offer.offerId),String(offer.offerDigest),JSON.stringify(offer),input.createdAt] }; + return { query: `INSERT INTO execution_capability_offers (capacity_provider_id,execution_provider_id,offer_id,offer_digest,offer_json,status,last_seen_at) SELECT ?,?,?,?,?,'active',?${guardClause} ON CONFLICT (capacity_provider_id,offer_id) DO UPDATE SET execution_provider_id=EXCLUDED.execution_provider_id,offer_digest=EXCLUDED.offer_digest,offer_json=EXCLUDED.offer_json,status=CASE WHEN execution_capability_offers.status='context_overflow' AND execution_capability_offers.offer_digest=EXCLUDED.offer_digest THEN 'context_overflow' ELSE 'active' END,last_seen_at=EXCLUDED.last_seen_at`, params: [input.providerId,id,String(offer.offerId),String(offer.offerDigest),JSON.stringify(offer),input.createdAt,...guardParams] }; }) : []; return [executionProviderOperation, ...laneOperations, ...offerOperations]; }); diff --git a/src/api/capacity/repositories/support/audit.ts b/src/api/capacity/repositories/support/audit.ts index d4d1df712..43175d513 100644 --- a/src/api/capacity/repositories/support/audit.ts +++ b/src/api/capacity/repositories/support/audit.ts @@ -146,6 +146,18 @@ export class CapacityAuditRepository { limit?: unknown; cursor?: unknown; } = {}): Promise> { + const result=await this.listRowsPage(teamId,input); + return {...result,items:result.items.map(auditEvent)}; + } + async listRowsPage(teamId: string, input: { + providerId?: unknown; + membershipId?: unknown; + action?: unknown; + resourceType?: unknown; + resourceId?: unknown; + limit?: unknown; + cursor?: unknown; + } = {}): Promise>> { await this.database.ensureInitialized(); let limit: number; let cursor; @@ -167,17 +179,17 @@ export class CapacityAuditRepository { } } if (cursor) { - clauses.push('(created_at < ? OR (created_at = ? AND id > ?))'); + clauses.push('(created_at < ? OR (created_at = ? AND id < ?))'); values.push(cursor.createdAt, cursor.createdAt, cursor.id); } const rows = await this.database.all( - `SELECT * FROM capacity_audit_events WHERE ${clauses.join(' AND ')} ORDER BY created_at DESC, id ASC LIMIT ?`, + `SELECT * FROM capacity_audit_events WHERE ${clauses.join(' AND ')} ORDER BY created_at DESC, id DESC LIMIT ?`, [...values, limit + 1], ); const selected = rows.slice(0, limit); const last = selected.at(-1); return { - items: selected.map(auditEvent), + items: selected, page: { limit, hasMore: rows.length > limit, diff --git a/src/api/capacity/services/accounts/availability-session-service.ts b/src/api/capacity/services/accounts/availability-session-service.ts index 312a94195..14ba323b4 100644 --- a/src/api/capacity/services/accounts/availability-session-service.ts +++ b/src/api/capacity/services/accounts/availability-session-service.ts @@ -78,12 +78,14 @@ export class AvailabilitySessionService { } async close(principal: ProviderAvailabilityPrincipal, sessionId: string) { - await this.assertMembership(principal); - const existing = await this.repository.get(principal.teamId, sessionId); + return capacityTransaction(this.database,async database=>{const owner=new AvailabilitySessionService(database); + await owner.assertMembership(principal); + const existing = await owner.repository.get(principal.teamId, sessionId); if (!existing || existing.membershipId !== principal.membershipId || existing.providerId !== principal.capacityProviderId) return null; if (existing.status === 'closed' || existing.status === 'expired') return existing; if (existing.status !== 'open' && existing.status !== 'draining') throw new CapacityGovernanceError('provider_availability_close_conflict', `Availability session in ${existing.status} state cannot be closed.`, 409, { sessionId }); - return this.repository.close(principal.teamId, principal.membershipId, sessionId); + return owner.repository.close(principal.teamId, principal.membershipId, sessionId); + }); } private async accountingTransaction(principal: ProviderAvailabilityPrincipal, write: AvailabilitySessionWrite, diff --git a/src/api/capacity/services/runtime/runtime-retention-service.ts b/src/api/capacity/services/runtime/runtime-retention-service.ts index 60d45418c..bf93956c7 100644 --- a/src/api/capacity/services/runtime/runtime-retention-service.ts +++ b/src/api/capacity/services/runtime/runtime-retention-service.ts @@ -1,3 +1,7 @@ +import {capacityTransaction} from '../../transaction.ts'; +import {AvailabilitySessionRepository} from '../../repositories/accounts/availability-session.ts'; +import {observeAvailabilitySession} from '../../repositories/accounts/availability-observation.ts'; +import {CapacityGovernanceError} from '../../database.ts'; import type { CapacityGovernanceDatabase } from '../../database.ts'; export interface CapacityRuntimeRetentionResult { @@ -21,28 +25,25 @@ export async function maintainCapacityRuntimeRetention( database: CapacityGovernanceDatabase, now = new Date().toISOString(), ): Promise { + return capacityTransaction(database,async database=>{ const [ expiredAccessTokens, - expiredAvailabilitySessions, expiredRegistrationRequests, deletedProofNonces, deletedRateLimitBuckets, ] = await Promise.all([ count(database, `SELECT COUNT(*) AS total FROM capacity_provider_access_tokens WHERE status = 'active' AND expires_at <= ?`, now), - count(database, `SELECT COUNT(*) AS total FROM capacity_provider_availability_sessions WHERE status IN ('open', 'draining') AND expires_at <= ?`, now), count(database, `SELECT COUNT(*) AS total FROM capacity_provider_registration_requests WHERE status = 'pending' AND expires_at <= ?`, now), count(database, `SELECT COUNT(*) AS total FROM capacity_provider_proof_nonces WHERE expires_at <= ?`, now), count(database, `SELECT COUNT(*) AS total FROM capacity_provider_registration_rate_limits WHERE expires_at <= ?`, now), ]); + const changed=await database.all<{id:string;team_id:string}>(`UPDATE capacity_provider_availability_sessions SET status = 'expired', closed_at = COALESCE(closed_at, ?), updated_at = ? WHERE status IN ('open', 'draining') AND expires_at <= ? RETURNING id,team_id`,[now,now,now]); + const expiredAvailabilitySessions=changed.length; await database.batch([ { query: `UPDATE capacity_provider_access_tokens SET status = 'expired', expired_at = COALESCE(expired_at, ?), updated_at = ? WHERE status = 'active' AND expires_at <= ?`, params: [now, now, now], }, - { - query: `UPDATE capacity_provider_availability_sessions SET status = 'expired', closed_at = COALESCE(closed_at, ?), updated_at = ? WHERE status IN ('open', 'draining') AND expires_at <= ?`, - params: [now, now, now], - }, { query: `UPDATE capacity_provider_registration_requests SET status = 'expired', updated_at = ? WHERE status = 'pending' AND expires_at <= ?`, params: [now, now], @@ -56,6 +57,7 @@ export async function maintainCapacityRuntimeRetention( params: [now], }, ]); + for(const value of changed){const session=await new AvailabilitySessionRepository(database).get(value.team_id,value.id);if(!session||session.status!=='expired')throw new CapacityGovernanceError('provider_availability_observation_conflict','Original expired session is unavailable.',500);await observeAvailabilitySession(database,session);} return { expiredAccessTokens, expiredAvailabilitySessions, @@ -63,4 +65,5 @@ export async function maintainCapacityRuntimeRetention( deletedProofNonces, deletedRateLimitBuckets, }; + }); } diff --git a/src/api/control-plane/repositories/capacity/capacity-query-service.ts b/src/api/control-plane/repositories/capacity/capacity-query-service.ts index a9994cb7e..51a7e72ad 100644 --- a/src/api/control-plane/repositories/capacity/capacity-query-service.ts +++ b/src/api/control-plane/repositories/capacity/capacity-query-service.ts @@ -1,3 +1,4 @@ +import {CapacityAuditRepository} from '../../../capacity/repositories/support/audit.ts'; import { decodeCapacityPageCursor, normalizeCapacityPageLimit } from '@treeseed/sdk/capacity-pagination'; import type { CapacityGrantStatus } from '@treeseed/sdk/agent-capacity'; import { optionalAvailabilityStatus } from '../../../capacity/services/accounts/availability-session-service.ts'; @@ -55,8 +56,7 @@ export function createCapacityQueryService(store: any) { }, async audit(principal: CapacityPrincipal, teamId: string, query: Record) { await authorizeCapacityTeam(store, principal, teamId, 'projects:read:team'); - const { limit } = page(query); - return { items: await store.all(`SELECT * FROM capacity_audit_events WHERE team_id = ? ORDER BY created_at DESC, id DESC LIMIT ?`, [teamId, limit]), page: { limit, hasMore: false, nextCursor: null } }; + try{return {...await new CapacityAuditRepository(store).listRowsPage(teamId,{limit:query.limit,cursor:query.cursor})};}catch(error){translate(error);} }, async lanes(principal: CapacityPrincipal, teamId: string) { await authorizeCapacityTeam(store, principal, teamId, 'projects:read:team'); diff --git a/tests/unit/control-plane/capacity/execution/graph/architecture/workday-closeout-persistence.test.ts b/tests/unit/control-plane/capacity/execution/graph/architecture/workday-closeout-persistence.test.ts index 98bf14726..9c8757861 100644 --- a/tests/unit/control-plane/capacity/execution/graph/architecture/workday-closeout-persistence.test.ts +++ b/tests/unit/control-plane/capacity/execution/graph/architecture/workday-closeout-persistence.test.ts @@ -149,13 +149,17 @@ describe('canonical single report at the real lifecycle and SQL boundary', () => try { const run = (await reads.get('team', 'workday'))!; const original = structuredClone(run); - const ddl=splitPostgresSqlStatements(readFileSync('drizzle/control-plane/0000_control_plane.sql','utf8')).filter(sql=>sql.startsWith('CREATE TABLE "capacity_provider_availability_sessions" ('));expect(ddl).toHaveLength(1);await db.exec(ddl[0]!); + const statements=splitPostgresSqlStatements(readFileSync('drizzle/control-plane/0000_control_plane.sql','utf8')); + for(const table of ['capacity_provider_availability_sessions','capacity_audit_events']){const ddl=statements.filter(sql=>sql.startsWith('CREATE TABLE "'+table+'" ('));expect(ddl).toHaveLength(1);await db.exec(ddl[0]!); } for(const id of ['sdk-session','unrelated-session'])await query('INSERT INTO capacity_provider_availability_sessions (id,membership_id,team_id,capacity_provider_id,status,opened_at,refreshed_at,expires_at,available_from,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)',[id,'membership','team','provider','open',now,now,'2099-10-02T21:01:00.000Z',now,now,now]); await query("UPDATE capacity_provider_assignments SET provider_session_id='sdk-session' WHERE id='assignment-report'"); const unrelated=(await query("SELECT * FROM capacity_provider_availability_sessions WHERE id='unrelated-session'")).rows; const result = await advanceLivingWorkday(store as never, run, now); expect((await query("SELECT status,closed_at FROM capacity_provider_availability_sessions WHERE id='sdk-session'")).rows).toEqual([{status:'closed',closed_at:now}]); expect((await query("SELECT * FROM capacity_provider_availability_sessions WHERE id='unrelated-session'")).rows).toEqual(unrelated); + const availabilityAudit=(await query("SELECT * FROM capacity_audit_events ORDER BY id")).rows;expect(availabilityAudit).toHaveLength(1); + expect(availabilityAudit[0]).toMatchObject({id:createHash('sha256').update('provider.availability.updated:sdk-session:1:closed').digest('base64url'),team_id:'team',capacity_provider_id:'provider',membership_id:'membership',actor_type:'system',actor_id:null,action:'provider.availability.updated',resource_type:'provider-availability-session',resource_id:'sdk-session',idempotency_key:'1:closed',created_at:now}); + expect(JSON.parse(String(availabilityAudit[0]!.metadata_json))).toMatchObject({availabilitySessionId:'sdk-session',sequence:1,status:'closed',openedAt:now,refreshedAt:now,closedAt:now}); expect(result).toMatchObject({ status: 'completed', plan: { state: 'ended', reportRef } }); expect(run).toEqual(original); const rows = (await query('SELECT parameters_json,report_refs_json FROM capacity_workday_runs')).rows; @@ -176,6 +180,7 @@ describe('canonical single report at the real lifecycle and SQL boundary', () => const repeated = await advanceLivingWorkday(store as never, settled, now); expect((await query("SELECT status,closed_at FROM capacity_provider_availability_sessions WHERE id='sdk-session'")).rows).toEqual([{status:'closed',closed_at:now}]); expect(repeated.changed).toBe(false); + expect((await query("SELECT * FROM capacity_audit_events ORDER BY id")).rows).toEqual(availabilityAudit); expect((await query('SELECT COUNT(*)::int AS count FROM capacity_workday_runs')).rows).toEqual([{ count: 1 }]); } finally { await db.close(); } }); diff --git a/tests/unit/control-plane/capacity/ontology/availability-history.test.ts b/tests/unit/control-plane/capacity/ontology/availability-history.test.ts new file mode 100644 index 000000000..aa1c82d2b --- /dev/null +++ b/tests/unit/control-plane/capacity/ontology/availability-history.test.ts @@ -0,0 +1,113 @@ +import {upsertCapacityExecutionProviderOperations} from '../../../../../src/api/capacity/repositories/capacity/providers/execution-provider.ts'; +import {translateControlPlaneSqlToPostgres} from '../../../../../src/api/support/control-plane-postgres.ts'; +import {maintainCapacityRuntimeRetention} from '../../../../../src/api/capacity/services/runtime/runtime-retention-service.ts'; +import {PGlite} from '@electric-sql/pglite'; +import {availabilityObservationFaultSql} from './fixtures/availability-observation.ts'; +import {observeAvailabilitySession} from '../../../../../src/api/capacity/repositories/accounts/availability-observation.ts'; +import {createCapacityQueryService} from '../../../../../src/api/control-plane/repositories/capacity/capacity-query-service.ts'; +import {expect,it,vi} from 'vitest'; +import {AvailabilitySessionRepository,type AvailabilitySessionWrite} from '../../../../../src/api/capacity/repositories/accounts/availability-session.ts'; +import type {CapacityGovernanceDatabase,CapacityDatabaseOperation} from '../../../../../src/api/capacity/database.ts'; +import {canonicalOfferBuildInput} from '../execution/fixtures/assignment-attempt-fixtures.ts'; + +vi.mock('../../../../../src/api/capacity/transaction.ts',()=>({capacityTransaction:async(database:CapacityGovernanceDatabase,apply:(value:CapacityGovernanceDatabase)=>Promise)=>apply(database)})); + +function availabilityHistoryInput(){ + const at='2026-10-17T00:00:00.000Z',provider=canonicalOfferBuildInput(at).providers[0]!; + const input:AvailabilitySessionWrite={id:'session',membershipId:'membership',teamId:'team',providerId:'provider',environment:'local',sequence:1, + openedAt:at,refreshedAt:at,expiresAt:'2026-10-17T00:01:00.000Z',availableFrom:at,availableUntil:null, + executionProviders:[{...provider,id:'configured-adapter',adapter:'codex',status:'active',maxConcurrentRunners:1, + observations:{credential:'PRIVATE_ADAPTER_OBSERVATION'},accountingObservation:{modelUsage:{day:'2026-10-17',observedAt:at,healthy:true,activeSeconds:0,reservedSeconds:0},capabilityUsage:{}}}], + capabilities:provider.capabilities,nativeLimits:{},runnerPressure:{},constraints:{},metadata:{}}; + const row={id:input.id,membership_id:input.membershipId,team_id:input.teamId,capacity_provider_id:input.providerId,status:'open',sequence:1, + opened_at:at,refreshed_at:at,expires_at:input.expiresAt,available_from:at,available_until:null,closed_at:null as string|null, + execution_providers_json:JSON.stringify(input.executionProviders),capabilities_json:JSON.stringify(input.capabilities), + native_limits_json:'{}',runner_pressure_json:'{}',constraints_json:'{}',metadata_json:'{}'}; + return {at,input,row}; +} + +it('publishes the committed provider availability transition into existing scoped capacity audit history without private adapter observations',async()=>{ + const {at,input,row}=availabilityHistoryInput(); + const operations:CapacityDatabaseOperation[]=[],before=structuredClone(input); + const all=vi.fn().mockResolvedValue([]); + const database:CapacityGovernanceDatabase={ensureInitialized:vi.fn(async()=>{}),run:vi.fn(async(query:string,params:unknown[]=[])=>{operations.push({query,params});if(query.includes('INSERT INTO capacity_audit_events')){const [id,team_id,capacity_provider_id,membership_id,actor_type,actor_id,action,resource_type,resource_id,request_id,idempotency_key,metadata_json,created_at]=params;all.mockResolvedValue([{id,team_id,capacity_provider_id,membership_id,actor_type,actor_id,action,resource_type,resource_id,request_id,idempotency_key,metadata_json,created_at}]);}}), + first:async>(query:string,params?:unknown[]):Promise=>{void query;void params;const result:Record=structuredClone(row);return result as T;}, + all:async>(query:string,params?:unknown[]):Promise=>await all(query,params) as T[], + batch:vi.fn(async batch=>{operations.push(...structuredClone(batch));return batch.map(()=>({results:[]}));})}; + const result=await new AvailabilitySessionRepository(database).open(input,[]); + expect(result).toMatchObject({id:'session',sequence:1,status:'open'}); + const observations=operations.filter(operation=>operation.query.includes('INSERT INTO capacity_audit_events')); + expect(observations).toHaveLength(1); + expect(JSON.stringify(observations)).toContain('provider.availability.updated'); + expect(JSON.stringify(observations)).toContain('session'); + expect(JSON.stringify(observations)).not.toContain('PRIVATE_ADAPTER_OBSERVATION'); + expect(input).toEqual(before); + if(!result)throw new Error('Original successful session required'); + for(const field of ['expiresAt','openedAt']as const){const bad=structuredClone(result);Reflect.set(bad,field,'not-a-clock');const count=operations.length;await expect(observeAvailabilitySession(database,bad)).rejects.toMatchObject({code:'provider_availability_observation_invalid'});expect(operations).toHaveLength(count);} + const invalidCalendar=structuredClone(result);invalidCalendar.openedAt='2026-02-30T00:00:00.000Z';const calendarWrites=operations.length;await expect(observeAvailabilitySession(database,invalidCalendar)).rejects.toMatchObject({code:'provider_availability_observation_invalid'});expect(operations).toHaveLength(calendarWrites); + const unmeasured=structuredClone(result);unmeasured.snapshot.availableWorkers=NaN;const count=operations.length;await expect(observeAvailabilitySession(database,unmeasured)).rejects.toMatchObject({code:'provider_availability_observation_invalid'});expect(operations).toHaveLength(count); +}); + +it('public capacity audit preserves raw records and complete cursor pagination including equal-clock identities',async()=>{ + const at='2026-10-17T00:00:00.000Z',rows=[{id:'audit-b',created_at:at,metadata_json:'{}'},{id:'audit-a',created_at:at,metadata_json:'{}'}]; + const all=vi.fn(async(sql:string,parameters:unknown[])=>rows.filter(row=>parameters.length<3||row.id{},all,principalCanAccessTeam:async()=>false}); + const principal={id:'operator',roles:['admin']}; + await expect(query.audit(undefined,'team',{limit:1})).rejects.toMatchObject({status:401,code:'authentication_required'});await expect(query.audit({id:'denied',roles:[]},'team',{limit:1})).rejects.toMatchObject({status:403,code:'team_access_denied'});expect(all).not.toHaveBeenCalled(); + await expect(query.audit(principal,'team',{limit:1,cursor:'not-a-valid-cursor'})).rejects.toMatchObject({status:400,code:'capacity_page_invalid'});expect(all).not.toHaveBeenCalled(); + const first=await query.audit(principal,'team',{limit:1});expect(first.items).toEqual([rows[0]]);expect(first.page.hasMore).toBe(true);expect(first.page.nextCursor).toBeTruthy(); + const second=await query.audit(principal,'team',{limit:1,cursor:first.page.nextCursor});expect(second.items).toEqual([rows[1]]);expect(second.page.hasMore).toBe(false);expect(second.page.nextCursor).toBeNull(); + expect(all.mock.calls[1]![0]).toContain('id < ?');expect(all.mock.calls[1]![1]).toContain('audit-b'); +}); + +it('executes the exact owning native availability fault SQL with missing interrupted and altered audit writes before original source retry',async()=>{ + const database=new PGlite();try{ + await database.exec('CREATE TABLE capacity_audit_events (action TEXT NOT NULL, metadata_json TEXT NOT NULL)'); + for(const mode of ['missing','interrupted','altered']as const){ + await database.exec(availabilityObservationFaultSql(mode)); + await database.exec('CREATE TRIGGER retained_availability_observation_fault BEFORE INSERT ON capacity_audit_events FOR EACH ROW EXECUTE FUNCTION retained_availability_observation_fault()'); + const insert=()=>database.query('INSERT INTO capacity_audit_events (action,metadata_json) VALUES ($1,$2)',['provider.availability.updated','{"original":true}']); + try{if(mode==='interrupted')await expect(insert()).rejects.toThrow('retained native availability observation interruption');else await insert();expect((await database.query('SELECT * FROM capacity_audit_events')).rows).toEqual(mode==='altered'?[{action:'provider.availability.updated',metadata_json:'{}'}]:[]);} + finally{await database.exec('DROP TRIGGER retained_availability_observation_fault ON capacity_audit_events; DROP FUNCTION retained_availability_observation_fault(); DELETE FROM capacity_audit_events');} + } + await database.query('INSERT INTO capacity_audit_events (action,metadata_json) VALUES ($1,$2)',['provider.availability.updated','{"original":true}']);expect((await database.query('SELECT * FROM capacity_audit_events')).rows).toEqual([{action:'provider.availability.updated',metadata_json:'{"original":true}'}]); + }finally{await database.close();} + // Controlled PostgreSQL grammar in WASM is a focused check, never native + // concurrency, complete prerequisite or managed evidence. +}); + +it('retains each actual runtime expiry in the original availability audit clock and reports only newly expired sessions through unchanged replay',async()=>{ + const {at,row:original}=availabilityHistoryInput(),now='2026-10-17T00:02:00.000Z',session={...original},audits:Array>=[]; + const expire=()=>{if(!['open','draining'].includes(session.status)||session.expires_at>now)return[];session.status='expired';session.closed_at=now;return[{id:session.id,team_id:session.team_id}];}; + const database:CapacityGovernanceDatabase={ensureInitialized:async()=>{}, + first:async>(query:string):Promise=>{const value:Record=query.includes('COUNT(*)')?{total:query.includes('capacity_provider_availability_sessions')&&['open','draining'].includes(session.status)?1:0}:structuredClone(session);return value as T;}, + all:async>(query:string):Promise=>{const values=query.startsWith('UPDATE capacity_provider_availability_sessions')?expire():query.includes('capacity_audit_events')?structuredClone(audits):[];return values as T[];}, + run:async(query,params=[])=>{if(query.includes('INSERT INTO capacity_audit_events')){const[id,team_id,capacity_provider_id,membership_id,actor_type,actor_id,action,resource_type,resource_id,request_id,idempotency_key,metadata_json,created_at]=params;audits.push({id,team_id,capacity_provider_id,membership_id,actor_type,actor_id,action,resource_type,resource_id,request_id,idempotency_key,metadata_json,created_at});}}, + batch:async operations=>{if(operations.some(value=>value.query.includes("SET status = 'expired'")&&value.query.includes('capacity_provider_availability_sessions')))expire();return operations.map(()=>({results:[]}));}}; + const result=await maintainCapacityRuntimeRetention(database,now);expect(result.expiredAvailabilitySessions).toBe(1);expect(audits).toHaveLength(1); + expect(audits[0]).toMatchObject({created_at:now,resource_id:session.id,idempotency_key:'1:expired'});expect(JSON.parse(String(audits[0]!.metadata_json))).toMatchObject({status:'expired',closedAt:now,refreshedAt:at}); + const bad=await new AvailabilitySessionRepository(database).get(session.team_id,session.id);if(!bad)throw new Error('Original expired session required');bad.closedAt=at;const writes=audits.length;await expect(observeAvailabilitySession(database,bad)).rejects.toMatchObject({code:'provider_availability_observation_invalid'});expect(audits).toHaveLength(writes); + const held=structuredClone({session,audits});expect((await maintainCapacityRuntimeRetention(database,now)).expiredAvailabilitySessions).toBe(0);expect({session,audits}).toEqual(held); + // Transaction mechanics and independent native concurrency are covered by + // the owning PostgreSQL case; this unit controls only the database protocol. +}); + +it('retains exact capability offer bytes on stale foreign or terminal availability publication before original guarded retry',async()=>{ + const {at,input}=availabilityHistoryInput(),database=new PGlite(),later='2026-10-17T00:00:10.000Z';try{ + await database.exec("CREATE TABLE capacity_provider_availability_sessions (id TEXT,membership_id TEXT,team_id TEXT,sequence INTEGER,status TEXT);CREATE TABLE execution_capability_offers (capacity_provider_id TEXT,execution_provider_id TEXT,offer_id TEXT,offer_digest TEXT,offer_json TEXT,status TEXT,last_seen_at TEXT,PRIMARY KEY(capacity_provider_id,offer_id));INSERT INTO capacity_provider_availability_sessions VALUES ('session','membership','team',2,'open')"); + const base={providerId:input.providerId,executionProviders:input.executionProviders,providerNativeLimits:input.nativeLimits,createdAt:at},guard={sessionId:'session',membershipId:'membership',teamId:'team',expectedSequence:2}; + const publish=async(value:Parameters[0])=>{const operations=upsertCapacityExecutionProviderOperations(value).filter(item=>item.query.startsWith('INSERT INTO execution_capability_offers'));expect(operations).toHaveLength(1);for(const operation of operations)await database.query(translateControlPlaneSqlToPostgres(operation.query),operation.params);}; + await publish(base);const held=(await database.query>('SELECT * FROM execution_capability_offers')).rows; + for(const denied of[{...guard,expectedSequence:1},{...guard,sessionId:'foreign'},{...guard,membershipId:'foreign'},{...guard,teamId:'foreign'}]){await publish({...base,createdAt:later,availabilityGuard:denied});expect((await database.query>('SELECT * FROM execution_capability_offers')).rows).toEqual(held);} + for(const status of['closed','expired','draining']){await database.query('UPDATE capacity_provider_availability_sessions SET status=$1',[status]);await publish({...base,createdAt:later,availabilityGuard:guard});expect((await database.query>('SELECT * FROM execution_capability_offers')).rows).toEqual(held);} + await database.query("UPDATE capacity_provider_availability_sessions SET status='open'");await publish({...base,createdAt:later,availabilityGuard:guard});expect((await database.query>('SELECT * FROM execution_capability_offers')).rows).toEqual(held.map(value=>({...value,last_seen_at:later}))); + await database.query("UPDATE execution_capability_offers SET status='context_overflow'");const exact=(await database.query>('SELECT * FROM execution_capability_offers')).rows;await publish({...base,createdAt:later,availabilityGuard:guard});expect((await database.query>('SELECT * FROM execution_capability_offers')).rows).toEqual(exact); + }finally{await database.close();} + // Real generated SQL grammar and owning guards, with controlled local data; + // independent PostgreSQL contention remains the original native case. +}); + +it('denies a missing current availability claim before every offer lane or seed grant mutation without changing original requests',async()=>{ + const {input}=availabilityHistoryInput(),before=structuredClone(input),first=vi.fn().mockResolvedValue(null),batch=vi.fn().mockResolvedValue([{results:[{id:input.id,sequence:2}]}]); + const database:CapacityGovernanceDatabase={ensureInitialized:async()=>{},first:async>(query:string,params?:unknown[]):Promise=>{await first(query,params);return null;},batch,run:vi.fn(async()=>{}),all:vi.fn(async()=>[])},operations=[{query:'UPDATE capacity_grants SET updated_at=? WHERE id=?',params:[input.refreshedAt,'controlled-original-seed-grant']}],held=structuredClone(operations); + await expect(new AvailabilitySessionRepository(database).refresh(input,1,operations)).resolves.toBeNull();expect(batch).not.toHaveBeenCalled();expect(database.run).not.toHaveBeenCalled();expect(database.all).not.toHaveBeenCalled();expect(first).toHaveBeenCalledTimes(1);expect(first.mock.calls[0]![1]).toEqual([input.id,input.membershipId,input.teamId,input.providerId,1]);expect(first.mock.calls[0]![0]).toContain("status = 'open'");expect(first.mock.calls[0]![0]).toContain('FOR NO KEY UPDATE');expect(input).toEqual(before);expect(operations).toEqual(held); +}); diff --git a/tests/unit/control-plane/capacity/ontology/availability-postgres.test.ts b/tests/unit/control-plane/capacity/ontology/availability-postgres.test.ts index 3c472b2ab..ef5d518f7 100644 --- a/tests/unit/control-plane/capacity/ontology/availability-postgres.test.ts +++ b/tests/unit/control-plane/capacity/ontology/availability-postgres.test.ts @@ -1,3 +1,6 @@ +import {maintainCapacityRuntimeRetention} from '../../../../../src/api/capacity/services/runtime/runtime-retention-service.ts'; +import {availabilityObservationFaultSql} from './fixtures/availability-observation.ts'; +import {createCapacityQueryService} from '../../../../../src/api/control-plane/repositories/capacity/capacity-query-service.ts'; import {providerAccountingStatus} from '../../../../../src/api/control-plane/repositories/providers/provider-runtime-service.ts'; import { randomUUID, generateKeyPairSync, createHash } from 'node:crypto'; import pg from 'pg'; @@ -32,7 +35,34 @@ describe('provider accounting in disposable PostgreSQL', () => { first: >(sql: string, params: unknown[] = []) => f.left.prepare(sql).bind(...params).first(), all: async >(sql: string, params: unknown[] = []) => (await f.left.prepare(sql).bind(...params).all()).results, batch: (operations: Array<{ query: string; params?: unknown[] }>) => f.left.batch(operations) }; + await f.left.pool.query("INSERT INTO projects (id,team_id,slug,name,created_at,updated_at) VALUES ('availability-seed-project','team','availability-seed-project','Controlled seed grant input',$1,$1)",[original.now]); + await f.left.pool.query("INSERT INTO capacity_grants (id,membership_id,capacity_provider_id,team_id,project_id,environment,status,metadata_json,created_at,updated_at) VALUES ('availability-seed-grant','membership','provider','team','availability-seed-project','local','active','{\"seedName\":\"controlled-native-publication\"}',$1,$1)",[original.now]); const principal = { teamId: 'team', membershipId: 'membership', capacityProviderId: 'provider' }; + await f.left.pool.query("INSERT INTO teams (id,slug,name,created_at,updated_at) VALUES ('availability-other-team','availability-other-team','Unrelated observation input',$1,$1)",[original.now]); + for(const[id,team,status]of[['availability-live','team','running'],['availability-ended','team','completed'],['availability-foreign','availability-other-team','running']]as const) + await f.left.pool.query("INSERT INTO capacity_workday_runs (id,team_id,status,execution_mode,execution_kind,started_at,created_at,updated_at) VALUES ($1,$2,$3,'simulation','workday',$4,$4,$4)",[id,team,status,original.now]); + for(const id of ['audit-pagination-a','audit-pagination-b','audit-pagination-c']) + await f.left.pool.query("INSERT INTO capacity_audit_events (id,team_id,actor_type,action,resource_type,metadata_json,created_at) VALUES ($1,'team','controlled-input','controlled.pagination.input','controlled-input','{}',$2)",[id,original.now]); + const observeAvailability=async(expected:Array<{sequence:number;status:string;at:string}>)=>{ + if(!opened)throw new Error('Original published session identity required before public history readback'); + const raw=(await f.right.pool.query("SELECT * FROM capacity_audit_events WHERE team_id='team' AND action='provider.availability.updated' ORDER BY created_at,id")).rows; + expect(raw).toHaveLength(expected.length); + const query=createCapacityQueryService(store),publicRows:typeof raw=[],seen=new Set();let cursor:string|undefined; + do{const page=await query.audit(f.principal,'team',{limit:1,cursor});publicRows.push(...page.items); + if(!page.page.hasMore){expect(page.page.nextCursor).toBeNull();break;} + expect(page.page.nextCursor).toBeTruthy();if(!page.page.nextCursor||seen.has(page.page.nextCursor)||seen.size>10)throw new Error('Complete original audit pagination required'); + seen.add(page.page.nextCursor);cursor=page.page.nextCursor; + }while(cursor); + const complete=(await f.right.pool.query("SELECT * FROM capacity_audit_events WHERE team_id='team' ORDER BY created_at DESC,id DESC")).rows; + expect(publicRows).toEqual(complete); + const identities=new Set();for(const value of raw){ + const context=JSON.parse(String(value.metadata_json)),match=expected.filter(item=>item.sequence===context.sequence&&item.status===context.status);expect(match).toHaveLength(1);expect(value.created_at).toBe(match[0]!.at);const key=context.sequence+':'+context.status;expect(identities.has(key)).toBe(false);identities.add(key); + expect(context).toMatchObject({providerId:'provider',membershipId:'membership',availabilitySessionId:opened.id}); + expect(JSON.stringify(value)).not.toContain('opaque-controlled-private-adapter-field'); + } + const unrelated=(await f.right.pool.query("SELECT id,next_event_index FROM capacity_workday_runs WHERE id IN ('availability-live','availability-ended','availability-foreign') ORDER BY id")).rows; + expect(unrelated).toEqual([{id:'availability-ended',next_event_index:0},{id:'availability-foreign',next_event_index:0},{id:'availability-live',next_event_index:0}]); + }; const modelPolicy={provider:'openai',model:'configured-luna',reasoningEffort:'low',capabilities:adapter.capabilities}; const input = { adapters: [{ id: adapter.id, runtimeBuild: adapter.runtimeBuild, offers: adapter.offers, capabilities: adapter.capabilities, status: 'available', maxConcurrentWorkers: 1, activeWorkers: 0, @@ -43,9 +73,10 @@ describe('provider accounting in disposable PostgreSQL', () => { priority: 1, capabilities: adapter.capabilities })) }; const inputBefore = structuredClone(input), service = new AvailabilitySessionService(store); const opened = await service.open(principal, input); if (!opened) throw new Error('Actual canonical session publication required'); + await observeAvailability([{sequence:opened.sequence,status:'open',at:opened.refreshedAt}]); const compileNow = new Date().toISOString(); const buildSource = { ...original, run: canonicalOfferBuildInput(compileNow).run, now: compileNow }; - const tables = ['capacity_providers', 'capacity_provider_team_memberships', 'capacity_provider_availability_sessions', + const tables = ['capacity_audit_events','capacity_grants','capacity_workday_runs','capacity_workday_events','capacity_providers', 'capacity_provider_team_memberships', 'capacity_provider_availability_sessions', 'capacity_execution_providers', 'capacity_provider_lanes', 'execution_capability_offers', 'capacity_provider_assignments', 'capacity_reservations', 'capacity_usage_actuals', 'capacity_ledger_entries', 'capacity_operation_receipts']; const snapshot = async (database: typeof f.left) => { @@ -158,9 +189,65 @@ describe('provider accounting in disposable PostgreSQL', () => { expect(outcome.cause, outcome.name).toMatchObject({ status: 409, code: 'capacity_execution_provider_unavailable' }); } expect(await compile()).toEqual(frozen); expect(await snapshot(f.right)).toEqual(baseline); - await service.close(principal, opened.id); const closed = await snapshot(f.right); + const refreshInput={...structuredClone(input),expectedSequence:opened.sequence}; + const originalHealth=refreshInput.adapters[0]!.accountingObservation; + if(!originalHealth)throw new Error('Original configured health observation required'); + Object.assign(originalHealth.modelUsage,{healthy:false}); + const refreshBefore=structuredClone(refreshInput),failedAvailability:Array<{mode:string;cause:unknown;unchanged:boolean}>=[]; + for(const mode of ['missing','interrupted','altered']as const){ + await f.left.pool.query(availabilityObservationFaultSql(mode)); + await f.left.pool.query('CREATE TRIGGER retained_availability_observation_fault BEFORE INSERT ON capacity_audit_events FOR EACH ROW EXECUTE FUNCTION retained_availability_observation_fault()'); + let cause:unknown; + try{await service.refresh(principal,opened.id,refreshInput);}catch(error){cause=error;} + finally{await f.left.pool.query('DROP TRIGGER retained_availability_observation_fault ON capacity_audit_events');await f.left.pool.query('DROP FUNCTION retained_availability_observation_fault()');} + expect(cause).toBeInstanceOf(Error);if(mode!=='interrupted')expect(cause).toMatchObject({code:'provider_availability_observation_conflict'}); + const unchanged=JSON.stringify(await snapshot(f.right))===JSON.stringify(baseline);expect(unchanged).toBe(true);expect(await snapshot(f.left)).toEqual(baseline);expect(refreshInput).toEqual(refreshBefore);failedAvailability.push({mode,cause,unchanged}); + } + for(const corruptedAt of ['not-a-clock','2026-02-30T00:00:00.000Z']){ + await f.left.pool.query('UPDATE capacity_provider_availability_sessions SET opened_at=$1 WHERE id=$2',[corruptedAt,opened.id]); + const malformedClock=await snapshot(f.right);let clockFailure:unknown; + try{await service.refresh(principal,opened.id,refreshInput);}catch(error){clockFailure=error;} + expect(clockFailure).toMatchObject({code:'provider_availability_observation_invalid'});expect(await snapshot(f.left)).toEqual(malformedClock);const unchanged=JSON.stringify(await snapshot(f.right))===JSON.stringify(malformedClock);expect(unchanged).toBe(true);failedAvailability.push({mode:'clock:'+corruptedAt,cause:clockFailure,unchanged}); + // Restore only the deliberately corrupted owning input; failed observations remain. + await f.left.pool.query('UPDATE capacity_provider_availability_sessions SET opened_at=$1 WHERE id=$2',[opened.openedAt,opened.id]);expect(await snapshot(f.right)).toEqual(baseline); + } + const peer={db:f.right,ensureInitialized:()=>f.right.migrate(),run:async(sql:string,params:unknown[]=[])=>{await f.right.prepare(sql).bind(...params).run();}, + first:>(sql:string,params:unknown[]=[])=>f.right.prepare(sql).bind(...params).first(), + all:async>(sql:string,params:unknown[]=[]) =>(await f.right.prepare(sql).bind(...params).all()).results, + batch:(operations:Array<{query:string;params?:unknown[]}>)=>f.right.batch(operations)}; + const competing=await Promise.all([service.refresh(principal,opened.id,refreshInput),new AvailabilitySessionService(peer).refresh(principal,opened.id,refreshInput)]),winners=competing.filter(value=>value!==null);expect(winners).toHaveLength(1);expect(competing.filter(value=>value===null)).toHaveLength(1); + const refreshed=winners[0]!;expect(refreshed.sequence).toBe(opened.sequence+1); + await observeAvailability([{sequence:opened.sequence,status:'open',at:opened.refreshedAt},{sequence:refreshed.sequence,status:'open',at:refreshed.refreshedAt}]); + const observedHealth=(await f.right.pool.query("SELECT metadata_json FROM capacity_audit_events WHERE resource_id=$1 AND idempotency_key=$2",[opened.id,refreshed.sequence+':open'])).rows;expect(observedHealth).toHaveLength(1);expect(JSON.parse(observedHealth[0].metadata_json).adapters[0].modelHealthy).toBe(false); + const refreshedBytes=await snapshot(f.right);expect(await service.refresh(principal,opened.id,refreshInput)).toBeNull();expect(await snapshot(f.left)).toEqual(refreshedBytes);expect(refreshInput).toEqual(refreshBefore);expect(failedAvailability).toHaveLength(5); + console.log(JSON.stringify({retainedAvailabilityFailures:failedAvailability.map(value=>({mode:value.mode,unchanged:value.unchanged,message:String(value.cause)}))})); + const terminalAvailability=await service.close(principal, opened.id);if(!terminalAvailability?.closedAt)throw new Error('Original terminal availability required'); + await observeAvailability([{sequence:opened.sequence,status:'open',at:opened.refreshedAt},{sequence:refreshed.sequence,status:'open',at:refreshed.refreshedAt},{sequence:terminalAvailability.sequence,status:'closed',at:terminalAvailability.closedAt}]); + expect((await new AvailabilitySessionService(peer).close(principal,opened.id))?.closedAt).toBe(terminalAvailability.closedAt); + const closed = await snapshot(f.right); await expect(compile()).rejects.toMatchObject({ status: 409, code: 'provider_synthesis_session_not_open' }); expect(await snapshot(f.left)).toEqual(closed); expect(input).toEqual(inputBefore); expect(original).toEqual(before); + const due=await service.open(principal,{...structuredClone(input),ttlSeconds:30});const unrelated=await service.open(qualificationPrincipal,{...structuredClone(qualifiedInput),ttlSeconds:300});if(!due||!unrelated)throw new Error('Original due and unrelated live sessions required'); + const expiryAt=new Date(Date.parse(due.expiresAt)+1).toISOString(),beforeExpiry=await snapshot(f.right);expect(Date.parse(expiryAt)).toBeLessThan(Date.parse(unrelated.expiresAt)); + const retainedExpiryFailures:Array<{mode:string;message:string}>=[]; + for(const mode of ['missing','interrupted','altered']as const){ + await f.left.pool.query(availabilityObservationFaultSql(mode));await f.left.pool.query('CREATE TRIGGER retained_availability_observation_fault BEFORE INSERT ON capacity_audit_events FOR EACH ROW EXECUTE FUNCTION retained_availability_observation_fault()');let failure:unknown; + try{await maintainCapacityRuntimeRetention(store,expiryAt);}catch(error){failure=error;} + finally{await f.left.pool.query('DROP TRIGGER retained_availability_observation_fault ON capacity_audit_events');await f.left.pool.query('DROP FUNCTION retained_availability_observation_fault()');} + expect(failure).toBeInstanceOf(Error);if(mode!=='interrupted')expect(failure).toMatchObject({code:'provider_availability_observation_conflict'});expect(await snapshot(f.left)).toEqual(beforeExpiry);expect(await snapshot(f.right)).toEqual(beforeExpiry);retainedExpiryFailures.push({mode,message:String(failure)}); + } + await f.left.pool.query('UPDATE capacity_provider_availability_sessions SET closed_at=$1 WHERE id=$2',[due.refreshedAt,due.id]);const malformedExpiry=await snapshot(f.right);await expect(maintainCapacityRuntimeRetention(store,expiryAt)).rejects.toMatchObject({code:'provider_availability_observation_invalid'});expect(await snapshot(f.left)).toEqual(malformedExpiry);expect(await snapshot(f.right)).toEqual(malformedExpiry); + // Restore only the deliberately corrupted source clock; no failed observation is removed. + await f.left.pool.query('UPDATE capacity_provider_availability_sessions SET closed_at=NULL WHERE id=$1',[due.id]);expect(await snapshot(f.right)).toEqual(beforeExpiry); + const expired=await Promise.all([maintainCapacityRuntimeRetention(store,expiryAt),maintainCapacityRuntimeRetention(peer,expiryAt)]);expect(expired.map(value=>value.expiredAvailabilitySessions).sort()).toEqual([0,1]); + const retainedExpiry=(await f.right.pool.query("SELECT * FROM capacity_audit_events WHERE action='provider.availability.updated' AND resource_id=$1 ORDER BY created_at,id",[due.id])).rows;expect(retainedExpiry).toHaveLength(2); + const expiry=retainedExpiry.filter(value=>value.idempotency_key===due.sequence+':expired');expect(expiry).toHaveLength(1);expect(expiry[0].created_at).toBe(expiryAt);expect(JSON.parse(String(expiry[0].metadata_json))).toMatchObject({availabilitySessionId:due.id,status:'expired',sequence:due.sequence,closedAt:expiryAt,refreshedAt:due.refreshedAt}); + const expiredSession=await service.get(principal.teamId,due.id);expect(expiredSession).toMatchObject({status:'expired',sequence:due.sequence,closedAt:expiryAt});expect(await service.get(qualificationPrincipal.teamId,unrelated.id)).toEqual(unrelated); + const query=createCapacityQueryService(store),publicHistory:Array>=[];let auditCursor:string|undefined; + for(let pageNumber=0;pageNumber<40;pageNumber++){const page=await query.audit(f.principal,'team',{limit:2,cursor:auditCursor});publicHistory.push(...page.items);if(!page.page.hasMore){expect(page.page.nextCursor).toBeNull();break;}if(!page.page.nextCursor)throw new Error('Original expiry audit continuation missing');auditCursor=page.page.nextCursor;} + expect(publicHistory).toEqual((await f.right.pool.query("SELECT * FROM capacity_audit_events WHERE team_id='team' ORDER BY created_at DESC,id DESC")).rows); + const afterExpiry=await snapshot(f.right);expect(await snapshot(f.left)).toEqual(afterExpiry);expect((await maintainCapacityRuntimeRetention(store,expiryAt)).expiredAvailabilitySessions).toBe(0);expect(await service.close(principal,due.id)).toEqual(expiredSession);expect(await snapshot(f.right)).toEqual(afterExpiry);expect(await snapshot(f.left)).toEqual(afterExpiry); + console.log(JSON.stringify({retainedExpiryFailures,expiredAvailabilitySessions:expired.map(value=>value.expiredAvailabilitySessions),unrelatedRetained:true,unchangedExpiryReplay:true})); // Original services, full migrations, two PostgreSQL pools and exact // persisted bytes. Supplied identity/conformance are not bearer auth, // native qualification, live model work or atomic assignment admission. diff --git a/tests/unit/control-plane/capacity/ontology/fixtures/availability-observation.ts b/tests/unit/control-plane/capacity/ontology/fixtures/availability-observation.ts new file mode 100644 index 000000000..b693982a5 --- /dev/null +++ b/tests/unit/control-plane/capacity/ontology/fixtures/availability-observation.ts @@ -0,0 +1,4 @@ +/** Original owning fault SQL, shared by grammar regression and native PostgreSQL. */ +export function availabilityObservationFaultSql(mode:'missing'|'interrupted'|'altered'):string{ + return "CREATE OR REPLACE FUNCTION retained_availability_observation_fault() RETURNS trigger LANGUAGE plpgsql AS $availability$ BEGIN IF NEW.action='provider.availability.updated' THEN "+(mode==='missing'?"RETURN NULL;":mode==='altered'?"NEW.metadata_json = '{}'; RETURN NEW;":"RAISE EXCEPTION 'retained native availability observation interruption';")+" END IF; RETURN NEW; END $availability$"; +}