refactor(gitops): split source acceptance from Direct target application (#1892)

* refactor(gitops): split source acceptance from Direct target application applied() bound source acceptance and Direct target application in one mutation, so a future dispatch path could not accept a candidate and defer binding a target until after promotion. Extract the mode-neutral and Direct-only mutations into shared helpers, and expose them as sourceAccepted and targetApplied. applied() now composes the same helpers in the same transaction and emits the same single history row, so its observable contract is unchanged. targetApplied refuses to bind a target to a generation the application has not accepted, so a target can never be bound to source content nothing authorized. * fix(gitops): move apply's cache invalidation and post-deploy scan into the service invalidateNodeCaches and triggerPostDeployScan were called by the manual apply route only, so a webhook-driven apply (and any future poll, retry, or resume trigger) never invalidated caches or ran a post-deploy scan. Move both into GitSourceService.apply() itself: cache invalidation fires once whenever promotion commits, apply-only and deploy-failed outcomes included, since the authoritative Compose files have already changed by then regardless of whether a deploy followed; the post-deploy scan still fires only after a successful deploy. Remove the now-redundant calls from the apply route so each still runs exactly once. * feat(gitops): add tri-state candidate policy evaluation The deploy-time policy gate deliberately fails open (ok: true, trivyMissing: true) when the scanner is unavailable, so an operator is never blocked from deploying. That fail-open behavior must not extend to automatic GitOps source acceptance: an unresolvable scanner state accepting a candidate nothing actually proved safe would defeat the policy gate entirely. Add evaluateCandidatePolicy, a thin tri-state (allowed | blocked | unavailable) wrapper around the existing enforcePolicyForImageRefs evaluator. It takes the candidate's own image refs directly rather than reading compose from disk, so a future caller evaluating a staged candidate (before it is promoted to the live stack) can reuse the same policy logic the deploy gate already uses. * feat(gitops): propagate the raw transport failure reason onto GitSourceError The native transport catch block classified a raw TransportFailure into a sanitized (code, message) pair for GitSourceError, discarding the structured reason (e.g. exit, timeout, target-unresolved) once it had been logged. A future GitOps retry/backoff classifier needs more than the public error code to tell a transient network condition from a permanent configuration one, since several distinct reasons collapse onto the same code (both a DNS failure and a connection reset map to NETWORK_TIMEOUT, and several permanent conditions map to GIT_ERROR). Add transportReason to GitSourceError.extras, carrying the original reason through both throw sites. * fix(gitops): give source suspension its own reason field sourceSuspended and sourceUnsuspended wrote and cleared pause_reason, the same column rolloutPaused and rolloutUnpaused use on the application row. Suspending a source and later pausing its rollout (or the reverse) would silently overwrite whichever reason was written first, since both events share the row but were sharing one field for two unrelated concerns. Add a distinct source_suspended_reason column, move sourceSuspended and sourceUnsuspended onto it, and surface it on the source_suspended projection facet so a suspended source's reason is visible independently of any rollout pause reason on the same application. * feat(notifications): add a GitOps operation reference and dedupe key notification_history had no way to link a notification back to the GitOps history/operation it reports on, and no way to detect a duplicate: it has only an autoincrement id and an unstructured message. A future GitOps fanout repair (re-running after a crash between a settled attempt commit and its notification) needs to be idempotent, which the table could not support. Add gitops_operation_id and dedupe_key columns, with a partial unique index on dedupe_key so a second insert with the same key is a no-op returning the existing row rather than a duplicate notification. Both columns are optional and every existing caller is unaffected: omitting dedupe_key keeps today's behavior exactly. * feat(gitops): classify suspend/resume/retry as stack:edit for remote routing An unclassified named-stack path fails closed with 403 on a remote node, so a future suspend/resume/retry endpoint would be unreachable there until its classification landed. Add the three suffix rules now, matching the existing git-source/pull and git-source/apply entries, so remote and scoped-permission routing already works correctly once those endpoints are added. * fix(gitops): address pre-commit review findings on the acceptance split Code review found five substantive issues across the prior six commits: - targetApplied took applicationId as a redundant positional parameter alongside AppliedArgs.applicationId, which every caller had to pass twice; the copy inside args was silently ignored. Drop the positional parameter. - targetApplied had no target_mode guard, unlike applied()'s existing Direct-only check, so a Blueprint target could in principle be bound through it. Add the guard. - The frontend's hand-written GitOps type mirror was not updated for the new suspendedReason field, which the file's own header warns is exactly the drift it does not detect on its own. - The dedupe unique index's creation failure was silently swallowed, but unlike a pure performance index, this one is the ON CONFLICT target every notification write depends on; a missing index would break every notification in the product with no diagnostic. Log it. - evaluateCandidatePolicy inherited the deploy-shaped default audit path from the evaluator it wraps, so a bypassed candidate evaluation would write an audit row claiming a deploy that never happened. Default the audit attribution to a candidate-evaluation path before delegating. Also: removed a test fixture in git-source-routes.test.ts duplicating the shared one in helpers/gitopsFixtures.ts, hoisted the repeated policy field out of CandidatePolicyEvaluation's union, and removed an unnecessary any cast. * fix(gitops): close three safety gaps found in the pre-merge audit An independent audit of PR #1892 found three release-blocking defects in the source acceptance split, each reproducible against the existing test suite: - sourceAccepted() accepted a candidate while its source was suspended. applied() (preserved byte-identical, predating suspension) shares this gap, but the plan's own suspension guarantee is specifically for the new entry point, so the check is added to sourceAccepted() directly rather than the shared guard applied() also uses. - evaluateCandidatePolicy() misclassified three safety cases: an image reference that failed validation was silently skipped and read as allowed; a scanner execution failure was treated as a genuine policy violation (blocked) rather than an inability to evaluate (unavailable); and an explicitly authorized bypass still returned unavailable when the scanner was absent, since that early-return path in the shared evaluator ignores the caller's bypass flag. Fixed by requesting fail-closed handling of invalid refs from the existing evaluator, distinguishing a genuine scanned violation from an evaluation failure by whether the violation carries an `error` field, and honoring bypass before returning unavailable. - targetApplied() validated only that its generation was still the application's accepted one, not that the target's own candidate still matched it or that the supplied acceptance reference was the one actually recorded. A delayed dispatch of a since-superseded (but still accepted) generation could erase a newer candidate already staged on the target, and a caller could bind a target to a nonexistent acceptance reference. Both are now validated before mutation. Also removed two explicit `any` callback parameters the audit flagged in the new dedupe test, typing the array instead so inference covers them.
2026-09-07 07:56:05 +00:00 · 2026-09-03 01:27:21 +00:00
parent f8cfcb547a
commit da905ab07c
32 changed files with 825 additions and 107 deletions
@@ -573,6 +573,14 @@ export interface NotificationHistory {
    container_name?: string;
    actor_username?: string | null;
    suppression_match?: string | null;
+    /** The GitOps operation this notification reports on, if any. */
+    gitops_operation_id?: string | null;
+    /**
+     * Unique across all notifications when set. Lets fanout repair re-run
+     * safely: inserting the same key again is a no-op that returns the
+     * existing row instead of creating a duplicate.
+     */
+    dedupe_key?: string | null;
 }

 export interface FleetSnapshot {
@@ -1164,6 +1172,7 @@ export class DatabaseService {
        this.migratePolicyEvaluationColumn();
        this.migrateNotificationCategory();
        this.migrateNotificationActor();
+        this.migrateNotificationGitOpsDedupe();
        this.migrateMeshTables();
        this.migrateNodeLabels();
        this.migrateBlueprints();
@@ -1963,6 +1972,10 @@ export class DatabaseService {
        // from the CREATE TABLE; older DBs need the additive column here.
        maybeAddCol('gitops_generations', 'resolved_ref_kind', 'TEXT NULL');
        maybeAddCol('gitops_applications', 'fetched_resolved_ref_kind', 'TEXT NULL');
+        // Source suspension reason, distinct from the rollout pause_reason
+        // existing installs already have. New installs get it from the
+        // CREATE TABLE; older DBs need the additive column here.
+        maybeAddCol('gitops_applications', 'source_suspended_reason', 'TEXT NULL');

        // Distributed API model columns
        maybeAddCol('nodes', 'api_url', "TEXT DEFAULT ''");
@@ -2799,6 +2812,29 @@ stmt.run('gitops_schema_version', '1');
        }
    }

+    /**
+     * A GitOps history/operation reference and a dedupe key, so notification
+     * fanout for a settled GitOps attempt can be repaired from durable state
+     * (retried at startup after a crash between commit and fanout) without
+     * ever inserting a duplicate notification for the same attempt.
+     */
+    private migrateNotificationGitOpsDedupe(): void {
+        this.tryAddColumn('notification_history', 'gitops_operation_id', 'TEXT');
+        this.tryAddColumn('notification_history', 'dedupe_key', 'TEXT');
+        try {
+            this.db.prepare(
+                'CREATE UNIQUE INDEX IF NOT EXISTS idx_notif_history_dedupe_key ON notification_history(dedupe_key) WHERE dedupe_key IS NOT NULL'
+            ).run();
+        } catch (err) {
+            // Unlike a pure performance index, this one is the ON CONFLICT
+            // target every addNotificationHistory() insert names. If it is
+            // missing, every notification write in the product fails, not
+            // just GitOps ones, so a silent catch here would turn into an
+            // unexplained total outage instead of a diagnosable startup log.
+            console.error('[DatabaseService] Failed to create notification dedupe index:', err);
+        }
+    }
+
    private migrateMeshTables(): void {
        try {
            if (isPilotMode()) {
@@ -4738,8 +4774,13 @@ stmt.run('gitops_schema_version', '1');
    }

    public addNotificationHistory(nodeId: number, notification: Omit<NotificationHistory, 'id' | 'is_read'>): NotificationHistory {
+        const dedupeKey = notification.dedupe_key ?? null;
        const stmt = this.db.prepare(
-            'INSERT INTO notification_history (node_id, level, message, timestamp, is_read, stack_name, container_name, category, actor_username) VALUES (?, ?, ?, ?, 0, ?, ?, ?, ?)'
+            `INSERT INTO notification_history (
+                node_id, level, message, timestamp, is_read, stack_name, container_name,
+                category, actor_username, gitops_operation_id, dedupe_key
+            ) VALUES (?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(dedupe_key) WHERE dedupe_key IS NOT NULL DO NOTHING`
        );
        const result = stmt.run(
            nodeId,
@@ -4750,8 +4791,20 @@ stmt.run('gitops_schema_version', '1');
            notification.container_name ?? null,
            notification.category ?? null,
            notification.actor_username ?? null,
+            notification.gitops_operation_id ?? null,
+            dedupeKey,
        );

+        // A repair replaying a settled GitOps attempt must not create a
+        // duplicate notification; the conflict is not an error, it is proof
+        // this exact attempt was already reported.
+        if (result.changes === 0 && dedupeKey !== null) {
+            const existing = this.db.prepare(
+                'SELECT * FROM notification_history WHERE dedupe_key = ?',
+            ).get(dedupeKey);
+            return this.mapNotificationRow(existing);
+        }
+
        return {
            id: result.lastInsertRowid as number,
            level: notification.level,
@@ -4762,6 +4815,8 @@ stmt.run('gitops_schema_version', '1');
            stack_name: notification.stack_name,
            container_name: notification.container_name,
            actor_username: notification.actor_username,
+            gitops_operation_id: notification.gitops_operation_id,
+            dedupe_key: dedupeKey,
        };
    }

@@ -11,7 +11,8 @@ import { ComposeService } from './ComposeService';
 import { StackOpLockService } from './StackOpLockService';
 import { HealthGateService } from './HealthGateService';
 import { NodeRegistry } from './NodeRegistry';
-import { assertPolicyGateAllows, buildSystemPolicyGateOptions } from '../helpers/policyGate';
+import { assertPolicyGateAllows, buildSystemPolicyGateOptions, triggerPostDeployScan } from '../helpers/policyGate';
+import { invalidateNodeCaches } from '../helpers/cacheInvalidation';
 import { isDebugEnabled } from '../utils/debug';
 import { sanitizeForLog } from '../utils/safeLog';
 import { isPathWithinBase, isValidRelativeStackPath } from '../utils/validation';
@@ -27,7 +28,7 @@ import type { ComposeInputEntry, GitProjectManifest, GitSourceManifestState, Inv
 import type { GitChangePlan, PublicGitChangePlan, GitChangePlanCounts, PublicGitChangePlanOperation } from '../types/gitChangePlan';
 import { GIT_CHANGE_PLAN_SCHEMA_VERSION } from '../types/gitChangePlan';
 import type { NotificationCategory } from './NotificationService';
-import { classifyGitFailure, isTransportFailure } from './git/errors';
+import { classifyGitFailure, isTransportFailure, type TransportFailureReason } from './git/errors';
 import type { RefKind, SshDeployKeyAuth } from './git/types';
 import { nativeGitTransport, verifyFastForward } from './git/nativeGitTransport';
 import { fingerprintFromKnownHostsLine } from './git/sshTrust';
@@ -82,7 +83,18 @@ export class GitSourceError extends Error {
    constructor(
        public code: GitSourceErrorCode,
        message: string,
-        public extras?: { plan?: PublicGitChangePlan; planFingerprint?: string },
+        public extras?: {
+            plan?: PublicGitChangePlan;
+            planFingerprint?: string;
+            /**
+             * The raw structured reason from the native transport failure,
+             * kept alongside the sanitized `code`/`message` operators see.
+             * Consumed by GitOps retry/backoff classification, which needs
+             * more than the public error code to tell a transient network
+             * condition from a permanent configuration one.
+             */
+            transportReason?: TransportFailureReason;
+        },
    ) {
        super(message);
        this.name = 'GitSourceError';
@@ -1259,9 +1271,9 @@ export class GitSourceService {
                // A ref that resolved before but no longer does is a deletion
                // or force-push, distinct from a mis-typed ref on first link.
                if (classified.code === 'REF_NOT_FOUND' && hasPriorHistory) {
-                    throw new GitSourceError('REF_DELETED', REF_DELETED_MESSAGE);
+                    throw new GitSourceError('REF_DELETED', REF_DELETED_MESSAGE, { transportReason: e.reason });
                }
-                throw new GitSourceError(classified.code, classified.message);
+                throw new GitSourceError(classified.code, classified.message, { transportReason: e.reason });
            }
            throw e;
        } finally {
@@ -2642,6 +2654,12 @@ export class GitSourceService {
                `Git apply succeeded for ${stackName} (${commitSha.slice(0, 7)}, op ${pending.operationId.slice(0, 8)}, plan ${plan.fingerprint.slice(0, 12)})`,
                actor,
            );
+            // Promotion has committed and rewritten the authoritative Compose
+            // files, so cached stats/statuses/project-name state is stale here
+            // whether or not a deploy follows. This must fire exactly once per
+            // successful promotion, from every trigger, not only the manual
+            // apply route (which used to invalidate here itself).
+            invalidateNodeCaches(nodeId);
        } else {
            throw new GitSourceError('PLAN_UNAVAILABLE', 'Pending update cannot be reviewed; pull again.');
        }
@@ -2725,6 +2743,12 @@ export class GitSourceService {
                    recoverySvc.linkGateOrRetain(recoveryId, healthGateId);
                }
                console.log(`[GitSource] Applied and deployed ${stackName} at ${commitSha.slice(0, 7)}`);
+                // Fire-and-forget, matching the manual apply route's prior
+                // placement: the scan runs only after a successful deploy and
+                // must never delay or fail the apply response.
+                triggerPostDeployScan(stackName, nodeId).catch((err) =>
+                    console.error(`[Security] Post-deploy scan failed for ${sanitizeForLog(stackName)}:`, err),
+                );
                return { applied: true, deployed: true, recoveryId };
            } catch (e) {
                // R1: do not auto-compensate. Keep applied files and leave the
@@ -74,6 +74,20 @@ export interface PolicyEnforcementResult {
    trivyMissing?: boolean;
 }

+/**
+ * Candidate (pre-acceptance) policy outcome. Unlike the deploy-time gate,
+ * which deliberately fails open when the scanner is unavailable so an
+ * operator is never blocked from deploying, an unresolvable scanner state
+ * here is its own outcome: automatic source acceptance must not read
+ * `unavailable` as `allowed`, or a GitOps source could accept a candidate
+ * nothing actually proved safe.
+ */
+export type CandidatePolicyEvaluation = { policy?: ScanPolicy } & (
+    | { status: 'allowed' }
+    | { status: 'blocked'; violations: PolicyViolation[] }
+    | { status: 'unavailable'; reason: string }
+);
+
 const TRIVY_MISSING_NOTIFY_COOLDOWN_MS = 60 * 60 * 1000;
 // Growth bounded by configured-policy fanout (only stacks with an enabled
 // block_on_deploy policy can land here), not by total stack churn. Cleared
@@ -472,3 +486,50 @@ export async function enforcePolicyForImageRefs(
    );
    return { ok: false, bypassed: false, policy, violations };
 }
+
+/**
+ * Tri-state candidate evaluation for GitOps source acceptance, built on the
+ * same evaluator the deploy-time gate uses, with the candidate's own image
+ * refs supplied directly rather than read from disk. Has side effects:
+ * writes a policy.bypass/policy.suppression_pass audit row when applicable,
+ * and may dispatch the once-per-hour Trivy-missing operator notification.
+ */
+export async function evaluateCandidatePolicy(
+    stackName: string,
+    nodeId: number,
+    imageRefs: string[],
+    opts: PolicyEnforcementOptions,
+): Promise<CandidatePolicyEvaluation> {
+    const result = await enforcePolicyForImageRefs(stackName, nodeId, imageRefs, {
+        ...opts,
+        // Undefaulted, these attribute the audit row to a deploy path
+        // (enforcePolicyForImageRefs's own default), which never happened
+        // for a pre-acceptance candidate.
+        auditMethod: opts.auditMethod ?? 'POST',
+        auditPath: opts.auditPath ?? `/api/stacks/${stackName}/git-source/candidate`,
+        // The deploy gate silently skips an unscannable ref (fail-open,
+        // since it must never block an operator's deploy on its own
+        // inability to evaluate). Candidate evaluation is the opposite: an
+        // unscannable ref must surface as evidence, not vanish, so it can be
+        // told apart from a genuinely clean scan below.
+    }, undefined, true);
+    // Only trivyMissing forgoes bypass consideration below it because it is
+    // the one path that returns bypassed: false unconditionally; every other
+    // branch of the shared evaluator already honors opts.bypass itself.
+    if (result.trivyMissing) {
+        if (opts.bypass) return { status: 'allowed', policy: result.policy };
+        return { status: 'unavailable', policy: result.policy, reason: 'Vulnerability scanner is unavailable' };
+    }
+    if (!result.ok) {
+        // A violation with no `error` is a genuine scanned policy match; one
+        // with `error` set is an invalid ref, a scan failure, or an
+        // evaluation failure -- evidence Sencho could not prove either way,
+        // not a proven violation. All-unproven must not read as `blocked`.
+        const hasGenuineViolation = result.violations.some((v) => !v.error);
+        if (!hasGenuineViolation) {
+            return { status: 'unavailable', policy: result.policy, reason: 'Candidate could not be fully evaluated' };
+        }
+        return { status: 'blocked', policy: result.policy, violations: result.violations };
+    }
+    return { status: 'allowed', policy: result.policy };
+}
@@ -381,6 +381,7 @@ export function blankInlineApplication(id: string, blueprintId: number, at: numb
    active_generation_id: null,
    pause_at: null,
    pause_reason: null,
+    source_suspended_reason: null,
    partial_json: null,
    failure_stage: null,
    failure_class: null,
@@ -234,7 +234,14 @@ function deriveSource(app: GitOpsApplicationRow, limitations: GitOpsLimitation[]
      interruptedGenerationId: app.interruption_generation_id,
    };
  }
-  if (app.suspended_at) return { ...identity, status: 'source_suspended', suspendedAt: app.suspended_at };
+  if (app.suspended_at) {
+    return {
+      ...identity,
+      status: 'source_suspended',
+      suspendedAt: app.suspended_at,
+      suspendedReason: app.source_suspended_reason,
+    };
+  }
  if (app.failure_stage === 'fetch' || app.failure_stage === 'validation' || app.failure_stage === 'apply' || app.failure_stage === 'create') {
    return {
      ...identity,
@@ -156,6 +156,7 @@ export function buildDirectApplicationRow(args: {
    active_generation_id: null,
    pause_at: null,
    pause_reason: null,
+    source_suspended_reason: null,
    partial_json: null,
    failure_stage: null,
    failure_class: null,
@@ -68,10 +68,12 @@ export type GitOpsHistoryStage =
  | 'rollout_candidate_opened'
  | 'rollout_paused'
  | 'rollout_unpaused'
+  | 'source_accepted'
  | 'source_conflict_blocker'
  | 'source_retry_scheduled'
  | 'source_suspended'
  | 'source_unsuspended'
+  | 'target_applied'
  | 'target_tombstoned';

 export type HistoryInsert = {
@@ -95,6 +95,11 @@ CREATE TABLE IF NOT EXISTS gitops_applications (
  active_generation_id TEXT NULL,
  pause_at INTEGER NULL,
  pause_reason TEXT NULL,
+  -- Distinct from pause_reason: sourceSuspended/sourceUnsuspended write this
+  -- field, not the one rolloutPaused/rolloutUnpaused share across app and
+  -- target rows, so suspending a source can never clobber an unrelated
+  -- rollout pause reason (or the reverse).
+  source_suspended_reason TEXT NULL,
  partial_json TEXT NULL,
  failure_stage TEXT NULL CHECK (
    failure_stage IS NULL OR failure_stage IN (
@@ -365,12 +365,12 @@ export class GitOpsStore {
        intent_revision_id, rollout_candidate_id, rollout_generation_id, source_acceptance_ref,
        placement_approval_ref, rollout_authorization_ref, legacy_combined_approval_ref,
        preflight_fingerprint, latest_operation_id, active_operation_id, active_operation_stage,
-        active_operation_at, active_generation_id, pause_at, pause_reason, partial_json,
+        active_operation_at, active_generation_id, pause_at, pause_reason, source_suspended_reason, partial_json,
        failure_stage, failure_class, failure_at, retry_at, retry_count, suspended_at,
        recovery_ref, recovery_phase, interruption_stage, interruption_at,
        interruption_operation_id, interruption_generation_id, evidence_fresh_at,
        evidence_limitations_json, created_at, updated_at
-      ) VALUES (${Array(55).fill('?').join(', ')})`,
+      ) VALUES (${Array(56).fill('?').join(', ')})`,
    ).run(
      row.id, row.lifecycle_key, row.lifecycle_status, row.target_mode, row.stack_name, row.blueprint_id,
      row.configured_repo_url, row.repo_identity_json, row.configured_ref, row.compose_paths_json,
@@ -380,7 +380,7 @@ export class GitOpsStore {
      row.intent_revision_id, row.rollout_candidate_id, row.rollout_generation_id, row.source_acceptance_ref,
      row.placement_approval_ref, row.rollout_authorization_ref, row.legacy_combined_approval_ref,
      row.preflight_fingerprint, row.latest_operation_id, row.active_operation_id, row.active_operation_stage,
-      row.active_operation_at, row.active_generation_id, row.pause_at, row.pause_reason, row.partial_json,
+      row.active_operation_at, row.active_generation_id, row.pause_at, row.pause_reason, row.source_suspended_reason, row.partial_json,
      row.failure_stage, row.failure_class, row.failure_at, row.retry_at, row.retry_count, row.suspended_at,
      row.recovery_ref, row.recovery_phase, row.interruption_stage, row.interruption_at,
      row.interruption_operation_id, row.interruption_generation_id, row.evidence_fresh_at,
@@ -372,28 +372,10 @@ export class GitOpsTransitions {
  applied(args: AppliedArgs): TransitionResult {
    return this.mutateApp(args.applicationId, args.envelope, 'applied', 'committed', (app) => {
      const targets = this.acceptanceTargets(app, args);
-      this.insertAcceptanceRecords(app, args);
-      app.accepted_generation_id = args.generationId;
-      app.artifact_set_id = args.artifactSetId;
-      app.latest_artifact_set_id = args.artifactSetId;
-      app.source_acceptance_ref = args.sourceAcceptanceId;
-      app.candidate_generation_id = null;
-      app.candidate_plan_blocked = 0;
-      app.review_required = 0;
-      if (args.activateCreating && app.lifecycle_status === 'creating') {
-        app.lifecycle_status = 'active';
-      }
-      this.clearActive(app);
-      this.clearAppFailure(app, ['apply', 'fetch', 'validation']);
-      this.clearInterruption(app, 'apply_started');
+      this.applySourceAcceptanceMutation(app, args);
      for (const target of targets) {
        if (app.target_mode === 'direct') {
-          target.desired_generation_id = args.generationId;
-          target.applied_generation_id = args.generationId;
-          target.expected_artifact_set_id = args.artifactSetId;
-          target.latest_artifact_set_id = args.artifactSetId;
-          target.source_acceptance_ref = args.sourceAcceptanceId;
-          target.candidate_generation_id = null;
+          this.applyTargetAcceptanceMutation(target, args);
        }
        this.store().upsertTarget(target);
      }
@@ -404,6 +386,63 @@ export class GitOpsTransitions {
    });
  }

+  /**
+   * Mode-neutral half of `applied`: accept the candidate at the application
+   * level without binding any target. A Direct dispatch calls this before
+   * promotion; `targetApplied` binds the target only after promotion commits.
+   */
+  sourceAccepted(args: AppliedArgs): TransitionResult {
+    return this.mutateApp(args.applicationId, args.envelope, 'source_accepted', 'committed', (app) => {
+      // Unlike applied() (preserved byte-identical, predates suspension),
+      // this new entry point is the one a suspended source must refuse: no
+      // new acceptance while suspended, so the check lives here rather than
+      // in the shared requireAcceptableCandidate guard.
+      if (app.suspended_at) throw new GitOpsTransitionError('source is suspended');
+      this.requireAcceptableCandidate(app, args);
+      this.applySourceAcceptanceMutation(app, args);
+    }, {
+      generationId: args.generationId,
+      artifactSetId: args.artifactSetId,
+      sourceAcceptanceRef: args.sourceAcceptanceId,
+    });
+  }
+
+  /**
+   * Direct-only half of `applied`: bind one target to an already-accepted
+   * generation. Refuses a generation the application has not accepted, so a
+   * dispatch cannot bind a target to source content nothing authorized.
+   */
+  targetApplied(nodeId: number, args: AppliedArgs): TransitionResult {
+    const app = this.requireApp(args.applicationId);
+    if (app.target_mode !== 'direct') {
+      throw new GitOpsTransitionError('target application is not direct');
+    }
+    if (app.accepted_generation_id !== args.generationId) {
+      throw new GitOpsTransitionError('generation is not accepted');
+    }
+    // The application's accepted_generation_id does not move again until a
+    // later sourceAccepted call, so a delayed dispatch for a superseded-but-
+    // still-accepted generation would otherwise pass the check above even
+    // after a newer candidate has already been staged for this target. Only
+    // an acceptance reference this application actually recorded may bind a
+    // target; a caller passing any other id would otherwise write
+    // unverifiable authorization evidence straight onto the target row.
+    if (app.source_acceptance_ref !== args.sourceAcceptanceId) {
+      throw new GitOpsTransitionError('source acceptance reference does not match the accepted generation');
+    }
+    return this.mutateTarget(args.applicationId, nodeId, args.envelope, 'target_applied', args.generationId, (target) => {
+      if (target.target_status !== 'active') {
+        throw new GitOpsTransitionError('cannot apply to a tombstoned target');
+      }
+      if (target.candidate_generation_id !== args.generationId) {
+        throw new GitOpsTransitionError('target candidate does not match applied generation');
+      }
+      const before = { appliedGenerationId: target.applied_generation_id };
+      this.applyTargetAcceptanceMutation(target, args);
+      return { before, after: { appliedGenerationId: args.generationId } };
+    });
+  }
+
  /**
   * The single transaction that makes a create-from-Git durable.
   *
@@ -841,7 +880,7 @@ export class GitOpsTransitions {
        this.clearActive(app);
      }
      app.suspended_at = envelope.at;
-      app.pause_reason = reason;
+      app.source_suspended_reason = reason;
    });
  }

@@ -850,7 +889,7 @@ export class GitOpsTransitions {
    return this.mutateApp(applicationId, envelope, 'source_unsuspended', 'committed', (app) => {
      if (!app.suspended_at) throw new GitOpsTransitionError('source is not suspended');
      app.suspended_at = null;
-      app.pause_reason = null;
+      app.source_suspended_reason = null;
    });
  }

@@ -1893,7 +1932,8 @@ export class GitOpsTransitions {
   * Guard every precondition of `applied` and return the active targets the
   * acceptance has to bind.
   */
-  private acceptanceTargets(app: GitOpsApplicationRow, args: AppliedArgs): GitOpsTargetCurrentRow[] {
+  /** Guard every application-level precondition of accepting a candidate. */
+  private requireAcceptableCandidate(app: GitOpsApplicationRow, args: AppliedArgs): void {
    if (app.candidate_generation_id !== args.generationId) {
      throw new GitOpsTransitionError('applied generation is not the current candidate');
    }
@@ -1913,6 +1953,10 @@ export class GitOpsTransitions {
        throw new GitOpsTransitionError('live apply belongs to a different operation');
      }
    }
+  }
+
+  private acceptanceTargets(app: GitOpsApplicationRow, args: AppliedArgs): GitOpsTargetCurrentRow[] {
+    this.requireAcceptableCandidate(app, args);
    const targets = this.store().listTargets(app.id).filter((row) => row.target_status === 'active');
    if (app.target_mode === 'direct') {
      for (const target of targets) {
@@ -1924,6 +1968,34 @@ export class GitOpsTransitions {
    return targets;
  }

+  /** The mode-neutral application-row mutation `applied` and `sourceAccepted` share. */
+  private applySourceAcceptanceMutation(app: GitOpsApplicationRow, args: AppliedArgs): void {
+    this.insertAcceptanceRecords(app, args);
+    app.accepted_generation_id = args.generationId;
+    app.artifact_set_id = args.artifactSetId;
+    app.latest_artifact_set_id = args.artifactSetId;
+    app.source_acceptance_ref = args.sourceAcceptanceId;
+    app.candidate_generation_id = null;
+    app.candidate_plan_blocked = 0;
+    app.review_required = 0;
+    if (args.activateCreating && app.lifecycle_status === 'creating') {
+      app.lifecycle_status = 'active';
+    }
+    this.clearActive(app);
+    this.clearAppFailure(app, ['apply', 'fetch', 'validation']);
+    this.clearInterruption(app, 'apply_started');
+  }
+
+  /** The Direct-only target-row mutation `applied` and `targetApplied` share. */
+  private applyTargetAcceptanceMutation(target: GitOpsTargetCurrentRow, args: AppliedArgs): void {
+    target.desired_generation_id = args.generationId;
+    target.applied_generation_id = args.generationId;
+    target.expected_artifact_set_id = args.artifactSetId;
+    target.latest_artifact_set_id = args.artifactSetId;
+    target.source_acceptance_ref = args.sourceAcceptanceId;
+    target.candidate_generation_id = null;
+  }
+
  /** Seed the unresolved artifact row and the source acceptance this apply proves. */
  private insertAcceptanceRecords(app: GitOpsApplicationRow, args: AppliedArgs): void {
    const artifact: GitOpsArtifactSetRow = {
@@ -2247,7 +2319,7 @@ export class GitOpsTransitions {
        legacy_combined_approval_ref=?, preflight_fingerprint=?,
        latest_operation_id=?, active_operation_id=?,
        active_operation_stage=?, active_operation_at=?, active_generation_id=?,
-        pause_at=?, pause_reason=?, partial_json=?,
+        pause_at=?, pause_reason=?, source_suspended_reason=?, partial_json=?,
        failure_stage=?, failure_class=?, failure_at=?, retry_at=?, retry_count=?,
        suspended_at=?, recovery_ref=?, recovery_phase=?,
        interruption_stage=?, interruption_at=?, interruption_operation_id=?,
@@ -2264,7 +2336,7 @@ export class GitOpsTransitions {
      app.legacy_combined_approval_ref, app.preflight_fingerprint,
      app.latest_operation_id, app.active_operation_id,
      app.active_operation_stage, app.active_operation_at, app.active_generation_id,
-      app.pause_at, app.pause_reason, app.partial_json,
+      app.pause_at, app.pause_reason, app.source_suspended_reason, app.partial_json,
      app.failure_stage, app.failure_class, app.failure_at, app.retry_at, app.retry_count,
      app.suspended_at, app.recovery_ref, app.recovery_phase,
      app.interruption_stage, app.interruption_at, app.interruption_operation_id,
@@ -71,6 +71,8 @@ export type GitOpsApplicationRow = {
  active_generation_id: string | null;
  pause_at: number | null;
  pause_reason: string | null;
+  /** sourceSuspended/sourceUnsuspended's own reason field; independent of pause_reason. */
+  source_suspended_reason: string | null;
  partial_json: string | null;
  failure_stage: ApplicationFailureStage | null;
  failure_class: string | null;
@@ -402,7 +404,7 @@ export type SourceFacet =
  | (SourceIdentityFields & { status: 'source_superseded'; supersededGenerationId: string })
  | (SourceIdentityFields & { status: 'applying'; activeOperationId: string; activeGenerationId: string })
  | (SourceIdentityFields & { status: 'source_retry_scheduled'; retryAt: number; retryCount: number })
-  | (SourceIdentityFields & { status: 'source_suspended'; suspendedAt: number })
+  | (SourceIdentityFields & { status: 'source_suspended'; suspendedAt: number; suspendedReason: string | null })
  | (SourceIdentityFields & {
      status: 'source_failed';
      failureStage: 'fetch' | 'validation' | 'apply' | 'create';