Deploy a Verified Readiness Model
Package a verified CPU ML API with strict validation and artifact integrity.
Introduction
30 Second Summary
A prediction can look convincing even when one of the files behind it has changed. Without proof of where each file came from, that result is hard to trust.
In this project, you will package the already-selected Level 6 running-readiness model as a local FastAPI service inside a Docker image. SHA-256 lineage checks form fail-closed gates before packaging and at service startup.
What You'll Build
You will demo a local running-readiness API that returns a Level 6-parity probability only after its lineage and packaged artifacts pass integrity checks.
By the end of this project, you'll have:
- A verified lineage check that reports status as passed before packaging can begin.
- Level 6-parity inference that returns a finite probability with decision 0 or 1. The response preserves threshold 0.5 on device cpu with approach partial.
- A strict local API that rejects malformed or prediction-time-ineligible inputs. Historical datasets and consumed holdout evidence stay outside its allowlisted image.
- Secret Mission: Corrupt a temporary checkpoint copy to prove packaging stops before Docker runs.
Are there any prerequisites?
This project requires the completed local triscravello/level-1-data checkout from Level 6. The three gitignored historical files must still exist at their original paths.
The inherited running-readiness-level6:2.15.0 CPU image must be available through Docker Desktop.
Before We Start
Your local repository root plus Docker Desktop provide the launch point for the trusted running-readiness-level6:2.15.0 CPU image. Before deployment work begins, commit to serving the already-selected immutable Level 6 model without retraining, reselection, threshold tuning, model comparison, feature selection, or reading the consumed v1 holdout so deployment evidence stays separate from selection evidence.
Verify the Local Lineage
Every later parity claim depends on the exact Level 6 artifacts remaining unchanged. A missing file or altered checkpoint would make the packaged result untrustworthy.
This verifier reads expected SHA-256 fingerprints from committed contracts. It fails closed before Docker can package an untrusted model.
In this step, get ready to:
- Create a standard-library verifier for the inherited artifact chain.
- Enforce the consumed-holdout policy from the committed contracts.
- Prove packaging readiness with both lineage gates.
Build the fingerprint helpers
A cryptographic fingerprint represents the exact bytes in a file. The verifier compares each current fingerprint with its committed expected value.
- In the Explorer sidebar in VS Code, create src/level7_lineage.py inside the src folder.
- Add the imports plus the immutable policy constants by pasting this first section:
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Mapping
PROJECT_ROOT = Path(__file__).resolve().parents[1]
FORBIDDEN_USES = frozenset(
{"feature_selection", "model_comparison", "model_selection", "threshold_tuning"}
)
class VerificationError(RuntimeError):
pass
def read_json(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
What does this code do?
- PROJECT_ROOT resolves the repository from the verifier's location.
- FORBIDDEN_USES preserves the four prohibited ways of reusing the consumed holdout.
- read_json() loads committed contracts without regenerating their evidence.
- sha256() hashes the exact file bytes.
- Add the fail-closed file check directly below sha256() by pasting this function:
def verify_file(path: Path, expected: str) -> str:
if not path.is_file():
raise VerificationError(f"Missing required artifact: {path}")
actual = sha256(path)
if actual != expected:
raise VerificationError(
f"SHA-256 mismatch for {path}: expected {expected}, got {actual}"
)
return actual
Why does the verifier fail closed?
verify_file() raises VerificationError when an artifact is missing. It also stops when the current fingerprint differs from the expected fingerprint.
The function returns a fingerprint only after both checks pass. Unverified bytes never receive a success result.
- Save src/level7_lineage.py.
- Use the VS Code Outline panel to confirm that VerificationError, read_json(), sha256(), and verify_file() appear in the file structure.
Missing a helper in Outline?
Check that each function starts at the left edge of the file. Accidental indentation can nest one helper inside another.
Confirm that the closing parenthesis in FORBIDDEN_USES appears before VerificationError.
Help me find the structural error in my lineage helper code.
Enforce the lineage contract
The committed contracts form a chain from Level 3 through Level 7. The audit collects every expected fingerprint before it verifies any local artifact.
- Append the path override helpers plus the contract-loading section below verify_file() by pasting this code:
def resolved_path(
root: Path, relative: Path, overrides: Mapping[Path, Path] | None
) -> Path:
return overrides[relative] if overrides and relative in overrides else root / relative
def add_check(checks: dict[Path, str], relative: Path, expected: str) -> None:
if relative in checks and checks[relative] != expected:
raise VerificationError(f"Conflicting expected hashes for {relative}")
checks[relative] = expected
def audit_lineage(
root: Path = PROJECT_ROOT,
overrides: Mapping[Path, Path] | None = None,
) -> dict:
level3 = read_json(root / "artifacts/v1/level3_evaluation_contract.json")
level4 = read_json(root / "artifacts/v1/level4_features_contract.json")
level6_input = read_json(root / "artifacts/v3/level6_finetune_handoff.json")
level7 = read_json(root / "artifacts/v4/level7_deploy_handoff.json")
How does contract loading stay trustworthy?
- resolved_path() normally keeps every artifact at its original repository path.
- add_check() rejects conflicting fingerprints for the same path.
- audit_lineage() reads the committed Level 3, Level 4, Level 6, and Level 7 contracts.
- Continue audit_lineage() by adding the required artifact map directly below the level7 assignment:
checks: dict[Path, str] = {}
for record in level3["input_fingerprints"].values():
add_check(checks, Path(record["path"]), record["sha256"])
add_check(
checks,
Path(level4["holdout_policy"]["marker_path"]),
level4["holdout_policy"]["marker_sha256"],
)
add_check(
checks,
Path(level4["evaluation_evidence"]["json_report"]["path"]),
level4["evaluation_evidence"]["json_report"]["sha256"],
)
level7_paths = {
"development_data": Path("data/v1/processed/train.csv"),
"level6_input_handoff": Path("artifacts/v3/level6_finetune_handoff.json"),
"level6_report": Path("reports/v4/finetune_experiment.json"),
"source_checkpoint": Path("artifacts/v4/level6_source_pretrained.pt"),
"selected_checkpoint": Path("artifacts/v4/level6_selected_model.pt"),
"preprocessor": Path("artifacts/v4/level6_preprocessor.json"),
}
for label, relative in level7_paths.items():
add_check(checks, relative, level7["fingerprints"][label])
Which artifacts enter the gate?
The earlier contracts contribute the local development data fingerprint plus the consumed-holdout marker and Level 3 evaluation report. The Level 7 handoff contributes the Level 6 input handoff, report, source checkpoint, selected checkpoint, and preprocessor.
Every path remains in its original repository location. The gate never reads data/v1/processed/test.csv.
- Save src/level7_lineage.py.
- Use Find in the current file to confirm that level7_paths contains six labeled Level 6 inputs.
- Add the verification loop plus the local policy inputs directly below the level7_paths loop:
verified = {}
for relative, expected in sorted(checks.items(), key=lambda item: str(item[0])):
verified[relative.as_posix()] = verify_file(
resolved_path(root, relative, overrides), expected
)
marker = read_json(
resolved_path(root, Path(level4["holdout_policy"]["marker_path"]), overrides)
)
level6_report = read_json(root / "reports/v4/finetune_experiment.json")
What does the verification loop prove?
The sorted loop sends every required path through verify_file(). One missing file or mismatched fingerprint stops the audit immediately.
The marker and Level 6 report are then loaded for policy checks. Their content must agree with the immutable lineage.
- Finish audit_lineage() by adding the policy assertions plus the passed result below level6_report assignment:
if level4["holdout_policy"]["holdout_consumed"] is not True:
raise VerificationError("Level 4 no longer marks the holdout consumed")
if marker.get("holdout_consumed") is not True:
raise VerificationError("Local holdout marker does not mark it consumed")
if set(marker.get("forbidden_future_uses", [])) != FORBIDDEN_USES:
raise VerificationError("Local holdout restrictions changed")
if set(level4["holdout_policy"]["forbidden_future_uses"]) != FORBIDDEN_USES:
raise VerificationError("Level 4 holdout restrictions changed")
if set(level6_input["forbidden_uses"]) != FORBIDDEN_USES:
raise VerificationError("Level 6 input restrictions changed")
if set(level7["forbidden_uses"]) != FORBIDDEN_USES:
raise VerificationError("Level 7 handoff restrictions changed")
if level7.get("consumed_v1_holdout_read") is not False:
raise VerificationError("Level 7 handoff records a holdout read")
if level6_report.get("consumed_v1_holdout_read") is not False:
raise VerificationError("Level 6 report records a holdout read")
if level7.get("next_level") != "Level 7: Deploy":
raise VerificationError("Unexpected Level 7 handoff destination")
if level7.get("threshold") != 0.5:
raise VerificationError("The locked Level 6 threshold changed")
return {
"status": "passed",
"verified": verified,
"holdout_consumed": True,
"consumed_v1_holdout_read": False,
"forbidden_uses": sorted(FORBIDDEN_USES),
}
What does the policy gate enforce?
The gate requires consumed_v1_holdout_read to remain false. It also requires the locked threshold to remain 0.5.
The four forbidden uses must remain feature_selection, model_comparison, model_selection, and threshold_tuning. The handoff destination must remain Level 7: Deploy.
- Save src/level7_lineage.py.
- Use Find in the current file to confirm that every VerificationError branch appears before the passed result.
- Make the verifier executable as a Python module by adding this final section below audit_lineage():
def main() -> None:
print(json.dumps(audit_lineage(), indent=2, sort_keys=True))
if __name__ == "__main__":
main()
How does the module entry point work?
main() runs the complete audit. It prints sorted JSON only after audit_lineage() returns successfully.
Any VerificationError interrupts that path. Packaging receives no passed attestation from an incomplete lineage.
- Save src/level7_lineage.py.
✔️ Awesome, I've got everything!
Your saved verifier now contains the fingerprint gate plus the consumed-holdout policy gate.
ⓧ I'd like to double check the full code
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Mapping
PROJECT_ROOT = Path(__file__).resolve().parents[1]
FORBIDDEN_USES = frozenset(
{"feature_selection", "model_comparison", "model_selection", "threshold_tuning"}
)
class VerificationError(RuntimeError):
pass
def read_json(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def verify_file(path: Path, expected: str) -> str:
if not path.is_file():
raise VerificationError(f"Missing required artifact: {path}")
actual = sha256(path)
if actual != expected:
raise VerificationError(
f"SHA-256 mismatch for {path}: expected {expected}, got {actual}"
)
return actual
def resolved_path(
root: Path, relative: Path, overrides: Mapping[Path, Path] | None
) -> Path:
return overrides[relative] if overrides and relative in overrides else root / relative
def add_check(checks: dict[Path, str], relative: Path, expected: str) -> None:
if relative in checks and checks[relative] != expected:
raise VerificationError(f"Conflicting expected hashes for {relative}")
checks[relative] = expected
def audit_lineage(
root: Path = PROJECT_ROOT,
overrides: Mapping[Path, Path] | None = None,
) -> dict:
level3 = read_json(root / "artifacts/v1/level3_evaluation_contract.json")
level4 = read_json(root / "artifacts/v1/level4_features_contract.json")
level6_input = read_json(root / "artifacts/v3/level6_finetune_handoff.json")
level7 = read_json(root / "artifacts/v4/level7_deploy_handoff.json")
checks: dict[Path, str] = {}
for record in level3["input_fingerprints"].values():
add_check(checks, Path(record["path"]), record["sha256"])
add_check(
checks,
Path(level4["holdout_policy"]["marker_path"]),
level4["holdout_policy"]["marker_sha256"],
)
add_check(
checks,
Path(level4["evaluation_evidence"]["json_report"]["path"]),
level4["evaluation_evidence"]["json_report"]["sha256"],
)
level7_paths = {
"development_data": Path("data/v1/processed/train.csv"),
"level6_input_handoff": Path("artifacts/v3/level6_finetune_handoff.json"),
"level6_report": Path("reports/v4/finetune_experiment.json"),
"source_checkpoint": Path("artifacts/v4/level6_source_pretrained.pt"),
"selected_checkpoint": Path("artifacts/v4/level6_selected_model.pt"),
"preprocessor": Path("artifacts/v4/level6_preprocessor.json"),
}
for label, relative in level7_paths.items():
add_check(checks, relative, level7["fingerprints"][label])
verified = {}
for relative, expected in sorted(checks.items(), key=lambda item: str(item[0])):
verified[relative.as_posix()] = verify_file(
resolved_path(root, relative, overrides), expected
)
marker = read_json(
resolved_path(root, Path(level4["holdout_policy"]["marker_path"]), overrides)
)
level6_report = read_json(root / "reports/v4/finetune_experiment.json")
if level4["holdout_policy"]["holdout_consumed"] is not True:
raise VerificationError("Level 4 no longer marks the holdout consumed")
if marker.get("holdout_consumed") is not True:
raise VerificationError("Local holdout marker does not mark it consumed")
if set(marker.get("forbidden_future_uses", [])) != FORBIDDEN_USES:
raise VerificationError("Local holdout restrictions changed")
if set(level4["holdout_policy"]["forbidden_future_uses"]) != FORBIDDEN_USES:
raise VerificationError("Level 4 holdout restrictions changed")
if set(level6_input["forbidden_uses"]) != FORBIDDEN_USES:
raise VerificationError("Level 6 input restrictions changed")
if set(level7["forbidden_uses"]) != FORBIDDEN_USES:
raise VerificationError("Level 7 handoff restrictions changed")
if level7.get("consumed_v1_holdout_read") is not False:
raise VerificationError("Level 7 handoff records a holdout read")
if level6_report.get("consumed_v1_holdout_read") is not False:
raise VerificationError("Level 6 report records a holdout read")
if level7.get("next_level") != "Level 7: Deploy":
raise VerificationError("Unexpected Level 7 handoff destination")
if level7.get("threshold") != 0.5:
raise VerificationError("The locked Level 6 threshold changed")
return {
"status": "passed",
"verified": verified,
"holdout_consumed": True,
"consumed_v1_holdout_read": False,
"forbidden_uses": sorted(FORBIDDEN_USES),
}
def main() -> None:
print(json.dumps(audit_lineage(), indent=2, sort_keys=True))
if __name__ == "__main__":
main()
Run both fail-closed gates
The standard-library gate checks exact artifact identity plus holdout policy. The inherited Level 6 audit independently checks the earlier experiment chain inside its trusted CPU environment.
- Predict whether every required local artifact still matches its committed fingerprint.
- From the repository root, run the lineage verifier with this command:
python -m src.level7_lineage
What does this command prove?
The module hashes all eight required local artifacts at their original paths. It then checks the consumed-holdout restrictions plus the locked deployment contract.
A successful result proves the local bytes match the committed lineage. It does not regenerate historical evidence.
You should see JSON with status set to passed. The output should also show consumed_v1_holdout_read set to false.
Did the lineage gate stop?
A missing-artifact message means one of the required gitignored historical files is absent from its original path. Restore that file from your trusted local project history.
A fingerprint mismatch means the local bytes changed. Do not regenerate the contract or replace its expected fingerprint.
Help me diagnose my failed Level 7 lineage gate.
The local fingerprint chain now passes. That clears the first fail-closed boundary before packaging.
- Predict whether the inherited audit will confirm the same Level 6 chain inside the trusted CPU image.
- Run the inherited audit in running-readiness-level6:2.15.0 with 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
Why run the inherited audit too?
The container uses the trusted Level 6 CPU image. Its audit checks the inherited experiment evidence before Level 7 stages any serving context.
The audit does not read, copy, fingerprint, or stage data/v1/processed/test.csv.
You should see the inherited audit complete successfully. The terminal returns to the prompt with a zero exit status.
Did the container audit fail?
Confirm that Docker Desktop is running before retrying the command. Confirm that the running-readiness-level6:2.15.0 image exists locally.
Run the command from the repository root so the mounted workspace contains src/finetune_experiment.py and the inherited contracts.
Help me diagnose the inherited Level 6 audit without touching the consumed holdout.
Both gates now trust the inherited model chain without reopening the consumed holdout. Next, you will reconstruct the exact Level 6 preprocessing and CPU inference path.
Reconstruct CPU Inference
Your lineage gate proved that the selected checkpoint belongs to the trusted historical chain. The next challenge is preserving its behavior when those verified bytes become a running prediction.
You will reconstruct the Level 6 architecture with PyTorch. The runtime will preserve the inherited feature order, preprocessing sequence, probability conversion, and locked decision threshold.
In this step, get ready to:
- Add a runtime integrity boundary for every packaged serving artifact.
- Recreate the selected Level 6 architecture with strict CPU checkpoint loading.
- Run a synthetic five-feature request through the complete inference path.
Why Inherit the Level 6 Image?
The trusted Level 6 image already contains the CPU environment used by the selected checkpoint. Reusing it keeps environment fidelity tied to the verified model lineage.
A fresh base image would introduce a second dependency environment to reconcile. The inherited image keeps this step focused on inference parity.
Add the artifact verification boundary
A packaged model can change after the local lineage audit. The runtime therefore hashes the checkpoint, preprocessor, and deployment handoff before model deserialization begins.
- Press Cmd+Space on macOS or the Windows key on Windows to open system search.
- Type Visual Studio Code into system search.
- Press Enter to open Visual Studio Code.
- Click File in the top menu bar.
- Click Open Folder.
- Select the local level-1-data repository folder.
- Click Open.
- Create src/level7_runtime.py from the Explorer sidebar.
The new file gives the serving runtime a separate boundary from the historical lineage verifier.
- Add the runtime imports, artifact paths, and synthetic request to src/level7_runtime.py by pasting this code:
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
import numpy as np
import torch
from torch import nn
HANDOFF_PATH = Path("artifacts/v4/level7_deploy_handoff.json")
CHECKPOINT_PATH = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_PATH = Path("artifacts/v4/level6_preprocessor.json")
MANIFEST_PATH = Path("serving-manifest.json")
SAMPLE_REQUEST = {
"weekly_km_to_date": 32.0,
"sleep_hours_last_night": 7.0,
"resting_hr_morning": 50.0,
"soreness_score_evening": 4.0,
"prior_injury": 0,
}
What Does This Code Establish?
- The path constants identify the deployment handoff, selected checkpoint, preprocessor, and serving manifest.
- The SAMPLE_REQUEST dictionary follows the inherited five-feature order.
- The imports provide byte hashing, JSON loading, numerical preprocessing, and model execution.
- Save src/level7_runtime.py.
- Confirm that src/level7_runtime.py appears beneath src in the Explorer sidebar.
File Missing From the Explorer?
Confirm that you created the file inside the repository's existing src folder. Check that its name ends with .py.
Help me check the runtime file location.
The serving manifest is an allowlist of expected artifact fingerprints. A missing path or changed fingerprint must stop execution before the checkpoint reaches the loader.
- Add the integrity exception, byte-hashing helper, and artifact verifier beneath SAMPLE_REQUEST by pasting this code:
class IntegrityError(RuntimeError):
pass
def file_sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def verify_serving_artifacts(root: Path) -> dict:
manifest_path = root / MANIFEST_PATH
if not manifest_path.is_file():
raise IntegrityError(f"Missing serving manifest: {manifest_path}")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
for relative_text, expected in manifest["artifacts"].items():
artifact = root / relative_text
if not artifact.is_file():
raise IntegrityError(f"Missing serving artifact: {artifact}")
actual = file_sha256(artifact)
if actual != expected:
raise IntegrityError(
f"Serving artifact mismatch for {artifact}: expected {expected}, got {actual}"
)
return manifest
How Does the Gate Fail Closed?
- The runtime requires serving-manifest.json before examining model data.
- The file_sha256() helper computes a SHA-256 fingerprint from the exact artifact bytes.
- Any missing artifact or fingerprint mismatch raises IntegrityError immediately.
- Save src/level7_runtime.py.
- Open the editor symbol list for src/level7_runtime.py.
- Confirm that IntegrityError, file_sha256, and verify_serving_artifacts appear as top-level symbols.
Verifier Symbols Missing?
Align the class and both function declarations with the left edge of the file. Extra indentation can hide them inside the preceding dictionary or function.
Help me repair the verifier structure.
Recreate the Level 6 model
Checkpoint weights depend on the architecture that produced them. The runtime must recreate the same encoder widths, activations, and classification head before loading the saved state dictionary.
- Add the Encoder class beneath verify_serving_artifacts() by pasting 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))))
What Does the Encoder Preserve?
- The first linear layer maps the ordered inputs to 16 values.
- The second linear layer maps those values to 8 encoded values.
- The nested rectified linear activations reproduce the Level 6 forward path.
- Save src/level7_runtime.py.
- Confirm that Encoder appears as a class in the editor symbol list.
Encoder Missing From the Symbol List?
Keep the Encoder declaration at the left edge of the file. Indent __init__() and forward() by one class level.
Help me correct the Encoder class.
- Add the TransferClassifier class beneath Encoder by pasting this code:
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)
What Does the Classifier Preserve?
The classifier passes each request through Encoder. Its single-output head returns the logit used for probability conversion.
- Save src/level7_runtime.py.
- Confirm that Encoder and TransferClassifier appear as separate classes in the editor symbol list.
Model Classes Appear Nested?
Align both class declarations with the left edge of the file. Keep TransferClassifier outside the body of Encoder.
Help me separate the model classes.
The runtime constructor connects integrity verification to restricted checkpoint loading. It also remaps tensors to the CPU before restoring the model state.
- Add the initial ModelRuntime constructor beneath TransferClassifier by pasting this code:
class ModelRuntime:
def __init__(self, root: Path):
self.root = root
self.manifest = verify_serving_artifacts(root)
self.integrity_status = "passed"
self.handoff = json.loads((root / HANDOFF_PATH).read_text(encoding="utf-8"))
self.preprocessor = json.loads(
(root / PREPROCESSOR_PATH).read_text(encoding="utf-8")
)
checkpoint: dict[str, Any] = torch.load(
root / CHECKPOINT_PATH,
map_location=torch.device("cpu"),
weights_only=True,
)
self.model = TransferClassifier(checkpoint["input_dim"])
self.model.load_state_dict(checkpoint["model_state"], strict=True)
self.model.eval()
self.threshold = float(checkpoint["threshold"])
self.feature_names = list(checkpoint["feature_names"])
self.device = "cpu"
What Does the Constructor Establish?
- Artifact verification completes before torch.load() receives the checkpoint.
- CPU remapping removes any dependency on an unavailable accelerator.
- Restricted loading accepts weight-compatible checkpoint data.
- Strict state-dict loading requires exact agreement between the saved keys and recreated architecture.
- Save src/level7_runtime.py.
- Confirm that ModelRuntime appears in the editor symbol list with an __init__ method.
Constructor Missing From ModelRuntime?
Keep every assignment from self.root through self.device inside __init__().
Help me repair the runtime constructor.
Hash agreement proves byte identity. Compatibility checks prove that the checkpoint, preprocessor, and handoff still describe the same serving contract.
- Find this model-construction line inside ModelRuntime.__init__():
self.model = TransferClassifier(checkpoint["input_dim"])
Why Locate This Line?
This is the first line that uses checkpoint metadata to construct the model. The contract checks belong immediately before it.
- Insert the compatibility checks directly above the model-construction line so the section looks like this:
if checkpoint.get("model_class") != "TransferClassifier":
raise IntegrityError("Unexpected Level 6 model class")
if checkpoint.get("feature_names") != self.handoff["feature_names"]:
raise IntegrityError("Checkpoint feature order differs from the handoff")
if self.preprocessor.get("feature_names") != self.handoff["feature_names"]:
raise IntegrityError("Preprocessor feature order differs from the handoff")
if checkpoint.get("threshold") != self.handoff["threshold"]:
raise IntegrityError("Checkpoint threshold differs from the handoff")
if checkpoint.get("approach") != self.handoff["selected_approach"]:
raise IntegrityError("Checkpoint approach differs from the handoff")
if checkpoint.get("input_dim") != len(self.handoff["feature_names"]):
raise IntegrityError("Checkpoint input dimension is inconsistent")
self.model = TransferClassifier(checkpoint["input_dim"])
What Do the Compatibility Checks Protect?
- The model class must remain TransferClassifier.
- The checkpoint and preprocessor must preserve the handoff's feature order.
- The selected approach and threshold must match the deployment handoff.
- The checkpoint input dimension must match the number of inherited features.
- Save src/level7_runtime.py.
- Review the editor diagnostics for the constructor.
- Confirm that the compatibility checks show no indentation or undefined-name errors.
Checks Appear Outside the Constructor?
Match the indentation of the existing self.model line. Keep all six checks inside __init__().
Help me place the contract checks.
Preserve preprocessing and run inference
Inference parity depends on the transformation order. The runtime reads the five fields in handoff order before applying mean imputation, standard scaling, and float32 conversion.
- Add the preprocess() method inside ModelRuntime beneath __init__() by pasting this code:
def preprocess(self, payload: dict[str, float | int]) -> np.ndarray:
values = np.asarray(
[[float(payload[name]) for name in self.feature_names]], dtype=np.float64
)
statistics = np.asarray(
self.preprocessor["imputer_statistics"], dtype=np.float64
)
means = np.asarray(self.preprocessor["scaler_mean"], dtype=np.float64)
scales = np.asarray(self.preprocessor["scaler_scale"], dtype=np.float64)
if not np.isfinite(values).all():
raise ValueError("Inputs must be finite")
if np.any(scales == 0.0):
raise IntegrityError("Preprocessor contains a zero scale")
imputed = np.where(np.isnan(values), statistics, values)
return ((imputed - means) / scales).astype(np.float32)
How Does Preprocessing Preserve Parity?
- The list comprehension reads each value through self.feature_names.
- The stored imputation statistics supply the inherited mean values.
- The stored means and scales reproduce standard scaling.
- The final conversion supplies the model with float32 values.
- Save src/level7_runtime.py.
- Confirm that preprocess appears beneath ModelRuntime in the editor symbol list.
Preprocess Appears as a Top-Level Function?
Indent preprocess() by one class level. Keep its body indented one additional level.
Help me place preprocessing correctly.
The prediction path converts the model's single logit into a probability exactly once. It compares that probability with the checkpoint threshold exactly once.
- Add the predict() method inside ModelRuntime beneath preprocess() by pasting this code:
def predict(self, payload: dict[str, float | int]) -> dict[str, Any]:
inputs = torch.from_numpy(self.preprocess(payload))
with torch.inference_mode():
probability = float(torch.sigmoid(self.model(inputs)).item())
return {
"probability": probability,
"decision": int(probability >= self.threshold),
"threshold": self.threshold,
"device": self.device,
"selected_approach": self.handoff["selected_approach"],
}
How Does Prediction Preserve Behavior?
- The preprocessed NumPy array becomes a tensor without another transformation.
- Inference mode disables gradient tracking for evaluation.
- The sigmoid operation converts the logit into one probability.
- The locked threshold produces a binary decision.
- Save src/level7_runtime.py.
- Review the editor diagnostics for preprocess() and predict().
- Confirm that both methods show no syntax or indentation errors.
Predict Cannot Access Preprocess?
Confirm that both methods share the same class indentation. Check that prediction calls self.preprocess(payload).
Help me connect prediction to preprocessing.
A module entry point turns the synthetic request into observable runtime evidence. Its output exposes the probability, decision, threshold, device, and selected approach.
- Add the module entry point beneath the ModelRuntime class by pasting this code:
def main() -> None:
runtime = ModelRuntime(Path.cwd())
print(json.dumps(runtime.predict(SAMPLE_REQUEST), indent=2, sort_keys=True))
if __name__ == "__main__":
main()
What Does the Entry Point Prove?
Running the module creates ModelRuntime from the current repository directory. Successful output proves that integrity verification, compatibility checks, preprocessing, model execution, and thresholding all completed.
- Save src/level7_runtime.py.
- Confirm that main appears as a top-level function in the editor symbol list.
Main Appears Inside ModelRuntime?
Align main() and the final conditional with the left edge of the file. Both belong outside ModelRuntime.
Help me correct the module entry point.
✔️ Awesome, I've got everything!
Your runtime file now contains the integrity boundary, Level 6 architecture, compatibility checks, preprocessing path, and prediction entry point. Confirm that src/level7_runtime.py is saved.
ⓧ I'd like to double check the full code
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
import numpy as np
import torch
from torch import nn
HANDOFF_PATH = Path("artifacts/v4/level7_deploy_handoff.json")
CHECKPOINT_PATH = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_PATH = Path("artifacts/v4/level6_preprocessor.json")
MANIFEST_PATH = Path("serving-manifest.json")
SAMPLE_REQUEST = {
"weekly_km_to_date": 32.0,
"sleep_hours_last_night": 7.0,
"resting_hr_morning": 50.0,
"soreness_score_evening": 4.0,
"prior_injury": 0,
}
class IntegrityError(RuntimeError):
pass
def file_sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def verify_serving_artifacts(root: Path) -> dict:
manifest_path = root / MANIFEST_PATH
if not manifest_path.is_file():
raise IntegrityError(f"Missing serving manifest: {manifest_path}")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
for relative_text, expected in manifest["artifacts"].items():
artifact = root / relative_text
if not artifact.is_file():
raise IntegrityError(f"Missing serving artifact: {artifact}")
actual = file_sha256(artifact)
if actual != expected:
raise IntegrityError(
f"Serving artifact mismatch for {artifact}: expected {expected}, got {actual}"
)
return manifest
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 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)
class ModelRuntime:
def __init__(self, root: Path):
self.root = root
self.manifest = verify_serving_artifacts(root)
self.integrity_status = "passed"
self.handoff = json.loads((root / HANDOFF_PATH).read_text(encoding="utf-8"))
self.preprocessor = json.loads(
(root / PREPROCESSOR_PATH).read_text(encoding="utf-8")
)
checkpoint: dict[str, Any] = torch.load(
root / CHECKPOINT_PATH,
map_location=torch.device("cpu"),
weights_only=True,
)
if checkpoint.get("model_class") != "TransferClassifier":
raise IntegrityError("Unexpected Level 6 model class")
if checkpoint.get("feature_names") != self.handoff["feature_names"]:
raise IntegrityError("Checkpoint feature order differs from the handoff")
if self.preprocessor.get("feature_names") != self.handoff["feature_names"]:
raise IntegrityError("Preprocessor feature order differs from the handoff")
if checkpoint.get("threshold") != self.handoff["threshold"]:
raise IntegrityError("Checkpoint threshold differs from the handoff")
if checkpoint.get("approach") != self.handoff["selected_approach"]:
raise IntegrityError("Checkpoint approach differs from the handoff")
if checkpoint.get("input_dim") != len(self.handoff["feature_names"]):
raise IntegrityError("Checkpoint input dimension is inconsistent")
self.model = TransferClassifier(checkpoint["input_dim"])
self.model.load_state_dict(checkpoint["model_state"], strict=True)
self.model.eval()
self.threshold = float(checkpoint["threshold"])
self.feature_names = list(checkpoint["feature_names"])
self.device = "cpu"
def preprocess(self, payload: dict[str, float | int]) -> np.ndarray:
values = np.asarray(
[[float(payload[name]) for name in self.feature_names]], dtype=np.float64
)
statistics = np.asarray(
self.preprocessor["imputer_statistics"], dtype=np.float64
)
means = np.asarray(self.preprocessor["scaler_mean"], dtype=np.float64)
scales = np.asarray(self.preprocessor["scaler_scale"], dtype=np.float64)
if not np.isfinite(values).all():
raise ValueError("Inputs must be finite")
if np.any(scales == 0.0):
raise IntegrityError("Preprocessor contains a zero scale")
imputed = np.where(np.isnan(values), statistics, values)
return ((imputed - means) / scales).astype(np.float32)
def predict(self, payload: dict[str, float | int]) -> dict[str, Any]:
inputs = torch.from_numpy(self.preprocess(payload))
with torch.inference_mode():
probability = float(torch.sigmoid(self.model(inputs)).item())
return {
"probability": probability,
"decision": int(probability >= self.threshold),
"threshold": self.threshold,
"device": self.device,
"selected_approach": self.handoff["selected_approach"],
}
def main() -> None:
runtime = ModelRuntime(Path.cwd())
print(json.dumps(runtime.predict(SAMPLE_REQUEST), indent=2, sort_keys=True))
if __name__ == "__main__":
main()
Before you run the runtime, what values do you expect for the device, threshold, and selected approach?
- Run the runtime inside the trusted Level 6 CPU image from the repository directory 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.level7_runtime
What Does This Command Isolate?
- The bind mount exposes the verified repository at /workspace inside the container.
- The working-directory option makes relative artifact paths resolve from the mounted repository.
- The inherited image supplies the trusted Level 6 CPU environment.
- The temporary container removes itself after the runtime exits.
You should see a finite probability with a decision of 0 or 1. The output also shows threshold as 0.5.
You should also see device as cpu with selected_approach as partial.
Runtime Stops Before Producing JSON?
A manifest failure means the runtime refused to deserialize an unverified bundle. Confirm that serving-manifest.json is available beside the packaged serving artifacts.
A compatibility failure means the checkpoint, preprocessor, or handoff differs from the selected Level 6 contract.
Help me diagnose the fail-closed runtime.
That completes the parity boundary. Your verified checkpoint now produces a CPU prediction through the inherited architecture, feature order, preprocessing sequence, sigmoid conversion, and locked threshold.
Your CPU inference path is reconstructed and observable. Next, you will place it behind a temporary API contract and expose the unsafe edge that strict validation must close.
Expose the Unsafe API Edge
The verified CPU runtime now reproduces the selected Level 6 prediction path. A callable model still needs an API boundary that decides which client inputs are eligible.
A temporary FastAPI wrapper lets you test that boundary with a generic dictionary body. You will send a normal request first before probing the contract with a prediction-time-ineligible field.
In this step, get ready to:
- Wrap the verified CPU runtime with temporary health and prediction routes.
- Confirm that a valid five-feature request returns a successful prediction.
- Probe the temporary contract with an ineligible target field and record its response.
Create the temporary API shell
The health route provides the smallest possible service check. It proves the API can construct the verified runtime before you expose prediction.
- In the VS Code Explorer sidebar, select the src folder.
- Create level7_unsafe_service.py inside the src folder.
- Build the temporary health route by pasting this code into src/level7_unsafe_service.py:
from __future__ import annotations
from pathlib import Path
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.level7_runtime import ModelRuntime, SAMPLE_REQUEST
# Reuse the verified CPU runtime from the repository root
runtime = ModelRuntime(Path.cwd())
app = FastAPI()
client = TestClient(app)
@app.get("/health")
def health() -> dict:
# Expose only a small runtime readiness signal
return {"status": "ok", "device": runtime.device}
# Exercise the temporary route in process
health_response = client.get("/health")
print("health", health_response.status_code, health_response.json())
What does this code do?
- The ModelRuntime constructor verifies the serving manifest before the API becomes usable.
- The TestClient sends requests to the temporary app inside the current Python process.
- The /health route exposes a readiness signal without returning model artifacts.
- Save src/level7_unsafe_service.py.
- Click Run Python File in the editor toolbar.
You should see health followed by status 200. The response body should report status ok with device cpu.
Health check not returning 200?
Make sure VS Code uses the Python environment that contains FastAPI and HTTPX. Run the file from the repository root so Path.cwd() can locate the serving manifest.
Check that serving-manifest.json still matches the checkpoint and preprocessing artifacts.
Help me diagnose the temporary health route.
Add the temporary prediction route
A valid request establishes the control case for your boundary test. Its response confirms that the API wrapper preserves the runtime behavior you verified earlier.
- In src/level7_unsafe_service.py, find the comment # Exercise the temporary route in process.
- Place the following prediction route directly below that comment:
@app.post("/predict")
def predict(payload: dict) -> dict:
# Forward the request body through the verified runtime
return runtime.predict(payload)
# Confirm that the inherited five-feature request succeeds
valid_response = client.post("/predict", json=SAMPLE_REQUEST)
print("valid", valid_response.status_code, valid_response.json())
How does the control request work?
- The temporary /predict route forwards its dictionary body to runtime.predict().
- The inherited SAMPLE_REQUEST keeps the feature order and values consistent with the CPU runtime check.
- The printed status and response body give you a baseline for the contract probe.
- Save src/level7_unsafe_service.py.
- Click Run Python File in the editor toolbar.
You should see valid followed by status 200. The body should contain a finite probability plus a decision of 0 or 1.
Valid prediction not succeeding?
Check that the route passes payload directly to runtime.predict(). Confirm that SAMPLE_REQUEST still comes from src.level7_runtime.
Help me debug the temporary valid prediction request.
Send the ineligible target field
The control request proved that eligible data reaches the model. The next request adds needs_recovery_next_day, which represents the future outcome the model is trying to predict.
- In src/level7_unsafe_service.py, find the line that prints valid_response.
- Add the contract probe directly below that line:
# Add a prediction-time-ineligible target to the valid request
unsafe_request = dict(SAMPLE_REQUEST, needs_recovery_next_day=1)
unsafe_response = client.post("/predict", json=unsafe_request)
print("unsafe", unsafe_response.status_code, unsafe_response.json())
What does this probe test?
The probe copies the known-good request before adding one field. It sends that expanded body through the same temporary prediction route.
- Save src/level7_unsafe_service.py.
Before you run the probe, do you think the API rejects the target field or still serves a prediction?
- Click Run Python File in the editor toolbar.
You will see unsafe followed by status 200. The response still contains a probability and decision.
This 200 Is the Designed Obstacle
The generic dictionary body accepts the ineligible field. ModelRuntime reads only the five names in feature_names, so needs_recovery_next_day never reaches the model tensor.
The returned prediction is numerically valid while the request contract is unsafe. A serving boundary must reject prediction-time-ineligible data before inference begins.
You have reproduced the contract failure on purpose. Remove the disposable route now so it cannot enter the serving bundle.
- In the VS Code Explorer sidebar, right-click src/level7_unsafe_service.py.
- Select Delete.
- Confirm the deletion when VS Code asks.
You should no longer see src/level7_unsafe_service.py in the src folder. Your verified lineage and CPU runtime files remain unchanged.
✔️ Awesome, I've got everything!
Your screenshot records the temporary 200 response. The permissive service file is deleted.
Double-check that src/level7_lineage.py and src/level7_runtime.py remain unchanged.
ⓧ I'd like to double check the full code
The temporary API file should no longer exist. Compare src/level7_runtime.py with this unchanged reference:
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
import numpy as np
import torch
from torch import nn
HANDOFF_PATH = Path("artifacts/v4/level7_deploy_handoff.json")
CHECKPOINT_PATH = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_PATH = Path("artifacts/v4/level6_preprocessor.json")
MANIFEST_PATH = Path("serving-manifest.json")
SAMPLE_REQUEST = {
"weekly_km_to_date": 32.0,
"sleep_hours_last_night": 7.0,
"resting_hr_morning": 50.0,
"soreness_score_evening": 4.0,
"prior_injury": 0,
}
class IntegrityError(RuntimeError):
pass
def file_sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def verify_serving_artifacts(root: Path) -> dict:
manifest_path = root / MANIFEST_PATH
if not manifest_path.is_file():
raise IntegrityError(f"Missing serving manifest: {manifest_path}")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
for relative_text, expected in manifest["artifacts"].items():
artifact = root / relative_text
if not artifact.is_file():
raise IntegrityError(f"Missing serving artifact: {artifact}")
actual = file_sha256(artifact)
if actual != expected:
raise IntegrityError(
f"Serving artifact mismatch for {artifact}: expected {expected}, got {actual}"
)
return manifest
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 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)
class ModelRuntime:
def __init__(self, root: Path):
self.root = root
self.manifest = verify_serving_artifacts(root)
self.integrity_status = "passed"
self.handoff = json.loads((root / HANDOFF_PATH).read_text(encoding="utf-8"))
self.preprocessor = json.loads(
(root / PREPROCESSOR_PATH).read_text(encoding="utf-8")
)
checkpoint: dict[str, Any] = torch.load(
root / CHECKPOINT_PATH,
map_location=torch.device("cpu"),
weights_only=True,
)
if checkpoint.get("model_class") != "TransferClassifier":
raise IntegrityError("Unexpected Level 6 model class")
if checkpoint.get("feature_names") != self.handoff["feature_names"]:
raise IntegrityError("Checkpoint feature order differs from the handoff")
if self.preprocessor.get("feature_names") != self.handoff["feature_names"]:
raise IntegrityError("Preprocessor feature order differs from the handoff")
if checkpoint.get("threshold") != self.handoff["threshold"]:
raise IntegrityError("Checkpoint threshold differs from the handoff")
if checkpoint.get("approach") != self.handoff["selected_approach"]:
raise IntegrityError("Checkpoint approach differs from the handoff")
if checkpoint.get("input_dim") != len(self.handoff["feature_names"]):
raise IntegrityError("Checkpoint input dimension is inconsistent")
self.model = TransferClassifier(checkpoint["input_dim"])
self.model.load_state_dict(checkpoint["model_state"], strict=True)
self.model.eval()
self.threshold = float(checkpoint["threshold"])
self.feature_names = list(checkpoint["feature_names"])
self.device = "cpu"
def preprocess(self, payload: dict[str, float | int]) -> np.ndarray:
values = np.asarray(
[[float(payload[name]) for name in self.feature_names]], dtype=np.float64
)
statistics = np.asarray(
self.preprocessor["imputer_statistics"], dtype=np.float64
)
means = np.asarray(self.preprocessor["scaler_mean"], dtype=np.float64)
scales = np.asarray(self.preprocessor["scaler_scale"], dtype=np.float64)
if not np.isfinite(values).all():
raise ValueError("Inputs must be finite")
if np.any(scales == 0.0):
raise IntegrityError("Preprocessor contains a zero scale")
imputed = np.where(np.isnan(values), statistics, values)
return ((imputed - means) / scales).astype(np.float32)
def predict(self, payload: dict[str, float | int]) -> dict[str, Any]:
inputs = torch.from_numpy(self.preprocess(payload))
with torch.inference_mode():
probability = float(torch.sigmoid(self.model(inputs)).item())
return {
"probability": probability,
"decision": int(probability >= self.threshold),
"threshold": self.threshold,
"device": self.device,
"selected_approach": self.handoff["selected_approach"],
}
def main() -> None:
runtime = ModelRuntime(Path.cwd())
print(json.dumps(runtime.predict(SAMPLE_REQUEST), indent=2, sort_keys=True))
if __name__ == "__main__":
main()
You now have evidence that a successful prediction can still cross an unsafe contract boundary. Next, you will replace that permissive boundary with explicit validation while preserving prediction parity.
Enforce Validation and Parity
Your temporary FastAPI route exposed the contract failure clearly. An ineligible target field reached the service and still produced a successful prediction.
This step replaces the permissive dictionary with strict Pydantic models. It also locks valid predictions to the verified Level 6 behavior.
In this step, get ready to:
- Define strict request and response contracts for prediction and health data.
- Connect both routes to the existing integrity-verified CPU runtime.
- Add tests for parity, invalid requests, corrupted artifacts, metadata, and bundle exclusions.
Define strict request and response models
The application initializes the existing ModelRuntime before it accepts requests. That startup path preserves the manifest verification that already runs before checkpoint loading.
- Return to the repository in VS Code.
- Create `level7_service.py` inside the existing `src/` folder.
- Initialize the runtime and application by adding this code:
from __future__ import annotations
from pathlib import Path
from typing import Annotated
from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, Field, FiniteFloat, StrictInt
try:
from src.level7_runtime import ModelRuntime
except ModuleNotFoundError:
from level7_runtime import ModelRuntime
BASE_DIR = Path(__file__).resolve().parent
runtime = ModelRuntime(BASE_DIR)
app = FastAPI(
title="Running Readiness Level 7",
version="4.0.0",
description=(
f"{runtime.manifest['model_notice']} "
f"{runtime.manifest['inherited_v1_preprocessing_limitation']}"
),
)
Why initialize the runtime first?
- `ModelRuntime(BASE_DIR)` verifies the serving manifest before the application begins handling traffic.
- The import fallback supports both repository imports and the flattened module layout inside the serving image.
- The application description publishes the model notice and inherited preprocessing limitation from verified metadata.
- Save `src/level7_service.py`.
- Return to the terminal at the repository checkout.
- Compile the application setup by running this command:
python -m py_compile src/level7_service.py
What does this check prove?
The command asks Python to compile the source without starting the service. A clean return to the terminal prompt confirms that this first section has valid Python syntax.
You should return to the terminal prompt without a traceback.
Seeing a compilation error?
Check the indentation inside the `try` block and the `FastAPI` call. Confirm that both formatted description strings remain inside the surrounding parentheses.
Ask for help with the exact failing line: Help me diagnose this Python compilation error in my FastAPI application setup.
The request model is the boundary that closes the unsafe dictionary gap. Every accepted field now has a strict type and an inclusive repository-defined range.
- Add the strict prediction request model below the application definition:
class PredictionRequest(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True, allow_inf_nan=False)
weekly_km_to_date: Annotated[FiniteFloat, Field(ge=0.0, le=200.0)]
sleep_hours_last_night: Annotated[FiniteFloat, Field(ge=0.0, le=14.0)]
resting_hr_morning: Annotated[FiniteFloat, Field(ge=30.0, le=220.0)]
soreness_score_evening: Annotated[FiniteFloat, Field(ge=0.0, le=10.0)]
prior_injury: Annotated[StrictInt, Field(ge=0, le=1)]
How does the request gate work?
- `extra="forbid"` rejects fields outside the five-feature contract.
- `strict=True` prevents type coercion from hiding malformed values.
- `allow_inf_nan=False` and `FiniteFloat` keep non-finite values outside the runtime.
- `StrictInt` limits `prior_injury` to an actual integer within the range from `0` to `1`.
- Add the prediction response contract below `PredictionRequest`:
class PredictionResponse(BaseModel):
probability: Annotated[FiniteFloat, Field(ge=0.0, le=1.0)]
decision: Annotated[StrictInt, Field(ge=0, le=1)]
threshold: Annotated[FiniteFloat, Field(ge=0.0, le=1.0)]
device: str
selected_approach: str
What does the prediction response guarantee?
The response model constrains the probability and decision to their valid domains. It also makes the locked threshold, CPU device, and selected approach part of the documented contract.
- Save `src/level7_service.py`.
- Compile both prediction models by running this command:
python -m py_compile src/level7_service.py
What does this compilation check?
Python now parses the strict request model and the prediction response model as part of the same module. Returning to the prompt without a traceback confirms their annotations are syntactically valid.
You should return to the terminal prompt without a traceback.
Are the model annotations failing to compile?
Confirm that each field uses `Annotated` with matching square brackets. Check that `Field` remains inside each annotation.
Get help locating the malformed annotation: Help me find the syntax error in my strict Pydantic prediction models.
- Add the health response contract below `PredictionResponse`:
class HealthResponse(BaseModel):
status: str
integrity_status: str
device: str
selected_approach: str
evidence_scope: str
model_notice: str
inherited_v1_preprocessing_limitation: str
Why publish health metadata?
The health contract exposes operational integrity beside the scientific evidence limits. A caller can confirm what the service loaded without assuming the model is validated for coaching or medical use.
- Save `src/level7_service.py`.
- Compile the completed model section by running this command:
python -m py_compile src/level7_service.py
What does the clean return mean?
The module now contains all three API models. A clean terminal return confirms that Python can compile the complete contract layer.
You should return to the terminal prompt without a traceback.
Is the health model causing an error?
Check that every health field stays inside `HealthResponse`. Confirm that each annotation is `str`.
Ask for a focused comparison: Help me fix the HealthResponse definition in my FastAPI service.
Connect the verified runtime to both routes
The health route reports whether startup integrity verification passed. Its metadata also keeps the service's development-only evidence scope visible to every caller.
- Add the health route below `HealthResponse`:
@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
return HealthResponse(
status="ok",
integrity_status=runtime.integrity_status,
device=runtime.device,
selected_approach=runtime.handoff["selected_approach"],
evidence_scope=runtime.manifest["evidence_scope"],
model_notice=runtime.manifest["model_notice"],
inherited_v1_preprocessing_limitation=runtime.manifest[
"inherited_v1_preprocessing_limitation"
],
)
What does the health route expose?
- `integrity_status` reports the result established when `ModelRuntime` verified the packaged artifacts.
- `selected_approach` identifies the inherited Level 6 approach as `partial`.
- `evidence_scope` identifies the evidence as `development_only`.
- The final two fields preserve the synthetic educational notice and inherited preprocessing limitation.
- Save `src/level7_service.py`.
- Compile the health route by running this command:
python -m py_compile src/level7_service.py
What does this check cover?
The compilation check now includes the response model construction and manifest lookups inside `health()`. A clean return confirms the route is valid Python.
You should return to the terminal prompt without a traceback.
Is the health route failing to compile?
Check the closing bracket around `inherited_v1_preprocessing_limitation`. Confirm that the closing parenthesis for `HealthResponse` remains aligned with `return`.
Share the traceback without sharing artifact contents: Help me fix the bracket or indentation error in my health route.
The prediction route receives a validated model instead of a raw dictionary. Only the five approved fields reach the runtime through `request.model_dump()`.
- Add the strict prediction route below `health()`:
@app.post("/predict", response_model=PredictionResponse)
def predict(request: PredictionRequest) -> PredictionResponse:
return PredictionResponse(**runtime.predict(request.model_dump()))
Why does this route preserve parity?
The route delegates preprocessing and inference to the existing `ModelRuntime`. The response model documents the result without changing its probability or threshold decision.
- Save `src/level7_service.py`.
- Compile the completed service by running this command:
python -m py_compile src/level7_service.py
What does the final service check prove?
Python now compiles the complete strict service as one module. This catches malformed decorators, annotations, calls, and indentation before the service enters the packaged test stage.
You should return to the terminal prompt without a traceback.
Is the prediction route failing to compile?
Confirm that the route accepts `PredictionRequest`. Check that `request.model_dump()` remains inside `runtime.predict()`.
Ask for help tracing the nested call: Help me fix the strict prediction route in my FastAPI service.
✔️ Awesome, I've got everything!
Your final service now validates requests before inference. Save `src/level7_service.py` before continuing.
ⓧ I'd like to double check the full code
from __future__ import annotations
from pathlib import Path
from typing import Annotated
from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, Field, FiniteFloat, StrictInt
try:
from src.level7_runtime import ModelRuntime
except ModuleNotFoundError:
from level7_runtime import ModelRuntime
BASE_DIR = Path(__file__).resolve().parent
runtime = ModelRuntime(BASE_DIR)
app = FastAPI(
title="Running Readiness Level 7",
version="4.0.0",
description=(
f"{runtime.manifest['model_notice']} "
f"{runtime.manifest['inherited_v1_preprocessing_limitation']}"
),
)
class PredictionRequest(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True, allow_inf_nan=False)
weekly_km_to_date: Annotated[FiniteFloat, Field(ge=0.0, le=200.0)]
sleep_hours_last_night: Annotated[FiniteFloat, Field(ge=0.0, le=14.0)]
resting_hr_morning: Annotated[FiniteFloat, Field(ge=30.0, le=220.0)]
soreness_score_evening: Annotated[FiniteFloat, Field(ge=0.0, le=10.0)]
prior_injury: Annotated[StrictInt, Field(ge=0, le=1)]
class PredictionResponse(BaseModel):
probability: Annotated[FiniteFloat, Field(ge=0.0, le=1.0)]
decision: Annotated[StrictInt, Field(ge=0, le=1)]
threshold: Annotated[FiniteFloat, Field(ge=0.0, le=1.0)]
device: str
selected_approach: str
class HealthResponse(BaseModel):
status: str
integrity_status: str
device: str
selected_approach: str
evidence_scope: str
model_notice: str
inherited_v1_preprocessing_limitation: str
@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
return HealthResponse(
status="ok",
integrity_status=runtime.integrity_status,
device=runtime.device,
selected_approach=runtime.handoff["selected_approach"],
evidence_scope=runtime.manifest["evidence_scope"],
model_notice=runtime.manifest["model_notice"],
inherited_v1_preprocessing_limitation=runtime.manifest[
"inherited_v1_preprocessing_limitation"
],
)
@app.post("/predict", response_model=PredictionResponse)
def predict(request: PredictionRequest) -> PredictionResponse:
return PredictionResponse(**runtime.predict(request.model_dump()))
What should match?
Compare the imports, application metadata, three models, and two routes with your file. The order matters because the route decorators reference models defined above them.
Add parity and integrity tests
Contract tests must prove that bad inputs stop at the boundary. Independent parity tests must also prove that adding validation does not change valid Level 6 predictions.
- Create `test_level7_service.py` inside the existing `tests/` folder.
- Add the test dependencies and valid synthetic request:
from __future__ import annotations
import json
import shutil
import tempfile
import unittest
from pathlib import Path
import numpy as np
import torch
from fastapi.testclient import TestClient
from torch import nn
try:
from src.level7_runtime import CHECKPOINT_PATH, IntegrityError, ModelRuntime
from src.level7_service import app
except ModuleNotFoundError:
from level7_runtime import CHECKPOINT_PATH, IntegrityError, ModelRuntime
from level7_service import app
ROOT = Path(__file__).resolve().parents[1]
VALID_REQUEST = {
"weekly_km_to_date": 32.0,
"sleep_hours_last_night": 7.0,
"resting_hr_morning": 50.0,
"soreness_score_evening": 4.0,
"prior_injury": 0,
}
What does the test setup provide?
The imports support API requests, temporary corruption, independent numeric preprocessing, and checkpoint loading. `VALID_REQUEST` gives every positive-path test the same five-feature input.
- Add independent reference model classes below `VALID_REQUEST`:
class ReferenceEncoder(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 ReferenceTransferClassifier(nn.Module):
def __init__(self, input_dim: int):
super().__init__()
self.encoder = ReferenceEncoder(input_dim)
self.head = nn.Linear(8, 1)
def forward(self, inputs: torch.Tensor) -> torch.Tensor:
return self.head(self.encoder(inputs)).squeeze(1)
Why duplicate the architecture?
The reference classes reconstruct the Level 6 architecture independently of `ModelRuntime`. This prevents a parity test from passing merely because both sides call the same prediction implementation.
- Save `tests/test_level7_service.py`.
- Compile the test setup by running this command:
python -m py_compile tests/test_level7_service.py
What does this test compilation prove?
Python can parse the test imports, valid request, and independent reference classes. A clean return confirms that the architecture definitions have valid syntax.
You should return to the terminal prompt without a traceback.
Are the reference classes failing to compile?
Check the indentation beneath each class and method. Confirm that both nested `torch.relu` calls close before the final parenthesis.
Ask for help comparing the layers: Help me fix the independent reference model definitions in my service tests.
The reference prediction repeats preprocessing in the test suite. It reads the handoff, preprocessor, and checkpoint directly instead of calling the production preprocessing method.
- Start `reference_prediction()` below the reference classes with this code:
def reference_prediction(payload: dict) -> tuple[float, int, float]:
handoff = json.loads(
(ROOT / "artifacts/v4/level7_deploy_handoff.json").read_text(encoding="utf-8")
)
preprocessor = json.loads(
(ROOT / "artifacts/v4/level6_preprocessor.json").read_text(encoding="utf-8")
)
checkpoint = torch.load(
ROOT / CHECKPOINT_PATH, map_location=torch.device("cpu"), weights_only=True
)
values = np.asarray(
[[float(payload[name]) for name in handoff["feature_names"]]], dtype=np.float64
)
transformed = (
(
np.where(
np.isnan(values),
np.asarray(preprocessor["imputer_statistics"], dtype=np.float64),
values,
)
- np.asarray(preprocessor["scaler_mean"], dtype=np.float64)
)
/ np.asarray(preprocessor["scaler_scale"], dtype=np.float64)
).astype(np.float32)
What does this half calculate?
This section loads trusted artifacts and preserves the stored feature order. It repeats mean imputation, standard scaling, and `float32` conversion independently.
- Continue inside `reference_prediction()` with this indented code:
model = ReferenceTransferClassifier(checkpoint["input_dim"])
model.load_state_dict(checkpoint["model_state"], strict=True)
model.eval()
with torch.inference_mode():
probability = float(torch.sigmoid(model(torch.from_numpy(transformed))).item())
threshold = float(checkpoint["threshold"])
return probability, int(probability >= threshold), threshold
How is the reference result produced?
The reference loads exact state-dict keys into its own model classes. It applies sigmoid once before comparing the probability with the stored threshold.
- Save `tests/test_level7_service.py`.
- Compile the completed reference function by running this command:
python -m py_compile tests/test_level7_service.py
What does the reference compilation confirm?
The complete independent prediction path now compiles as one function. A clean return confirms that the multiline transformation and model operations form valid Python.
You should return to the terminal prompt without a traceback.
Is the reference function failing to compile?
Check that the model lines retain four spaces of indentation. Confirm that the transformation closes before `model` is created.
Ask for help matching the function boundary: Help me fix the reference_prediction function in my parity tests.
The first service tests cover health metadata and valid inference parity. Together they verify both the documented evidence boundary and the unchanged prediction path.
- Start `Level7ServiceTests` below `reference_prediction()` with these tests:
class Level7ServiceTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.client = TestClient(app)
def test_health_documents_integrity_and_limitations(self):
response = self.client.get("/health")
self.assertEqual(response.status_code, 200)
body = response.json()
self.assertEqual(body["status"], "ok")
self.assertEqual(body["integrity_status"], "passed")
self.assertEqual(body["device"], "cpu")
self.assertEqual(body["selected_approach"], "partial")
self.assertEqual(body["evidence_scope"], "development_only")
self.assertIn("Synthetic educational", body["model_notice"])
self.assertIn("not claim", body["inherited_v1_preprocessing_limitation"])
def test_prediction_matches_independent_reference(self):
probability, decision, threshold = reference_prediction(VALID_REQUEST)
response = self.client.post("/predict", json=VALID_REQUEST)
self.assertEqual(response.status_code, 200)
self.assertAlmostEqual(response.json()["probability"], probability, places=7)
self.assertEqual(response.json()["decision"], decision)
self.assertEqual(response.json()["threshold"], threshold)
What do the positive tests lock?
The health test requires passed integrity, CPU execution, the partial approach, and development-only evidence. The parity test compares the API probability to the independent reference at seven decimal places.
- Add the invalid-request tests inside `Level7ServiceTests`:
def test_invalid_requests_are_rejected(self):
cases = []
missing = dict(VALID_REQUEST)
missing.pop("sleep_hours_last_night")
cases.append(("missing", missing))
extra = dict(VALID_REQUEST, needs_recovery_next_day=1)
cases.append(("extra", extra))
wrong_type = dict(VALID_REQUEST, resting_hr_morning="50.0")
cases.append(("wrong type", wrong_type))
out_of_range = dict(VALID_REQUEST, soreness_score_evening=11.0)
cases.append(("out of range", out_of_range))
for label, payload in cases:
with self.subTest(label=label):
self.assertEqual(self.client.post("/predict", json=payload).status_code, 422)
def test_nonfinite_value_is_rejected(self):
payload = dict(VALID_REQUEST, weekly_km_to_date=float("nan"))
response = self.client.post(
"/predict",
content=json.dumps(payload),
headers={"content-type": "application/json"},
)
self.assertEqual(response.status_code, 422)
Which unsafe inputs are covered?
The table-driven test covers missing, extra, wrong-type, and out-of-range inputs. The separate non-finite test proves that `nan` also receives a `422` response.
- Save `tests/test_level7_service.py`.
- Compile the API behavior tests by running this command:
python -m py_compile tests/test_level7_service.py
What does this compilation check include?
Python now parses the positive API tests and every invalid-payload case. A clean return confirms that the class methods and nested subtests are syntactically valid.
You should return to the terminal prompt without a traceback.
Are the service tests failing to compile?
Check that every test method remains indented inside `Level7ServiceTests`. Confirm that the `for` loop and `with` block each add one indentation level.
Ask for help with the failing method: Help me fix the indentation or syntax in my Level 7 API tests.
The corruption test exercises the second fail-closed boundary. It changes one byte in a temporary checkpoint copy before constructing a new runtime.
- Add the temporary checkpoint corruption test inside `Level7ServiceTests`:
def test_startup_rejects_corrupted_temporary_checkpoint(self):
with tempfile.TemporaryDirectory() as directory:
temporary_root = Path(directory)
for relative in (
"serving-manifest.json",
"artifacts/v4/level7_deploy_handoff.json",
"artifacts/v4/level6_preprocessor.json",
"artifacts/v4/level6_selected_model.pt",
):
target = temporary_root / relative
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(ROOT / relative, target)
checkpoint = temporary_root / CHECKPOINT_PATH
data = bytearray(checkpoint.read_bytes())
data[len(data) // 2] ^= 0x01
checkpoint.write_bytes(data)
with self.assertRaises(IntegrityError):
ModelRuntime(temporary_root)
Why corrupt a temporary copy?
The test preserves the original selected checkpoint byte for byte. The expected `IntegrityError` proves that startup hashes the artifact before loading it.
The final bundle test confirms that the serving context contains only deployment evidence. Historical datasets and consumed-holdout evidence stay outside the image boundary.
- Finish `Level7ServiceTests` with the bundle exclusions and test runner:
def test_bundle_excludes_historical_and_holdout_files(self):
for relative in (
"data/v1/processed/train.csv",
"data/v1/processed/test.csv",
"artifacts/v1/holdout_consumed.json",
"reports/v1/level3_evaluation.json",
"reports/v4/finetune_experiment.json",
):
self.assertFalse((ROOT / relative).exists(), relative)
self.assertTrue((ROOT / "lineage-attestation.json").is_file())
self.assertTrue((ROOT / "serving-manifest.json").is_file())
if __name__ == "__main__":
unittest.main()
What does the bundle test enforce?
The test rejects development data, the consumed test split, the holdout marker, and historical reports inside the serving bundle. It also requires the lineage attestation and serving manifest.
- Save `tests/test_level7_service.py`.
Before you run the final syntax check, do you expect both completed files to compile cleanly?
- Compile the final service and test suite together by running this command:
python -m py_compile src/level7_service.py tests/test_level7_service.py
What does the final check establish?
The command compiles both completed files in one pass. A clean return confirms that the strict service and its full test suite are ready for the mandatory packaged test stage.
You should return to the terminal prompt without a traceback. That clean result confirms that the complete contract and parity suite is syntactically ready for packaging.
Is one file still failing to compile?
Use the traceback path to identify whether the service or test file failed. Compare the reported line with the corresponding full-code tab.
Ask for help with the exact traceback: Help me diagnose the final compilation failure in my Level 7 service or tests.
✔️ Awesome, I've got everything!
Your service tests now cover strict validation, independent parity, startup corruption, health metadata, and bundle exclusions. Save `tests/test_level7_service.py`.
ⓧ I'd like to double check the full code
from __future__ import annotations
import json
import shutil
import tempfile
import unittest
from pathlib import Path
import numpy as np
import torch
from fastapi.testclient import TestClient
from torch import nn
try:
from src.level7_runtime import CHECKPOINT_PATH, IntegrityError, ModelRuntime
from src.level7_service import app
except ModuleNotFoundError:
from level7_runtime import CHECKPOINT_PATH, IntegrityError, ModelRuntime
from level7_service import app
ROOT = Path(__file__).resolve().parents[1]
VALID_REQUEST = {
"weekly_km_to_date": 32.0,
"sleep_hours_last_night": 7.0,
"resting_hr_morning": 50.0,
"soreness_score_evening": 4.0,
"prior_injury": 0,
}
class ReferenceEncoder(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 ReferenceTransferClassifier(nn.Module):
def __init__(self, input_dim: int):
super().__init__()
self.encoder = ReferenceEncoder(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 reference_prediction(payload: dict) -> tuple[float, int, float]:
handoff = json.loads(
(ROOT / "artifacts/v4/level7_deploy_handoff.json").read_text(encoding="utf-8")
)
preprocessor = json.loads(
(ROOT / "artifacts/v4/level6_preprocessor.json").read_text(encoding="utf-8")
)
checkpoint = torch.load(
ROOT / CHECKPOINT_PATH, map_location=torch.device("cpu"), weights_only=True
)
values = np.asarray(
[[float(payload[name]) for name in handoff["feature_names"]]], dtype=np.float64
)
transformed = (
(
np.where(
np.isnan(values),
np.asarray(preprocessor["imputer_statistics"], dtype=np.float64),
values,
)
- np.asarray(preprocessor["scaler_mean"], dtype=np.float64)
)
/ np.asarray(preprocessor["scaler_scale"], dtype=np.float64)
).astype(np.float32)
model = ReferenceTransferClassifier(checkpoint["input_dim"])
model.load_state_dict(checkpoint["model_state"], strict=True)
model.eval()
with torch.inference_mode():
probability = float(torch.sigmoid(model(torch.from_numpy(transformed))).item())
threshold = float(checkpoint["threshold"])
return probability, int(probability >= threshold), threshold
class Level7ServiceTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.client = TestClient(app)
def test_health_documents_integrity_and_limitations(self):
response = self.client.get("/health")
self.assertEqual(response.status_code, 200)
body = response.json()
self.assertEqual(body["status"], "ok")
self.assertEqual(body["integrity_status"], "passed")
self.assertEqual(body["device"], "cpu")
self.assertEqual(body["selected_approach"], "partial")
self.assertEqual(body["evidence_scope"], "development_only")
self.assertIn("Synthetic educational", body["model_notice"])
self.assertIn("not claim", body["inherited_v1_preprocessing_limitation"])
def test_prediction_matches_independent_reference(self):
probability, decision, threshold = reference_prediction(VALID_REQUEST)
response = self.client.post("/predict", json=VALID_REQUEST)
self.assertEqual(response.status_code, 200)
self.assertAlmostEqual(response.json()["probability"], probability, places=7)
self.assertEqual(response.json()["decision"], decision)
self.assertEqual(response.json()["threshold"], threshold)
def test_invalid_requests_are_rejected(self):
cases = []
missing = dict(VALID_REQUEST)
missing.pop("sleep_hours_last_night")
cases.append(("missing", missing))
extra = dict(VALID_REQUEST, needs_recovery_next_day=1)
cases.append(("extra", extra))
wrong_type = dict(VALID_REQUEST, resting_hr_morning="50.0")
cases.append(("wrong type", wrong_type))
out_of_range = dict(VALID_REQUEST, soreness_score_evening=11.0)
cases.append(("out of range", out_of_range))
for label, payload in cases:
with self.subTest(label=label):
self.assertEqual(self.client.post("/predict", json=payload).status_code, 422)
def test_nonfinite_value_is_rejected(self):
payload = dict(VALID_REQUEST, weekly_km_to_date=float("nan"))
response = self.client.post(
"/predict",
content=json.dumps(payload),
headers={"content-type": "application/json"},
)
self.assertEqual(response.status_code, 422)
def test_startup_rejects_corrupted_temporary_checkpoint(self):
with tempfile.TemporaryDirectory() as directory:
temporary_root = Path(directory)
for relative in (
"serving-manifest.json",
"artifacts/v4/level7_deploy_handoff.json",
"artifacts/v4/level6_preprocessor.json",
"artifacts/v4/level6_selected_model.pt",
):
target = temporary_root / relative
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(ROOT / relative, target)
checkpoint = temporary_root / CHECKPOINT_PATH
data = bytearray(checkpoint.read_bytes())
data[len(data) // 2] ^= 0x01
checkpoint.write_bytes(data)
with self.assertRaises(IntegrityError):
ModelRuntime(temporary_root)
def test_bundle_excludes_historical_and_holdout_files(self):
for relative in (
"data/v1/processed/train.csv",
"data/v1/processed/test.csv",
"artifacts/v1/holdout_consumed.json",
"reports/v1/level3_evaluation.json",
"reports/v4/finetune_experiment.json",
):
self.assertFalse((ROOT / relative).exists(), relative)
self.assertTrue((ROOT / "lineage-attestation.json").is_file())
self.assertTrue((ROOT / "serving-manifest.json").is_file())
if __name__ == "__main__":
unittest.main()
What should match?
Compare the import fallback, independent model, reference prediction, six test methods, and final test runner with your file. The corruption test must modify only its temporary checkpoint copy.
Your API contract now rejects the unsafe inputs that the temporary dictionary accepted. Next, you will place this service behind a gated Docker build that must pass the full test suite before producing the runtime image.
Build the Gated Serving Image
Your strict FastAPI service now rejects invalid requests. It also refuses corrupted serving artifacts before model loading.
The remaining risk sits outside the API. An oversized Docker build context could still receive historical datasets or evaluation evidence. This step creates an explicit allowlist behind two fail-closed gates.
In this step, get ready to:
- Generate a serving context only after the local lineage gate and inherited Level 6 audit pass.
- Build a multi-stage image that requires the service tests to pass.
- Run the packaged API locally to verify its health metadata and prediction response.
Stage only trusted serving files
A build context defines every file Docker can receive during a build. An allowlist gives the packaging command a small set of approved serving files to copy.
Why an allowlist?
Building from the repository directory would expose every local path to Docker. The generated context contains only runtime code, test code, dependency pins, selected artifacts, and integrity metadata.
The lineage gate runs before the old context is replaced. A failed gate preserves the previous trusted context and prevents the build command from running.
- In the VS Code Explorer, select the repository-level .gitignore file.
- Replace its contents with the following ignore policy:
# Python environments and cache
.venv/
venv/
__pycache__/
*.py[cod]
.pytest_cache/
.mypy_cache/
.ruff_cache/
# macOS and VS Code
.DS_Store
.vscode/
# Secrets
.env
.env.*
!.env.example
# Local datasets
data/v1/raw/
data/v1/cleaned/
data/v1/processed/
# Consumed holdout and evaluation evidence
artifacts/v1/holdout_consumed.json
reports/v1/level3_errors.csv
reports/v1/level3_evaluation.json
reports/v1/level3_evaluation.md
# Serialized models
models/v1/*.joblib
# Temporary evidence bundles
level4-evidence/
level4-evidence.zip
# Generated Level 7 serving context
build/level7-context/
What does this ignore policy protect?
- The dataset rules keep local development data outside version control.
- The evidence rules keep consumed holdout files and historical evaluation reports local.
- The final rule prevents the generated serving context from becoming a committed artifact.
- Save .gitignore.
- Confirm that build/level7-context/ appears under the generated Level 7 serving context comment.
Missing the Level 7 ignore rule?
Check that the path is spelled build/level7-context/ with the trailing slash. Place it below the generated Level 7 serving context comment.
Ask for help checking the ignore file.
- In the repository directory shown in the VS Code Explorer, create requirements-level7.txt.
- Add the runtime dependency pins below:
fastapi==0.162.0
numpy==2.5.4
pydantic==2.14.2
uvicorn==0.54.0
What belongs in the runtime image?
- The FastAPI pin provides the request routes and generated API documentation.
- The NumPy pin preserves the preprocessing operations used by the runtime.
- The Pydantic pin preserves the strict request and response schemas.
- The Uvicorn pin serves the application inside the final image.
- Save requirements-level7.txt.
- Confirm that the file contains exactly four pinned runtime dependencies.
- In the repository directory shown in the VS Code Explorer, create requirements-level7-test.txt.
- Add the test-only dependency pin below:
httpx==0.28.1
Why isolate the test dependency?
FastAPI's test client uses HTTPX to exercise the routes. Keeping it in a separate file lets the test stage install it without carrying it into the runtime stage.
- Save requirements-level7-test.txt.
- Confirm that the file contains only httpx==0.28.1.
- In the VS Code Explorer, create level7_package.py inside the existing src directory.
- Add the imports, image names, artifact paths, and allowlist below:
from __future__ import annotations
import argparse
import json
import shutil
import subprocess
from pathlib import Path
from typing import Mapping
from src.level7_lineage import PROJECT_ROOT, audit_lineage, read_json, sha256
CONTEXT_PATH = Path("build/level7-context")
IMAGE = "running-readiness-level7:4.0.0"
LEVEL6_IMAGE = "running-readiness-level6:2.15.0"
HANDOFF_RELATIVE = Path("artifacts/v4/level7_deploy_handoff.json")
CHECKPOINT_RELATIVE = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_RELATIVE = Path("artifacts/v4/level6_preprocessor.json")
ALLOWLIST = {
Path("Dockerfile.level7"): Path("Dockerfile"),
Path("requirements-level7.txt"): Path("requirements-level7.txt"),
Path("requirements-level7-test.txt"): Path("requirements-level7-test.txt"),
Path("src/level7_runtime.py"): Path("level7_runtime.py"),
Path("src/level7_service.py"): Path("level7_service.py"),
Path("tests/test_level7_service.py"): Path("test_level7_service.py"),
CHECKPOINT_RELATIVE: CHECKPOINT_RELATIVE,
PREPROCESSOR_RELATIVE: PREPROCESSOR_RELATIVE,
HANDOFF_RELATIVE: HANDOFF_RELATIVE,
}
What does this packaging map do?
- The image constants lock the inherited audit environment and the new serving image tag.
- The artifact constants identify the selected checkpoint, preprocessor, and deployment handoff.
- The ALLOWLIST map separates repository source paths from their smaller context destinations.
- Save src/level7_package.py.
- Confirm that the VS Code Explorer lists level7_package.py under src.
Packaging file missing from src?
Check that level7_package.py sits beside level7_lineage.py. A repository-level copy cannot import the module through the expected package path.
Ask for help checking the packaging file location.
- In src/level7_package.py, place the following functions below ALLOWLIST:
def run_level6_audit(root: Path, runner=subprocess.run) -> None:
runner(
[
"docker", "run", "--rm", "--platform", "linux/amd64", "--mount",
f"type=bind,src={root},dst=/workspace", "-w", "/workspace",
LEVEL6_IMAGE, "python", "-m", "src.finetune_experiment", "audit",
],
cwd=root,
check=True,
)
def create_serving_manifest(root: Path) -> dict:
handoff = read_json(root / HANDOFF_RELATIVE)
level5 = read_json(root / "artifacts/v2/level5_features_handoff.json")
return {
"artifacts": {
CHECKPOINT_RELATIVE.as_posix(): handoff["fingerprints"]["selected_checkpoint"],
PREPROCESSOR_RELATIVE.as_posix(): handoff["fingerprints"]["preprocessor"],
HANDOFF_RELATIVE.as_posix(): sha256(root / HANDOFF_RELATIVE),
},
"evidence_scope": handoff["evidence_scope"],
"model_notice": level5["scientific_status"],
"inherited_v1_preprocessing_limitation": handoff["limitations"][0],
}
What do these functions enforce?
- The inherited audit runs inside the trusted Level 6 CPU image with the repository mounted read-only by policy.
- The serving manifest roots checkpoint and preprocessor hashes in the verified Level 7 handoff.
- The manifest carries the evidence scope, model notice, and inherited preprocessing limitation into the service.
- Save src/level7_package.py.
- Open the VS Code Outline view for the file.
- Confirm that run_level6_audit and create_serving_manifest appear as function symbols.
Function missing from the outline?
Check the indentation below each function declaration. Also confirm that the new functions begin after the closing brace of ALLOWLIST.
Ask for help finding the syntax issue.
- In src/level7_package.py, place the context-staging function below create_serving_manifest:
def stage_context(
root: Path = PROJECT_ROOT,
overrides: Mapping[Path, Path] | None = None,
runner=subprocess.run,
) -> Path:
attestation = audit_lineage(root, overrides)
run_level6_audit(root, runner)
context = root / CONTEXT_PATH
if context.exists():
shutil.rmtree(context)
context.mkdir(parents=True)
for source_relative, target_relative in ALLOWLIST.items():
source = root / source_relative
if not source.is_file():
raise FileNotFoundError(f"Missing allowlisted package file: {source}")
target = context / target_relative
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
(context / "lineage-attestation.json").write_text(
json.dumps(attestation, indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
(context / "serving-manifest.json").write_text(
json.dumps(create_serving_manifest(root), indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
return context
Where does the fail-closed order live?
- The local audit_lineage call runs first.
- The inherited run_level6_audit call runs second.
- The existing context is removed only after both calls return successfully.
- The copy loop rejects any missing allowlisted source before writing the integrity files.
- Save src/level7_package.py.
- Confirm that stage_context appears in the VS Code Outline view.
Staging function not recognized?
Confirm that stage_context starts at the left edge of the file. Check that each write_text call closes before the next statement.
Ask for help checking the function structure.
- In src/level7_package.py, place the build and command-dispatch functions below stage_context:
def build_image(root: Path = PROJECT_ROOT, runner=subprocess.run) -> None:
context = stage_context(root, runner=runner)
runner(
["docker", "build", "--platform", "linux/amd64", "-t", IMAGE, str(context)],
cwd=root,
check=True,
)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=("audit", "stage", "build"))
args = parser.parse_args()
if args.command == "audit":
print(json.dumps(audit_lineage(), indent=2, sort_keys=True))
elif args.command == "stage":
print(stage_context())
else:
build_image()
print(IMAGE)
if __name__ == "__main__":
main()
How does the build command stay gated?
- The build_image function receives its context from stage_context.
- Docker receives the generated context only after staging succeeds.
- The command parser exposes separate audit, stage, and build paths for testing the workflow.
- Save src/level7_package.py.
- Confirm that the VS Code Outline view lists build_image and main.
Build functions missing from the outline?
Check that build_image follows the closing line of stage_context. Confirm that the module guard sits at the left edge.
Ask for help checking the final packaging functions.
Make image tests mandatory
A multi-stage build gives testing and runtime separate dependency sets. The final stage can inherit proof that tests passed without inheriting test packages or test source.
- In the repository directory shown in the VS Code Explorer, create Dockerfile.level7.
- Add the complete multi-stage build below:
# syntax=docker/dockerfile:1
FROM running-readiness-level6:2.15.0 AS base
WORKDIR /app
COPY requirements-level7.txt /tmp/requirements-level7.txt
RUN python -m pip install --no-cache-dir -r /tmp/requirements-level7.txt
COPY level7_runtime.py level7_service.py ./
COPY artifacts/v4/level6_selected_model.pt artifacts/v4/level6_selected_model.pt
COPY artifacts/v4/level6_preprocessor.json artifacts/v4/level6_preprocessor.json
COPY artifacts/v4/level7_deploy_handoff.json artifacts/v4/level7_deploy_handoff.json
COPY lineage-attestation.json serving-manifest.json ./
FROM base AS test
COPY requirements-level7-test.txt /tmp/requirements-level7-test.txt
RUN python -m pip install --no-cache-dir -r /tmp/requirements-level7-test.txt
RUN mkdir -p /app/tests
COPY test_level7_service.py /app/tests/test_level7_service.py
RUN python -m unittest -v tests/test_level7_service.py && \
touch /tmp/level7-tests-passed
FROM base AS runtime
COPY --from=test /tmp/level7-tests-passed /app/level7-tests-passed
EXPOSE 8000
CMD ["uvicorn", "level7_service:app", "--host", "0.0.0.0", "--port", "8000"]
How do the three stages work?
- The base stage inherits the verified Level 6 PyTorch CPU environment. It adds only runtime dependencies and serving artifacts.
- The test stage adds HTTPX and the service test source. It creates the pass marker only after the test suite succeeds.
- The runtime stage copies the pass marker from the test stage. It does not inherit the test stage filesystem.
- The final command starts the strict API on container port 8000.
- Save Dockerfile.level7.
- Confirm that the file contains the base, test, and runtime stages.
Dockerfile shows a syntax warning?
Check the continuation character after the test command. Confirm that the runtime command remains a valid JSON array.
Ask for help comparing the stages.
Use the reference below if you want to compare every packaging file before the gated build.
✔️ Awesome, I've got everything!
Save all five packaging files. Your project is ready for the gated build.
ⓧ I'd like to double check the full code
# Python environments and cache
.venv/
venv/
__pycache__/
*.py[cod]
.pytest_cache/
.mypy_cache/
.ruff_cache/
# macOS and VS Code
.DS_Store
.vscode/
# Secrets
.env
.env.*
!.env.example
# Local datasets
data/v1/raw/
data/v1/cleaned/
data/v1/processed/
# Consumed holdout and evaluation evidence
artifacts/v1/holdout_consumed.json
reports/v1/level3_errors.csv
reports/v1/level3_evaluation.json
reports/v1/level3_evaluation.md
# Serialized models
models/v1/*.joblib
# Temporary evidence bundles
level4-evidence/
level4-evidence.zip
# Generated Level 7 serving context
build/level7-context/
fastapi==0.162.0
numpy==2.5.4
pydantic==2.14.2
uvicorn==0.54.0
httpx==0.28.1
from __future__ import annotations
import argparse
import json
import shutil
import subprocess
from pathlib import Path
from typing import Mapping
from src.level7_lineage import PROJECT_ROOT, audit_lineage, read_json, sha256
CONTEXT_PATH = Path("build/level7-context")
IMAGE = "running-readiness-level7:4.0.0"
LEVEL6_IMAGE = "running-readiness-level6:2.15.0"
HANDOFF_RELATIVE = Path("artifacts/v4/level7_deploy_handoff.json")
CHECKPOINT_RELATIVE = Path("artifacts/v4/level6_selected_model.pt")
PREPROCESSOR_RELATIVE = Path("artifacts/v4/level6_preprocessor.json")
ALLOWLIST = {
Path("Dockerfile.level7"): Path("Dockerfile"),
Path("requirements-level7.txt"): Path("requirements-level7.txt"),
Path("requirements-level7-test.txt"): Path("requirements-level7-test.txt"),
Path("src/level7_runtime.py"): Path("level7_runtime.py"),
Path("src/level7_service.py"): Path("level7_service.py"),
Path("tests/test_level7_service.py"): Path("test_level7_service.py"),
CHECKPOINT_RELATIVE: CHECKPOINT_RELATIVE,
PREPROCESSOR_RELATIVE: PREPROCESSOR_RELATIVE,
HANDOFF_RELATIVE: HANDOFF_RELATIVE,
}
def run_level6_audit(root: Path, runner=subprocess.run) -> None:
runner(
[
"docker", "run", "--rm", "--platform", "linux/amd64", "--mount",
f"type=bind,src={root},dst=/workspace", "-w", "/workspace",
LEVEL6_IMAGE, "python", "-m", "src.finetune_experiment", "audit",
],
cwd=root,
check=True,
)
def create_serving_manifest(root: Path) -> dict:
handoff = read_json(root / HANDOFF_RELATIVE)
level5 = read_json(root / "artifacts/v2/level5_features_handoff.json")
return {
"artifacts": {
CHECKPOINT_RELATIVE.as_posix(): handoff["fingerprints"]["selected_checkpoint"],
PREPROCESSOR_RELATIVE.as_posix(): handoff["fingerprints"]["preprocessor"],
HANDOFF_RELATIVE.as_posix(): sha256(root / HANDOFF_RELATIVE),
},
"evidence_scope": handoff["evidence_scope"],
"model_notice": level5["scientific_status"],
"inherited_v1_preprocessing_limitation": handoff["limitations"][0],
}
def stage_context(
root: Path = PROJECT_ROOT,
overrides: Mapping[Path, Path] | None = None,
runner=subprocess.run,
) -> Path:
attestation = audit_lineage(root, overrides)
run_level6_audit(root, runner)
context = root / CONTEXT_PATH
if context.exists():
shutil.rmtree(context)
context.mkdir(parents=True)
for source_relative, target_relative in ALLOWLIST.items():
source = root / source_relative
if not source.is_file():
raise FileNotFoundError(f"Missing allowlisted package file: {source}")
target = context / target_relative
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
(context / "lineage-attestation.json").write_text(
json.dumps(attestation, indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
(context / "serving-manifest.json").write_text(
json.dumps(create_serving_manifest(root), indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
return context
def build_image(root: Path = PROJECT_ROOT, runner=subprocess.run) -> None:
context = stage_context(root, runner=runner)
runner(
["docker", "build", "--platform", "linux/amd64", "-t", IMAGE, str(context)],
cwd=root,
check=True,
)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=("audit", "stage", "build"))
args = parser.parse_args()
if args.command == "audit":
print(json.dumps(audit_lineage(), indent=2, sort_keys=True))
elif args.command == "stage":
print(stage_context())
else:
build_image()
print(IMAGE)
if __name__ == "__main__":
main()
# syntax=docker/dockerfile:1
FROM running-readiness-level6:2.15.0 AS base
WORKDIR /app
COPY requirements-level7.txt /tmp/requirements-level7.txt
RUN python -m pip install --no-cache-dir -r /tmp/requirements-level7.txt
COPY level7_runtime.py level7_service.py ./
COPY artifacts/v4/level6_selected_model.pt artifacts/v4/level6_selected_model.pt
COPY artifacts/v4/level6_preprocessor.json artifacts/v4/level6_preprocessor.json
COPY artifacts/v4/level7_deploy_handoff.json artifacts/v4/level7_deploy_handoff.json
COPY lineage-attestation.json serving-manifest.json ./
FROM base AS test
COPY requirements-level7-test.txt /tmp/requirements-level7-test.txt
RUN python -m pip install --no-cache-dir -r /tmp/requirements-level7-test.txt
RUN mkdir -p /app/tests
COPY test_level7_service.py /app/tests/test_level7_service.py
RUN python -m unittest -v tests/test_level7_service.py && \
touch /tmp/level7-tests-passed
FROM base AS runtime
COPY --from=test /tmp/level7-tests-passed /app/level7-tests-passed
EXPOSE 8000
CMD ["uvicorn", "level7_service:app", "--host", "0.0.0.0", "--port", "8000"]
Before you run the build, predict where it stops if a local historical file has changed. Will Docker receive a context?
- Build the verified linux/amd64 image from the repository directory by running:
python -m src.level7_package build
What happens during this command?
- The standard-library lineage gate verifies the complete inherited chain first.
- The Level 6 image runs its inherited audit second.
- The packaging command replaces build/level7-context only after both gates pass.
- The Docker test stage runs the service suite before the runtime image can be produced.
The first build can take several minutes while Docker installs the pinned dependencies. The terminal eventually prints running-readiness-level7:4.0.0 after the mandatory tests pass.
- In the VS Code Explorer, expand build/level7-context.
- Confirm that the context contains the three selected artifacts under artifacts/v4.
- Confirm that the context contains lineage-attestation.json and serving-manifest.json.
- Confirm that the context has no data directory.
- Confirm that the context has no reports directory.
- Confirm that the context has no artifacts/v1 directory.
Build stopped before Docker?
A lineage mismatch or failed Level 6 audit is a safe stop. Read the first reported path before changing anything.
Confirm that the three gitignored historical files remain at their original repository paths. Restore trusted bytes from your existing Level 1 through Level 6 checkout if a fingerprint changed.
Ask for help interpreting the failed gate.
Run and exercise the packaged API
The final check crosses both integrity boundaries. The local gate has approved the bundle. The service must now re-hash its packaged artifacts before serving requests.
Before you start the container, predict what startup should do before loading the checkpoint. Which three packaged artifacts should it verify?
- Start the image with a localhost-only port binding by running:
docker run --rm --platform linux/amd64 -p 127.0.0.1:8000:8000 running-readiness-level7:4.0.0
What does this container command do?
- The platform option runs the image built for the Intel-compatible target.
- The port mapping exposes container port 8000 only through local address 127.0.0.1.
- The removal option deletes the temporary container when you stop it.
You'll see Uvicorn startup logs with no integrity failure. Leave this terminal running while you check the API.
Container exits during startup?
A missing or mismatched serving artifact causes a deliberate startup refusal. Rebuild the image from the verified generated context before retrying.
If port 8000 is already occupied, stop the earlier local container before starting this image.
Ask for help interpreting the startup logs.
- Open http://127.0.0.1:8000/health in your browser.
You'll see status set to ok. You'll also see integrity_status set to passed.
The response also reports CPU execution, the partial approach, development-only evidence, the synthetic educational notice, and the inherited preprocessing limitation.
- Open http://127.0.0.1:8000/docs in your browser.
- Expand the /predict route.
- Select Try it out.
- Replace the request body with this synthetic five-feature payload:
{
"weekly_km_to_date": 32.0,
"sleep_hours_last_night": 7.0,
"resting_hr_morning": 50.0,
"soreness_score_evening": 4.0,
"prior_injury": 0
}
What does this request prove?
The payload follows the locked five-feature order and the strict repository-defined ranges. It contains no target field or historical holdout value.
The response passes through the packaged preprocessor and selected Level 6 checkpoint. This closes the parity loop through the actual HTTP boundary.
Before you execute the request, predict which response fields prove the threshold and runtime stayed locked.
- Select Execute.
You'll receive a successful response with a finite probability. The decision is 0 or 1. The threshold is 0.5.
The same response reports cpu as the device and partial as the selected approach.
Prediction request did not succeed?
Confirm that all five request fields are present. Check that prior_injury is the integer 0.
A validation response means the strict API rejected the submitted body before inference. Compare the highlighted field with the payload above.
Ask for help reading the response.
That's the complete serving boundary working. Your API now preserves Level 6 behavior while refusing unverified lineage, corrupted artifacts, and invalid requests.
Secret mission
Corrupt a Checkpoint and Prove Fail-Closed Packaging
A successful build proves the trusted path. This challenge corrupts a temporary checkpoint copy to prove packaging stops before Docker receives altered bytes while the original checkpoint remains unchanged.
Clean Up Your Resources
Clean Up Your Resources
Choose whether to keep your local resources running, pause the service for later, or remove the generated resources entirely. Every resource stays on your machine, so this project creates no ongoing cloud or service costs.
Resources you used:
- A verified Docker image tagged running-readiness-level7:4.0.0.
- An optional local container bound to port 8000.
- A generated serving context stored at build/level7-context.
Keep everything running
No action is needed. Choose this option if you plan to keep demonstrating the verified API or extending its deployment controls.
- Your optional container stays available at http://127.0.0.1:8000/docs while Docker Desktop is running.
- Your verified image remains ready for another local container.
- Your generated context remains available for inspecting the allowlisted serving bundle.
- These local resources create no ongoing financial cost.
Pause - I'll come back to this later
Pausing stops the local service while preserving the verified image and generated context. This frees the container's active CPU and memory use without discarding your packaged work.
- Locate the optional container bound to host port 8000 in Docker Desktop.
- Stop the container with the control beside its entry.
- Visit http://127.0.0.1:8000/health in your browser.
Your browser should report that it cannot reach the local service. The verified image and build/level7-context remain available for your next session.
Delete - I don't want to use this again
Deleting these resources removes only the Level 7 container, image, and generated serving context. Your Level 1 through Level 6 artifacts remain unchanged.
Finding the optional container can be fiddly because the project does not assign it a fixed name. Its host port 8000 identifies the correct entry.
- Locate the optional container bound to host port 8000 in Docker Desktop.
- Stop the container with the control beside its entry.
- Remove the stopped container with its deletion control.
- Locate the image tagged running-readiness-level7:4.0.0 in Docker Desktop's local image list.
- Remove the verified image with its deletion control.
Docker Desktop should no longer list the Level 7 container or image. The final generated resource is the allowlisted context inside your local repository.
macOS
- Return to the repository root in your terminal.
- Delete the generated serving context by running this command:
rm -rf build/level7-context
What Does This Remove?
The command targets only build/level7-context. Your committed source files and historical artifacts stay in place.
Windows
- Return to the repository root in Command Prompt.
- Delete the generated serving context by running this command:
rmdir /s /q build\level7-context
What Does This Remove?
The command targets only build/level7-context. Your committed source files and historical artifacts stay in place.
- Check Docker Desktop to confirm that no container remains bound to host port 8000.
- Check Docker Desktop to confirm that running-readiness-level7:4.0.0 no longer appears.
- Check the repository file tree in VS Code to confirm that build/level7-context no longer appears.
Cleanup is complete. Your verified historical artifacts and the original selected checkpoint remain untouched.
Nice Work!
Nice Work!
You did it! Your local running-readiness service now preserves verified Level 6 behavior through a fail-closed deployment path.
What you learned:
- Built a fail-closed lineage gate that validates inherited artifacts with exact SHA-256 hashes before packaging. The gate preserves the consumed-holdout policy. It verifies excluded evidence at its original local paths.
- Recreated the Level 6 CPU inference path behind a strict FastAPI contract. Independent tests confirm preprocessing parity. They also confirm probability parity at the locked threshold. Invalid prediction inputs now receive validation errors.
- Packaged the service into a tested Docker image from an explicit allowlist. Startup integrity checks refuse altered serving artifacts. The generated context contains serving files only.
- Secret Mission: Added a negative supply-chain test that flips one byte in a temporary checkpoint. The corruption stops packaging before Docker runs. The original selected checkpoint remains unchanged.
Ready to quiz yourself?