Fine-Tune a Leakage-Safe PyTorch Encoder

Introduction

30 Second Summary

A model can look successful while hiding where its knowledge came from. Once training examples leak into evaluation, its results stop being trustworthy.

In this project, you will use PyTorch to pretrain a denoising encoder without readiness labels. You will run a fair three-way transfer learning comparison on identical runner-safe folds before packaging the selected CPU model for Level 7.

What You'll Build

Your final local CPU run ends with a defensible result that ranks scratch training against two transfer strategies on the same unseen runners.

By the end of this project, you'll have:

• A self-supervised source checkpoint you can inspect to confirm that its encoder learned through denoising without target labels or target-validation runners.
• A three-way evidence report that ranks scratch training against frozen transfer and partial fine-tuning under the same 80-epoch budget. Its validation metrics make performance visible. Its generalization gaps expose overfitting. Its separate timing categories reveal compute cost.
• A fingerprinted Level 7 handoff that reloads the selected CPU model with its preprocessing metadata and evidence lineage intact.
• Secret Mission: Compare excessive and controlled encoder learning rates against one immutable pretrained encoder to measure catastrophic forgetting.

Are there any prerequisites?

This level continues from your completed Level 5 repository with artifacts/v3/level6_finetune_handoff.json plus its fingerprinted dependencies. You also need the local running-readiness-level5:2.15.0 image available through Docker Desktop.

Before We Start

Level 6 is a development-only controlled experiment in transfer learning. Its evidence boundary separates label-free feature denoising from running-readiness classification while keeping the consumed v1 holdout permanently unavailable and excluding the Level 5 readiness checkpoint from pretraining claims.

Audit Level 5 and Extend the CPU Container

Fine-tuning depends on an unbroken evidence lineage. A changed checkpoint or misidentified artifact would undermine the experiment before the first source-task weight is learned.

You will extend the inherited CPU environment with a thin Docker image. A fail-closed audit will verify the handoff before Level 6 can continue.

In this step, get ready to:

• Extend the inherited Level 5 CPU image without reinstalling its dependencies.
• Create the Level 6 command entry point with fingerprint and checkpoint safeguards.
• Run the audit inside the new image and confirm that the inherited evidence passes.

Extend the inherited CPU image

A thin container extension preserves the pinned PyTorch environment from Level 5. The new image adds a Level 6 default command while reusing every installed dependency.

• Create a file named Dockerfile.level6 in the top-level repository folder using VS Code.
• Add the Level 6 image definition by copying this code into the file:

# syntax=docker/dockerfile:1
FROM running-readiness-level5:2.15.0

WORKDIR /workspace
CMD ["python", "-m", "src.finetune_experiment", "audit"]

What does this image definition do?

• The FROM instruction extends the existing Level 5 CPU image with its pinned runtime intact.
• The WORKDIR instruction places commands inside the bind-mounted repository.
• The CMD instruction makes the Level 6 audit the container's default action.

• Save Dockerfile.level6.
• Confirm that Dockerfile.level6 appears beside the repository's existing top-level files.

Can't find the Dockerfile?

Check that the file is in the repository's top-level folder. A file created inside src will not be found by the build command.

Make sure the filename ends with .level6 and has no extra extension. Help me check the Level 6 Dockerfile location and filename.

✔️ Awesome, I've got everything!

Great. Save Dockerfile.level6 before continuing.

ⓧ I'd like to double check the full code

# syntax=docker/dockerfile:1
FROM running-readiness-level5:2.15.0

WORKDIR /workspace
CMD ["python", "-m", "src.finetune_experiment", "audit"]

What should match?

Your file should contain one inherited base image. It should also contain the workspace and default audit command shown above.

Add the fail-closed audit entry point

The audit treats the Level 5 artifacts as immutable evidence. Each fingerprint links an expected digest to the file currently present in the repository.

The entry point also restricts checkpoint loading to CPU tensors. It rejects changed handoff metadata before any Level 6 training command can use it.

• Create finetune_experiment.py inside the existing src folder using VS Code.
• Add the audit imports by copying this first section into src/finetune_experiment.py:

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any

import pandas as pd
import torch

from src.feature_experiment import (
    FORBIDDEN_HOLDOUT_USES,
    LEVEL5_HANDOFF_PATH,
    TRAIN_PATH,
    V2_SCHEMA_PATH,
    fingerprint,
    fold_signature,
    read_json,
)
from src.neural_network_experiment import (
    CHECKPOINT_PATH as LEVEL5_CHECKPOINT_PATH,
    LEVEL6_HANDOFF_PATH,
    PREPROCESSOR_PATH as LEVEL5_PREPROCESSOR_PATH,
    REPORT_PATH as LEVEL5_REPORT_PATH,
    approved_matrix,
    audit_level4,
)
from src.train_models import SCHEMA_PATH, make_group_folds

What do these imports connect?

• The inherited feature module supplies the restricted holdout policy and artifact fingerprint helpers.
• The inherited neural network module supplies the Level 5 checkpoint paths and base audit.
• The training module supplies the grouped-fold constructor used to compare inherited fold signatures.

• Save src/finetune_experiment.py.
• Confirm that the file appears inside the existing src folder.

Seeing unresolved inherited imports?

Confirm that you are editing the completed Level 5 repository. The imported modules must already exist inside its src folder.

Check every imported name against the snippet above. Help me identify a mismatched inherited import in finetune_experiment.py.

• Add the Level 6 artifact paths and restricted checkpoint helpers below the imports by copying this code:

REPORT_PATH = Path("reports/v4/finetune_experiment.json")
SECRET_REPORT_PATH = Path("reports/v4/catastrophic_forgetting.json")
PREVIEW_SOURCE_PATH = Path("artifacts/v4/level6_source_preview.pt")
SOURCE_CHECKPOINT_PATH = Path("artifacts/v4/level6_source_pretrained.pt")
SELECTED_CHECKPOINT_PATH = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_PATH = Path("artifacts/v4/level6_preprocessor.json")
LEVEL7_HANDOFF_PATH = Path("artifacts/v4/level7_deploy_handoff.json")
DOC_PATH = Path("docs/level-6-finetune.md")


def verify_fingerprint(path: Path, expected: dict[str, Any]) -> None:
    observed = fingerprint(path)
    if observed.get("sha256") != expected.get("sha256"):
        raise ValueError(f"Fingerprint mismatch: {path}")


def load_level5_checkpoint(root: Path) -> dict[str, Any]:
    return torch.load(
        root / LEVEL5_CHECKPOINT_PATH,
        map_location="cpu",
        weights_only=True,
    )

How do the safeguards work?

• The path constants give every later Level 6 artifact one fixed location.
• The verify_fingerprint() helper compares the observed SHA-256 value with the handoff's expected value.
• The load_level5_checkpoint() helper restricts checkpoint loading and remaps its tensors to the CPU.

• Save src/finetune_experiment.py.
• Confirm that verify_fingerprint() and load_level5_checkpoint() appear below the path constants.

Checkpoint helper looks incomplete?

Check the indentation inside torch.load(). Its checkpoint path and loading restrictions must remain inside the function call.

Confirm that the closing parenthesis aligns with return. Help me check the checkpoint-loading helper's Python indentation.

• Add the initial audit_level5() structure below load_level5_checkpoint() by copying this code:

def audit_level5(root: Path) -> dict[str, Any]:
    base = audit_level4(root)
    handoff = read_json(root / LEVEL6_HANDOFF_PATH)
    paths = {
        "level4_handoff": LEVEL5_HANDOFF_PATH,
        "development_data": TRAIN_PATH,
        "feature_schema": V2_SCHEMA_PATH,
        "level5_report": LEVEL5_REPORT_PATH,
        "checkpoint": LEVEL5_CHECKPOINT_PATH,
        "preprocessor": LEVEL5_PREPROCESSOR_PATH,
    }
    for label, relative in paths.items():
        verify_fingerprint(root / relative, handoff["fingerprints"][label])
    return {
        "status": "passed",
        "holdout_reads": base["holdout_reads"],
        "verified_fingerprints": sorted(paths),
        "checkpoint_device": "cpu",
        "approved_features": handoff["approved_features"],
    }

What does the audit cover first?

The inherited audit_level4() call preserves the earlier target-label and consumed-holdout checks. The six-entry path map connects the Level 6 handoff to every dependency it fingerprints.

The returned payload exposes the evidence you need to inspect. A passing status alone never replaces the fingerprint list.

• In audit_level5(), find the line that loads handoff.
• Insert these handoff-policy checks directly below that line:

    if handoff.get("handoff_version") != "3.0.0":
        raise ValueError("Expected Level 6 handoff version 3.0.0")
    if handoff.get("created_by") != "Level 5: Networks":
        raise ValueError("Unexpected Level 6 handoff creator")
    if handoff.get("next_level") != "Level 6: Fine-tune":
        raise ValueError("Handoff does not target Level 6: Fine-tune")
    if handoff.get("consumed_v1_holdout_read") is not False:
        raise ValueError("Level 5 handoff does not preserve zero holdout reads")
    if handoff.get("forbidden_uses") != list(FORBIDDEN_HOLDOUT_USES):
        raise ValueError("Consumed-holdout restrictions changed")

Why fail on handoff metadata?

These checks establish the handoff's version and creator before trusting its fingerprints. They also preserve the permanent rule that the consumed v1 holdout has zero reads.

Any changed forbidden-use policy stops the command. Level 6 cannot silently weaken an inherited evidence boundary.

• In audit_level5(), find the fingerprint loop.
• Insert these checkpoint and preprocessing checks between the fingerprint loop and the return statement:

    checkpoint = load_level5_checkpoint(root)
    preprocessor = read_json(root / LEVEL5_PREPROCESSOR_PATH)
    if checkpoint["feature_names"] != handoff["approved_features"]:
        raise ValueError("Checkpoint features differ from the handoff")
    if preprocessor["feature_names"] != handoff["approved_features"]:
        raise ValueError("Preprocessor features differ from the handoff")
    for field in ("threshold", "seed", "best_epoch"):
        if checkpoint[field] != handoff[field]:
            raise ValueError(f"Checkpoint metadata mismatch: {field}")
    if checkpoint["config"] != handoff["selected_config"]:
        raise ValueError("Checkpoint configuration differs from the handoff")

What do the metadata checks prove?

The checkpoint and serialized preprocessor must preserve the approved feature order recorded by the handoff. This prevents a valid file from being used with a mismatched input layout.

The threshold and training metadata must also match. The audit rejects a checkpoint that belongs to a different Level 5 run.

• Add the inherited-data check and command entry point below audit_level5() by copying this code:

def inherited_data(root: Path):
    x, y, groups, _ = approved_matrix(root)
    frame = pd.read_csv(root / TRAIN_PATH)
    folds = make_group_folds(frame, read_json(root / SCHEMA_PATH))
    report = read_json(root / LEVEL5_REPORT_PATH)
    signatures = [fold_signature(groups, fit, valid) for fit, valid in folds]
    if signatures != report["fold_signatures"]:
        raise ValueError("Inherited fold signatures changed")
    return x, y, groups, folds, signatures


