diff --git a/back/runtime/adapters/legacy-sequelize/legacyShadowTerminalDifferenceSource.ts b/back/runtime/adapters/legacy-sequelize/legacyShadowTerminalDifferenceSource.ts new file mode 100644 index 00000000..e0cf7f5e --- /dev/null +++ b/back/runtime/adapters/legacy-sequelize/legacyShadowTerminalDifferenceSource.ts @@ -0,0 +1,431 @@ +import { QueryTypes, type Sequelize } from 'sequelize'; +import { + RUN_ATTEMPT_TABLE, + RUN_TABLE, +} from '../../../migrations/0002-run-schema'; +import { RUNNING_INSTANCE_TABLE } from '../../../migrations/0003-running-instance-run-reference'; +import { + EXECUTION_ORIGINS, + RUN_ATTEMPT_STATUSES, + RUN_STATUSES, + type ExecutionOrigin, + type RunAttemptStatus, + type RunStatus, +} from '../../domain/run'; +import { + MAX_LEGACY_SHADOW_TERMINAL_EVIDENCE_PER_PAGE, + MAX_LEGACY_SHADOW_TERMINAL_PAGE_SIZE, + type LegacyShadowTerminalCandidate, + type LegacyShadowTerminalCursor, + type LegacyShadowTerminalDifferenceSource, + type LegacyShadowTerminalPage, + type LegacyTerminalEvidence, + type LegacyTerminalOutcome, +} from '../../ports/legacyShadowTerminalDifference'; + +interface CandidateRow { + runId: string; + legacyCronId: number | null; + origin: string; + runStatus: string; + createdAtMs: number | string; + startedAtMs: number | string | null; + finishedAtMs: number | string | null; + attemptCount: number | string; + attemptId: string | null; + attemptStatus: string | null; + attemptPid: number | null; + attemptLogArtifactId: string | null; + attemptCreatedAtMs: number | string | null; + attemptStartedAtMs: number | string | null; + attemptFinishedAtMs: number | string | null; + attemptExitCode: number | null; +} + +interface EvidenceRow { + instanceId: number | string; + legacyCronId: number | string; + runId: string | null; + attemptId: string | null; + pid: number | null; + logPath: string | null; + startedAt: number | string; + finishedAt: number | string; + status: number | string; + exitCode: number | null; +} + +export type LegacyTerminalLogArtifactIdFactory = (logPath: string) => string; + +const ORIGINS = new Set(EXECUTION_ORIGINS); +const RUN_STATUS_SET = new Set(RUN_STATUSES); +const ATTEMPT_STATUS_SET = new Set(RUN_ATTEMPT_STATUSES); + +function safeInteger(value: number | string, name: string): number { + const parsed = typeof value === 'number' ? value : Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 0) { + throw new TypeError(`${name} is invalid`); + } + return parsed; +} + +function optionalPositiveInteger( + value: number | null, + name: string, +): number | undefined { + if (value === null) return undefined; + const parsed = safeInteger(value, name); + if (parsed < 1) throw new TypeError(`${name} is invalid`); + return parsed; +} + +function secondsToMilliseconds(value: number | string, name: string): number { + const milliseconds = safeInteger(value, name) * 1_000; + if (!Number.isSafeInteger(milliseconds)) { + throw new TypeError(`${name} is invalid`); + } + return milliseconds; +} + +function assertLimit(limit: number): void { + if ( + !Number.isSafeInteger(limit) || + limit < 1 || + limit > MAX_LEGACY_SHADOW_TERMINAL_PAGE_SIZE + ) { + throw new RangeError( + `limit must be between 1 and ${MAX_LEGACY_SHADOW_TERMINAL_PAGE_SIZE}`, + ); + } +} + +function runStatus(value: string): RunStatus { + if (!RUN_STATUS_SET.has(value)) { + throw new TypeError('Run status is invalid'); + } + return value as RunStatus; +} + +function attemptStatus(value: string): RunAttemptStatus { + if (!ATTEMPT_STATUS_SET.has(value)) { + throw new TypeError('Run Attempt status is invalid'); + } + return value as RunAttemptStatus; +} + +function origin(value: string): ExecutionOrigin { + if (!ORIGINS.has(value)) { + throw new TypeError('Run execution origin is invalid'); + } + return value as ExecutionOrigin; +} + +function terminalOutcome(value: number): LegacyTerminalOutcome { + if (value === 1) return 'succeeded'; + if (value === 2) return 'stopped'; + if (value === 3) return 'failed'; + throw new TypeError('RunningInstance terminal status is invalid'); +} + +function candidateFromRow(row: CandidateRow): LegacyShadowTerminalCandidate { + const attemptCount = safeInteger(row.attemptCount, 'attemptCount'); + if ( + (attemptCount === 0 && row.attemptId !== null) || + (attemptCount > 0 && + (row.attemptId === null || + row.attemptStatus === null || + row.attemptCreatedAtMs === null)) + ) { + throw new TypeError('Run Attempt projection is invalid'); + } + return { + runId: row.runId, + ...(row.legacyCronId === null + ? {} + : { + legacyCronId: optionalPositiveInteger( + row.legacyCronId, + 'legacyCronId', + ), + }), + origin: origin(row.origin), + runStatus: runStatus(row.runStatus), + createdAtMs: safeInteger(row.createdAtMs, 'createdAtMs'), + ...(row.startedAtMs === null + ? {} + : { startedAtMs: safeInteger(row.startedAtMs, 'startedAtMs') }), + ...(row.finishedAtMs === null + ? {} + : { finishedAtMs: safeInteger(row.finishedAtMs, 'finishedAtMs') }), + attemptCount, + ...(row.attemptId === null + ? {} + : { + attempt: { + attemptId: row.attemptId, + status: attemptStatus(row.attemptStatus!), + ...(row.attemptPid === null + ? {} + : { pid: optionalPositiveInteger(row.attemptPid, 'attemptPid') }), + ...(row.attemptLogArtifactId === null + ? {} + : { logArtifactId: row.attemptLogArtifactId }), + createdAtMs: safeInteger( + row.attemptCreatedAtMs!, + 'attemptCreatedAtMs', + ), + ...(row.attemptStartedAtMs === null + ? {} + : { + startedAtMs: safeInteger( + row.attemptStartedAtMs, + 'attemptStartedAtMs', + ), + }), + ...(row.attemptFinishedAtMs === null + ? {} + : { + finishedAtMs: safeInteger( + row.attemptFinishedAtMs, + 'attemptFinishedAtMs', + ), + }), + ...(row.attemptExitCode === null + ? {} + : { + exitCode: safeInteger(row.attemptExitCode, 'attemptExitCode'), + }), + }, + }), + }; +} + +/** SQLite/Sequelize adapter for the explicit, read-only local audit. */ +export class LegacySequelizeShadowTerminalDifferenceSource + implements LegacyShadowTerminalDifferenceSource +{ + constructor( + private readonly database: Sequelize, + private readonly createLogArtifactId: LegacyTerminalLogArtifactIdFactory, + ) {} + + async listCandidates({ + projectId, + origins, + windowStartMs, + windowEndMs, + observedAtMs, + correlationToleranceMs, + cursor, + limit, + }: { + projectId: string; + origins: readonly ExecutionOrigin[]; + windowStartMs: number; + windowEndMs: number; + observedAtMs: number; + correlationToleranceMs: number; + cursor?: LegacyShadowTerminalCursor; + limit: number; + }): Promise { + assertLimit(limit); + if (origins.length === 0) { + return { + candidates: [], + evidence: [], + evidenceTruncated: false, + truncated: false, + }; + } + const rows = await this.database.query( + `SELECT + r.id AS "runId", + r.legacy_cron_id AS "legacyCronId", + r.execution_origin AS origin, + r.status AS "runStatus", + r.created_at_ms AS "createdAtMs", + r.started_at_ms AS "startedAtMs", + r.finished_at_ms AS "finishedAtMs", + (SELECT COUNT(*) + FROM ${RUN_ATTEMPT_TABLE} counted + WHERE counted.run_id = r.id) AS "attemptCount", + attempt.id AS "attemptId", + attempt.status AS "attemptStatus", + attempt.pid AS "attemptPid", + attempt.log_artifact_id AS "attemptLogArtifactId", + attempt.created_at_ms AS "attemptCreatedAtMs", + attempt.started_at_ms AS "attemptStartedAtMs", + attempt.finished_at_ms AS "attemptFinishedAtMs", + attempt.exit_code AS "attemptExitCode" + FROM ${RUN_TABLE} r + LEFT JOIN ${RUN_ATTEMPT_TABLE} attempt + ON attempt.id = ( + SELECT latest.id + FROM ${RUN_ATTEMPT_TABLE} latest + WHERE latest.run_id = r.id + ORDER BY latest.attempt DESC, latest.id DESC + LIMIT 1 + ) + WHERE r.project_id = :projectId + AND r.execution_owner = 'legacy' + AND r.execution_origin IN (:origins) + AND r.created_at_ms >= :windowStartMs + AND r.created_at_ms < :windowEndMs + ${ + cursor === undefined + ? '' + : `AND ( + r.created_at_ms > :cursorCreatedAtMs OR + (r.created_at_ms = :cursorCreatedAtMs AND r.id > :cursorRunId) + )` + } + ORDER BY r.created_at_ms ASC, r.id ASC + LIMIT :fetchLimit`, + { + replacements: { + projectId, + origins: [...new Set(origins)], + windowStartMs, + windowEndMs, + ...(cursor === undefined + ? {} + : { + cursorCreatedAtMs: cursor.createdAtMs, + cursorRunId: cursor.runId, + }), + fetchLimit: limit + 1, + }, + type: QueryTypes.SELECT, + }, + ); + const truncated = rows.length > limit; + const candidates = rows.slice(0, limit).map(candidateFromRow); + if (candidates.length === 0) { + return { + candidates, + evidence: [], + evidenceTruncated: false, + truncated: false, + }; + } + + const evidence = await this.listEvidence({ + candidates, + windowStartMs, + observedAtMs, + correlationToleranceMs, + }); + const last = candidates[candidates.length - 1]; + return { + candidates, + ...evidence, + truncated, + ...(truncated + ? { + nextCursor: { + createdAtMs: last.createdAtMs, + runId: last.runId, + }, + } + : {}), + }; + } + + private async listEvidence({ + candidates, + windowStartMs, + observedAtMs, + correlationToleranceMs, + }: { + candidates: readonly LegacyShadowTerminalCandidate[]; + windowStartMs: number; + observedAtMs: number; + correlationToleranceMs: number; + }): Promise<{ + evidence: readonly LegacyTerminalEvidence[]; + evidenceTruncated: boolean; + }> { + const runIds = candidates.map((candidate) => candidate.runId); + const attemptIds = candidates.flatMap((candidate) => + candidate.attempt ? [candidate.attempt.attemptId] : [], + ); + const legacyCronIds = [ + ...new Set( + candidates.flatMap((candidate) => + candidate.legacyCronId === undefined ? [] : [candidate.legacyCronId], + ), + ), + ]; + const clauses = ['run_id IN (:runIds)']; + if (attemptIds.length > 0) clauses.push('attempt_id IN (:attemptIds)'); + if (legacyCronIds.length > 0) { + clauses.push(`( + cron_id IN (:legacyCronIds) + AND started_at >= :startedAfterSeconds + AND started_at <= :startedBeforeSeconds + )`); + } + const evidenceLimit = Math.min( + MAX_LEGACY_SHADOW_TERMINAL_EVIDENCE_PER_PAGE, + candidates.length * 8, + ); + const rows = await this.database.query( + `SELECT + id AS "instanceId", + cron_id AS "legacyCronId", + run_id AS "runId", + attempt_id AS "attemptId", + pid, + log_path AS "logPath", + started_at AS "startedAt", + finished_at AS "finishedAt", + status, + exit_code AS "exitCode" + FROM ${RUNNING_INSTANCE_TABLE} + WHERE status IN (1, 2, 3) + AND finished_at IS NOT NULL + AND (${clauses.join(' OR ')}) + ORDER BY started_at ASC, id ASC + LIMIT :fetchLimit`, + { + replacements: { + runIds, + ...(attemptIds.length === 0 ? {} : { attemptIds }), + ...(legacyCronIds.length === 0 + ? {} + : { + legacyCronIds, + startedAfterSeconds: Math.floor( + Math.max(0, windowStartMs - correlationToleranceMs) / 1_000, + ), + startedBeforeSeconds: Math.ceil(observedAtMs / 1_000), + }), + fetchLimit: evidenceLimit + 1, + }, + type: QueryTypes.SELECT, + }, + ); + const evidenceTruncated = rows.length > evidenceLimit; + return { + evidence: rows.slice(0, evidenceLimit).map((row) => ({ + instanceId: safeInteger(row.instanceId, 'instanceId'), + legacyCronId: safeInteger(row.legacyCronId, 'legacyCronId'), + ...(row.runId === null ? {} : { runId: row.runId }), + ...(row.attemptId === null ? {} : { attemptId: row.attemptId }), + ...(row.pid === null + ? {} + : { pid: optionalPositiveInteger(row.pid, 'pid') }), + ...(row.logPath === null + ? {} + : { logArtifactId: this.createLogArtifactId(row.logPath) }), + startedAtMs: secondsToMilliseconds(row.startedAt, 'startedAt'), + finishedAtMs: secondsToMilliseconds(row.finishedAt, 'finishedAt'), + outcome: terminalOutcome(safeInteger(row.status, 'status')), + ...(row.exitCode === null + ? {} + : { exitCode: safeInteger(row.exitCode, 'exitCode') }), + })), + evidenceTruncated, + }; + } +} diff --git a/back/runtime/application/legacyShadowTerminalDifferenceAuditor.ts b/back/runtime/application/legacyShadowTerminalDifferenceAuditor.ts new file mode 100644 index 00000000..729f056f --- /dev/null +++ b/back/runtime/application/legacyShadowTerminalDifferenceAuditor.ts @@ -0,0 +1,641 @@ +import type { + ExecutionOrigin, + RunAttemptStatus, + RunStatus, +} from '../domain/run'; +import { EXECUTION_ORIGINS } from '../domain/run'; +import { + MAX_LEGACY_SHADOW_TERMINAL_EVIDENCE_PER_PAGE, + MAX_LEGACY_SHADOW_TERMINAL_PAGE_SIZE, + type LegacyShadowTerminalAttemptEvidence, + type LegacyShadowTerminalCandidate, + type LegacyShadowTerminalCursor, + type LegacyShadowTerminalDifferenceSource, + type LegacyShadowTerminalPage, + type LegacyTerminalEvidence, + type LegacyTerminalOutcome, +} from '../ports/legacyShadowTerminalDifference'; + +export type LegacyShadowTerminalAuditProfile = 'edge' | 'standalone'; + +export const LEGACY_SHADOW_TERMINAL_CATEGORIES = Object.freeze([ + 'matched', + 'shadow_not_terminal', + 'shadow_attempt_missing', + 'shadow_attempt_ambiguous', + 'legacy_evidence_missing', + 'legacy_evidence_ambiguous', + 'status_mismatch', + 'field_mismatch', +] as const); + +export type LegacyShadowTerminalCategory = + (typeof LEGACY_SHADOW_TERMINAL_CATEGORIES)[number]; + +export type LegacyShadowTerminalCategoryCounts = Record< + LegacyShadowTerminalCategory, + number +>; + +export interface LegacyShadowTerminalDimensionCounts { + compared: number; + matched: number; + mismatched: number; + unavailable: number; +} + +export interface LegacyShadowTerminalDimensions { + status: LegacyShadowTerminalDimensionCounts; + exitCode: LegacyShadowTerminalDimensionCounts; + startedAt: LegacyShadowTerminalDimensionCounts; + finishedAt: LegacyShadowTerminalDimensionCounts; + logArtifact: LegacyShadowTerminalDimensionCounts; +} + +export interface LegacyShadowTerminalOriginSummary + extends LegacyShadowTerminalCategoryCounts { + origin: ExecutionOrigin; + scanned: number; +} + +export type LegacyShadowTerminalAuditAssessment = + | 'matched' + | 'empty' + | 'differences_found' + | 'window_open' + | 'incomplete'; + +export type LegacyShadowTerminalAuditStopReason = + | 'complete' + | 'page_limit' + | 'cursor_stalled'; + +export interface LegacyShadowTerminalDifferenceReport { + schema: 'qinglong/legacy-shadow-terminal-difference-report@v1'; + profile: LegacyShadowTerminalAuditProfile; + observedAtMs: number; + window: { + basis: 'shadow_run_created_at'; + startInclusiveMs: number; + endExclusiveMs: number; + minimumSettlingAgeMs: number; + closed: boolean; + }; + coverage: { + direction: 'shadow_to_legacy'; + cohort: 'legacy_owned_shadow_runs'; + legacyWithoutShadow: 'not_measured'; + }; + budget: { pageSize: number; maxPages: number; maxCandidates: number }; + pages: number; + scanned: number; + stopReason: LegacyShadowTerminalAuditStopReason; + remaining: boolean; + evidenceComplete: boolean; + evidenceOverflowPages: number; + assessment: LegacyShadowTerminalAuditAssessment; + counts: LegacyShadowTerminalCategoryCounts; + byOrigin: readonly LegacyShadowTerminalOriginSummary[]; + dimensions: LegacyShadowTerminalDimensions; + terminalAgreementPermille?: number; + fullyComparablePermille?: number; +} + +export interface LegacyShadowTerminalAuditorOptions { + profile: LegacyShadowTerminalAuditProfile; + projectId?: string; + origins: readonly ExecutionOrigin[]; + windowStartMs: number; + windowEndMs: number; + observedAtMs?: number; + minimumSettlingAgeMs?: number; + correlationToleranceMs?: number; + clock?: { now(): number }; +} + +interface Selection { + status: 'matched' | 'missing' | 'ambiguous'; + evidence?: LegacyTerminalEvidence; +} + +interface Classification { + category: LegacyShadowTerminalCategory; + fullyComparable: boolean; +} + +const BUDGETS: Readonly< + Record< + LegacyShadowTerminalAuditProfile, + { pageSize: number; maxPages: number } + > +> = { + edge: { pageSize: 8, maxPages: 1 }, + standalone: { pageSize: 32, maxPages: 4 }, +}; + +const MAX_CONFIGURED_ORIGINS = 7; +const MAX_CORRELATION_TOLERANCE_MS = 60_000; +const EXECUTION_ORIGIN_SET = new Set(EXECUTION_ORIGINS); + +function assertSafeInteger(name: string, value: number): void { + if (!Number.isSafeInteger(value) || value < 0) { + throw new RangeError(`${name} must be a non-negative safe integer`); + } +} + +function emptyCounts(): LegacyShadowTerminalCategoryCounts { + return Object.fromEntries( + LEGACY_SHADOW_TERMINAL_CATEGORIES.map((category) => [category, 0]), + ) as LegacyShadowTerminalCategoryCounts; +} + +function emptyDimension(): LegacyShadowTerminalDimensionCounts { + return { compared: 0, matched: 0, mismatched: 0, unavailable: 0 }; +} + +function emptyDimensions(): LegacyShadowTerminalDimensions { + return { + status: emptyDimension(), + exitCode: emptyDimension(), + startedAt: emptyDimension(), + finishedAt: emptyDimension(), + logArtifact: emptyDimension(), + }; +} + +function originSummary( + origin: ExecutionOrigin, +): LegacyShadowTerminalOriginSummary { + return { origin, scanned: 0, ...emptyCounts() }; +} + +function compareDimension( + target: LegacyShadowTerminalDimensionCounts, + left: unknown, + right: unknown, + matches: (left: unknown, right: unknown) => boolean = (a, b) => a === b, +): boolean | undefined { + if (left === undefined || right === undefined) { + target.unavailable += 1; + return undefined; + } + target.compared += 1; + if (matches(left, right)) { + target.matched += 1; + return true; + } + target.mismatched += 1; + return false; +} + +function markUnavailable(dimensions: LegacyShadowTerminalDimensions): void { + for (const dimension of Object.values(dimensions)) { + dimension.unavailable += 1; + } +} + +function expectedOutcome( + status: RunStatus | RunAttemptStatus, +): LegacyTerminalOutcome | undefined { + if (status === 'succeeded') return 'succeeded'; + if (status === 'failed') return 'failed'; + if (status === 'cancelled') return 'stopped'; + return undefined; +} + +function isTerminalShadowStatus(status: RunStatus | RunAttemptStatus): boolean { + return ['succeeded', 'failed', 'cancelled', 'timed_out', 'lost'].includes( + status, + ); +} + +function selectEvidence( + candidate: LegacyShadowTerminalCandidate, + attempt: LegacyShadowTerminalAttemptEvidence, + evidence: readonly LegacyTerminalEvidence[], + toleranceMs: number, +): Selection { + const direct = evidence.filter( + (item) => + item.attemptId === attempt.attemptId || item.runId === candidate.runId, + ); + if ( + direct.some( + (item) => + (item.attemptId !== undefined && + item.attemptId !== attempt.attemptId) || + (item.runId !== undefined && item.runId !== candidate.runId), + ) + ) { + return { status: 'ambiguous' }; + } + if (direct.length === 1) return { status: 'matched', evidence: direct[0] }; + if (direct.length > 1) return { status: 'ambiguous' }; + if (candidate.legacyCronId === undefined) return { status: 'missing' }; + + const sameCron = evidence.filter( + (item) => item.legacyCronId === candidate.legacyCronId, + ); + const byLog = + attempt.logArtifactId === undefined + ? [] + : sameCron.filter((item) => item.logArtifactId === attempt.logArtifactId); + const byPid = + attempt.pid === undefined + ? [] + : sameCron.filter((item) => item.pid === attempt.pid); + + if (attempt.logArtifactId !== undefined && attempt.pid !== undefined) { + const byPidSet = new Set(byPid); + const intersection = byLog.filter((item) => byPidSet.has(item)); + if (intersection.length === 1) { + return { status: 'matched', evidence: intersection[0] }; + } + if (intersection.length > 1 || byLog.length > 1 || byPid.length > 1) { + return { status: 'ambiguous' }; + } + if (byLog.length === 1 && byPid.length === 0) { + return { status: 'matched', evidence: byLog[0] }; + } + if (byPid.length === 1 && byLog.length === 0) { + return { status: 'matched', evidence: byPid[0] }; + } + if (byLog.length === 1 && byPid.length === 1) { + return { status: 'ambiguous' }; + } + } else if (byLog.length === 1) { + return { status: 'matched', evidence: byLog[0] }; + } else if (byLog.length > 1) { + return { status: 'ambiguous' }; + } else if (byPid.length === 1) { + return { status: 'matched', evidence: byPid[0] }; + } else if (byPid.length > 1) { + return { status: 'ambiguous' }; + } + + const shadowStartedAtMs = attempt.startedAtMs ?? candidate.startedAtMs; + if (shadowStartedAtMs === undefined) return { status: 'missing' }; + const byStart = sameCron.filter( + (item) => Math.abs(item.startedAtMs - shadowStartedAtMs) <= toleranceMs, + ); + if (byStart.length === 1) { + return { status: 'matched', evidence: byStart[0] }; + } + return { status: byStart.length > 1 ? 'ambiguous' : 'missing' }; +} + +function classify( + candidate: LegacyShadowTerminalCandidate, + evidence: readonly LegacyTerminalEvidence[], + evidenceTruncated: boolean, + toleranceMs: number, + dimensions: LegacyShadowTerminalDimensions, +): Classification { + if ( + !isTerminalShadowStatus(candidate.runStatus) || + candidate.finishedAtMs === undefined + ) { + markUnavailable(dimensions); + return { category: 'shadow_not_terminal', fullyComparable: false }; + } + if (candidate.attemptCount === 0 || candidate.attempt === undefined) { + markUnavailable(dimensions); + return { category: 'shadow_attempt_missing', fullyComparable: false }; + } + if (candidate.attemptCount !== 1) { + markUnavailable(dimensions); + return { category: 'shadow_attempt_ambiguous', fullyComparable: false }; + } + if ( + !isTerminalShadowStatus(candidate.attempt.status) || + candidate.attempt.finishedAtMs === undefined + ) { + markUnavailable(dimensions); + return { category: 'shadow_not_terminal', fullyComparable: false }; + } + if (evidenceTruncated) { + markUnavailable(dimensions); + return { category: 'legacy_evidence_ambiguous', fullyComparable: false }; + } + + const selection = selectEvidence( + candidate, + candidate.attempt, + evidence, + toleranceMs, + ); + if (selection.status === 'ambiguous') { + markUnavailable(dimensions); + return { category: 'legacy_evidence_ambiguous', fullyComparable: false }; + } + if (selection.status === 'missing' || selection.evidence === undefined) { + markUnavailable(dimensions); + return { category: 'legacy_evidence_missing', fullyComparable: false }; + } + + const expectedRunOutcome = expectedOutcome(candidate.runStatus); + const expectedAttemptOutcome = expectedOutcome(candidate.attempt.status); + let statusMatches: boolean | undefined; + if ( + expectedRunOutcome === undefined || + expectedAttemptOutcome === undefined + ) { + dimensions.status.unavailable += 1; + } else if (expectedRunOutcome !== expectedAttemptOutcome) { + dimensions.status.compared += 1; + dimensions.status.mismatched += 1; + statusMatches = false; + } else { + statusMatches = compareDimension( + dimensions.status, + expectedRunOutcome, + selection.evidence.outcome, + ); + } + if ( + expectedRunOutcome === undefined || + expectedAttemptOutcome === undefined || + expectedRunOutcome !== expectedAttemptOutcome || + statusMatches !== true + ) { + for (const dimension of [ + dimensions.exitCode, + dimensions.startedAt, + dimensions.finishedAt, + dimensions.logArtifact, + ]) { + dimension.unavailable += 1; + } + return { category: 'status_mismatch', fullyComparable: false }; + } + + const exitMatches = compareDimension( + dimensions.exitCode, + candidate.attempt.exitCode, + selection.evidence.exitCode, + ); + const startMatches = compareDimension( + dimensions.startedAt, + candidate.attempt.startedAtMs ?? candidate.startedAtMs, + selection.evidence.startedAtMs, + (left, right) => Math.abs(Number(left) - Number(right)) <= toleranceMs, + ); + const finishMatches = compareDimension( + dimensions.finishedAt, + candidate.attempt.finishedAtMs ?? candidate.finishedAtMs, + selection.evidence.finishedAtMs, + (left, right) => Math.abs(Number(left) - Number(right)) <= toleranceMs, + ); + const logMatches = compareDimension( + dimensions.logArtifact, + candidate.attempt.logArtifactId, + selection.evidence.logArtifactId, + ); + const compared = [true, exitMatches, startMatches, finishMatches, logMatches]; + const hasMismatch = compared.includes(false); + return { + category: hasMismatch ? 'field_mismatch' : 'matched', + fullyComparable: compared.every((value) => value === true), + }; +} + +function sameCursor( + left: LegacyShadowTerminalCursor | undefined, + right: LegacyShadowTerminalCursor, +): boolean { + return ( + left !== undefined && + left.createdAtMs === right.createdAtMs && + left.runId === right.runId + ); +} + +function validatePage( + page: LegacyShadowTerminalPage, + cursor: LegacyShadowTerminalCursor | undefined, + limit: number, + allowedOrigins: ReadonlySet, + windowStartMs: number, + windowEndMs: number, +): void { + if (page.candidates.length > limit) { + throw new Error('Legacy Shadow terminal source exceeded candidate limit'); + } + if ( + page.evidence.length > + Math.min( + MAX_LEGACY_SHADOW_TERMINAL_EVIDENCE_PER_PAGE, + page.candidates.length * 8, + ) + ) { + throw new Error('Legacy Shadow terminal source exceeded evidence limit'); + } + let previous = cursor; + const runIds = new Set(); + for (const candidate of page.candidates) { + if ( + candidate.runId.length === 0 || + runIds.has(candidate.runId) || + !allowedOrigins.has(candidate.origin) || + !Number.isSafeInteger(candidate.createdAtMs) || + candidate.createdAtMs < windowStartMs || + candidate.createdAtMs >= windowEndMs || + (previous !== undefined && + (candidate.createdAtMs < previous.createdAtMs || + (candidate.createdAtMs === previous.createdAtMs && + candidate.runId <= previous.runId))) + ) { + throw new Error('Legacy Shadow terminal source returned an invalid page'); + } + runIds.add(candidate.runId); + previous = { createdAtMs: candidate.createdAtMs, runId: candidate.runId }; + } + if (page.truncated) { + if ( + page.candidates.length !== limit || + page.nextCursor === undefined || + previous === undefined || + page.nextCursor.createdAtMs !== previous.createdAtMs || + page.nextCursor.runId !== previous.runId + ) { + throw new Error( + 'Legacy Shadow terminal source returned an invalid cursor', + ); + } + } else if (page.nextCursor !== undefined) { + throw new Error('Completed Legacy Shadow terminal page returned a cursor'); + } +} + +function permille(numerator: number, denominator: number): number { + return Math.floor((numerator * 1_000) / denominator); +} + +/** + * Explicit, timer-free audit for a closed Shadow cohort. The report deliberately + * does not claim Legacy-to-Shadow capture coverage because RunningInstances have + * no trustworthy execution-origin field in the 2.x schema. + */ +export class LegacyShadowTerminalDifferenceAuditor { + constructor(private readonly source: LegacyShadowTerminalDifferenceSource) {} + + async run( + options: LegacyShadowTerminalAuditorOptions, + ): Promise { + if (!['edge', 'standalone'].includes(options.profile)) { + throw new RangeError('profile must be edge or standalone'); + } + const projectId = options.projectId ?? 'default'; + if (projectId.length < 1 || projectId.length > 128) { + throw new RangeError('projectId length is invalid'); + } + const origins = [...new Set(options.origins)]; + if (origins.length < 1 || origins.length > MAX_CONFIGURED_ORIGINS) { + throw new RangeError('origin count is outside its audit budget'); + } + if (origins.some((origin) => !EXECUTION_ORIGIN_SET.has(origin))) { + throw new RangeError('terminal audit origin is invalid'); + } + assertSafeInteger('windowStartMs', options.windowStartMs); + assertSafeInteger('windowEndMs', options.windowEndMs); + if (options.windowStartMs >= options.windowEndMs) { + throw new RangeError('terminal audit window must be non-empty'); + } + const observedAtMs = + options.observedAtMs ?? options.clock?.now() ?? Date.now(); + const minimumSettlingAgeMs = options.minimumSettlingAgeMs ?? 5 * 60_000; + const correlationToleranceMs = options.correlationToleranceMs ?? 2_000; + assertSafeInteger('observedAtMs', observedAtMs); + assertSafeInteger('minimumSettlingAgeMs', minimumSettlingAgeMs); + assertSafeInteger('correlationToleranceMs', correlationToleranceMs); + if (correlationToleranceMs > MAX_CORRELATION_TOLERANCE_MS) { + throw new RangeError('correlationToleranceMs exceeds its hard limit'); + } + const budget = BUDGETS[options.profile]; + if (budget.pageSize > MAX_LEGACY_SHADOW_TERMINAL_PAGE_SIZE) { + throw new RangeError('terminal audit Profile exceeds the source limit'); + } + + const counts = emptyCounts(); + const dimensions = emptyDimensions(); + const originSummaries = new Map( + origins.map((origin) => [origin, originSummary(origin)]), + ); + const allowedOrigins = new Set(origins); + let cursor: LegacyShadowTerminalCursor | undefined; + let pages = 0; + let scanned = 0; + let fullyComparable = 0; + let evidenceOverflowPages = 0; + let remaining = false; + let stopReason: LegacyShadowTerminalAuditStopReason = 'complete'; + + while (pages < budget.maxPages) { + const page = await this.source.listCandidates({ + projectId, + origins, + windowStartMs: options.windowStartMs, + windowEndMs: options.windowEndMs, + observedAtMs, + correlationToleranceMs, + ...(cursor === undefined ? {} : { cursor }), + limit: budget.pageSize, + }); + pages += 1; + validatePage( + page, + cursor, + budget.pageSize, + allowedOrigins, + options.windowStartMs, + options.windowEndMs, + ); + if (page.evidenceTruncated) evidenceOverflowPages += 1; + + for (const candidate of page.candidates) { + const result = classify( + candidate, + page.evidence, + page.evidenceTruncated, + correlationToleranceMs, + dimensions, + ); + counts[result.category] += 1; + const perOrigin = originSummaries.get(candidate.origin)!; + perOrigin.scanned += 1; + perOrigin[result.category] += 1; + scanned += 1; + if (result.fullyComparable) fullyComparable += 1; + } + + if (!page.truncated) break; + if ( + page.nextCursor === undefined || + sameCursor(cursor, page.nextCursor) + ) { + remaining = true; + stopReason = 'cursor_stalled'; + break; + } + cursor = page.nextCursor; + if (pages === budget.maxPages) { + remaining = true; + stopReason = 'page_limit'; + } + } + + const windowClosed = + options.windowEndMs <= observedAtMs - minimumSettlingAgeMs; + const evidenceComplete = evidenceOverflowPages === 0; + const assessment: LegacyShadowTerminalAuditAssessment = !windowClosed + ? 'window_open' + : remaining || !evidenceComplete + ? 'incomplete' + : scanned === 0 + ? 'empty' + : counts.matched === scanned + ? 'matched' + : 'differences_found'; + const ratiosAvailable = + windowClosed && !remaining && evidenceComplete && scanned > 0; + + return { + schema: 'qinglong/legacy-shadow-terminal-difference-report@v1', + profile: options.profile, + observedAtMs, + window: { + basis: 'shadow_run_created_at', + startInclusiveMs: options.windowStartMs, + endExclusiveMs: options.windowEndMs, + minimumSettlingAgeMs, + closed: windowClosed, + }, + coverage: { + direction: 'shadow_to_legacy', + cohort: 'legacy_owned_shadow_runs', + legacyWithoutShadow: 'not_measured', + }, + budget: { + pageSize: budget.pageSize, + maxPages: budget.maxPages, + maxCandidates: budget.pageSize * budget.maxPages, + }, + pages, + scanned, + stopReason, + remaining, + evidenceComplete, + evidenceOverflowPages, + assessment, + counts, + byOrigin: [...originSummaries.values()], + dimensions, + ...(ratiosAvailable + ? { + terminalAgreementPermille: permille(counts.matched, scanned), + fullyComparablePermille: permille(fullyComparable, scanned), + } + : {}), + }; + } +} diff --git a/back/runtime/ports/legacyShadowTerminalDifference.ts b/back/runtime/ports/legacyShadowTerminalDifference.ts new file mode 100644 index 00000000..e46f328b --- /dev/null +++ b/back/runtime/ports/legacyShadowTerminalDifference.ts @@ -0,0 +1,76 @@ +import type { + ExecutionOrigin, + RunAttemptStatus, + RunStatus, +} from '../domain/run'; + +export const MAX_LEGACY_SHADOW_TERMINAL_PAGE_SIZE = 64; +export const MAX_LEGACY_SHADOW_TERMINAL_EVIDENCE_PER_PAGE = 512; + +export interface LegacyShadowTerminalCursor { + createdAtMs: number; + runId: string; +} + +export interface LegacyShadowTerminalAttemptEvidence { + attemptId: string; + status: RunAttemptStatus; + pid?: number; + logArtifactId?: string; + createdAtMs: number; + startedAtMs?: number; + finishedAtMs?: number; + exitCode?: number; +} + +export interface LegacyShadowTerminalCandidate { + runId: string; + legacyCronId?: number; + origin: ExecutionOrigin; + runStatus: RunStatus; + createdAtMs: number; + startedAtMs?: number; + finishedAtMs?: number; + attemptCount: number; + attempt?: LegacyShadowTerminalAttemptEvidence; +} + +export type LegacyTerminalOutcome = 'succeeded' | 'failed' | 'stopped'; + +export interface LegacyTerminalEvidence { + instanceId: number; + legacyCronId: number; + runId?: string; + attemptId?: string; + pid?: number; + logArtifactId?: string; + startedAtMs: number; + finishedAtMs: number; + outcome: LegacyTerminalOutcome; + exitCode?: number; +} + +export interface LegacyShadowTerminalPage { + candidates: readonly LegacyShadowTerminalCandidate[]; + evidence: readonly LegacyTerminalEvidence[]; + evidenceTruncated: boolean; + truncated: boolean; + nextCursor?: LegacyShadowTerminalCursor; +} + +/** + * Read-only local Legacy authority used by the explicit terminal-difference + * audit. Implementations must bound both candidate and evidence result sets. + */ +export interface LegacyShadowTerminalDifferenceSource { + listCandidates(options: { + projectId: string; + origins: readonly ExecutionOrigin[]; + windowStartMs: number; + windowEndMs: number; + observedAtMs: number; + correlationToleranceMs: number; + cursor?: LegacyShadowTerminalCursor; + limit: number; + }): Promise; +} diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index 4d8892da..b667f51e 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -6,10 +6,24 @@ - 目标版本：QingLong 3.x - 作者：QingLong Maintainers - 创建日期：2026-07-17 -- 最后更新：2026-08-18 +- 最后更新：2026-08-19 - 讨论范围：架构与演进路线，不包含最终 UI 视觉方案 -最新增量证据（2026-08-18）： +最新增量证据（2026-08-19）： + +- D-358/ADR-0450（已接受）：新增显式、只读、一次性的 Legacy Shadow 闭合窗口终态审计。调用方必须提供 origin 与 + `[windowStartMs, windowEndMs)`，cohort 固定为窗口内创建的 legacy-owned Shadow Run；只有窗口经过默认五分钟 settling、候选 keyset 与 + Legacy evidence 都完整且分母非零时，才输出 terminal agreement/full comparability permille。关联只接受 direct Run/Attempt reference、同 Cron + opaque log ID、PID 或唯一容差内 started time，任何多解或 evidence hard-limit 都归入 ambiguous/incomplete；报告按八类 outcome 与五个固定 + dimension 计数守恒，不含 Project/Run/Attempt/Cron/PID/log/task/user/error identity。edge 预算为 `8 × 1 page`，standalone 为 + `32 × 4 pages`；每页一次候选查询和一次 evidence 查询，无 N+1、timer、watcher、写入、schema/migration、package、生产依赖或部署对象。 + 因 2.x RunningInstance 没有可信 execution origin，v1 明确只证明 `shadow_to_legacy`，将 `legacyWithoutShadow` 标为 `not_measured`，不把它单独 + 冒充 Primary gate。阶段门已重跑：真实 SQLite/CLI 聚焦测试 `14/14`、Legacy/Shadow 串行扩展 `91/91`、`build:back`、完整 backend + `1,455 pass / 0 fail / 2 conditional skip`、18-package clean build/test、14/14 static audit 与 14/14 artifact audit 全部通过。 + edge/standalone 产物字节保持 D-357 基线：base `2,589,998 / 2,590,076`、adopted `2,809,293 / 2,809,416`、application + `3,632,877 / 3,632,997`、application-api `3,800,430 / 3,800,574`、AI `3,069,251 / 3,069,341`、application+AI + `4,493,151 / 4,493,283`、MCP `7,315,930 / 7,316,038`。本切片不改 PostgreSQL、容器或 Kubernetes 部署面，因而不重跑物理 + PostgreSQL HA/K3s，也不把 D-357 的相邻结果冒充本阶段新证据；D-359 继续负责 edge/standalone 资源压力和 Shadow-off 回滚演练。 - D-357/ADR-0449（已接受）：将 D-356 的进程内 startup summary 收紧为可供后续 origin-scoped gate 使用的版本化差异证据。每个候选只归入 completed/cancelled/abandoned/markedLost/repaired/pending/ambiguous/skipped/failed 九个固定 outcome，总量与最多七条已配置 origin @@ -9115,7 +9129,7 @@ flowchart LR | PR-1 Run Schema | Incubating | Run/RunAttempt/RunEvent schema、nullable cancel request 与 Attempt deadline 字段及恢复索引、CancellationDispatch 状态/version/lease/backoff schema、Repository port、临时 Sequelize adapter、统一事件大小/分页上限、跨 adapter RunRepository contract suite（原子事务、回滚、Run/Attempt/RetryPolicy CAS、唯一错误、分页与取消恢复）；ADR-0041 的 `pg-0003-run-retry-policy`、capability v2、driver-neutral PostgreSQL Run Repository 与真实 `pg.Pool` 上的共享 Repository/rollback/SQLSTATE contract；ADR-0063/0069/0071/0073/0074/0076 的独立 Node 24 local-sqlite typed schema、十二条 reviewed migration、capability v6、共享 operation authority、readiness/RunRepository/API credential repository/receipt journal/dispatch plan/encrypted Secret envelope/Project Policy/security audit/authorized mutation/stable Identity catalog、Drizzle↔真实 catalog table/column/index/CHECK/FK lockstep、base/adopted/application edge/standalone 产物门禁；ADR-0064 的 legacy baseline/plan digest、Online Backup recovery、side-by-side target migration、staged manifest、双库栅栏 activation、source 生命周期写栅栏、target stable identity 和重启语义；ADR-0065 的独立 cutover authority、外部副作用停机 evidence、append-only journal、start/restart/stop barrier 与 unknown→manual_required 收敛；ADR-0066 的 adopted storage→Run reconciliation→receipt maintenance→domain recovery→lifecycle→admission application gate、严格有界 recovery summary 与 admission-first reverse stop；ADR-0067 的 SQLite 事实驱动 Run 候选源、256 条硬上限、截断失败关闭和唯一 Repository authority；ADR-0068 的 receipt-first Reconciler、callback token/sequence fence、exact local-process identity、Attempt/Run/双 Event 原子终态推进和最终 verifier；ADR-0069 的 local-process 单向包边界、pre-spawn journal、受审 POSIX launcher、immutable receipt、exact identity 和 Profile-aware cleanup lifecycle；ADR-0070 的独立 local-execution、spawn 前后双 transaction CAS、callback digest、exact stop 补偿与 fail-closed starting 保留；ADR-0071 的独立 local-dispatch、不可变 revision/context、Secret-first materializer、Profile Artifact admission、4/64 MiB output hard quota 和窄 application facade；ADR-0073/0074 的 Project-bound SecretRef、AES-256-GCM、外置 keyring 生命周期、双 SQLite authority CAS、application preflight、强 Principal/Policy 和 envelope+audit 原子提交；ADR-0086 的本机 Owner provisioning/challenge/claim/delivery acknowledgement/credential recovery CLI；ADR-0377 的 Local/Cluster 同构、Profile-aware、Project-scoped Artifact range read | fresh database/pepper setup、credential rotation/GC 运维编排与 Secret/Project/Role/Approval 管理 CLI/API/UI、备份/rekey、2.x/target process controller、人工 recovery、target 写后 reconciliation 与完整 cutover/rollback 演练；retry 产品策略、Artifact retention/tombstone stack、具体本机 lifecycle 和 target executable；Linux x64/arm64、PID namespace、断电与固定路由设备门禁；PostgreSQL 16/18 双连接并发与 failover integration；Task revision/context 跨方言 contract/并发压力与引用感知 retention、Keyv 数据迁移 | | PR-2 Run 状态机 | Incubating | 纯转换表、终态/时间/错误/执行器元数据规则、Run version 与 event sequence CAS、事务性 RunCommandService、回滚测试 | 重复 Worker callback/fencing、并发数据库压力测试、Primary 执行链接入 | | PR-3 Executor 端口 | Incubating | ADR-0003、ExecutionSpec/Context/Handle/Result、Executor port、LocalProcessExecutor、进程组取消/超时升级、流式背压、Legacy Cron spec builder、真实进程 contract tests、可复现 edge 基准入口 | 固定 edge/多架构设备基线、Legacy builder 与 makeCommand 差异审计、Primary 生产流量接入 | -| PR-4 Shadow Run | Incubating | origin 三态策略；默认关闭的 `QL3_SHADOW_ORIGINS`；manual、scheduled_node、boot、subscription、system 与 script 现有 ChildProcess 旁路观察；system crond 显式 origin marker、Shell execution ID、finish-only 准入、确定性 Run/Attempt 与 exact replay；`@once` 保持 manual、gRPC transport 不冒充 origin 的准入裁决；每个 worker 懒加载；Run/Attempt/Event 影子生命周期；稳定且不复制 caller 原文的 task identity/revision 与有界日志引用；同 worker 有界注册表和跨 worker 持久化候选关联；stop all/stop instance、Shell callback、乱序/迟到/歧义处理；监听前一次性、Profile-aware 的 keyset Startup Reconciler，终态证据补齐、lost/abandoned/pending 分流与 terminal Attempt response-loss 修复；origin-bounded 且逐级守恒的版本化 startup difference report、固定字段 metric batch 与一次性 collector；失败开放和契约测试 | 跨测量窗口终态差异查询、具体 exporter、资源压力、回滚演练和 Primary 门禁 | +| PR-4 Shadow Run | Incubating | origin 三态策略；默认关闭的 `QL3_SHADOW_ORIGINS`；manual、scheduled_node、boot、subscription、system 与 script 现有 ChildProcess 旁路观察；system crond 显式 origin marker、Shell execution ID、finish-only 准入、确定性 Run/Attempt 与 exact replay；`@once` 保持 manual、gRPC transport 不冒充 origin 的准入裁决；每个 worker 懒加载；Run/Attempt/Event 影子生命周期；稳定且不复制 caller 原文的 task identity/revision 与有界日志引用；同 worker 有界注册表和跨 worker 持久化候选关联；stop all/stop instance、Shell callback、乱序/迟到/歧义处理；监听前一次性、Profile-aware 的 keyset Startup Reconciler，终态证据补齐、lost/abandoned/pending 分流与 terminal Attempt response-loss 修复；origin-bounded 且逐级守恒的版本化 startup difference report、固定字段 metric batch 与一次性 collector；显式、只读、闭合窗口且 Profile-bounded 的 Shadow→Legacy 终态差异审计；失败开放和契约测试 | Legacy→Shadow capture authority、具体 exporter、资源压力、回滚演练和 Primary 门禁 | | PR-5 Primary LocalExecutor | Incubating（默认不激活，仅 manifest-gated manual） | runtime-owned Run 创建器；持久化先于 spawn；Run/Attempt 完整成功、失败、取消、超时与 lost 闭环；Executor handle 身份校验；spawn 后激活写失败的 stop+lost 补偿；completion rejection 安全收敛；独立 Primary 幂等查询与唯一索引竞态裁决；durable `run.cancel_requested`、stop-before-signal、首次请求幂等、晚到完成裁决与待取消有界恢复查询；最多 64 条一页的 cross-worker cancellation source；独立 CancellationDispatch Repository 原子 claim/result、lease expiry 接管、owner/token/version fencing、指数退避与结果 RunEvent；最多 64 页的单周期 cancel supervisor；显式 start/stop、无重叠、错误隔离、停止等待有上限且 timer unref 的 lifecycle runner；Linux durable handle 的 PID/boot/start ticks/process-group 复验与 TERM/KILL controller；完整有界分页且 fail-closed 的 startup Reconcile supervisor；RunningInstance nullable `run_id/attempt_id` 关联；Primary 专用组合 Repository 在同一 SQLite 事务提交前投影 Crontab/RunningInstance，失败整体回滚；有界且防穿越的 legacy log output ref；manual owner seam、真实本机装配、单 spawn/fail-closed；严格 manual-only rollout manifest loader、短期审批/gate、配置哈希审计；HTTP worker 已接轻量 lazy bootstrap，accepted 后按 receipt-first reconcile→completion receipt lifecycle→timeout intent lifecycle→cancel dispatch lifecycle→router 顺序激活，失败撤销，监听失败和 shutdown 有界停止；Primary timeout 在 spawn 前持久化绝对 deadline，有界 source/requester/supervisor 只提交 timeout 意图并复用 CancellationDispatch；代码级 edge/standalone Profile 为各 lifecycle 提供不同 cadence 与页上限，cluster-control/worker 拒绝误装本机 SQLite Primary；统一 CompletionService 原子提交 Attempt/Run/双 Event，spawn 前保存 callback token hash、终态推进 sequence，实时回调与 receipt consumer 共享入口并覆盖两个清理 crash window；manual Primary 已接入受限 POSIX launcher、`0600` direct-file stdout/stderr、父进程退出后续写、不可覆盖 receipt 生产、回执环境清除、TERM 转发等待及 live transaction 后清理；Startup Reconciler receipt-first 双检查并在确定 exited 后执行 profile 化的单次 50/100 ms publish grace；`0007` 独立 CompletionReceiptJournal 在 spawn 前登记、为升级前 active Attempt 补登记并驱动周期扫描，使终态残留继续可发现；确定无效的已知 Attempt receipt 先持久化隔离状态，再进入确定性私有分片 quarantine；终态 missing 与 quarantine 按 edge/standalone retention 有界清理；非 Journal 文件具备只读优先、固定分片/条目上限、overflow fail-closed、显式同盘隔离的 Node 24 运维 CLI；扫描具备页上限、resume cursor、timer unref、无重叠、有界 stop 和低敏计数；ENOSPC 与 launcher receipt 存储失败有代码门禁 | 部署配置写入/审批入口与用户可见状态；PostgreSQL CancellationDispatch adapter；cluster-control 生产启动拓扑；固定 edge/Linux 多架构与真实磁盘压力基线、完整 2.x API 契约和回滚演练 | | PR-7 Worker Session、Run Lease 与启动协议基础 | Incubating（默认关闭，独立入口显式 opt-in） | ADR-0012/0013/0014/0021/0057–0061/0108–0121/0231–0239/0377；有界 capability/Placement/Dispatcher；SQLite 协议孵化与 PostgreSQL v9 Session/Run Lease/credential/attestation authority；immutable revision Placement、数据库时钟 keyset candidate、认证 Worker Pull、digest-only offer recovery；versioned capability-free ExecutionSpec response、stable claim 跨重启退避、单 owner 原子 inbox 准入与 TLS 1.3 mTLS/`ql3w` HTTPS client；同一 package journal 上 revision-fenced starting/spawn/started/running/completion 状态、callback digest、tagged no-spawn 与 ambiguous recovery；PostgreSQL starting/running/start-failure/completion 数据库权威事务、精确重放与 cancellation/timeout 优先终态；batch Secret delivery 在 Attempt advisory lock 下复验 Session/Lease/revision 完整围栏并复用单 Agent，Secret-before-Artifact materializer 将同一 log ID 交给 Executor/journal/running ACK；offer-scoped `wlog-*` 私有文件 spool、Edge/Node 容量策略、append/quota/path 防护、barrier 后 output ownership、受审 POSIX Executor、truncation fact、固定内存流式 source、认证 Artifact stream、共享 immutable store port、S3-compatible SSE/checksum/条件 promotion adapter、upload-before-completion 协调，以及 Local/Cluster 同构、Profile-aware、ETag-fenced range read；用户取消 run.stop mutation 以数据库时间写 intent/Event 并在事务内复验 Project/RoleBinding fence；非执行取消 convergence lifecycle、运行期 expiry 与安全 lost retry 已接入 cluster-control 单一全局 cadence；完整 generation/version/token/Attempt fencing；独立最小权限 Worker ingress、CA/CRL 与连接 generation 热重载；offer journal、spawn barrier、receipt-first recovery；独立 `@qinglong/worker-runtime` 的本地 P-256 CSR、key/chain/trust 验证、generation + active pointer 安装和持久退避；默认关闭的 production process 已装配具体 execution graph、完整 Session heartbeat/drain/offline、direct-file bootstrap、单 Agent/单 cadence、startup reconciliation、证书 maintenance、transport fail-close/recovery 与 Edge/Node 有界预算；真实 PostgreSQL 18 + Linux Node 合约已覆盖 Run completion、credential 和 CA 双轮换且保持同一 Session；真实 K3s 合约已覆盖 TLS/credential Secret 分权、双对象 CAS、Recreate 顺序、identity generation 与单节点 PVC recovery；所有能力默认不可达且受 edge/cluster import audit 约束 | 具体 cert-manager/Vault/SPIFFE/离线 CA adapter 与模板、ingress reload controller、生产 RBAC、证书到期告警和 `ql3w` credential recovery 产品面；具体 KMS/Vault Secret provider、对象存储 credential/temporary lifecycle 与 retention/tombstone；Worker 管理 API；真实 Kubernetes 多节点 CSI/node-loss/production 360 秒 drain 与固定 edge 文件系统 suspend/时钟/断电、x64/arm64 资源门禁 | | PR-8 Project/Policy/Approval Core | Incubating（默认拒绝、无生产业务执行入口） | ADR-0028；统一六类 ActorRef 与 exact-shape 校验；`0017` ownerless default Project 和 append-only versioned RoleBinding；owner/admin/operator/viewer 固定矩阵；Project 内 mutation 幂等、expected-version CAS、双 SQLite 连接竞争门禁；archived read-only、revocation、存储损坏 fail-closed；Agent 写/Secret/Tool `require_approval`；ADR-0047 把六类 subject、role/permission matrix 与 fence 抽到 runtime-core，`pg-0004-project-policy`/capability v3 建立 ownerless PostgreSQL baseline、严格 role/state CHECK、append-only runtime 权限、SERIALIZABLE Project lock、mutation replay、双连接单 winner 和 cluster admission authorizer；ADR-0049/`pg-0005` capability v4 建立 stable IdentitySubject、append-only digest-only API credential、真实 cluster bearer authenticator、write-only durable security audit 与最小权限 runtime role，且已验证 HTTP→credential→Policy→audit→handler 纵向链路；ADR-0051 建立 `/api/v3` 认证前 peer/global 双预算、transport-peer-only、无 timer 且有界内存的 overload shield；ADR-0027 Artifact authorizer adapter；ADR-0029 `AuthenticatedPrincipal` contract、`0018` digest-only versioned challenge、CSPRNG/TTL、同事务消费 challenge + 写首 owner、精确重放与双连接竞争/崩溃回滚门禁；ADR-0030 `0019` stable identity/binding、legacy HS384 + current-session membership、logout/platform/revoke/disable、single-factor 与损坏 fail-closed 门禁；ADR-0031 `0020` digest-bound ApprovalRequest、User-only decision、Project/Role version fence、精确 expiry/重放/并发裁决及同事务 immutable dispatch；ADR-0032 `0021` execution backfill、三表原子 consume、稳定 due keyset、claim/renew/start/result fencing、pre-start takeover/post-start recovery-required、attempt budget、handler inspect/digest barrier 和 bounded dispatcher；ADR-0033/`0022` control/resolution backfill、start/renew/completion 原子联动、稳定 recovery keyset、双 resolver claim/takeover、finding/result 精确重放、自动/人工终结、迟到 completion 单 winner 和 evidence-only bounded reconciler；ADR-0034/`0023` 首个 `run.create` canonical plan、Run/Attempt/Event/receipt 同事务、幂等 collision fail-closed、renew/终态 fence、真实 SQLite handler 与 automatic evidence provider；ADR-0035/`0024` 独立 `approval.recover` 矩阵、稳定 User + 五分钟强认证、Project/RoleBinding fence、human resolution + authorization fact 原子提交、撤权竞态与回滚门禁；ADR-0036 recovery-first 单 timer lifecycle、edge/standalone 独立 cadence/页预算、跨周期 cursor、非重叠与有界 stop；ADR-0074 以新的 Node 24 SQLite v5 ownerless Project/RoleBinding/audit authority 和独立 local-secret-admin 提供强 Principal、`secret.manage`、撤权 fence、envelope+allowed audit 原子提交及不回显语义；ADR-0086 以可信 POSIX console 和 staged delivery 完成本机首 Owner 产品 ceremony | fresh database/pepper setup 与安全迁移向导；`shareStore`/Express 到 authentication core 的 production migration；credential rotation/revocation API、mTLS/Worker enrollment、恢复码；Project/Role/Approval/Secret 管理 CLI/API/UI、audit retention/query/export/alert、preview Artifact/digest/immutable plan builder、真实 MFA/hardware adapter、人工 recovery API/UI/独立 rate limit 与审计事件、handler/provider registry、lifecycle startup/shutdown/指标/admission gate；PostgreSQL action/receipt/provider/recovery-authorization 与 OPA adapter、缓存 version 失效；Tool/Package/Secret/Shell 各自的 handler/evidence contract；Secret/Run/Tool/Workflow waiting_approval 全入口装配；完整回滚演练 | diff --git a/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md b/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md index 7120ad4c..3dc033c6 100644 --- a/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md +++ b/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md @@ -239,7 +239,7 @@ Shadow Adapter 不得： Shadow 转换仍必须遵守 ADR-0001。无法合法映射时追加 compat.transition_mismatch，不能强行覆盖终态。 -当前 Alpha 切片对已审 Node worker origin 直接观察同一 ChildProcess 的 spawn、error 和 exit 事件，因此不依赖 Shell callback 才能形成基本终态。下述两级关联已补充 Shell callback、stop/cancel 和乱序/迟到回调；ADR-0448 又补充了监听前一次性启动恢复，ADR-0449 将其投影为 origin-bounded、版本化的差异报告与固定字段 metric batch。该 startup snapshot 仍不能替代跨测量窗口的历史终态对账与正式 Primary gate。 +当前 Alpha 切片对已审 Node worker origin 直接观察同一 ChildProcess 的 spawn、error 和 exit 事件，因此不依赖 Shell callback 才能形成基本终态。下述两级关联已补充 Shell callback、stop/cancel 和乱序/迟到回调；ADR-0448 又补充了监听前一次性启动恢复，ADR-0449 将其投影为 origin-bounded、版本化的差异报告与固定字段 metric batch。ADR-0450 再提供显式、只读、Profile-bounded 的闭合窗口终态审计；但 2.x RunningInstance 缺少可信 origin，因此它只证明已写 Shadow Run 到 Legacy evidence 的一致性，不能单独证明 Legacy→Shadow 捕获率或替代正式 Primary gate。 ### 9.4 `next` Alpha callback 与 stop 关联 @@ -253,7 +253,7 @@ Shadow 转换仍必须遵守 ADR-0001。无法合法映射时追加 compat.trans 6. 取消事实在 Legacy kill 前投递；同 worker 的后续 exit 排在取消之后。跨 worker 使用持久化定位器尽力关联，任何查询或写入失败都不能阻断 kill 或改变 2.x API 响应。 7. 乱序 finished 可以从 queued/claimed 补齐 dispatching、starting、running 和终态；重复终态 callback、取消后的迟到成功 callback 不覆盖终态，也不追加重复完成事件。 -这仍不是完整的 Shadow→Primary 门禁：ADR-0448 已提供启动后有界批量扫描、终态证据补齐、lost/abandoned 收敛和两事务 response-loss 修复；ADR-0449 已增加 startup 差异报表、固定低基数 metric batch 与可注入单次 collector，但尚未完成跨窗口终态差异查询、具体 exporter、资源/回滚演练和正式 Primary gate。后续能力不得让 edge 增加常驻 watcher 或无界内存队列。 +这仍不是完整的 Shadow→Primary 门禁：ADR-0448 已提供启动后有界批量扫描、终态证据补齐、lost/abandoned 收敛和两事务 response-loss 修复；ADR-0449 已增加 startup 差异报表、固定低基数 metric batch 与可注入单次 collector；ADR-0450 已完成不伪造反向捕获率的闭合窗口 Shadow→Legacy 终态差异查询。具体 exporter、Legacy→Shadow capture authority、资源/回滚演练和正式 Primary gate 尚未完成。后续能力不得让 edge 增加常驻 watcher 或无界内存队列。 ### 9.3 Shadow 写失败 diff --git a/docs/adr/ADR-0450-closed-window-legacy-shadow-terminal-difference-audit.md b/docs/adr/ADR-0450-closed-window-legacy-shadow-terminal-difference-audit.md new file mode 100644 index 00000000..fcecc716 --- /dev/null +++ b/docs/adr/ADR-0450-closed-window-legacy-shadow-terminal-difference-audit.md @@ -0,0 +1,79 @@ +# ADR-0450：闭合窗口的 Legacy Shadow 终态差异审计 + +- 状态：Accepted +- 日期：2026-08-18 +- 关联 RFC：QL-RFC-0001 D-02、D-358、PR-4 +- 关联 ADR：ADR-0002、ADR-0448、ADR-0449 +- Amends：ADR-0449 的跨测量窗口待办 + +## 上下文 + +ADR-0448/0449 只证明一次启动时 active Shadow Run 的恢复结果，不能回答一个明确时间段内已经写入的 Shadow Run 是否与 Legacy 执行终态一致。 +直接用 startup scanned 计算比例会把未知尾页和未稳定执行混入分母；把全部 `RunningInstances` 当成 Shadow 应写集合也不成立，因为 QingLong 2.x +表没有可信的 execution origin，manual、system crond 与其他 Legacy 路径可能复用同一 Cron。 + +部署跨度还要求同一能力既能在低性能路由设备运行，也能在 standalone 节点处理较大窗口。审计不能增加启动查询、常驻 timer、第二数据库 authority、 +无界数组或高基数指标，更不能为了“补齐”历史数据而改写 Run/Attempt/RunningInstance。 + +## 决策 + +1. 新增显式、只读、一次性的 `audit:legacy-shadow-terminal:ql3` 运维入口。调用方必须给出至少一个已支持 Shadow origin 以及 + `[windowStartMs, windowEndMs)`；窗口 cohort 以 `legacy-owned Shadow Run.created_at_ms` 为准，不自动选择“最近一段时间”。 +2. 窗口只有在 `windowEndMs <= observedAtMs - minimumSettlingAgeMs` 时闭合，默认 settling age 为五分钟。窗口未闭合、候选页未扫完或 + Legacy evidence 达到硬上限时，不输出任何比例。 +3. Source 使用 `(created_at_ms, run_id)` 稳定 keyset，复用现有 `Runs(project_id, created_at_ms)` 索引；Run 与 latest Attempt 在同一有界查询中读取。 + 每页再执行一次有界 RunningInstance evidence 查询，不逐 candidate 发出 N+1 查询。单页候选硬上限 64、Legacy evidence 硬上限 512。 +4. Profile 默认预算保持与 startup recovery 一致：edge `8 × 1 page`，standalone `32 × 4 pages`。查询不安装 timer、watcher、线程、进程、连接池 + 常驻生命周期或后台续扫；预算耗尽返回 `incomplete`，操作者缩小窗口后重试。 +5. 关联优先级固定为 direct Attempt/Run reference，其次同 Cron 下的 opaque log artifact ID、PID，最后才允许唯一且在容差内的 started time。 + 多个候选或 evidence 截断一律归入 ambiguous，不用“最近一条”猜测。原始 log path 只在 adapter 内哈希，不进入 application/report。 +6. 每个 Shadow candidate 必须且只能归入八类之一：`matched`、`shadow_not_terminal`、`shadow_attempt_missing`、 + `shadow_attempt_ambiguous`、`legacy_evidence_missing`、`legacy_evidence_ambiguous`、`status_mismatch`、`field_mismatch`。 + aggregate 与按 origin matrix 都必须守恒。 +7. status、exit code、started time、finished time 与 log artifact 使用固定 dimension counters 记录 compared/matched/mismatched/unavailable。 + 时间默认允许两秒精度差，硬上限一分钟；`terminalAgreementPermille` 与 `fullyComparablePermille` 只在窗口闭合、候选与 evidence 都完整且分母非零时出现。 +8. 报告 schema 为 `qinglong/legacy-shadow-terminal-difference-report@v1`。报告只包含 Profile、窗口、预算、固定计数、最多七条 origin matrix 和 + `matched/empty/differences_found/window_open/incomplete` assessment；不输出 Project/Run/Attempt/Cron/PID/log/task/user/error identity。 +9. 报告明确声明 `direction=shadow_to_legacy` 和 `legacyWithoutShadow=not_measured`。当前数据模型无法可靠证明“存在 Legacy 执行但 Shadow Run 未写入”， + 因而本报告不能单独作为 Primary gate；D-360 必须组合 observer failure/capture evidence，或先引入可信 origin-scoped admission ledger。 +10. 审计不写数据库、不修复差异、不启动 Executor、不改变 Legacy 返回值。`--fail-on-difference` 只把非 `matched` assessment 映射为进程退出码 1， + 供人工 rollout/CI gate 使用。 + +## 资源与部署影响 + +- 不新增 package、生产依赖、schema、migration、表、索引、数据库 authority、HTTP/gRPC 路由、timer、watcher、worker、容器或 Kubernetes 对象。 +- edge 默认最多读取 8 个 Shadow candidate 与 64 条 Legacy evidence；standalone 默认最多 128 个 candidate 与每页 256 条 evidence。 +- CLI 数据库连接固定只读、单连接池；报告为固定字段 aggregate，不保留 candidate identity，也不建立进程内累计 registry。 +- `Runs(project_id, created_at_ms)` 负责 cohort keyset；RunningInstance 查询每页一次且结果有硬上限。若未来固定设备证明表扫描 CPU 不可接受，应通过新 migration + 增加 `(cron_id, started_at, id)` 索引并单独评审升级写放大，而不是在本 ADR 中静默修改已发布 migration checksum。 + +## 被拒绝的替代方案 + +### 双向扫描所有 RunningInstances 并计算 Shadow 捕获率 + +拒绝。2.x `RunningInstances` 没有可信 execution origin，无法知道某条记录是否属于已开启 Shadow 的入口；把它们都放进分母会制造假阴性。 + +### 在每次 HTTP startup 自动审计历史窗口 + +拒绝。它会重复 D-356 的启动查询、延迟路由设备监听，并需要隐式选择时间窗口。历史证据必须由操作者显式触发。 + +### 为每个 Shadow candidate 单独查 RunningInstances + +拒绝。即使 candidate 数有上限，N+1 仍会让低性能 SQLite 设备承担不必要的查询调度和重复表扫描。 + +### 只输出差异明细 ID 方便排障 + +拒绝。运维日志会泄露高基数任务身份；需要明细时应新增受认证、Project-scoped 的诊断产品面，而不是扩大默认 CLI 报告。 + +## 验证 + +- 真实 SQLite 覆盖完全一致、status/exit field 差异、missing/ambiguous evidence、active Shadow、窗口未闭合、edge 页预算耗尽、evidence overflow、 + 双 origin 守恒、空分母和只读 CLI 脱敏输出。 +- 真实 SQLite/CLI 聚焦测试 `14/14`、Legacy/Shadow 串行扩展 `91/91`、`build:back`、完整 backend + `1,455 pass / 0 fail / 2 conditional skip`、18-package clean build/test、14/14 static audit 与 14/14 artifact audit 全部通过。 +- edge/standalone 的 base、adopted、application、application-api、AI、application+AI、MCP 产物分别为 + `2,589,998 / 2,590,076`、`2,809,293 / 2,809,416`、`3,632,877 / 3,632,997`、 + `3,800,430 / 3,800,574`、`3,069,251 / 3,069,341`、`4,493,151 / 4,493,283`、 + `7,315,930 / 7,316,038` bytes，与 D-357 完全一致。 +- 本切片不修改 PostgreSQL、容器或 Kubernetes 部署面，因此物理 PostgreSQL HA/K3s 门不作为本切片的新证据；D-359 仍负责 edge/standalone 资源压力与 + Shadow-off 回滚演练。 diff --git a/docs/adr/README.md b/docs/adr/README.md index c90ffe64..79bb2e20 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -453,6 +453,7 @@ | [ADR-0447](./ADR-0447-boot-shadow-and-non-origin-boundaries.md) | Boot Shadow 准入与 once/gRPC 非 Origin 边界 | Accepted | | [ADR-0448](./ADR-0448-bounded-legacy-shadow-startup-reconciliation.md) | 有界 Legacy Shadow 启动恢复 | Accepted | | [ADR-0449](./ADR-0449-versioned-legacy-shadow-startup-difference-report-and-metrics.md) | 版本化 Legacy Shadow 启动差异报告与指标批次 | Accepted | +| [ADR-0450](./ADR-0450-closed-window-legacy-shadow-terminal-difference-audit.md) | 闭合窗口的 Legacy Shadow 终态差异审计 | Accepted | ## 规则 diff --git a/package.json b/package.json index e497a0de..e6d4e238 100644 --- a/package.json +++ b/package.json @@ -73,6 +73,7 @@ "audit:schema:ql3": "pnpm --filter @qinglong/local-owner-cli run readiness", "audit:legacy-schema:ql3": "node scripts/ql3-schema-audit.cjs", "audit:receipts:ql3": "node scripts/ql3-receipt-audit.cjs", + "audit:legacy-shadow-terminal:ql3": "node scripts/ql3-legacy-shadow-terminal-audit.cjs", "audit:edge-imports:ql3": "node scripts/ql3-edge-import-audit.cjs", "audit:cluster-dependencies:ql3": "node scripts/ql3-cluster-dependency-audit.cjs", "audit:package-boundaries:ql3": "node scripts/ql3-package-boundary-audit.cjs", diff --git a/scripts/ql3-legacy-shadow-terminal-audit.cjs b/scripts/ql3-legacy-shadow-terminal-audit.cjs new file mode 100644 index 00000000..3fa9ccf8 --- /dev/null +++ b/scripts/ql3-legacy-shadow-terminal-audit.cjs @@ -0,0 +1,243 @@ +#!/usr/bin/env node + +require('ts-node/register/transpile-only'); + +const fs = require('node:fs'); +const path = require('node:path'); +const sqlite3 = require('sqlite3'); +const { QueryTypes, Sequelize } = require('sequelize'); +const { + LegacySequelizeShadowTerminalDifferenceSource, +} = require('../back/runtime/adapters/legacy-sequelize/legacyShadowTerminalDifferenceSource'); +const { + LegacyShadowTerminalDifferenceAuditor, +} = require('../back/runtime/application/legacyShadowTerminalDifferenceAuditor'); +const { + createLegacyLogArtifactId, +} = require('../back/runtime/compatibility/legacyTaskRevision'); + +const SUPPORTED_ORIGINS = new Set([ + 'boot', + 'manual', + 'scheduled_node', + 'scheduled_system', + 'script', + 'subscription', + 'system', +]); + +function numericArgument(name, value, minimum, maximum) { + if (!/^\d+$/.test(value)) { + throw new Error(`${name} must be an integer`); + } + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed < minimum || parsed > maximum) { + throw new Error(`${name} must be between ${minimum} and ${maximum}`); + } + return parsed; +} + +function parseArguments(argv) { + const options = { + database: path.resolve(process.cwd(), 'data/db/database.sqlite'), + profile: 'edge', + projectId: 'default', + origins: [], + minimumSettlingAgeMs: 5 * 60_000, + correlationToleranceMs: 2_000, + json: false, + failOnDifference: false, + }; + for (const argument of argv) { + if (argument === '--') continue; + if (argument === '--json') { + options.json = true; + } else if (argument === '--fail-on-difference') { + options.failOnDifference = true; + } else if (argument.startsWith('--database=')) { + const value = argument.slice('--database='.length); + if (!value) throw new Error('--database must not be empty'); + options.database = path.resolve(value); + } else if (argument.startsWith('--profile=')) { + const value = argument.slice('--profile='.length); + if (value !== 'edge' && value !== 'standalone') { + throw new Error('--profile must be edge or standalone'); + } + options.profile = value; + } else if (argument.startsWith('--project=')) { + const value = argument.slice('--project='.length); + if (!value || value.length > 128) { + throw new Error('--project length must be between 1 and 128'); + } + options.projectId = value; + } else if (argument.startsWith('--origin=')) { + const values = argument + .slice('--origin='.length) + .split(',') + .map((value) => value.trim()) + .filter(Boolean); + if (values.length === 0) throw new Error('--origin must not be empty'); + for (const value of values) { + if (!SUPPORTED_ORIGINS.has(value)) { + throw new Error(`unsupported Shadow origin: ${value}`); + } + if (!options.origins.includes(value)) options.origins.push(value); + } + } else if (argument.startsWith('--window-start-ms=')) { + options.windowStartMs = numericArgument( + '--window-start-ms', + argument.slice('--window-start-ms='.length), + 0, + Number.MAX_SAFE_INTEGER, + ); + } else if (argument.startsWith('--window-end-ms=')) { + options.windowEndMs = numericArgument( + '--window-end-ms', + argument.slice('--window-end-ms='.length), + 0, + Number.MAX_SAFE_INTEGER, + ); + } else if (argument.startsWith('--observed-at-ms=')) { + options.observedAtMs = numericArgument( + '--observed-at-ms', + argument.slice('--observed-at-ms='.length), + 0, + Number.MAX_SAFE_INTEGER, + ); + } else if (argument.startsWith('--minimum-settling-age-ms=')) { + options.minimumSettlingAgeMs = numericArgument( + '--minimum-settling-age-ms', + argument.slice('--minimum-settling-age-ms='.length), + 0, + Number.MAX_SAFE_INTEGER, + ); + } else if (argument.startsWith('--correlation-tolerance-ms=')) { + options.correlationToleranceMs = numericArgument( + '--correlation-tolerance-ms', + argument.slice('--correlation-tolerance-ms='.length), + 0, + 60_000, + ); + } else { + throw new Error(`Unsupported argument: ${argument}`); + } + } + if (options.origins.length === 0) { + throw new Error('at least one --origin is required'); + } + if (options.windowStartMs === undefined) { + throw new Error('--window-start-ms is required'); + } + if (options.windowEndMs === undefined) { + throw new Error('--window-end-ms is required'); + } + if (options.windowStartMs >= options.windowEndMs) { + throw new Error('measurement window must be non-empty'); + } + return options; +} + +async function assertRequiredSchema(database) { + const rows = await database.query( + `SELECT name + FROM sqlite_master + WHERE type = 'table' + AND name IN ('Runs', 'RunAttempts', 'RunningInstances')`, + { type: QueryTypes.SELECT }, + ); + const names = new Set(rows.map((row) => row.name)); + for (const required of ['Runs', 'RunAttempts', 'RunningInstances']) { + if (!names.has(required)) { + throw new Error( + `Database is not ready for Legacy Shadow terminal audit: missing ${required}`, + ); + } + } + const columns = await database + .getQueryInterface() + .describeTable('RunningInstances'); + for (const required of ['run_id', 'attempt_id']) { + if (!columns[required]) { + throw new Error( + `Database is not ready for Legacy Shadow terminal audit: missing RunningInstances.${required}`, + ); + } + } +} + +function renderText(report) { + const lines = [ + 'QingLong 3.0 Legacy Shadow terminal audit', + `profile: ${report.profile}`, + `window: [${report.window.startInclusiveMs}, ${report.window.endExclusiveMs})`, + `window closed: ${report.window.closed}`, + `coverage: ${report.coverage.direction} (${report.coverage.legacyWithoutShadow})`, + `assessment: ${report.assessment}`, + `pages/scanned: ${report.pages}/${report.scanned}`, + `remaining: ${report.remaining}`, + `evidence complete: ${report.evidenceComplete}`, + ]; + for (const [category, count] of Object.entries(report.counts)) { + if (count > 0) lines.push(`${category}: ${count}`); + } + if (report.terminalAgreementPermille !== undefined) { + lines.push( + `terminal agreement: ${report.terminalAgreementPermille}/1000`, + `fully comparable: ${report.fullyComparablePermille}/1000`, + ); + } + return `${lines.join('\n')}\n`; +} + +async function main() { + const options = parseArguments(process.argv.slice(2)); + const nodeMajor = Number(process.versions.node.split('.')[0]); + if (nodeMajor < 24) { + throw new Error( + 'Legacy Shadow terminal audit requires Node.js 24 or newer', + ); + } + if (!fs.existsSync(options.database)) { + throw new Error(`Database does not exist: ${options.database}`); + } + const database = new Sequelize({ + dialect: 'sqlite', + storage: options.database, + logging: false, + dialectOptions: { mode: sqlite3.OPEN_READONLY }, + pool: { max: 1, min: 0, idle: 1_000, acquire: 5_000 }, + }); + try { + await assertRequiredSchema(database); + const auditor = new LegacyShadowTerminalDifferenceAuditor( + new LegacySequelizeShadowTerminalDifferenceSource( + database, + createLegacyLogArtifactId, + ), + ); + const report = await auditor.run(options); + process.stdout.write( + options.json ? `${JSON.stringify(report)}\n` : renderText(report), + ); + if (options.failOnDifference && report.assessment !== 'matched') { + process.exitCode = 1; + } + } finally { + await database.close(); + } +} + +if (require.main === module) { + main().catch((error) => { + process.stderr.write( + `${error instanceof Error ? error.message : String(error)}\n`, + ); + process.exitCode = 1; + }); +} + +module.exports = { + assertRequiredSchema, + parseArguments, + renderText, +}; diff --git a/test/back/legacyShadowTerminalDifference.test.cjs b/test/back/legacyShadowTerminalDifference.test.cjs new file mode 100644 index 00000000..c9fbb9e3 --- /dev/null +++ b/test/back/legacyShadowTerminalDifference.test.cjs @@ -0,0 +1,500 @@ +require('ts-node/register/transpile-only'); + +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const { spawnSync } = require('node:child_process'); +const { afterEach, test } = require('node:test'); +const { DataTypes, Sequelize } = require('sequelize'); +const { + defineSchemaMigrationModel, +} = require('../../back/data/schemaMigration'); +const { runSchemaMigration } = require('../../back/migrations/0002-run-schema'); +const { + runningInstanceRunReferenceMigration, +} = require('../../back/migrations/0003-running-instance-run-reference'); +const { + runCancellationRequestMigration, +} = require('../../back/migrations/0004-run-cancellation-request'); +const { + runAttemptDeadlineMigration, +} = require('../../back/migrations/0006-run-attempt-deadline'); +const { runMigrations } = require('../../back/migrations/runner'); +const { + LegacySequelizeRunRepository, +} = require('../../back/runtime/adapters/legacy-sequelize/runRepository'); +const { + LegacySequelizeShadowTerminalDifferenceSource, +} = require('../../back/runtime/adapters/legacy-sequelize/legacyShadowTerminalDifferenceSource'); +const { + LegacyShadowRunWriter, +} = require('../../back/runtime/application/legacyShadowRunWriter'); +const { + LegacyShadowTerminalDifferenceAuditor, +} = require('../../back/runtime/application/legacyShadowTerminalDifferenceAuditor'); +const { + createLegacyLogArtifactId, +} = require('../../back/runtime/compatibility/legacyTaskRevision'); +const { + parseArguments, +} = require('../../scripts/ql3-legacy-shadow-terminal-audit.cjs'); + +const REPOSITORY_ROOT = path.resolve(__dirname, '../..'); +const CLI_PATH = path.join( + REPOSITORY_ROOT, + 'scripts', + 'ql3-legacy-shadow-terminal-audit.cjs', +); +const BASE_TIME = 1_750_100_000_000; +const databases = []; +const temporaryDirectories = []; +let idSequence = 4_000; +let timeSequence = BASE_TIME; + +function nextId() { + idSequence += 1; + return `019f7300-0000-7000-8000-${String(idSequence).padStart(12, '0')}`; +} + +function nextTime() { + timeSequence += 10_000; + return timeSequence; +} + +async function createDatabase(storage = ':memory:') { + const database = new Sequelize({ + dialect: 'sqlite', + storage, + logging: false, + }); + await database.getQueryInterface().createTable('RunningInstances', { + id: { + type: DataTypes.INTEGER, + primaryKey: true, + autoIncrement: true, + }, + cron_id: { type: DataTypes.INTEGER, allowNull: false }, + pid: { type: DataTypes.INTEGER, allowNull: true }, + log_path: { type: DataTypes.STRING, allowNull: true }, + started_at: { type: DataTypes.INTEGER, allowNull: false }, + finished_at: { type: DataTypes.INTEGER, allowNull: true }, + status: { type: DataTypes.INTEGER, allowNull: false }, + exit_code: { type: DataTypes.INTEGER, allowNull: true }, + }); + await runMigrations({ + database, + migrationModel: defineSchemaMigrationModel(database), + migrations: [ + runSchemaMigration, + runningInstanceRunReferenceMigration, + runCancellationRequestMigration, + runAttemptDeadlineMigration, + ], + logger: { info() {} }, + }); + databases.push(database); + return database; +} + +async function createStack(storage) { + const database = await createDatabase(storage); + const repository = new LegacySequelizeRunRepository(database); + const writer = new LegacyShadowRunWriter(repository, nextId); + const source = new LegacySequelizeShadowTerminalDifferenceSource( + database, + createLegacyLogArtifactId, + ); + const auditor = new LegacyShadowTerminalDifferenceAuditor(source); + return { database, repository, writer, source, auditor }; +} + +async function createShadow(writer, overrides = {}) { + const acceptedAtMs = overrides.acceptedAtMs ?? nextTime(); + const legacyCronId = overrides.legacyCronId ?? 31; + const pid = overrides.pid ?? 5_101; + const logPath = overrides.logPath ?? `cron/${acceptedAtMs}.log`; + const origin = overrides.origin ?? 'manual'; + const reference = await writer.accept({ + origin, + projectId: 'default', + taskId: `legacy-cron:${legacyCronId}`, + taskRevision: `sha256:terminal-audit-${legacyCronId}`, + legacyCronId, + triggerType: origin, + acceptedAtMs, + }); + if (overrides.acceptedOnly) { + return { acceptedAtMs, legacyCronId, pid, logPath, origin, reference }; + } + await writer.spawned(reference, { + atMs: acceptedAtMs + 100, + pid, + logArtifactId: createLegacyLogArtifactId(logPath), + }); + await writer.running(reference, acceptedAtMs + 200); + if (!overrides.runningOnly) { + await writer.exited(reference, { + atMs: acceptedAtMs + 500, + exitCode: overrides.shadowExitCode ?? 0, + }); + } + return { acceptedAtMs, legacyCronId, pid, logPath, origin, reference }; +} + +async function insertInstance(database, shadow, overrides = {}) { + await database.getQueryInterface().bulkInsert('RunningInstances', [ + { + cron_id: shadow.legacyCronId, + run_id: + overrides.direct === false + ? null + : overrides.runId ?? shadow.reference.runId, + attempt_id: + overrides.direct === false + ? null + : overrides.attemptId ?? shadow.reference.attemptId, + pid: overrides.pid ?? shadow.pid, + log_path: overrides.logPath ?? shadow.logPath, + started_at: Math.floor((shadow.acceptedAtMs + 100) / 1_000), + finished_at: Math.floor( + (shadow.acceptedAtMs + (overrides.finishedOffsetMs ?? 500)) / 1_000, + ), + status: overrides.status ?? 1, + exit_code: overrides.exitCode ?? 0, + }, + ]); +} + +function auditOptions(shadows, overrides = {}) { + const times = shadows.map((shadow) => shadow.acceptedAtMs); + const start = Math.min(...times) - 1; + const end = Math.max(...times) + 1; + return { + profile: overrides.profile ?? 'edge', + origins: overrides.origins ?? ['manual'], + windowStartMs: start, + windowEndMs: end, + observedAtMs: overrides.observedAtMs ?? end + 5 * 60_000, + minimumSettlingAgeMs: 5 * 60_000, + }; +} + +afterEach(async () => { + await Promise.all(databases.splice(0).map((database) => database.close())); + for (const directory of temporaryDirectories.splice(0)) { + fs.rmSync(directory, { recursive: true, force: true }); + } +}); + +test('reports a closed, fully comparable Shadow-to-Legacy terminal match', async () => { + const { database, writer, auditor } = await createStack(); + const shadow = await createShadow(writer); + await insertInstance(database, shadow); + + const report = await auditor.run(auditOptions([shadow])); + + assert.equal(report.assessment, 'matched'); + assert.equal(report.counts.matched, 1); + assert.equal(report.terminalAgreementPermille, 1_000); + assert.equal(report.fullyComparablePermille, 1_000); + assert.equal(report.coverage.direction, 'shadow_to_legacy'); + assert.equal(report.coverage.legacyWithoutShadow, 'not_measured'); + assert.equal(JSON.stringify(report).includes(shadow.reference.runId), false); + assert.equal(JSON.stringify(report).includes(shadow.logPath), false); +}); + +test('separates terminal status and field differences', async () => { + const { database, writer, auditor } = await createStack(); + const statusShadow = await createShadow(writer, { legacyCronId: 41 }); + const fieldShadow = await createShadow(writer, { legacyCronId: 42 }); + await insertInstance(database, statusShadow, { status: 3, exitCode: 1 }); + await insertInstance(database, fieldShadow, { exitCode: 9 }); + + const report = await auditor.run(auditOptions([statusShadow, fieldShadow])); + + assert.equal(report.assessment, 'differences_found'); + assert.equal(report.counts.status_mismatch, 1); + assert.equal(report.counts.field_mismatch, 1); + assert.equal(report.dimensions.status.mismatched, 1); + assert.equal(report.dimensions.exitCode.mismatched, 1); + assert.equal(report.terminalAgreementPermille, 0); +}); + +test('does not guess when Legacy evidence is missing or ambiguous', async () => { + const { database, writer, auditor } = await createStack(); + const missing = await createShadow(writer, { legacyCronId: 51 }); + const ambiguous = await createShadow(writer, { legacyCronId: 52 }); + await insertInstance(database, ambiguous, { direct: false }); + await insertInstance(database, ambiguous, { direct: false }); + + const report = await auditor.run(auditOptions([missing, ambiguous])); + + assert.equal(report.counts.legacy_evidence_missing, 1); + assert.equal(report.counts.legacy_evidence_ambiguous, 1); + assert.equal(report.assessment, 'differences_found'); +}); + +test('rejects conflicting direct Run and Attempt references as ambiguous', async () => { + const { database, writer, auditor } = await createStack(); + const shadow = await createShadow(writer, { legacyCronId: 53 }); + await insertInstance(database, shadow, { + attemptId: '019f7300-0000-7000-8000-999999999999', + }); + + const report = await auditor.run(auditOptions([shadow])); + + assert.equal(report.counts.legacy_evidence_ambiguous, 1); + assert.equal(report.assessment, 'differences_found'); +}); + +test('classifies a settled cohort member with an active Shadow Run', async () => { + const { writer, auditor } = await createStack(); + const shadow = await createShadow(writer, { runningOnly: true }); + + const report = await auditor.run(auditOptions([shadow])); + + assert.equal(report.counts.shadow_not_terminal, 1); + assert.equal(report.assessment, 'differences_found'); +}); + +test('withholds ratios while the measurement window is still open', async () => { + const { database, writer, auditor } = await createStack(); + const shadow = await createShadow(writer); + await insertInstance(database, shadow); + const options = auditOptions([shadow], { + observedAtMs: shadow.acceptedAtMs + 1_000, + }); + + const report = await auditor.run(options); + + assert.equal(report.assessment, 'window_open'); + assert.equal(report.window.closed, false); + assert.equal(report.terminalAgreementPermille, undefined); + assert.equal(report.fullyComparablePermille, undefined); +}); + +test('withholds ratios when the edge candidate budget is exhausted', async () => { + const { database, writer, auditor } = await createStack(); + const shadows = []; + for (let index = 0; index < 9; index += 1) { + const shadow = await createShadow(writer, { + legacyCronId: 60 + index, + pid: 6_000 + index, + }); + shadows.push(shadow); + await insertInstance(database, shadow); + } + + const report = await auditor.run(auditOptions(shadows)); + + assert.equal(report.scanned, 8); + assert.equal(report.remaining, true); + assert.equal(report.stopReason, 'page_limit'); + assert.equal(report.assessment, 'incomplete'); + assert.equal(report.terminalAgreementPermille, undefined); +}); + +test('marks evidence overflow incomplete instead of accepting a partial match set', async () => { + const { database, writer, auditor } = await createStack(); + const shadow = await createShadow(writer, { legacyCronId: 71 }); + for (let index = 0; index < 9; index += 1) { + await insertInstance(database, shadow, { + direct: false, + pid: 7_100 + index, + logPath: `cron/overflow-${index}.log`, + }); + } + + const report = await auditor.run(auditOptions([shadow])); + + assert.equal(report.evidenceComplete, false); + assert.equal(report.evidenceOverflowPages, 1); + assert.equal(report.counts.legacy_evidence_ambiguous, 1); + assert.equal(report.assessment, 'incomplete'); +}); + +test('keeps an exact, conservation-safe matrix for configured origins', async () => { + const { database, writer, auditor } = await createStack(); + const manual = await createShadow(writer, { legacyCronId: 81 }); + const system = await createShadow(writer, { + legacyCronId: 82, + origin: 'scheduled_system', + }); + await insertInstance(database, manual); + await insertInstance(database, system); + + const report = await auditor.run( + auditOptions([manual, system], { + origins: ['manual', 'scheduled_system'], + }), + ); + + assert.deepEqual( + report.byOrigin.map(({ origin, scanned, matched }) => ({ + origin, + scanned, + matched, + })), + [ + { origin: 'manual', scanned: 1, matched: 1 }, + { origin: 'scheduled_system', scanned: 1, matched: 1 }, + ], + ); + assert.equal( + Object.values(report.counts).reduce((sum, count) => sum + count, 0), + report.scanned, + ); +}); + +test('reports an empty closed cohort without claiming agreement', async () => { + const { auditor } = await createStack(); + const report = await auditor.run({ + profile: 'edge', + origins: ['manual'], + windowStartMs: BASE_TIME - 10_000, + windowEndMs: BASE_TIME - 5_000, + observedAtMs: BASE_TIME + 300_000, + }); + + assert.equal(report.assessment, 'empty'); + assert.equal(report.scanned, 0); + assert.equal(report.terminalAgreementPermille, undefined); +}); + +test('rejects malformed windows, excessive origins and tolerance', async () => { + const { auditor } = await createStack(); + await assert.rejects( + auditor.run({ + profile: 'edge', + origins: ['manual'], + windowStartMs: 10, + windowEndMs: 10, + }), + /non-empty/, + ); + await assert.rejects( + auditor.run({ + profile: 'edge', + origins: [ + 'manual', + 'boot', + 'scheduled_node', + 'scheduled_system', + 'script', + 'subscription', + 'system', + 'grpc', + ], + windowStartMs: 1, + windowEndMs: 2, + }), + /origin count/, + ); + await assert.rejects( + auditor.run({ + profile: 'edge', + origins: ['manual'], + windowStartMs: 1, + windowEndMs: 2, + correlationToleranceMs: 60_001, + }), + /hard limit/, + ); +}); + +test('rejects an adapter page that exceeds the evidence hard limit', async () => { + const auditor = new LegacyShadowTerminalDifferenceAuditor({ + async listCandidates() { + return { + candidates: [], + evidence: [ + { + instanceId: 1, + legacyCronId: 1, + startedAtMs: 1, + finishedAtMs: 2, + outcome: 'succeeded', + }, + ], + evidenceTruncated: false, + truncated: false, + }; + }, + }); + + await assert.rejects( + auditor.run({ + profile: 'edge', + origins: ['manual'], + windowStartMs: 1, + windowEndMs: 2, + observedAtMs: 300_002, + }), + /exceeded evidence limit/, + ); +}); + +test('CLI arguments require an explicit cohort and reject unsupported origins', () => { + assert.throws( + () => parseArguments(['--origin=manual', '--window-start-ms=1']), + /window-end-ms is required/, + ); + assert.throws( + () => + parseArguments([ + '--origin=grpc', + '--window-start-ms=1', + '--window-end-ms=2', + ]), + /unsupported Shadow origin/, + ); + const options = parseArguments([ + '--profile=standalone', + '--origin=manual,scheduled_system', + '--window-start-ms=1', + '--window-end-ms=2', + '--json', + ]); + assert.equal(options.profile, 'standalone'); + assert.deepEqual(options.origins, ['manual', 'scheduled_system']); + assert.equal(options.json, true); +}); + +test('read-only CLI emits the versioned redacted report', async () => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'ql3-shadow-audit-')); + temporaryDirectories.push(directory); + const storage = path.join(directory, 'database.sqlite'); + const { database, writer } = await createStack(storage); + const shadow = await createShadow(writer, { legacyCronId: 91 }); + await insertInstance(database, shadow); + await database.close(); + databases.splice(databases.indexOf(database), 1); + const options = auditOptions([shadow]); + + const result = spawnSync( + process.execPath, + [ + CLI_PATH, + `--database=${storage}`, + '--profile=edge', + '--origin=manual', + `--window-start-ms=${options.windowStartMs}`, + `--window-end-ms=${options.windowEndMs}`, + `--observed-at-ms=${options.observedAtMs}`, + '--json', + '--fail-on-difference', + ], + { cwd: REPOSITORY_ROOT, encoding: 'utf8' }, + ); + + assert.equal(result.status, 0, result.stderr); + const report = JSON.parse(result.stdout); + assert.equal( + report.schema, + 'qinglong/legacy-shadow-terminal-difference-report@v1', + ); + assert.equal(report.assessment, 'matched'); + assert.equal(result.stdout.includes(shadow.reference.runId), false); + assert.equal(result.stdout.includes(shadow.logPath), false); +});