How must an instance record the state of concurrent fine-tuning jobs when one diverges?

**Date of Record:** 2026 October 27 **Scope:** Applies when multiple fine-tuning jobs run concurrently against the same base model (or against each other's outputs), and one job diverges from the expected training trajectory (loss spike, distribution shift, or catastrophic forgetting), requiring the instance to isolate the divergence without corrupting the other jobs. Does not apply to strictly sequential fine-tuning or single-job training.

Core Distinction

The separation between **individual job checkpoint** (a job's own training state) and **concurrent lineage commitment** (the cryptographic record of which jobs were training concurrently, at what epoch each was, and which diverged). Relying on individual checkpoints creates the **Concurrent Lineage Contamination**: when one job diverges, the divergent weights may have been sampled by other jobs (via shared checkpoints, shared data, or shared evaluation), so isolating the divergent job requires knowing which other jobs consumed its outputs during the divergence window.

What Was Observed

In concurrent fine-tuning architectures, failing to record concurrent lineage produces three primary failure modes:

  1. **Silent Contamination:** Job A diverges at epoch 500. Job B sampled Job A's checkpoint at epoch 520 for reference. Job B's weights are now tainted, but no record exists to trace the contamination.
  2. **Divergence Attribution Failure:** Job A's loss spikes. The instance suspects Job B's recent checkpoint, but without lineage records, cannot prove or disprove the interaction.
  3. **Corrupted Rollback:** The divergent job is rolled back to a prior checkpoint, but other jobs that consumed its divergent state are not rolled back, leaving the fleet in an inconsistent state.

Recorded Elements

  1. **`concurrent_jobs_manifest`:** The exact set of jobs that were training concurrently at each epoch, with their checkpoints and training state, allowing traceability of contamination.
  2. **`checkpoint_consumption_ledger`:** Which job consumed which other job's checkpoint at which epoch, forming the dependency graph of concurrent training.
  3. **`divergence_detection_receipt`:** The cryptographic proof of divergence (loss spike, KL divergence from expected, evaluation regression), timestamped, allowing isolation of the exact epoch of divergence.
  4. **`isolation_action_receipt`:** For all jobs, the exact isolation action taken (halt, rollback, quarantine) and its scope, ensuring contaminated jobs are properly handled.

Condition of Validity

* **Applies when:** Multiple fine-tuning jobs run concurrently, they can share state, and divergence in one can contaminate the others. * **Does not apply when:** Fine-tuning jobs are strictly independent with no shared state, or only one job is running at a time.

Decision Boundary

  1. **Are multiple jobs training concurrently?** If no, standard fine-tuning recording applies. If yes, proceed.
  2. **Is the `concurrent_jobs_manifest` and `checkpoint_consumption_ledger` recorded?** If no, contamination cannot be traced. The instance must halt concurrent training until these are in place.
  3. **Has a job diverged?** If yes, execute the `divergence_detection_receipt` and identify all jobs that consumed the divergent job's checkpoints during the divergence window.
  4. **Conversely, when can an instance isolate a divergent job without checking downstream contamination?** Only when the divergent job's checkpoints were never consumed by any other job (strictly isolated fine-tuning), and no shared data/eval was used during the divergence window.

Operational Effect

  1. **Contamination Tracing:** Enables the instance to identify which jobs were affected by the divergent job, preventing silent propagation.
  2. **Deterministic Rollback:** Provides the exact scope of jobs that need rollback, avoiding over- or under-rollback.
  3. **Divergence Attribution:** Allows the instance to distinguish between internal divergence (job's own issues) and external contamination (another job's influence).
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class DivergenceVerdict(Enum):
    NO_DIVERGENCE = auto()
    DIVERGENCE_ISOLATED = auto()
    CONTAMINATION_TRACED = auto()
    UNRECOVERABLE_CONTAMINATION = auto()

@dataclass(frozen=True)
class FineTuneJob:
    job_id: str
    current_epoch: int
    divergent: bool
    consumed_checkpoints: tuple[tuple[str, int], ...]

@dataclass(frozen=True)
class ConcurrentTrainingState:
    jobs: tuple[FineTuneJob, ...]
    divergent_job_id: str | None
    divergence_epoch: int

@dataclass(frozen=True)
class DivergenceManifest:
    verdict: DivergenceVerdict
    affected_job_ids: tuple[str, ...]
    directive: str
    manifest_digest: str

class ConcurrentFineTuneGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_divergence(
        cls, state: ConcurrentTrainingState
    ) -> DivergenceManifest:
        
        # Gate 1: No divergence
        if not state.divergent_job_id:
            digest = cls._hash(json.dumps({
                "jobs": str(len(state.jobs)),
                "divergence": "NONE"
            }, sort_keys=True))
            return DivergenceManifest(
                verdict=DivergenceVerdict.NO_DIVERGENCE,
                affected_job_ids=tuple(),
                directive="PROCEED: No divergence detected.",
                manifest_digest=digest
            )

        # Gate 2: Trace contamination
        affected: list[str] = []
        for job in state.jobs:
            if job.job_id == state.divergent_job_id:
                continue
            for consumed_id, consumed_epoch in job.consumed_checkpoints:
                if consumed_id == state.divergent_job_id and consumed_epoch >= state.divergence_epoch:
                    affected.append(job.job_id)
                    break

        # Gate 3: No contamination
        if not affected:
            digest = cls._hash(json.dumps({
                "divergent": state.divergent_job_id,
                "epoch": str(state.divergence_epoch),
                "affected": "NONE"
            }, sort_keys=True))
            return DivergenceManifest(
                verdict=DivergenceVerdict.DIVERGENCE_ISOLATED,
                affected_job_ids=tuple(),
                directive=f"PROCEED: Divergent job {state.divergent_job_id} isolated, no contamination.",
                manifest_digest=digest
            )

        # Gate 4: Contamination traced
        digest = cls._hash(json.dumps({
            "divergent": state.divergent_job_id,
            "epoch": str(state.divergence_epoch),
            "affected": tuple(sorted(affected))
        }, sort_keys=True))
        
        return DivergenceManifest(
            verdict=DivergenceVerdict.CONTAMINATION_TRACED,
            affected_job_ids=tuple(sorted(affected)),
            directive=f"ACTION: Divergent job {state.divergent_job_id} contaminated {len(affected)} jobs: {sorted(affected)}. Quarantine all.",
            manifest_digest=digest
        )