def main() -> None:
    root = Path.cwd()
    command = sys.argv[1] if len(sys.argv) > 1 else "audit"
    if command == "audit":
        payload = audit_level5(root)
    else:
        raise SystemExit("Use: audit | assess | pretrain | run | secret | document")
    print(json.dumps(payload, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

How does the entry point close the loop?

The inherited_data() helper reconstructs the grouped folds and compares their signatures with the Level 5 report. Later experiment commands use this helper before training.

The main() function routes the first Level 6 command to audit_level5() and prints inspectable JSON. Unsupported commands stop with the complete command list.

• Save src/finetune_experiment.py.
• Confirm that main() is followed by the final module guard.

Seeing a Python syntax problem?

Check that each handoff and checkpoint check remains indented inside audit_level5(). The inherited_data() definition must return to the left edge.

Compare the closing module guard with the full-file reference below. Help me locate an indentation or missing-parenthesis issue in finetune_experiment.py.

✔️ Awesome, I've got everything!

Your audit entry point is ready. Save src/finetune_experiment.py before building the image.

ⓧ I'd like to double check the full code

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any

import pandas as pd
import torch

from src.feature_experiment import (
    FORBIDDEN_HOLDOUT_USES,
    LEVEL5_HANDOFF_PATH,
    TRAIN_PATH,
    V2_SCHEMA_PATH,
    fingerprint,
    fold_signature,
    read_json,
)
from src.neural_network_experiment import (
    CHECKPOINT_PATH as LEVEL5_CHECKPOINT_PATH,
    LEVEL6_HANDOFF_PATH,
    PREPROCESSOR_PATH as LEVEL5_PREPROCESSOR_PATH,
    REPORT_PATH as LEVEL5_REPORT_PATH,
    approved_matrix,
    audit_level4,
)
from src.train_models import SCHEMA_PATH, make_group_folds

REPORT_PATH = Path("reports/v4/finetune_experiment.json")
SECRET_REPORT_PATH = Path("reports/v4/catastrophic_forgetting.json")
PREVIEW_SOURCE_PATH = Path("artifacts/v4/level6_source_preview.pt")
SOURCE_CHECKPOINT_PATH = Path("artifacts/v4/level6_source_pretrained.pt")
SELECTED_CHECKPOINT_PATH = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_PATH = Path("artifacts/v4/level6_preprocessor.json")
LEVEL7_HANDOFF_PATH = Path("artifacts/v4/level7_deploy_handoff.json")
DOC_PATH = Path("docs/level-6-finetune.md")


def verify_fingerprint(path: Path, expected: dict[str, Any]) -> None:
    observed = fingerprint(path)
    if observed.get("sha256") != expected.get("sha256"):
        raise ValueError(f"Fingerprint mismatch: {path}")


def load_level5_checkpoint(root: Path) -> dict[str, Any]:
    return torch.load(
        root / LEVEL5_CHECKPOINT_PATH,
        map_location="cpu",
        weights_only=True,
    )


def audit_level5(root: Path) -> dict[str, Any]:
    base = audit_level4(root)
    handoff = read_json(root / LEVEL6_HANDOFF_PATH)
    if handoff.get("handoff_version") != "3.0.0":
        raise ValueError("Expected Level 6 handoff version 3.0.0")
    if handoff.get("created_by") != "Level 5: Networks":
        raise ValueError("Unexpected Level 6 handoff creator")
    if handoff.get("next_level") != "Level 6: Fine-tune":
        raise ValueError("Handoff does not target Level 6: Fine-tune")
    if handoff.get("consumed_v1_holdout_read") is not False:
        raise ValueError("Level 5 handoff does not preserve zero holdout reads")
    if handoff.get("forbidden_uses") != list(FORBIDDEN_HOLDOUT_USES):
        raise ValueError("Consumed-holdout restrictions changed")
    paths = {
        "level4_handoff": LEVEL5_HANDOFF_PATH,
        "development_data": TRAIN_PATH,
        "feature_schema": V2_SCHEMA_PATH,
        "level5_report": LEVEL5_REPORT_PATH,
        "checkpoint": LEVEL5_CHECKPOINT_PATH,
        "preprocessor": LEVEL5_PREPROCESSOR_PATH,
    }
    for label, relative in paths.items():
        verify_fingerprint(root / relative, handoff["fingerprints"][label])
    checkpoint = load_level5_checkpoint(root)
    preprocessor = read_json(root / LEVEL5_PREPROCESSOR_PATH)
    if checkpoint["feature_names"] != handoff["approved_features"]:
        raise ValueError("Checkpoint features differ from the handoff")
    if preprocessor["feature_names"] != handoff["approved_features"]:
        raise ValueError("Preprocessor features differ from the handoff")
    for field in ("threshold", "seed", "best_epoch"):
        if checkpoint[field] != handoff[field]:
            raise ValueError(f"Checkpoint metadata mismatch: {field}")
    if checkpoint["config"] != handoff["selected_config"]:
        raise ValueError("Checkpoint configuration differs from the handoff")
    return {
        "status": "passed",
        "holdout_reads": base["holdout_reads"],
        "verified_fingerprints": sorted(paths),
        "checkpoint_device": "cpu",
        "approved_features": handoff["approved_features"],
    }


def inherited_data(root: Path):
    x, y, groups, _ = approved_matrix(root)
    frame = pd.read_csv(root / TRAIN_PATH)
    folds = make_group_folds(frame, read_json(root / SCHEMA_PATH))
    report = read_json(root / LEVEL5_REPORT_PATH)
    signatures = [fold_signature(groups, fit, valid) for fit, valid in folds]
    if signatures != report["fold_signatures"]:
        raise ValueError("Inherited fold signatures changed")
    return x, y, groups, folds, signatures


def main() -> None:
    root = Path.cwd()
    command = sys.argv[1] if len(sys.argv) > 1 else "audit"
    if command == "audit":
        payload = audit_level5(root)
    else:
        raise SystemExit("Use: audit | assess | pretrain | run | secret | document")
    print(json.dumps(payload, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

What should match?

Use this cumulative file to compare imports and indentation. The audit must finish before inherited_data() and main() begin.

Build the image and verify the lineage

The image build packages the thin Level 6 command layer on top of the inherited runtime. Docker can reuse the existing base image because this Dockerfile installs no replacement dependencies.

• Build the Level 6 CPU image from the repository folder by running this command:

docker build --platform linux/amd64 -f Dockerfile.level6 -t running-readiness-level6:2.15.0 .

What does this build create?

Docker targets the inherited CPU platform and reads Dockerfile.level6. The resulting local image is tagged running-readiness-level6:2.15.0 for the remaining Level 6 commands.

• Wait for the build command to finish.
• Confirm that the build completes without an error.

Image build did not complete?

Confirm that Docker Desktop is still running. The build also requires the inherited running-readiness-level5:2.15.0 image to remain available locally.

Run the command from the repository folder containing Dockerfile.level6. Help me diagnose why the thin Level 6 Docker image is not building.

Before you run the audit, which inherited condition do you expect to stop the command first if one artifact has changed?

• Run the Level 6 audit inside the new image by using this command:

docker run --rm --platform linux/amd64 --mount type=bind,src="$(pwd)",dst=/workspace -w /workspace running-readiness-level6:2.15.0 python -m src.finetune_experiment audit

What does this audit command do?

The bind mount gives the container read access to the current repository through /workspace. The command then invokes the Level 6 audit entry point inside the pinned CPU environment.

The container is removed after the command exits. The local image and repository files remain available.

You will see JSON with status set to passed. You will also see holdout_reads set to 0.

The verified_fingerprints list will contain six inherited dependencies. The checkpoint_device value will be cpu.

That audit is the first Level 6 safeguard complete. Your transfer experiment now starts from a verified checkpoint lineage and an unchanged holdout policy.

Audit stopped before passing?

Treat the failure as a protected boundary. Restore the changed Level 5 artifact from the completed repository instead of editing the handoff to match it.

Check the reported artifact name before replacing anything. Help me trace a Level 6 audit failure without weakening its lineage checks.

Your Level 6 CPU image can now defend every inherited input before training begins. Next, you will make the deliberate case that the Level 5 readiness checkpoint is target-task evidence instead of a pretrained source model.

Reject the False Pretraining Shortcut

The local Docker image now proves that the inherited Level 5 evidence has not changed. That intact lineage still leaves one question about what the checkpoint actually learned.

Controlled transfer learning needs a source objective that differs from the target objective. The Level 5 PyTorch checkpoint learned the readiness target. This step tests that lineage before any transfer comparison can proceed.

In this step, get ready to:

• Assess whether the inherited checkpoint records a separate source task.
• Validate the feature count plus every grouped development fold.
• Run the deliberate rejection before enabling the controlled experiment.

Reject the inherited checkpoint

Checkpoint lineage determines which claim the experiment can defend. The assessment looks for a recorded source task instead of inferring pretraining from the existence of saved weights.

• Return to the existing src/finetune_experiment.py tab in VS Code.
• Locate the end of inherited_data().
• Add the assessment below inherited_data() by copying this code:

def assess_transfer_support(root: Path) -> dict[str, Any]:
    audit_level5(root)
    checkpoint = load_level5_checkpoint(root)
    x, y, groups, folds, signatures = inherited_data(root)
    reasons: list[str] = []
    if x.shape[1] < 2:
        reasons.append("Fewer than two approved numeric features")
    if len(folds) != 4 or len(signatures) != 4:
        reasons.append("Exactly four inherited folds are required")
    for number, (fit_rows, validation_rows) in enumerate(folds, start=1):
        fit_groups = set(groups.iloc[fit_rows])
        validation_groups = set(groups.iloc[validation_rows])
        if fit_groups & validation_groups:
            reasons.append(f"Fold {number} has runner overlap")
        if y.iloc[fit_rows].nunique() < 2 or y.iloc[validation_rows].nunique() < 2:
            reasons.append(f"Fold {number} lacks both target classes")
    has_source_lineage = "source_task" in checkpoint
    return {
        "level5_checkpoint_is_pretrained": has_source_lineage,
        "level5_checkpoint_reason": (
            "A source-task lineage exists." if has_source_lineage else
            "The checkpoint records only readiness-target training and no separate source task."
        ),
    }

What does this code assess?

• The call to audit_level5() verifies the inherited evidence before the assessment trusts it.
• The feature check requires at least two approved numeric inputs.
• The fold checks require four inherited signatures.
• The runner intersections expose grouped leakage inside any outer fold.
• The class checks require both readiness classes on each side of every fold.
• The source_task lookup tests whether the checkpoint records a distinct source objective.

• Save src/finetune_experiment.py.
• Confirm that your editor does not underline the new function as invalid Python.

Seeing invalid Python in the function?

Check that assess_transfer_support() starts after the final line of inherited_data(). Match the indentation of every line inside the new function.

Check the closing parenthesis under level5_checkpoint_reason plus the closing brace for the returned dictionary.

Help me repair the assessment function without changing its evidence gates.

Expose the deliberate rejection

The function now separates checkpoint lineage from the grouped-data requirements. The command entry point needs an assessment branch so you can inspect that decision as JSON.

• Scroll to main() near the bottom of src/finetune_experiment.py.
• Find the existing audit branch:

    if command == "audit":
        payload = audit_level5(root)

Why find this branch?

This branch already routes the default command through the inherited audit. The assessment branch belongs directly below it so both commands share the same entry point.

• Replace the audit branch with this expanded branch:

    if command == "audit":
        payload = audit_level5(root)
    elif command == "assess":
        payload = assess_transfer_support(root)

What does the new branch do?

The assess branch calls assess_transfer_support() against the repository at the current path. The existing JSON print exposes the returned decision.

• Save src/finetune_experiment.py.

Before you run the assessment, do you expect target-task weights to qualify as pretrained evidence without a recorded source objective?

• Switch back to the terminal from the previous step.
• Run the assessment from the repository by entering this command:

python -m src.finetune_experiment assess

What does this command test?

The command reruns the inherited audit before loading the checkpoint on the CPU. It then evaluates the recorded lineage plus the inherited development folds.

You will see level5_checkpoint_is_pretrained set to false. The reason states that the checkpoint contains readiness-target training with no separate source task.

That rejection is the intended result. Your experiment now refuses to present same-task supervised weights as pretrained evidence.

Add the fail-closed result

Checkpoint rejection answers the lineage question. Repository suitability needs a separate decision that blocks target comparison whenever any feature or fold requirement fails.

• Return to assess_transfer_support() in src/finetune_experiment.py.
• Find the current return dictionary:

    return {
        "level5_checkpoint_is_pretrained": has_source_lineage,
        "level5_checkpoint_reason": (
            "A source-task lineage exists." if has_source_lineage else
            "The checkpoint records only readiness-target training and no separate source task."
        ),
    }

What is still missing?

This dictionary reports the checkpoint-lineage decision. It does not yet expose whether the development data passed every controlled-transfer gate.

• Replace that return dictionary with the complete fail-closed result:

    return {
        "level5_checkpoint_is_pretrained": has_source_lineage,
        "level5_checkpoint_reason": (
            "A source-task lineage exists." if has_source_lineage else
            "The checkpoint records only readiness-target training and no separate source task."
        ),
        "controlled_transfer_supported": not reasons,
        "blocking_reasons": reasons,
        "approved_feature_count": x.shape[1],
        "runner_count": int(groups.nunique()),
        "fold_count": len(folds),
        "fallback_proposal": (
            None if not reasons else
            "Run a separate all-synthetic denoising-source and shifted-classification-target exercise; do not compare against readiness labels."
        ),
    }

How does this fail closed?

• An empty reasons list makes controlled_transfer_supported true.
• Any failed requirement adds a specific entry to blocking_reasons.
• Any blocking reason makes controlled_transfer_supported false.
• A blocked repository receives a separate all-synthetic fallback proposal.
• The fallback keeps readiness labels out of an unsupported target comparison.

• Save src/finetune_experiment.py.

✔️ Awesome, I've got everything!

Your assessment now rejects the false checkpoint claim. It also blocks the controlled comparison whenever a development-data gate fails.

ⓧ I'd like to double check the full code

from __future__ import annotations

import copy
import hashlib
import json
import sys
import time
from pathlib import Path
from statistics import fmean, pstdev
from typing import Any

import numpy as np
import pandas as pd
import torch
from torch import nn

from src.feature_experiment import (
    FORBIDDEN_HOLDOUT_USES,
    LEVEL5_HANDOFF_PATH,
    TRAIN_PATH,
    V2_SCHEMA_PATH,
    fingerprint,
    fold_signature,
    read_json,
    write_json_exclusive,
)
from src.neural_network_experiment import (
    CHECKPOINT_PATH as LEVEL5_CHECKPOINT_PATH,
    LEVEL6_HANDOFF_PATH,
    PREPROCESSOR_PATH as LEVEL5_PREPROCESSOR_PATH,
    REPORT_PATH as LEVEL5_REPORT_PATH,
    SEED,
    THRESHOLD,
    approved_matrix,
    audit_level4,
    evaluate_model,
    fit_preprocessor,
    seed_everything,
    serialize_preprocessor,
    transform,
)
from src.train_models import SCHEMA_PATH, make_group_folds

REPORT_PATH = Path("reports/v4/finetune_experiment.json")
SECRET_REPORT_PATH = Path("reports/v4/catastrophic_forgetting.json")
PREVIEW_SOURCE_PATH = Path("artifacts/v4/level6_source_preview.pt")
SOURCE_CHECKPOINT_PATH = Path("artifacts/v4/level6_source_pretrained.pt")
SELECTED_CHECKPOINT_PATH = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_PATH = Path("artifacts/v4/level6_preprocessor.json")
LEVEL7_HANDOFF_PATH = Path("artifacts/v4/level7_deploy_handoff.json")
DOC_PATH = Path("docs/level-6-finetune.md")
APPROACHES = ("scratch", "frozen", "partial")
SOURCE_EPOCHS = 60
SOURCE_MULTIPLIER = 8
SOURCE_NOISE = 0.15
TARGET_EPOCHS = 80
ENCODER_LEARNING_RATE = 0.001
HEAD_LEARNING_RATE = 0.01


def verify_fingerprint(path: Path, expected: dict[str, Any]) -> None:
    observed = fingerprint(path)
    if observed.get("sha256") != expected.get("sha256"):
        raise ValueError(f"Fingerprint mismatch: {path}")


def load_level5_checkpoint(root: Path) -> dict[str, Any]:
    return torch.load(
        root / LEVEL5_CHECKPOINT_PATH,
        map_location="cpu",
        weights_only=True,
    )


def audit_level5(root: Path) -> dict[str, Any]:
    base = audit_level4(root)
    handoff = read_json(root / LEVEL6_HANDOFF_PATH)
    if handoff.get("handoff_version") != "3.0.0":
        raise ValueError("Expected Level 6 handoff version 3.0.0")
    if handoff.get("created_by") != "Level 5: Networks":
        raise ValueError("Unexpected Level 6 handoff creator")
    if handoff.get("next_level") != "Level 6: Fine-tune":
        raise ValueError("Handoff does not target Level 6: Fine-tune")
    if handoff.get("consumed_v1_holdout_read") is not False:
        raise ValueError("Level 5 handoff does not preserve zero holdout reads")
    if handoff.get("forbidden_uses") != list(FORBIDDEN_HOLDOUT_USES):
        raise ValueError("Consumed-holdout restrictions changed")
    paths = {
        "level4_handoff": LEVEL5_HANDOFF_PATH,
        "development_data": TRAIN_PATH,
        "feature_schema": V2_SCHEMA_PATH,
        "level5_report": LEVEL5_REPORT_PATH,
        "checkpoint": LEVEL5_CHECKPOINT_PATH,
        "preprocessor": LEVEL5_PREPROCESSOR_PATH,
    }
    for label, relative in paths.items():
        verify_fingerprint(root / relative, handoff["fingerprints"][label])
    checkpoint = load_level5_checkpoint(root)
    preprocessor = read_json(root / LEVEL5_PREPROCESSOR_PATH)
    if checkpoint["feature_names"] != handoff["approved_features"]:
        raise ValueError("Checkpoint features differ from the handoff")
    if preprocessor["feature_names"] != handoff["approved_features"]:
        raise ValueError("Preprocessor features differ from the handoff")
    for field in ("threshold", "seed", "best_epoch"):
        if checkpoint[field] != handoff[field]:
            raise ValueError(f"Checkpoint metadata mismatch: {field}")
    if checkpoint["config"] != handoff["selected_config"]:
        raise ValueError("Checkpoint configuration differs from the handoff")
    return {
        "status": "passed",
        "holdout_reads": base["holdout_reads"],
        "verified_fingerprints": sorted(paths),
        "checkpoint_device": "cpu",
        "approved_features": handoff["approved_features"],
    }


def inherited_data(root: Path):
    x, y, groups, _ = approved_matrix(root)
    frame = pd.read_csv(root / TRAIN_PATH)
    folds = make_group_folds(frame, read_json(root / SCHEMA_PATH))
    report = read_json(root / LEVEL5_REPORT_PATH)
    signatures = [fold_signature(groups, fit, valid) for fit, valid in folds]
    if signatures != report["fold_signatures"]:
        raise ValueError("Inherited fold signatures changed")
    return x, y, groups, folds, signatures


def assess_transfer_support(root: Path) -> dict[str, Any]:
    audit_level5(root)
    checkpoint = load_level5_checkpoint(root)
    x, y, groups, folds, signatures = inherited_data(root)
    reasons: list[str] = []
    if x.shape[1] < 2:
        reasons.append("Fewer than two approved numeric features")
    if len(folds) != 4 or len(signatures) != 4:
        reasons.append("Exactly four inherited folds are required")
    for number, (fit_rows, validation_rows) in enumerate(folds, start=1):
        fit_groups = set(groups.iloc[fit_rows])
        validation_groups = set(groups.iloc[validation_rows])
        if fit_groups & validation_groups:
            reasons.append(f"Fold {number} has runner overlap")
        if y.iloc[fit_rows].nunique() < 2 or y.iloc[validation_rows].nunique() < 2:
            reasons.append(f"Fold {number} lacks both target classes")
    has_source_lineage = "source_task" in checkpoint
    return {
        "level5_checkpoint_is_pretrained": has_source_lineage,
        "level5_checkpoint_reason": (
            "A source-task lineage exists." if has_source_lineage else
            "The checkpoint records only readiness-target training and no separate source task."
        ),
        "controlled_transfer_supported": not reasons,
        "blocking_reasons": reasons,
        "approved_feature_count": x.shape[1],
        "runner_count": int(groups.nunique()),
        "fold_count": len(folds),
        "fallback_proposal": (
            None if not reasons else
            "Run a separate all-synthetic denoising-source and shifted-classification-target exercise; do not compare against readiness labels."
        ),
    }


def main() -> None:
    root = Path.cwd()
    command = sys.argv[1] if len(sys.argv) > 1 else "audit"
    if command == "audit":
        payload = audit_level5(root)
    elif command == "assess":
        payload = assess_transfer_support(root)
    print(json.dumps(payload, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

How to use this reference

Compare the function order plus indentation with your file. Confirm that assess_transfer_support() appears after inherited_data().

Confirm that the assess branch appears inside main().

Before the final run, which fields do you expect to prove that the checkpoint is rejected while the controlled experiment remains supported?

• Switch back to the terminal from the assessment run.
• Run the completed assessment by entering the same command:

python -m src.finetune_experiment assess

What does the completed result prove?

The JSON reports the checkpoint-lineage decision independently from the repository-support decision. This prevents valid grouped data from disguising an invalid pretraining claim.

You will see level5_checkpoint_is_pretrained set to false. You will also see controlled_transfer_supported set to true.

You will see at least 2 approved features. You will see 4 inherited folds.

You will see an empty blocking_reasons list. The empty list confirms runner separation plus class coverage across every outer partition.

Strong result. Your CLI now rejects the false pretraining shortcut while preserving a valid path to the controlled transfer experiment.

Assessment blocked or failing?

If the command fails before printing JSON, check that the inherited audit still passes. A changed Level 5 artifact must remain a blocking failure.

If blocking_reasons contains entries, use each entry to identify the failed feature or fold requirement. Keep every gate in place.

Help me diagnose the assessment without weakening its safeguards.

The shortcut is now blocked by executable evidence. Next, you will create a separate denoising source task that learns without readiness labels or target-validation runners.

Pretrain a Denoising Source Encoder

Your assessment proved that the Level 5 checkpoint learned the readiness target. A legitimate self-supervised pretraining stage needs a separate objective before readiness labels become available.

This step gives the PyTorch encoder that objective. It learns to reconstruct clean standardized features from seeded corrupted copies while every validation runner stays outside the source-training boundary.

In this step, get ready to:

• Define the shared encoder architecture for source pretraining and target classification.
• Generate deterministic denoising examples from fold-training runners only.
• Train a preview source model without readiness labels.

Define the source and target networks

A shared two-layer encoder makes the later comparison fair. The source model connects that encoder to a reconstruction decoder. The target model connects the same encoder shape to a fresh binary head.

• In the existing constants section of src/finetune_experiment.py, add the preview path and source-training settings with this code:

PREVIEW_SOURCE_PATH = Path("artifacts/v4/level6_source_preview.pt")
SOURCE_EPOCHS = 60
SOURCE_MULTIPLIER = 8
SOURCE_NOISE = 0.15

What do these settings control?

• The preview path keeps this checkpoint separate from the final source checkpoint created during the three-way comparison.
• The source trainer runs for 60 epochs.
• The multiplier produces a larger synthetic source dataset from the approved fold-training rows.
• The noise value controls the standard deviation of the seeded corruption.

• In src/finetune_experiment.py, locate the closing brace of assess_transfer_support().
• Add the three network classes below that function with this code:

class Encoder(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.layer1 = nn.Linear(input_dim, 16)
        self.layer2 = nn.Linear(16, 8)

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return torch.relu(self.layer2(torch.relu(self.layer1(inputs))))


class DenoisingAutoencoder(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.encoder = Encoder(input_dim)
        self.decoder = nn.Sequential(
            nn.Linear(8, 16),
            nn.ReLU(),
            nn.Linear(16, input_dim),
        )

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return self.decoder(self.encoder(inputs))


class TransferClassifier(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.encoder = Encoder(input_dim)
        self.head = nn.Linear(8, 1)

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return self.head(self.encoder(inputs)).squeeze(1)

How do the networks fit together?

• Encoder compresses each approved feature vector through layers with 16 units and 8 units.
• DenoisingAutoencoder asks the encoder to preserve enough structure for the decoder to reconstruct the clean input.
• TransferClassifier reuses the same encoder shape with a one-output classification head.
• The shared architecture lets later experiments isolate initialization and freezing strategy as the changing variables.

• Save src/finetune_experiment.py.
• Confirm that the expanded module still imports cleanly by running this assessment command:

python -m src.finetune_experiment assess

What does this check prove?

The assessment should still complete with level5_checkpoint_is_pretrained set to false. That result confirms the new classes load without weakening the earlier lineage rejection.

Module no longer imports?

Check that all three classes sit at the file's top level. Confirm that their indentation matches the code block.

Use the traceback line number to inspect the nearest class boundary. Help me diagnose an import failure after adding the Level 6 network classes.

Generate deterministic denoising data

The source task needs inputs that differ from its reconstruction targets. Seeded bootstrapping creates repeatable row selections. Seeded noise turns those selected clean rows into corrupted inputs.

• Below TransferClassifier, add the deterministic source-array generator with this code:

def source_arrays(x_train: np.ndarray, seed: int) -> tuple[np.ndarray, np.ndarray]:
    rng = np.random.default_rng(seed)
    count = max(len(x_train) * SOURCE_MULTIPLIER, 64)
    rows = rng.integers(0, len(x_train), size=count)
    clean = x_train[rows] + rng.normal(0.0, 0.03, size=(count, x_train.shape[1]))
    corrupted = clean + rng.normal(0.0, SOURCE_NOISE, size=clean.shape)
    return corrupted.astype(np.float32), clean.astype(np.float32)

What does this generator protect?

• The function receives an already transformed training array. It has no parameter for labels or validation rows.
• The bootstrap indices select only rows from that supplied training array.
• The clean output becomes the reconstruction target. The more heavily corrupted output becomes the model input.
• The seed makes the same source examples reproducible inside the pinned environment.

• Below source_arrays(), add the denoising training loop with this code:

def pretrain_source(
    x_train: np.ndarray,
    seed: int,
    epochs: int = SOURCE_EPOCHS,
) -> tuple[DenoisingAutoencoder, list[float], float]:
    seed_everything(seed)
    corrupted, clean = source_arrays(x_train, seed)
    tx = torch.tensor(corrupted, dtype=torch.float32)
    ty = torch.tensor(clean, dtype=torch.float32)
    model = DenoisingAutoencoder(x_train.shape[1])
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    history: list[float] = []
    started = time.perf_counter()
    for _ in range(epochs):
        model.train()
        optimizer.zero_grad(set_to_none=True)
        loss = (model(tx) - ty).pow(2).mean()
        loss.backward()
        optimizer.step()
        history.append(float(loss.item()))
    return model, history, time.perf_counter() - started

What does the source trainer learn?

• The trainer reseeds execution before generating data or initializing weights.
• The autoencoder receives corrupted feature vectors. Its loss measures reconstruction error against the clean vectors.
• The optimizer updates both the encoder and decoder for the fixed source epoch budget.
• The returned history makes the first and final reconstruction losses inspectable.

• Save src/finetune_experiment.py.
• Check that the source helpers preserve the existing assessment path by running:

python -m src.finetune_experiment assess

What should remain unchanged?

The assessment should still reject the Level 5 checkpoint as pretrained. The new source task has its own objective. It does not change the inherited checkpoint's lineage.

Assessment stopped after the edit?

Check that source_arrays() and pretrain_source() remain outside every class. A function accidentally nested inside a class can prevent later commands from finding it.

Confirm that time is imported at the top of the file. Help me fix the source-training helper definitions.

Save and verify the preview checkpoint

The preview command applies the leakage boundary before source training begins. It fits imputation and scaling on the first fold's outer-training runners. It transforms only those rows before synthetic bootstrapping.

• Below pretrain_source(), add the overwrite guard with this code:

def ensure_outputs_absent(root: Path, paths: tuple[Path, ...]) -> None:
    for relative in paths:
        if (root / relative).exists():
            raise FileExistsError(f"Refusing to overwrite {relative}")

Why refuse overwrites?

The guard keeps an existing evidence artifact immutable. A repeated experiment must start from a deliberate cleanup decision instead of silently replacing a checkpoint.

• Below ensure_outputs_absent(), add the checkpoint serializer with this code:

def save_source_checkpoint(path: Path, model: DenoisingAutoencoder, feature_names: list[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    torch.save(
        {
            "source_task": "denoise_corrupted_synthetic_feature_bootstraps",
            "pretraining_type": "self_supervised",
            "external_domain_pretraining": False,
            "target_labels_used": False,
            "encoder_state": model.encoder.state_dict(),
            "decoder_state": model.decoder.state_dict(),
            "input_dim": len(feature_names),
            "feature_names": feature_names,
            "seed": SEED,
            "epochs": SOURCE_EPOCHS,
            "noise_standard_deviation": SOURCE_NOISE,
        },
        path,
    )

What does the checkpoint prove?

• The metadata names the denoising objective as a separate source task.
• The source type is recorded as self-supervised. External-domain pretraining is explicitly excluded.
• The checkpoint records that target labels were not used.
• The encoder state and decoder state remain independently inspectable.
• The feature order and source recipe travel with the learned weights.

• Save src/finetune_experiment.py.
• Confirm that the serializer introduces no import errors by running:

python -m src.finetune_experiment assess

What does this regression check cover?

The assessment should complete with the same passed eligibility checks. That result confirms the checkpoint code loads before the new command invokes it.

• Below save_source_checkpoint(), add the preview workflow with this code:

def pretrain_preview(root: Path) -> dict[str, Any]:
    audit_level5(root)
    assessment = assess_transfer_support(root)
    if not assessment["controlled_transfer_supported"]:
        raise ValueError(assessment["fallback_proposal"])
    ensure_outputs_absent(root, (PREVIEW_SOURCE_PATH,))
    x, _, groups, folds, _ = inherited_data(root)
    fit_rows, validation_rows = folds[0]
    imputer, scaler = fit_preprocessor(x.iloc[fit_rows])
    x_fit = transform(x.iloc[fit_rows], imputer, scaler)
    model, history, seconds = pretrain_source(x_fit, SEED)
    save_source_checkpoint(root / PREVIEW_SOURCE_PATH, model, list(x.columns))
    overlap = set(groups.iloc[fit_rows]) & set(groups.iloc[validation_rows])
    return {
        "checkpoint": str(PREVIEW_SOURCE_PATH),
        "source_task": "denoise_corrupted_synthetic_feature_bootstraps",
        "pretraining_type": "self_supervised",
        "external_domain_pretraining": False,
        "target_labels_used": False,
        "source_validation_runner_overlap": sorted(str(value) for value in overlap),
        "first_reconstruction_loss": history[0],
        "last_reconstruction_loss": history[-1],
        "source_pretraining_seconds": seconds,
    }

Where is the leakage boundary enforced?

• The inherited audit and transfer assessment run before training.
• The first inherited fold supplies separate training and validation row indices.
• The imputer and scaler fit only on outer-training rows.
• Synthetic bootstrapping receives only the transformed outer-training array.
• The overlap check compares source-training runner groups with target-validation runner groups.

• In main(), locate the existing assess command branch.
• Add the new command branch directly below it with this code:

    elif command == "pretrain":
        payload = pretrain_preview(root)

How is the workflow exposed?

The branch routes the pretrain command to pretrain_preview(). The existing JSON printer then exposes the source-task metadata and boundary checks.

• Save src/finetune_experiment.py.

Before you run the command, do you expect the final reconstruction loss to be higher or lower than the first loss?

• Create the preview source checkpoint by running this command:

python -m src.finetune_experiment pretrain

What should the result show?

You should see JSON with target_labels_used set to false. The source_validation_runner_overlap array should be empty.

The final reconstruction loss should be below the first reconstruction loss. The output should point to artifacts/v4/level6_source_preview.pt.

You have now crossed the important line from same-task warm-starting to a separately defined source task. The encoder has learned reusable feature structure without seeing readiness labels or target-validation runners.

Preview command did not complete?

If the command refuses to overwrite the preview checkpoint, remove the existing preview only when you intentionally want a fresh run. The refusal protects an earlier artifact from silent replacement.

If the transfer assessment blocks pretraining, keep the printed all-synthetic fallback proposal. Do not bypass the gate or run the target comparison.

Use the traceback to identify whether the failure occurred during preprocessing, source-array generation, or checkpoint saving. Help me debug the Level 6 pretrain command.

✔️ Awesome, I've got everything!

Great work. Save src/finetune_experiment.py and keep the preview checkpoint for the comparison step.

ⓧ I'd like to double check the full code

from __future__ import annotations

import json
import sys
import time
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
import torch
from torch import nn

from src.feature_experiment import (
    FORBIDDEN_HOLDOUT_USES,
    LEVEL5_HANDOFF_PATH,
    TRAIN_PATH,
    V2_SCHEMA_PATH,
    fingerprint,
    fold_signature,
    read_json,
    write_json_exclusive,
)
from src.neural_network_experiment import (
    CHECKPOINT_PATH as LEVEL5_CHECKPOINT_PATH,
    LEVEL6_HANDOFF_PATH,
    PREPROCESSOR_PATH as LEVEL5_PREPROCESSOR_PATH,
    REPORT_PATH as LEVEL5_REPORT_PATH,
    SEED,
    THRESHOLD,
    approved_matrix,
    audit_level4,
    fit_preprocessor,
    seed_everything,
    transform,
)
from src.train_models import SCHEMA_PATH, make_group_folds

PREVIEW_SOURCE_PATH = Path("artifacts/v4/level6_source_preview.pt")
SOURCE_EPOCHS = 60
SOURCE_MULTIPLIER = 8
SOURCE_NOISE = 0.15


def verify_fingerprint(path: Path, expected: dict[str, Any]) -> None:
    observed = fingerprint(path)
    if observed.get("sha256") != expected.get("sha256"):
        raise ValueError(f"Fingerprint mismatch: {path}")


def load_level5_checkpoint(root: Path) -> dict[str, Any]:
    return torch.load(
        root / LEVEL5_CHECKPOINT_PATH,
        map_location="cpu",
        weights_only=True,
    )


def audit_level5(root: Path) -> dict[str, Any]:
    base = audit_level4(root)
    handoff = read_json(root / LEVEL6_HANDOFF_PATH)
    if handoff.get("handoff_version") != "3.0.0":
        raise ValueError("Expected Level 6 handoff version 3.0.0")
    if handoff.get("created_by") != "Level 5: Networks":
        raise ValueError("Unexpected Level 6 handoff creator")
    if handoff.get("next_level") != "Level 6: Fine-tune":
        raise ValueError("Handoff does not target Level 6: Fine-tune")
    if handoff.get("consumed_v1_holdout_read") is not False:
        raise ValueError("Level 5 handoff does not preserve zero holdout reads")
    if handoff.get("forbidden_uses") != list(FORBIDDEN_HOLDOUT_USES):
        raise ValueError("Consumed-holdout restrictions changed")
    paths = {
        "level4_handoff": LEVEL5_HANDOFF_PATH,
        "development_data": TRAIN_PATH,
        "feature_schema": V2_SCHEMA_PATH,
        "level5_report": LEVEL5_REPORT_PATH,
        "checkpoint": LEVEL5_CHECKPOINT_PATH,
        "preprocessor": LEVEL5_PREPROCESSOR_PATH,
    }
    for label, relative in paths.items():
        verify_fingerprint(root / relative, handoff["fingerprints"][label])
    checkpoint = load_level5_checkpoint(root)
    preprocessor = read_json(root / LEVEL5_PREPROCESSOR_PATH)
    if checkpoint["feature_names"] != handoff["approved_features"]:
        raise ValueError("Checkpoint features differ from the handoff")
    if preprocessor["feature_names"] != handoff["approved_features"]:
        raise ValueError("Preprocessor features differ from the handoff")
    for field in ("threshold", "seed", "best_epoch"):
        if checkpoint[field] != handoff[field]:
            raise ValueError(f"Checkpoint metadata mismatch: {field}")
    if checkpoint["config"] != handoff["selected_config"]:
        raise ValueError("Checkpoint configuration differs from the handoff")
    return {
        "status": "passed",
        "holdout_reads": base["holdout_reads"],
        "verified_fingerprints": sorted(paths),
        "checkpoint_device": "cpu",
        "approved_features": handoff["approved_features"],
    }


def inherited_data(root: Path):
    x, y, groups, _ = approved_matrix(root)
    frame = pd.read_csv(root / TRAIN_PATH)
    folds = make_group_folds(frame, read_json(root / SCHEMA_PATH))
    report = read_json(root / LEVEL5_REPORT_PATH)
    signatures = [fold_signature(groups, fit, valid) for fit, valid in folds]
    if signatures != report["fold_signatures"]:
        raise ValueError("Inherited fold signatures changed")
    return x, y, groups, folds, signatures


def assess_transfer_support(root: Path) -> dict[str, Any]:
    audit_level5(root)
    checkpoint = load_level5_checkpoint(root)
    x, y, groups, folds, signatures = inherited_data(root)
    reasons: list[str] = []
    if x.shape[1] < 2:
        reasons.append("Fewer than two approved numeric features")
    if len(folds) != 4 or len(signatures) != 4:
        reasons.append("Exactly four inherited folds are required")
    for number, (fit_rows, validation_rows) in enumerate(folds, start=1):
        fit_groups = set(groups.iloc[fit_rows])
        validation_groups = set(groups.iloc[validation_rows])
        if fit_groups & validation_groups:
            reasons.append(f"Fold {number} has runner overlap")
        if y.iloc[fit_rows].nunique() < 2 or y.iloc[validation_rows].nunique() < 2:
            reasons.append(f"Fold {number} lacks both target classes")
    has_source_lineage = "source_task" in checkpoint
    return {
        "level5_checkpoint_is_pretrained": has_source_lineage,
        "level5_checkpoint_reason": (
            "A source-task lineage exists." if has_source_lineage else
            "The checkpoint records only readiness-target training and no separate source task."
        ),
        "controlled_transfer_supported": not reasons,
        "blocking_reasons": reasons,
        "approved_feature_count": x.shape[1],
        "runner_count": int(groups.nunique()),
        "fold_count": len(folds),
        "fallback_proposal": (
            None if not reasons else
            "Run a separate all-synthetic denoising-source and shifted-classification-target exercise; do not compare against readiness labels."
        ),
    }


class Encoder(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.layer1 = nn.Linear(input_dim, 16)
        self.layer2 = nn.Linear(16, 8)

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return torch.relu(self.layer2(torch.relu(self.layer1(inputs))))


class DenoisingAutoencoder(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.encoder = Encoder(input_dim)
        self.decoder = nn.Sequential(
            nn.Linear(8, 16),
            nn.ReLU(),
            nn.Linear(16, input_dim),
        )

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return self.decoder(self.encoder(inputs))


class TransferClassifier(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.encoder = Encoder(input_dim)
        self.head = nn.Linear(8, 1)

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return self.head(self.encoder(inputs)).squeeze(1)


def source_arrays(x_train: np.ndarray, seed: int) -> tuple[np.ndarray, np.ndarray]:
    rng = np.random.default_rng(seed)
    count = max(len(x_train) * SOURCE_MULTIPLIER, 64)
    rows = rng.integers(0, len(x_train), size=count)
    clean = x_train[rows] + rng.normal(0.0, 0.03, size=(count, x_train.shape[1]))
    corrupted = clean + rng.normal(0.0, SOURCE_NOISE, size=clean.shape)
    return corrupted.astype(np.float32), clean.astype(np.float32)


def pretrain_source(
    x_train: np.ndarray,
    seed: int,
    epochs: int = SOURCE_EPOCHS,
) -> tuple[DenoisingAutoencoder, list[float], float]:
    seed_everything(seed)
    corrupted, clean = source_arrays(x_train, seed)
    tx = torch.tensor(corrupted, dtype=torch.float32)
    ty = torch.tensor(clean, dtype=torch.float32)
    model = DenoisingAutoencoder(x_train.shape[1])
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    history: list[float] = []
    started = time.perf_counter()
    for _ in range(epochs):
        model.train()
        optimizer.zero_grad(set_to_none=True)
        loss = (model(tx) - ty).pow(2).mean()
        loss.backward()
        optimizer.step()
        history.append(float(loss.item()))
    return model, history, time.perf_counter() - started


def ensure_outputs_absent(root: Path, paths: tuple[Path, ...]) -> None:
    for relative in paths:
        if (root / relative).exists():
            raise FileExistsError(f"Refusing to overwrite {relative}")


def save_source_checkpoint(path: Path, model: DenoisingAutoencoder, feature_names: list[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    torch.save(
        {
            "source_task": "denoise_corrupted_synthetic_feature_bootstraps",
            "pretraining_type": "self_supervised",
            "external_domain_pretraining": False,
            "target_labels_used": False,
            "encoder_state": model.encoder.state_dict(),
            "decoder_state": model.decoder.state_dict(),
            "input_dim": len(feature_names),
            "feature_names": feature_names,
            "seed": SEED,
            "epochs": SOURCE_EPOCHS,
            "noise_standard_deviation": SOURCE_NOISE,
        },
        path,
    )


def pretrain_preview(root: Path) -> dict[str, Any]:
    audit_level5(root)
    assessment = assess_transfer_support(root)
    if not assessment["controlled_transfer_supported"]:
        raise ValueError(assessment["fallback_proposal"])
    ensure_outputs_absent(root, (PREVIEW_SOURCE_PATH,))
    x, _, groups, folds, _ = inherited_data(root)
    fit_rows, validation_rows = folds[0]
    imputer, scaler = fit_preprocessor(x.iloc[fit_rows])
    x_fit = transform(x.iloc[fit_rows], imputer, scaler)
    model, history, seconds = pretrain_source(x_fit, SEED)
    save_source_checkpoint(root / PREVIEW_SOURCE_PATH, model, list(x.columns))
    overlap = set(groups.iloc[fit_rows]) & set(groups.iloc[validation_rows])
    return {
        "checkpoint": str(PREVIEW_SOURCE_PATH),
        "source_task": "denoise_corrupted_synthetic_feature_bootstraps",
        "pretraining_type": "self_supervised",
        "external_domain_pretraining": False,
        "target_labels_used": False,
        "source_validation_runner_overlap": sorted(str(value) for value in overlap),
        "first_reconstruction_loss": history[0],
        "last_reconstruction_loss": history[-1],
        "source_pretraining_seconds": seconds,
    }


def main() -> None:
    root = Path.cwd()
    command = sys.argv[1] if len(sys.argv) > 1 else "audit"
    if command == "audit":
        payload = audit_level5(root)
    elif command == "assess":
        payload = assess_transfer_support(root)
    elif command == "pretrain":
        payload = pretrain_preview(root)
    else:
        raise SystemExit("Use: audit | assess | pretrain | run | secret | document")
    print(json.dumps(payload, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

Use this reference to compare function names, metadata keys, constants, and command routing. Keep the existing audited code unchanged outside the additions from this step.

Your repository now has a label-free pretrained encoder with inspectable source-task lineage. Next up, you will compare random initialization, frozen transfer, and partial fine-tuning on the same inherited folds.

Compare Transfer Approaches

Your self-supervised encoder now learns a separate denoising source task without target labels or validation runners. The next question is whether those learned representations help the readiness classifier.

A fair transfer learning comparison must isolate the initialization strategy as the experimental variable. Every approach therefore uses the same unseen runners, architecture, loss, threshold, seed policy, head learning rate, preprocessing boundaries, and downstream epoch budget.

In this step, get ready to:

• Define identical downstream training rules for scratch, frozen, and partial approaches.
• Evaluate all three approaches across four inherited runner-safe folds.
• Produce the selected development model with its fingerprinted Level 7 handoff.

Add the fair comparison runner

The three arms answer different questions. Scratch training tests random initialization. Frozen feature extraction tests the pretrained representation without changing it. Partial fine-tuning preserves the first encoder layer while adapting the second.

What makes the comparison fair?

• Every arm trains the same target architecture for exactly 80 downstream epochs.
• Every arm uses a head learning rate of 0.01.
• Scratch and partial fine-tuning use an encoder learning rate of 0.001.
• Frozen transfer excludes the encoder from its torch.optim.Adam parameter groups.

The fold boundary also applies to pretraining. Each source encoder learns from synthetic variants of that fold's outer-training rows before any target-validation row is transformed or evaluated.

• Return to src/finetune_experiment.py in VS Code.
• Synchronize the file with the complete reference in the second tab below.
• Keep the existing pretrain_preview() implementation unchanged.

✔️ Awesome, I've got everything!

Great. Save src/finetune_experiment.py before running the comparison.

ⓧ I'd like to double check the full code

from __future__ import annotations

import copy
import hashlib
import json
import sys
import time
from pathlib import Path
from statistics import fmean, pstdev
from typing import Any

import numpy as np
import pandas as pd
import torch
from torch import nn

from src.feature_experiment import (
    FORBIDDEN_HOLDOUT_USES,
    LEVEL5_HANDOFF_PATH,
    TRAIN_PATH,
    V2_SCHEMA_PATH,
    fingerprint,
    fold_signature,
    read_json,
    write_json_exclusive,
)
from src.neural_network_experiment import (
    CHECKPOINT_PATH as LEVEL5_CHECKPOINT_PATH,
    LEVEL6_HANDOFF_PATH,
    PREPROCESSOR_PATH as LEVEL5_PREPROCESSOR_PATH,
    REPORT_PATH as LEVEL5_REPORT_PATH,
    SEED,
    THRESHOLD,
    approved_matrix,
    audit_level4,
    evaluate_model,
    fit_preprocessor,
    seed_everything,
    serialize_preprocessor,
    transform,
)
from src.train_models import SCHEMA_PATH, make_group_folds

REPORT_PATH = Path("reports/v4/finetune_experiment.json")
SECRET_REPORT_PATH = Path("reports/v4/catastrophic_forgetting.json")
PREVIEW_SOURCE_PATH = Path("artifacts/v4/level6_source_preview.pt")
SOURCE_CHECKPOINT_PATH = Path("artifacts/v4/level6_source_pretrained.pt")
SELECTED_CHECKPOINT_PATH = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_PATH = Path("artifacts/v4/level6_preprocessor.json")
LEVEL7_HANDOFF_PATH = Path("artifacts/v4/level7_deploy_handoff.json")
DOC_PATH = Path("docs/level-6-finetune.md")
APPROACHES = ("scratch", "frozen", "partial")
SOURCE_EPOCHS = 60
SOURCE_MULTIPLIER = 8
SOURCE_NOISE = 0.15
TARGET_EPOCHS = 80
ENCODER_LEARNING_RATE = 0.001
HEAD_LEARNING_RATE = 0.01


def verify_fingerprint(path: Path, expected: dict[str, Any]) -> None:
    observed = fingerprint(path)
    if observed.get("sha256") != expected.get("sha256"):
        raise ValueError(f"Fingerprint mismatch: {path}")


def load_level5_checkpoint(root: Path) -> dict[str, Any]:
    return torch.load(
        root / LEVEL5_CHECKPOINT_PATH,
        map_location="cpu",
        weights_only=True,
    )


def audit_level5(root: Path) -> dict[str, Any]:
    base = audit_level4(root)
    handoff = read_json(root / LEVEL6_HANDOFF_PATH)
    if handoff.get("handoff_version") != "3.0.0":
        raise ValueError("Expected Level 6 handoff version 3.0.0")
    if handoff.get("created_by") != "Level 5: Networks":
        raise ValueError("Unexpected Level 6 handoff creator")
    if handoff.get("next_level") != "Level 6: Fine-tune":
        raise ValueError("Handoff does not target Level 6: Fine-tune")
    if handoff.get("consumed_v1_holdout_read") is not False:
        raise ValueError("Level 5 handoff does not preserve zero holdout reads")
    if handoff.get("forbidden_uses") != list(FORBIDDEN_HOLDOUT_USES):
        raise ValueError("Consumed-holdout restrictions changed")
    paths = {
        "level4_handoff": LEVEL5_HANDOFF_PATH,
        "development_data": TRAIN_PATH,
        "feature_schema": V2_SCHEMA_PATH,
        "level5_report": LEVEL5_REPORT_PATH,
        "checkpoint": LEVEL5_CHECKPOINT_PATH,
        "preprocessor": LEVEL5_PREPROCESSOR_PATH,
    }
    for label, relative in paths.items():
        verify_fingerprint(root / relative, handoff["fingerprints"][label])
    checkpoint = load_level5_checkpoint(root)
    preprocessor = read_json(root / LEVEL5_PREPROCESSOR_PATH)
    if checkpoint["feature_names"] != handoff["approved_features"]:
        raise ValueError("Checkpoint features differ from the handoff")
    if preprocessor["feature_names"] != handoff["approved_features"]:
        raise ValueError("Preprocessor features differ from the handoff")
    for field in ("threshold", "seed", "best_epoch"):
        if checkpoint[field] != handoff[field]:
            raise ValueError(f"Checkpoint metadata mismatch: {field}")
    if checkpoint["config"] != handoff["selected_config"]:
        raise ValueError("Checkpoint configuration differs from the handoff")
    return {
        "status": "passed",
        "holdout_reads": base["holdout_reads"],
        "verified_fingerprints": sorted(paths),
        "checkpoint_device": "cpu",
        "approved_features": handoff["approved_features"],
    }


def inherited_data(root: Path):
    x, y, groups, _ = approved_matrix(root)
    frame = pd.read_csv(root / TRAIN_PATH)
    folds = make_group_folds(frame, read_json(root / SCHEMA_PATH))
    report = read_json(root / LEVEL5_REPORT_PATH)
    signatures = [fold_signature(groups, fit, valid) for fit, valid in folds]
    if signatures != report["fold_signatures"]:
        raise ValueError("Inherited fold signatures changed")
    return x, y, groups, folds, signatures


def assess_transfer_support(root: Path) -> dict[str, Any]:
    audit_level5(root)
    checkpoint = load_level5_checkpoint(root)
    x, y, groups, folds, signatures = inherited_data(root)
    reasons: list[str] = []
    if x.shape[1] < 2:
        reasons.append("Fewer than two approved numeric features")
    if len(folds) != 4 or len(signatures) != 4:
        reasons.append("Exactly four inherited folds are required")
    for number, (fit_rows, validation_rows) in enumerate(folds, start=1):
        fit_groups = set(groups.iloc[fit_rows])
        validation_groups = set(groups.iloc[validation_rows])
        if fit_groups & validation_groups:
            reasons.append(f"Fold {number} has runner overlap")
        if y.iloc[fit_rows].nunique() < 2 or y.iloc[validation_rows].nunique() < 2:
            reasons.append(f"Fold {number} lacks both target classes")
    has_source_lineage = "source_task" in checkpoint
    return {
        "level5_checkpoint_is_pretrained": has_source_lineage,
        "level5_checkpoint_reason": (
            "A source-task lineage exists." if has_source_lineage else
            "The checkpoint records only readiness-target training and no separate source task."
        ),
        "controlled_transfer_supported": not reasons,
        "blocking_reasons": reasons,
        "approved_feature_count": x.shape[1],
        "runner_count": int(groups.nunique()),
        "fold_count": len(folds),
        "fallback_proposal": (
            None if not reasons else
            "Run a separate all-synthetic denoising-source and shifted-classification-target exercise; do not compare against readiness labels."
        ),
    }


class Encoder(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.layer1 = nn.Linear(input_dim, 16)
        self.layer2 = nn.Linear(16, 8)

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return torch.relu(self.layer2(torch.relu(self.layer1(inputs))))


class DenoisingAutoencoder(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.encoder = Encoder(input_dim)
        self.decoder = nn.Sequential(
            nn.Linear(8, 16),
            nn.ReLU(),
            nn.Linear(16, input_dim),
        )

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return self.decoder(self.encoder(inputs))


class TransferClassifier(nn.Module):
    def __init__(self, input_dim: int):
        super().__init__()
        self.encoder = Encoder(input_dim)
        self.head = nn.Linear(8, 1)

    def forward(self, inputs: torch.Tensor) -> torch.Tensor:
        return self.head(self.encoder(inputs)).squeeze(1)


def source_arrays(x_train: np.ndarray, seed: int) -> tuple[np.ndarray, np.ndarray]:
    rng = np.random.default_rng(seed)
    count = max(len(x_train) * SOURCE_MULTIPLIER, 64)
    rows = rng.integers(0, len(x_train), size=count)
    clean = x_train[rows] + rng.normal(0.0, 0.03, size=(count, x_train.shape[1]))
    corrupted = clean + rng.normal(0.0, SOURCE_NOISE, size=clean.shape)
    return corrupted.astype(np.float32), clean.astype(np.float32)


def pretrain_source(
    x_train: np.ndarray,
    seed: int,
    epochs: int = SOURCE_EPOCHS,
) -> tuple[DenoisingAutoencoder, list[float], float]:
    seed_everything(seed)
    corrupted, clean = source_arrays(x_train, seed)
    tx = torch.tensor(corrupted, dtype=torch.float32)
    ty = torch.tensor(clean, dtype=torch.float32)
    model = DenoisingAutoencoder(x_train.shape[1])
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    history: list[float] = []
    started = time.perf_counter()
    for _ in range(epochs):
        model.train()
        optimizer.zero_grad(set_to_none=True)
        loss = (model(tx) - ty).pow(2).mean()
        loss.backward()
        optimizer.step()
        history.append(float(loss.item()))
    return model, history, time.perf_counter() - started


def build_target(
    input_dim: int,
    approach: str,
    seed: int,
    encoder_state: dict[str, torch.Tensor] | None,
) -> TransferClassifier:
    seed_everything(seed)
    model = TransferClassifier(input_dim)
    if approach != "scratch":
        if encoder_state is None:
            raise ValueError("Transferred approaches require source weights")
        model.encoder.load_state_dict(encoder_state)
    if approach == "frozen":
        for parameter in model.encoder.parameters():
            parameter.requires_grad = False
    elif approach == "partial":
        for parameter in model.encoder.layer1.parameters():
            parameter.requires_grad = False
    elif approach != "scratch":
        raise ValueError(f"Unknown approach: {approach}")
    return model


def optimizer_for(model: TransferClassifier, approach: str) -> torch.optim.Adam:
    groups: list[dict[str, Any]] = [
        {"params": model.head.parameters(), "lr": HEAD_LEARNING_RATE}
    ]
    if approach == "scratch":
        groups.insert(
            0,
            {"params": model.encoder.parameters(), "lr": ENCODER_LEARNING_RATE},
        )
    elif approach == "partial":
        groups.insert(
            0,
            {"params": model.encoder.layer2.parameters(), "lr": ENCODER_LEARNING_RATE},
        )
    elif approach != "frozen":
        raise ValueError(f"Unknown approach: {approach}")
    return torch.optim.Adam(groups)


def target_loss(model: TransferClassifier, x: torch.Tensor, y: torch.Tensor) -> float:
    model.eval()
    with torch.no_grad():
        return float(nn.BCEWithLogitsLoss()(model(x), y).item())


def train_target(
    x_fit: np.ndarray,
    y_fit: np.ndarray,
    approach: str,
    seed: int,
    encoder_state: dict[str, torch.Tensor] | None,
    epochs: int = TARGET_EPOCHS,
) -> tuple[TransferClassifier, list[dict[str, float]], int, float]:
    model = build_target(x_fit.shape[1], approach, seed, encoder_state)
    optimizer = optimizer_for(model, approach)
    criterion = nn.BCEWithLogitsLoss()
    tx = torch.tensor(x_fit, dtype=torch.float32)
    ty = torch.tensor(y_fit, dtype=torch.float32)
    history: list[dict[str, float]] = []
    started = time.perf_counter()
    for epoch in range(1, epochs + 1):
        model.train()
        optimizer.zero_grad(set_to_none=True)
        loss = criterion(model(tx), ty)
        loss.backward()
        optimizer.step()
        history.append(
            {"epoch": float(epoch), "train_loss": target_loss(model, tx, ty)}
        )
    return model, history, epochs, time.perf_counter() - started


def gap_values(train: dict[str, float], validation: dict[str, float]) -> dict[str, float]:
    return {
        metric: (
            validation[metric] - train[metric]
            if metric == "loss"
            else train[metric] - validation[metric]
        )
        for metric in train
    }


def evaluate_fold(root: Path, fold_number: int, fit_rows, validation_rows) -> dict[str, Any]:
    x, y, groups, _, _ = inherited_data(root)
    fit_groups = set(groups.iloc[fit_rows])
    validation_groups = set(groups.iloc[validation_rows])
    if fit_groups & validation_groups:
        raise ValueError("Outer runner leakage detected")
    imputer, scaler = fit_preprocessor(x.iloc[fit_rows])
    x_fit = transform(x.iloc[fit_rows], imputer, scaler)
    x_validation = transform(x.iloc[validation_rows], imputer, scaler)
    source, source_history, source_seconds = pretrain_source(
        x_fit, SEED + 200 + fold_number
    )
    results: dict[str, Any] = {}
    for approach in APPROACHES:
        model, _, downstream_epochs, downstream_seconds = train_target(
            x_fit,
            y.iloc[fit_rows].to_numpy(np.float32),
            approach,
            SEED + fold_number,
            None if approach == "scratch" else source.encoder.state_dict(),
            epochs=TARGET_EPOCHS,
        )
        train_metrics = evaluate_model(
            model, x_fit, y.iloc[fit_rows].to_numpy(np.float32)
        )
        validation_metrics = evaluate_model(
            model,
            x_validation,
            y.iloc[validation_rows].to_numpy(np.float32),
        )
        source_time = 0.0 if approach == "scratch" else source_seconds
        results[approach] = {
            "downstream_epochs": downstream_epochs,
            "train": train_metrics,
            "validation": validation_metrics,
            "generalization_gap": gap_values(train_metrics, validation_metrics),
            "source_pretraining_seconds": source_time,
            "downstream_training_seconds": downstream_seconds,
            "total_training_seconds": source_time + downstream_seconds,
        }
    return {
        "fold": fold_number,
        "fold_signature": fold_signature(groups, fit_rows, validation_rows),
        "source_training_runners": sorted(str(value) for value in fit_groups),
        "target_validation_runners": sorted(str(value) for value in validation_groups),
        "source_validation_runner_overlap": [],
        "source_target_labels_used": False,
        "source_reconstruction_loss": {
            "first": source_history[0],
            "last": source_history[-1],
        },
        "approaches": results,
    }


def summarize(folds: list[dict[str, Any]], approach: str) -> dict[str, Any]:
    metrics = ("accuracy", "precision", "recall", "f1", "loss")
    return {
        "validation_mean": {
            metric: fmean(row["approaches"][approach]["validation"][metric] for row in folds)
            for metric in metrics
        },
        "validation_stdev": {
            metric: pstdev(row["approaches"][approach]["validation"][metric] for row in folds)
            for metric in metrics
        },
        "gap_mean": {
            metric: fmean(row["approaches"][approach]["generalization_gap"][metric] for row in folds)
            for metric in metrics
        },
        "source_pretraining_seconds_mean": fmean(
            row["approaches"][approach]["source_pretraining_seconds"] for row in folds
        ),
        "downstream_training_seconds_mean": fmean(
            row["approaches"][approach]["downstream_training_seconds"] for row in folds
        ),
        "total_training_seconds_mean": fmean(
            row["approaches"][approach]["total_training_seconds"] for row in folds
        ),
    }


def ensure_outputs_absent(root: Path, paths: tuple[Path, ...]) -> None:
    for relative in paths:
        if (root / relative).exists():
            raise FileExistsError(f"Refusing to overwrite {relative}")


def save_source_checkpoint(path: Path, model: DenoisingAutoencoder, feature_names: list[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    torch.save(
        {
            "source_task": "denoise_corrupted_synthetic_feature_bootstraps",
            "pretraining_type": "self_supervised",
            "external_domain_pretraining": False,
            "target_labels_used": False,
            "encoder_state": model.encoder.state_dict(),
            "decoder_state": model.decoder.state_dict(),
            "input_dim": len(feature_names),
            "feature_names": feature_names,
            "seed": SEED,
            "epochs": SOURCE_EPOCHS,
            "noise_standard_deviation": SOURCE_NOISE,
        },
        path,
    )


def pretrain_preview(root: Path) -> dict[str, Any]:
    audit_level5(root)
    assessment = assess_transfer_support(root)
    if not assessment["controlled_transfer_supported"]:
        raise ValueError(assessment["fallback_proposal"])
    ensure_outputs_absent(root, (PREVIEW_SOURCE_PATH,))
    x, _, groups, folds, _ = inherited_data(root)
    fit_rows, validation_rows = folds[0]
    imputer, scaler = fit_preprocessor(x.iloc[fit_rows])
    x_fit = transform(x.iloc[fit_rows], imputer, scaler)
    model, history, seconds = pretrain_source(x_fit, SEED)
    save_source_checkpoint(root / PREVIEW_SOURCE_PATH, model, list(x.columns))
    overlap = set(groups.iloc[fit_rows]) & set(groups.iloc[validation_rows])
    return {
        "checkpoint": str(PREVIEW_SOURCE_PATH),
        "source_task": "denoise_corrupted_synthetic_feature_bootstraps",
        "pretraining_type": "self_supervised",
        "external_domain_pretraining": False,
        "target_labels_used": False,
        "source_validation_runner_overlap": sorted(str(value) for value in overlap),
        "first_reconstruction_loss": history[0],
        "last_reconstruction_loss": history[-1],
        "source_pretraining_seconds": seconds,
    }


def run_experiment(root: Path) -> dict[str, Any]:
    audit = audit_level5(root)
    assessment = assess_transfer_support(root)
    if not assessment["controlled_transfer_supported"]:
        raise ValueError(assessment["fallback_proposal"])
    outputs = (
        REPORT_PATH,
        SOURCE_CHECKPOINT_PATH,
        SELECTED_CHECKPOINT_PATH,
        PREPROCESSOR_PATH,
        LEVEL7_HANDOFF_PATH,
    )
    ensure_outputs_absent(root, outputs)
    x, y, _, folds, signatures = inherited_data(root)
    records = [
        evaluate_fold(root, number, fit_rows, validation_rows)
        for number, (fit_rows, validation_rows) in enumerate(folds, start=1)
    ]
    aggregates = {approach: summarize(records, approach) for approach in APPROACHES}
    selected = max(
        APPROACHES,
        key=lambda name: (
            aggregates[name]["validation_mean"]["f1"],
            -abs(aggregates[name]["gap_mean"]["f1"]),
            -aggregates[name]["downstream_training_seconds_mean"],
        ),
    )
    imputer, scaler = fit_preprocessor(x)
    x_all = transform(x, imputer, scaler)
    source, source_history, source_seconds = pretrain_source(x_all, SEED + 400)
    final_model, _, downstream_epochs, downstream_seconds = train_target(
        x_all,
        y.to_numpy(np.float32),
        selected,
        SEED,
        None if selected == "scratch" else source.encoder.state_dict(),
        epochs=TARGET_EPOCHS,
    )
    save_source_checkpoint(root / SOURCE_CHECKPOINT_PATH, source, list(x.columns))
    torch.save(
        {
            "model_state": final_model.state_dict(),
            "model_class": "TransferClassifier",
            "input_dim": x.shape[1],
            "feature_names": list(x.columns),
            "approach": selected,
            "threshold": THRESHOLD,
            "seed": SEED,
            "downstream_epochs": downstream_epochs,
            "source_task": None if selected == "scratch" else "denoise_corrupted_synthetic_feature_bootstraps",
        },
        root / SELECTED_CHECKPOINT_PATH,
    )
    write_json_exclusive(
        root / PREPROCESSOR_PATH,
        serialize_preprocessor(imputer, scaler, list(x.columns)),
    )
    report = {
        "report_version": "4.0.0",
        "evidence_scope": "development_only",
        "consumed_v1_holdout_read": False,
        "inherited_v1_preprocessing_limitation": (
            "The historical v1 process used development-wide mean imputation before splitting; Level 6 does not claim the full inherited history is leakage-free."
        ),
        "audit": audit,
        "assessment": assessment,
        "fair_comparison_contract": {
            "identical_fold_signatures": True,
            "identical_downstream_epochs": TARGET_EPOCHS,
            "identical_architecture": True,
            "identical_head_learning_rate": HEAD_LEARNING_RATE,
            "scratch_and_partial_encoder_learning_rate": ENCODER_LEARNING_RATE,
            "validation_used_for_fitting": False,
        },
        "source_task": {
            "name": "denoise_corrupted_synthetic_feature_bootstraps",
            "pretraining_type": "self_supervised",
            "external_domain_pretraining": False,
            "target": "clean standardized approved feature vector",
            "target_labels_used": False,
            "synthetic_multiplier": SOURCE_MULTIPLIER,
            "noise_standard_deviation": SOURCE_NOISE,
        },
        "target_task": "binary running readiness classification",
        "fold_signatures": signatures,
        "folds": records,
        "aggregates": aggregates,
        "selected_approach": selected,
        "final_development_fit": {
            "downstream_epochs": downstream_epochs,
            "source_first_loss": source_history[0],
            "source_last_loss": source_history[-1],
            "source_pretraining_seconds": source_seconds,
            "downstream_training_seconds": downstream_seconds,
            "train_metrics": evaluate_model(final_model, x_all, y.to_numpy(np.float32)),
        },
        "determinism": {
            "seed": SEED,
            "torch_use_deterministic_algorithms": True,
            "device": "cpu",
            "cross_platform_bitwise_guarantee": False,
        },
    }
    write_json_exclusive(root / REPORT_PATH, report)
    handoff = {
        "handoff_version": "4.0.0",
        "created_by": "Level 6: Fine-tune",
        "next_level": "Level 7: Deploy",
        "evidence_scope": "development_only",
        "consumed_v1_holdout_read": False,
        "forbidden_uses": list(FORBIDDEN_HOLDOUT_USES),
        "selected_approach": selected,
        "feature_names": list(x.columns),
        "threshold": THRESHOLD,
        "seed": SEED,
        "downstream_epochs": TARGET_EPOCHS,
        "limitations": [
            report["inherited_v1_preprocessing_limitation"],
            "The source task is controlled self-supervised synthetic denoising, not external-domain pretraining.",
            "Training times vary with host load even when model states and metrics reproduce in the same pinned environment.",
        ],
        "fingerprints": {
            "level6_input_handoff": fingerprint(root / LEVEL6_HANDOFF_PATH),
            "development_data": fingerprint(root / TRAIN_PATH),
            "level6_report": fingerprint(root / REPORT_PATH),
            "source_checkpoint": fingerprint(root / SOURCE_CHECKPOINT_PATH),
            "selected_checkpoint": fingerprint(root / SELECTED_CHECKPOINT_PATH),
            "preprocessor": fingerprint(root / PREPROCESSOR_PATH),
        },
    }
    write_json_exclusive(root / LEVEL7_HANDOFF_PATH, handoff)
    return {
        "selected_approach": selected,
        "report": str(REPORT_PATH),
        "checkpoint": str(SELECTED_CHECKPOINT_PATH),
        "handoff": str(LEVEL7_HANDOFF_PATH),
    }


def main() -> None:
    root = Path.cwd()
    command = sys.argv[1] if len(sys.argv) > 1 else "audit"
    if command == "audit":
        payload = audit_level5(root)
    elif command == "assess":
        payload = assess_transfer_support(root)
    elif command == "pretrain":
        payload = pretrain_preview(root)
    elif command == "run":
        payload = run_experiment(root)
    else:
        raise SystemExit("Use: audit | assess | pretrain | run | secret | document")
    print(json.dumps(payload, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

How does the runner work?

The new target helpers create identical classifiers while changing which encoder parameters can learn. Separate optimizer groups make the head rate and encoder rate inspectable.

Each fold fits its imputer and scaler on outer-training runners. The fold then creates its own source encoder before evaluating all three target arms against untouched validation runners.

The selection rule ranks mean validation F1 first. A smaller absolute F1 gap breaks a tie before lower downstream-training time becomes the final criterion.

• Save src/finetune_experiment.py.
• Confirm that your editor reports no syntax errors in the file.

Seeing a syntax error?

Check that build_target() begins after pretrain_source() has ended. Confirm that run_experiment() remains above main().

Compare the indentation around the report and handoff dictionaries with the full reference.

Help me find a syntax or indentation error in my comparison runner.

Run the comparison and inspect the evidence

This command trains a source encoder inside each fold before running all three downstream approaches. It then refits the selected approach on all development data without reading the consumed v1 holdout.

Expect the terminal to stay busy while the CPU completes the four-fold comparison. Host load affects wall-clock timing even when the model states and metrics reproduce inside the pinned environment.

Before you run it, which initialization strategy do you expect to achieve the strongest mean outer-fold F1?

• Start the development-only comparison by running:

python -m src.finetune_experiment run

The final JSON names the selected approach. It also prints the report, selected checkpoint, and Level 7 handoff paths.

What did the command create?

• The report stores fold-level F1, precision, recall, loss, generalization gaps, and timing categories.
• The source checkpoint stores the final all-development denoising encoder with its decoder and source-task metadata.
• The selected checkpoint stores the refitted TransferClassifier with its approach, feature order, threshold, seed, and 80-epoch budget.
• The Level 7 handoff fingerprints every model-serving dependency needed by the next project level.

• Return to the repository file tree in VS Code.
• Open reports/v4/finetune_experiment.json.
• Confirm that fold_signatures contains four entries.
• Confirm that identical_downstream_epochs is 80.
• Confirm that every source_validation_runner_overlap value is empty.
• Confirm that consumed_v1_holdout_read is false.
• Confirm that artifacts/v4/level7_deploy_handoff.json contains six dependency fingerprints.

Did the run stop early?

A refusal to overwrite an output means a previous comparison already created one of the protected v4 artifacts. Preserve those files if they are evidence from a completed run.

A fingerprint failure means an inherited Level 5 dependency changed after its handoff was created. Restore the immutable Level 5 artifact before rerunning this comparison.

Help me diagnose why the Level 6 comparison stopped before creating its report.

That is the full comparison complete. You now have runner-safe development evidence, a refitted target model, and a fingerprinted deployment handoff ready for executable safeguard tests.

Test Safeguards and Publish Evidence

Your four-fold comparison now has a selected approach and a fingerprinted deployment handoff. Those artifacts support a portfolio claim only when executable checks protect the boundaries behind them.

This step turns the critical assumptions into five Python unittest safeguards. You will also publish an evidence page that explains what the PyTorch experiment proves.

In this step, get ready to:

• Create five automated tests for checkpoint safety, frozen weights, grouped splits, artifact integrity, and deterministic target training.
• Add a report-driven documentation generator to the Level 6 experiment module.
• Run the safeguards and publish the Level 6 evidence page through the existing CPU container.

Build the executable safeguards

Each safeguard targets a claim that metrics alone cannot prove. Together they check safe checkpoint loading, frozen encoder invariance, runner isolation, fingerprint enforcement, and same-seed reproducibility.

• Switch back to the Level 5 repository in VS Code.
• Select the tests folder in the Explorer sidebar.
• Create a file named test_finetune.py inside the tests folder.
• Add the test dependencies and Level 6 functions by pasting this import group:

import copy
import tempfile
import unittest
from pathlib import Path

import numpy as np
import pandas as pd
import torch

from src.feature_experiment import fingerprint
from src.finetune_experiment import (
    load_level5_checkpoint,
    pretrain_source,
    train_target,
    verify_fingerprint,
)
from src.neural_network_experiment import inner_group_split

What do these imports support?

• The standard-library imports support state copies, temporary artifact mutation, test discovery, and repository-relative paths.
• NumPy and pandas create a small deterministic dataset for isolated safeguards.
• The project imports exercise the same loading, pretraining, target-training, fingerprint, and grouped-split functions used by the workflow.

• Save tests/test_finetune.py.
• Confirm that test_finetune.py appears beneath the tests folder in the Explorer sidebar.

Imports showing as unresolved?

Confirm that test_finetune.py is inside the repository's existing tests folder. The Docker test run uses the repository as its working directory.

If the file is in the correct location, help me diagnose unresolved imports in this Level 6 test file.

• Define the reusable feature matrix and binary labels by adding this class setup below the imports:

class FineTuneTests(unittest.TestCase):
    def setUp(self):
        self.x = np.array(
            [
                [-2.0, -1.0, 0.0],
                [-1.5, -0.5, 0.2],
                [-1.0, -1.5, 0.1],
                [-0.5, -2.0, 0.3],
                [0.5, 2.0, 0.7],
                [1.0, 1.5, 0.8],
                [1.5, 0.5, 0.9],
                [2.0, 1.0, 1.0],
            ],
            dtype=np.float32,
        )
        self.y = np.array([0, 0, 0, 0, 1, 1, 1, 1], dtype=np.float32)

Why use a fixed test fixture?

The fixture gives every model safeguard the same eight rows and three features. Its balanced labels make the tests small enough for CPU execution while still exercising binary training behavior.

• Save tests/test_finetune.py.
• Confirm that the class contains eight feature rows and eight corresponding labels.

• Add the restricted checkpoint-loading test inside FineTuneTests below setUp() by pasting:

    def test_level5_checkpoint_loads_on_cpu(self):
        checkpoint = load_level5_checkpoint(Path.cwd())
        self.assertIn("model_state", checkpoint)
        self.assertTrue(all(tensor.device.type == "cpu" for tensor in checkpoint["model_state"].values()))

What does this safeguard prove?

The test loads the inherited checkpoint through load_level5_checkpoint(). It verifies that the model state exists and every loaded tensor resides on the CPU.

• Save tests/test_finetune.py.

Before you run this first safeguard, do you expect the inherited checkpoint to satisfy both the restricted loading policy and the CPU device check?

• Run the current test file through the inherited container by using this command:

docker run --rm --platform linux/amd64 --mount type=bind,src="$(pwd)",dst=/workspace -w /workspace running-readiness-level6:2.15.0 python -m unittest -v tests/test_finetune.py

What should you see?

The checkpoint-loading test passes. This confirms that the Level 5 checkpoint can be inspected on the CPU without changing its tensors.

Checkpoint test failing?

Confirm that the terminal is still positioned in the completed repository. The bind mount resolves Path.cwd() against that location.

If the checkpoint exists but the test still fails, help me diagnose the restricted CPU checkpoint-loading safeguard.

• Add the frozen-transfer invariance test below the checkpoint-loading test by pasting:

    def test_frozen_encoder_does_not_change(self):
        source, _, _ = pretrain_source(self.x, 42, epochs=5)
        state = copy.deepcopy(source.encoder.state_dict())
        model, _, _, _ = train_target(
            self.x,
            self.y,
            "frozen",
            42,
            state,
            epochs=5,
        )
        for name, tensor in state.items():
            self.assertTrue(torch.equal(tensor, model.encoder.state_dict()[name]))
        self.assertTrue(all(not parameter.requires_grad for parameter in model.encoder.parameters()))

How does this test protect frozen transfer?

• A deep copy preserves the pretrained encoder state before downstream training begins.
• The test compares every original tensor with the corresponding post-training encoder tensor.
• The final assertion confirms that every encoder parameter has gradients disabled.

• Save tests/test_finetune.py.

• Add the grouped-split safeguard below the frozen-transfer test by pasting:

    def test_group_separation(self):
        frame = pd.DataFrame(self.x, columns=["a", "b", "c"])
        target = pd.Series(self.y)
        groups = pd.Series(["r1", "r1", "r2", "r2", "r3", "r3", "r4", "r4"])
        fit, stop = inner_group_split(frame, target, groups, 42)
        self.assertFalse(set(groups.iloc[fit]) & set(groups.iloc[stop]))

Why test the inner split separately?

The comparison report records runner-safe outer folds. This unit test protects the inner grouped split by asserting that no runner identifier appears in both partitions.

• Save tests/test_finetune.py.

Before you rerun the suite, do you expect downstream training to preserve every frozen encoder tensor while the grouped split keeps all runner identifiers separate?

• Run the expanded safeguard suite through the CPU container by using:

docker run --rm --platform linux/amd64 --mount type=bind,src="$(pwd)",dst=/workspace -w /workspace running-readiness-level6:2.15.0 python -m unittest -v tests/test_finetune.py

What has the suite proved?

You should now see the checkpoint, frozen-encoder, and group-separation tests pass. The result turns three experimental assumptions into executable checks.

Frozen or grouped safeguard failing?

For a frozen-state failure, confirm that the approach value is exactly frozen and that the copied state enters train_target().

For a runner-overlap failure, inspect the four repeated group labels in the test fixture. If the failure persists, help me debug the frozen encoder or grouped split test.

• Add the fingerprint-mutation test below the group-separation test by pasting:

    def test_fingerprint_verification_rejects_changes(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "artifact.txt"
            path.write_text("trusted", encoding="utf-8")
            expected = fingerprint(path)
            verify_fingerprint(path, expected)
            path.write_text("changed", encoding="utf-8")
            with self.assertRaises(ValueError):
                verify_fingerprint(path, expected)

How does the mutation test work?

The test fingerprints a temporary trusted artifact before changing its contents. It passes only when verify_fingerprint() rejects the modified artifact with a ValueError.

• Save tests/test_finetune.py.

• Add the same-seed reproducibility test below the fingerprint test by pasting:

    def test_same_seed_reproduces_history_and_state(self):
        first, history_a, epoch_a, _ = train_target(
            self.x,
            self.y,
            "scratch",
            42,
            None,
            epochs=12,
        )
        second, history_b, epoch_b, _ = train_target(
            self.x,
            self.y,
            "scratch",
            42,
            None,
            epochs=12,
        )
        self.assertEqual(epoch_a, epoch_b)
        self.assertEqual(history_a, history_b)
        for name, tensor in first.state_dict().items():
            self.assertTrue(torch.equal(tensor, second.state_dict()[name]))

What level of reproducibility is tested?

Two scratch models start with the same seed and train on the same CPU fixture. The test compares the selected epoch count, every history entry, and every final state tensor.

This is evidence for same-environment reproducibility. It does not claim bitwise equality across different PyTorch releases or hardware platforms.

• Save tests/test_finetune.py.
• Add the unittest entry point at the bottom of the file by pasting:

if __name__ == "__main__":
    unittest.main()

Why include the entry point?

The entry point lets the file invoke unittest discovery when executed directly. The container command can also discover the same five methods through the Python module runner.

• Save the completed tests/test_finetune.py file.

✔️ Awesome, I've got everything!

Your test file now contains five safeguards and the unittest entry point. Keep it saved for the final container run.

ⓧ I'd like to double check the full code

import copy
import tempfile
import unittest
from pathlib import Path

import numpy as np
import pandas as pd
import torch

from src.feature_experiment import fingerprint
from src.finetune_experiment import (
    load_level5_checkpoint,
    pretrain_source,
    train_target,
    verify_fingerprint,
)
from src.neural_network_experiment import inner_group_split


class FineTuneTests(unittest.TestCase):
    def setUp(self):
        self.x = np.array(
            [
                [-2.0, -1.0, 0.0],
                [-1.5, -0.5, 0.2],
                [-1.0, -1.5, 0.1],
                [-0.5, -2.0, 0.3],
                [0.5, 2.0, 0.7],
                [1.0, 1.5, 0.8],
                [1.5, 0.5, 0.9],
                [2.0, 1.0, 1.0],
            ],
            dtype=np.float32,
        )
        self.y = np.array([0, 0, 0, 0, 1, 1, 1, 1], dtype=np.float32)

    def test_level5_checkpoint_loads_on_cpu(self):
        checkpoint = load_level5_checkpoint(Path.cwd())
        self.assertIn("model_state", checkpoint)
        self.assertTrue(all(tensor.device.type == "cpu" for tensor in checkpoint["model_state"].values()))

    def test_frozen_encoder_does_not_change(self):
        source, _, _ = pretrain_source(self.x, 42, epochs=5)
        state = copy.deepcopy(source.encoder.state_dict())
        model, _, _, _ = train_target(
            self.x,
            self.y,
            "frozen",
            42,
            state,
            epochs=5,
        )
        for name, tensor in state.items():
            self.assertTrue(torch.equal(tensor, model.encoder.state_dict()[name]))
        self.assertTrue(all(not parameter.requires_grad for parameter in model.encoder.parameters()))

    def test_group_separation(self):
        frame = pd.DataFrame(self.x, columns=["a", "b", "c"])
        target = pd.Series(self.y)
        groups = pd.Series(["r1", "r1", "r2", "r2", "r3", "r3", "r4", "r4"])
        fit, stop = inner_group_split(frame, target, groups, 42)
        self.assertFalse(set(groups.iloc[fit]) & set(groups.iloc[stop]))

    def test_fingerprint_verification_rejects_changes(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "artifact.txt"
            path.write_text("trusted", encoding="utf-8")
            expected = fingerprint(path)
            verify_fingerprint(path, expected)
            path.write_text("changed", encoding="utf-8")
            with self.assertRaises(ValueError):
                verify_fingerprint(path, expected)

    def test_same_seed_reproduces_history_and_state(self):
        first, history_a, epoch_a, _ = train_target(
            self.x,
            self.y,
            "scratch",
            42,
            None,
            epochs=12,
        )
        second, history_b, epoch_b, _ = train_target(
            self.x,
            self.y,
            "scratch",
            42,
            None,
            epochs=12,
        )
        self.assertEqual(epoch_a, epoch_b)
        self.assertEqual(history_a, history_b)
        for name, tensor in first.state_dict().items():
            self.assertTrue(torch.equal(tensor, second.state_dict()[name]))


if __name__ == "__main__":
    unittest.main()

Add the evidence publisher

The experiment report is machine-readable evidence. The publisher converts its selected approach and aggregate measurements into a reviewable Markdown page without recomputing or changing any v4 experiment output.

• Select src/finetune_experiment.py in the VS Code Explorer sidebar.
• Find the existing def main() -> None: line near the bottom of the file.
• Insert the complete documentation function directly above main() by copying this reference:

def document_experiment(root: Path) -> None:
    report = read_json(root / REPORT_PATH)
    selected = report["selected_approach"]
    rows = []
    for approach in APPROACHES:
        aggregate = report["aggregates"][approach]
        values = aggregate["validation_mean"]
        rows.append(
            f"| {approach} | {values['f1']:.3f} | {values['precision']:.3f} | "
            f"{values['recall']:.3f} | {values['loss']:.3f} | "
            f"{aggregate['source_pretraining_seconds_mean']:.3f} | "
            f"{aggregate['downstream_training_seconds_mean']:.3f} |"
        )
    text = f"""# Level 6: Fine-Tune

## Result

Selected approach: `{selected}` using development-only runner-safe evidence.

| Approach | F1 | Precision | Recall | Loss | Source seconds | Downstream seconds |
| --- | ---: | ---: | ---: | ---: | ---: | ---: |
{chr(10).join(rows)}

## Fair comparison contract

All three approaches use the same inherited folds, encoder and head architecture, loss, threshold, seeds, and `{TARGET_EPOCHS}` downstream epochs. Scratch and partial fine-tuning use the same encoder learning rate, and all three use the same head learning rate. Source pretraining and downstream fitting times are reported separately.

## What counts as pretrained

The Level 5 readiness checkpoint is not called pretrained because it learned the same supervised target. Level 6 creates pretrained weights only after the encoder learns a separately defined denoising source task without readiness labels.

This is controlled self-supervised pretraining on synthetic variants of fold-training data. It is not same-task warm-starting and it is not external-domain transfer learning.

Frozen-feature transfer keeps the source encoder unchanged and trains a new head. Partial fine-tuning keeps the first encoder layer frozen while adapting the later encoder layer. Scratch training initializes the same target architecture randomly.

## Leakage and evidence boundary

All three methods use the same four inherited runner-safe folds. Source synthesis, fold-local preprocessing, and downstream fitting remain inside each outer-training boundary. Target-validation runners and labels are used only for evaluation. The consumed v1 holdout is not read.

The inherited v1 process used development-wide mean imputation before splitting. Level 6 preprocessing is boundary-local, but the full historical pipeline is not retroactively called leakage-free.

## Reproducibility

The workflow uses seed `{report['determinism']['seed']}`, CPU execution, and deterministic algorithms where supported. PyTorch does not guarantee bitwise equality across releases or platforms. Training time is measured separately and can vary with host load.

## Artifacts

- `{REPORT_PATH}`
- `{SOURCE_CHECKPOINT_PATH}`
- `{SELECTED_CHECKPOINT_PATH}`
- `{PREPROCESSOR_PATH}`
- `{LEVEL7_HANDOFF_PATH}`

## Reproduce

```bash
docker build --platform linux/amd64 -f Dockerfile.level6 -t running-readiness-level6:2.15.0 .
docker run --rm --platform linux/amd64 --mount type=bind,src="$(pwd)",dst=/workspace -w /workspace running-readiness-level6:2.15.0 python -m unittest -v tests/test_finetune.py
```
"""
    (root / DOC_PATH).parent.mkdir(parents=True, exist_ok=True)
    (root / DOC_PATH).write_text(text, encoding="utf-8")
    readme = root / "README.md"
    marker = "<!-- LEVEL_6_FINETUNE -->"
    existing = readme.read_text(encoding="utf-8") if readme.exists() else "# Running Readiness\n"
    line = f"{marker}\n- [Level 6: Fine-Tune]({DOC_PATH.as_posix()})\n"
    if marker in existing:
        existing = existing.split(marker)[0].rstrip() + "\n\n" + line
    else:
        existing = existing.rstrip() + "\n\n" + line
    readme.write_text(existing, encoding="utf-8")

How does the publisher preserve the evidence?

The function reads the existing reports/v4/finetune_experiment.json report and formats its aggregate measurements. It never retrains a model or rewrites the v4 checkpoints.

The README update uses the <!-- LEVEL_6_FINETUNE --> marker as an idempotent insertion point. Repeating the command replaces the existing Level 6 link instead of appending duplicates.

• Find the command-routing chain inside main().
• Insert the documentation branch directly before the final else: block by pasting:

    elif command == "document":
        document_experiment(root)
        payload = {"documentation": str(DOC_PATH), "status": "written"}

What does the new command route do?

The branch sends the document command to document_experiment(). It then emits the documentation path and a written status through the existing JSON output.

• Save src/finetune_experiment.py.
• Confirm that document_experiment() appears above main() and that the document branch appears before the final error branch.

Publisher code showing a syntax error?

Confirm that the Markdown template begins and ends with matching triple quotes. Check that document_experiment() remains at the file's top indentation level.

If the command branch is unreachable or the string does not close, help me compare my publisher with the Level 6 Complete Code.

Run the safeguards and publish the evidence

The final verification checks both forms of proof. The test suite validates behavior while the generated evidence page explains the experimental boundary and reproducibility scope.

Before you run the complete suite, do you expect all five safeguards to pass against the unchanged Level 5 and v4 artifacts?

• Run all five safeguards through the Level 6 CPU container by using:

docker run --rm --platform linux/amd64 --mount type=bind,src="$(pwd)",dst=/workspace -w /workspace running-readiness-level6:2.15.0 python -m unittest -v tests/test_finetune.py

What should the completed suite show?

You should see five tests finish with OK. The passing suite proves restricted CPU loading, frozen encoder invariance, runner separation, fingerprint rejection, and same-seed target-training reproducibility.

One of the five tests failing?

Use the failing test name to isolate the boundary under test. Avoid changing the v4 report or checkpoints to make a safeguard pass.

If the traceback does not identify the mismatch clearly, help me diagnose the failing Level 6 safeguard from its traceback.

Before you publish the page, which existing report fields do you expect the generator to turn into human-readable evidence?

• Generate the evidence page and README link through the existing container by running:

docker run --rm --platform linux/amd64 --mount type=bind,src="$(pwd)",dst=/workspace -w /workspace running-readiness-level6:2.15.0 python -m src.finetune_experiment document

What does publication change?

The command writes docs/level-6-finetune.md from the existing development-only report. It also updates README.md at the Level 6 marker.

Your report, source checkpoint, selected checkpoint, preprocessor, preview checkpoint, and Level 7 handoff remain unchanged.

• Select docs/level-6-finetune.md in the VS Code Explorer sidebar.
• Confirm that the evidence page names the selected approach and includes the aggregate comparison table.
• Confirm that the page explains the source-task boundary, frozen feature extraction, partial fine-tuning, scratch training, leakage boundary, inherited v1 limitation, and reproducibility scope.
• Select README.md in the Explorer sidebar.
• Confirm that the <!-- LEVEL_6_FINETUNE --> marker is followed by a link to docs/level-6-finetune.md.

Evidence page or README link missing?

Confirm that reports/v4/finetune_experiment.json still exists before rerunning the document command. The publisher reads that report to create the table and selected-result summary.

If the command reports success but either file is missing, help me trace the Level 6 documentation paths and marker update.

That is the evidence package complete. Your Level 6 claims now have automated safeguards, a report-driven explanation, and an unchanged fingerprinted handoff for deployment.

Secret mission

Secret Mission: Measure Catastrophic Forgetting

Compare excessive and controlled encoder learning rates against one immutable pretrained baseline. Measure target performance alongside source reconstruction loss and representation similarity before deciding whether either run shows catastrophic forgetting.

Clean Up Your Resources

Clean Up Your Resources

The project runs entirely locally with no cloud resources or recurring costs. The options below let you keep the Level 6 image and v4 outputs, pause Docker Desktop, or remove the Level 6 resources.

Resources you used:

• Local Level 6 Docker image: running-readiness-level6:2.15.0.
• Level 6 model artifacts in artifacts/v4/, including the source checkpoints, selected target checkpoint, preprocessing metadata, and Level 7 deployment handoff.
• Level 6 evidence reports in reports/v4/, including finetune_experiment.json and catastrophic_forgetting.json.

Keep everything running

This option preserves the complete Level 6 state for your Level 7 deployment work.

• The running-readiness-level6:2.15.0 image stays ready for future Level 6 commands.
• The artifacts/v4/ directory retains the source checkpoints, selected model, preprocessor, and Level 7 handoff.
• The reports/v4/ directory retains the transfer comparison and catastrophic-forgetting evidence.
• Your Level 6 implementation, tests, documentation, and README link remain unchanged.
• These local resources stay idle without creating ongoing costs.

Pause - I'll come back to this later

This option stops the local container runtime while preserving every Level 6 file and image for later use.

• Stop Docker Desktop from its application menu.
• Leave artifacts/v4/ unchanged.
• Leave reports/v4/ unchanged.
• Keep the running-readiness-level6:2.15.0 image installed for your return.

Delete - I don't want to use this again

This option removes only the Level 6 image and v4 output folders. It preserves the immutable v3 resources, source code, tests, documentation, and README link.

• Remove the Level 6 image and v4 output folders by running these commands:

docker image rm running-readiness-level6:2.15.0
rm -rf artifacts/v4 reports/v4

What Do These Commands Remove?

The first command removes the local Level 6 image. The second command recursively deletes only artifacts/v4/ and reports/v4/.

Your v3 handoff remains available because the cleanup does not target its directory.

• Check the local image list in Docker Desktop.
• Check the repository folders in VS Code.

You'll no longer see running-readiness-level6:2.15.0 in the local image list.

The artifacts/v4/ and reports/v4/ folders should be gone. Your v3 resources should remain.

That's the cleanup done. Your immutable v3 foundation stays ready if you rebuild Level 6.

Image Still Present?

A local container may still reference the Level 6 image. Stop the matching container in Docker Desktop before rerunning the cleanup block.

Help me diagnose this Level 6 cleanup issue.

Nice Work!

Nice Work!

You did it! Your leakage-safe PyTorch experiment now demonstrates controlled transfer learning with tested checkpoints and a fingerprinted Level 7 handoff.

What you learned:

• Protected the immutable Level 5 evidence lineage with fingerprint checks. You loaded the inherited checkpoint on the CPU. You proved that its supervised readiness training could not serve as a separate pretrained source task.
• Built a label-free denoising objective for self-supervised pretraining. You kept target labels outside source optimization. You also isolated every target-validation runner from source synthesis.
• Ran a fair three-way comparison across scratch training, frozen-feature transfer, and partial fine-tuning. You preserved identical inherited folds. You preserved the fixed downstream epoch budget. You published reloadable checkpoints and development-only evidence for the selected approach.
• Secret Mission: Measured catastrophic forgetting by comparing excessive fine-tuning with a controlled encoder learning rate. You reused one immutable pretrained encoder. You reused one fingerprinted probe. You claimed forgetting only when both retention measurements supported it.

Ready to quiz yourself?