from __future__ import annotations import hashlib import json from datetime import timedelta from pathlib import Path from typing import Any from django.db import transaction from django.utils import timezone from control_plane.events.bus import EventBus from control_plane.model_studio.backends import BackendResult, FakeTrainingBackend, TrainingBackend from control_plane.model_studio.models import ( BenchmarkResult, CheckpointType, CheckpointValidityStatus, Conclusion, Dataset, DatasetValidationStatus, DatasetVersion, DatasetCurationProposal, EvaluationRun, EvaluationRunStatus, EvaluationSuite, EvaluationSuiteVersion, ExperimentStatus, FailureCluster, ModelCheckpoint, ModelPromotionDecision, ModelPromotionPolicy, ModelStudioArtifact, OvernightResearchReport, OvernightTrainingProgram, ProgramStatus, PromotionDecision, TrainingExperiment, TrainingProject, TrainingProjectStatus, TrainingRecipe, TrainingRun, TrainingRunStatus, ) from control_plane.model_studio.profiles import GuardModelProfile, ModelProjectProfile, SparkGuardDatasetInventory, SparkGuardDatasetMaterializer from control_plane.projects.models import Project, ProjectStatus from model_router.providers import extract_json_object from model_router.router import ModelCapability, ModelRequestContract, ModelRouter class ModelStudioService: def __init__(self, *, profile: ModelProjectProfile | None = None, backend: TrainingBackend | None = None, bus: EventBus | None = None, router: ModelRouter | None = None, dataset_curator_model_hint: str = "qwen") -> None: self.profile = profile or GuardModelProfile() self.backend = backend or FakeTrainingBackend() self.bus = bus or EventBus() self.router = router self.dataset_curator_model_hint = dataset_curator_model_hint def import_guard(self, *, project: Project | None = None, repository_path: str, spark_working_directory: str = "", slug: str = "guard-3b") -> TrainingProject: project = project or Project.objects.create(name="Guard 3B Model Studio", project_type="MODEL", goal="Reconstruct and safely improve the Guard 3B model.", repository_path=repository_path, status=ProjectStatus.ARCHAEOLOGY) training_project, _ = TrainingProject.objects.update_or_create(slug=slug, defaults={"project": project, "name": "Guard 3B", "description": "ForgeGuard Qwen2.5-Coder-3B model-development program.", "goal": "Improve Guard only through reproducible, benchmarked scientific experiments.", "capability_target": "Source-grounded smart-contract security findings.", "model_family": "Qwen2.5-Coder", "model_size": "3B", "base_model": GuardModelProfile.base_model, "repository_path": repository_path, "working_directory": spark_working_directory, "default_profile": self.profile.name, "training_backend": type(self.backend).__name__, "status": TrainingProjectStatus.IMPORTING, "metadata": {"spark_working_directory_verified": False, "local_archaeology_repository": repository_path}}) self._event("TRAINING_PROJECT_IMPORTED", training_project, {"repository_path": repository_path}) ModelPromotionPolicy.objects.get_or_create(training_project=training_project, name="Guard conservative V0.1", version="v0.1", defaults={"criteria": {"primary_metric": "strict_canonical_match", "minimum_delta": 0.0, "max_regression": {}, "requires_fresh_baseline": True, "note": "Thresholds remain unconfigured until the imported Guard benchmark exposes comparable metrics."}}) return training_project def declare_base_champion(self, training_project: TrainingProject) -> ModelCheckpoint: """Record the human-mandated starting base model without claiming benchmark evidence.""" checkpoint, _ = ModelCheckpoint.objects.get_or_create(training_project=training_project, reference=training_project.base_model, defaults={"name": "Guard starting base model", "checkpoint_type": CheckpointType.CHAMPION, "content_hash": self._hash_text(training_project.base_model), "validity_status": CheckpointValidityStatus.UNKNOWN, "load_verified": False, "metadata": {"selection": "HUMAN_MANDATED_STARTING_CHAMPION", "evidence_status": "PENDING_FRESH_SPARK_BASELINE"}}) training_project.current_champion = checkpoint training_project.status = TrainingProjectStatus.BASELINING training_project.save(update_fields=["current_champion", "status", "updated_at"]) self._event("CHAMPION_SELECTED", training_project, {"checkpoint": str(checkpoint.id), "selection": "HUMAN_MANDATED_STARTING_CHAMPION", "evidence_status": "PENDING_FRESH_SPARK_BASELINE"}) return checkpoint def archaeology(self, training_project: TrainingProject) -> dict[str, Any]: training_project.status = TrainingProjectStatus.ARCHAEOLOGY training_project.save(update_fields=["status", "updated_at"]) self._event("ARCHAEOLOGY_STARTED", training_project, {}) report = self.profile.archaeology(training_project.repository_path) for item in report["datasets"]: dataset, _ = Dataset.objects.get_or_create(training_project=training_project, name=item["name"]) DatasetVersion.objects.update_or_create(dataset=dataset, version=item["hash"][:12], defaults={"manifest_reference": item["reference"], "content_hash": item["hash"], "record_count": item["record_count"], "source_metadata": {"archaeology_confidence": "CONFIRMED"}, "tags": item["tags"], "validation_status": DatasetValidationStatus.WARNING, "contamination_status": DatasetValidationStatus.UNKNOWN}) for item in report["checkpoints"]: ModelCheckpoint.objects.update_or_create(training_project=training_project, reference=item["reference"], defaults={"name": item["name"], "checkpoint_type": CheckpointType.IMPORTED, "content_hash": item["hash"], "adapter_type": item["adapter_type"], "validity_status": CheckpointValidityStatus.VALID, "load_verified": False, "metadata": {"archaeology_confidence": item["confidence"]}}) suite, _ = EvaluationSuite.objects.get_or_create(training_project=training_project, name="Guard EVMBench") benchmark_reference = report["benchmark_reference"] suite_version, _ = EvaluationSuiteVersion.objects.update_or_create(suite=suite, version="imported-local", defaults={"reference": benchmark_reference, "content_hash": self._hash_text(benchmark_reference), "command_template": [report["evaluator_reference"]], "groups": ["PRIMARY", "HOLDOUT", "REGRESSION"], "integrity_status": DatasetValidationStatus.UNKNOWN, "integrity_evidence": {"archaeology": report["findings"]}}) artifact = self._artifact(training_project, "TRAINING_ARCHAEOLOGY_REPORT", "Guard archaeology", report) training_project.metadata = {**training_project.metadata, "archaeology_artifact": str(artifact.id), "historical_reports": report["historical_reports"], "evaluation_suite_version": str(suite_version.id)} training_project.status = TrainingProjectStatus.NEEDS_REPAIR training_project.save(update_fields=["metadata", "status", "updated_at"]) self._event("ARCHAEOLOGY_COMPLETED", training_project, {"checkpoints": len(report["checkpoints"]), "datasets": len(report["datasets"]), "status": training_project.status}) return report def validate_benchmark(self, training_project: TrainingProject) -> EvaluationSuiteVersion: suite_version = EvaluationSuiteVersion.objects.filter(suite__training_project=training_project).order_by("-created_at").first() if suite_version is None: raise ValueError("Run archaeology before benchmark validation.") result = self.profile.validate_evaluation_suite(training_project.repository_path, suite_version.reference) suite_version.integrity_status = result["status"] suite_version.integrity_evidence = result["evidence"] suite_version.save(update_fields=["integrity_status", "integrity_evidence", "updated_at"]) training_project.status = TrainingProjectStatus.READY if result["status"] == DatasetValidationStatus.VALID else TrainingProjectStatus.NEEDS_REPAIR training_project.save(update_fields=["status", "updated_at"]) return suite_version def curate_guard_datasets(self, training_project: TrainingProject) -> dict[str, Any]: """Classify imported manifests without mutating source data or inferring missing provenance.""" rows = [] for version in DatasetVersion.objects.filter(dataset__training_project=training_project).select_related("dataset"): path = Path(version.manifest_reference) tags = set(version.tags) status = DatasetValidationStatus.WARNING contamination = DatasetValidationStatus.UNKNOWN evidence: dict[str, Any] = {"manifest": str(path)} if path.name.endswith("_summary.json"): status, contamination = DatasetValidationStatus.BLOCKED, DatasetValidationStatus.UNKNOWN tags.update(["summary", "not_training"]) evidence["reason"] = "Summary artifacts are evidence, not trainable records." else: summary_path = path.with_name(path.stem + "_summary.json") summary = self._read_json(summary_path) if summary_path.exists() else {} payload = self._read_json(path) record_count = len(payload) if isinstance(payload, list) else None evidence["summary_reference"] = str(summary_path) if summary_path.exists() else "" evidence["record_count_observed"] = record_count if not isinstance(payload, list): status = DatasetValidationStatus.BLOCKED tags.update(["not_training", "invalid_manifest_shape"]) evidence["reason"] = "Training manifest must be a JSON record list." elif summary.get("candidate_only") or "candidate" in path.name.lower() or "provisional" in path.name.lower(): status = DatasetValidationStatus.BLOCKED tags.update(["candidate_only", "not_training"]) evidence["reason"] = "Candidate/provisional corpus is explicitly not training-approved." elif "evmbench" in path.name.lower() or summary.get("evmbench_source_included") is True or summary.get("benchmark_source_included") is True: status = DatasetValidationStatus.BLOCKED tags.update(["benchmark_exclusion", "not_training"]) contamination = DatasetValidationStatus.WARNING evidence["reason"] = "Possible benchmark reference requires manual contamination review." elif summary.get("evmbench_source_included") is False and summary.get("weak_label_data_included") is False: tags.update(["pilot", "source_disjoint_claimed"]) status = DatasetValidationStatus.WARNING contamination = DatasetValidationStatus.WARNING evidence["reason"] = "Source-disjointness is declared, but full schema/provenance/coverage validation remains required." else: tags.update(["imported", "manual_provenance_review_required"]) evidence["reason"] = "No sufficient adjacent evidence to mark this manifest training-valid." if record_count is not None: version.record_count = record_count version.tags = sorted(tags) version.validation_status = status version.contamination_status = contamination version.source_metadata = {**version.source_metadata, "curation": evidence} version.save(update_fields=["record_count", "tags", "validation_status", "contamination_status", "source_metadata", "updated_at"]) rows.append({"dataset": version.dataset.name, "version": version.version, "reference": version.manifest_reference, "validation_status": status, "contamination_status": contamination, "tags": version.tags, "evidence": evidence}) summary = {"total": len(rows), "valid": sum(row["validation_status"] == DatasetValidationStatus.VALID for row in rows), "warning": sum(row["validation_status"] == DatasetValidationStatus.WARNING for row in rows), "blocked": sum(row["validation_status"] == DatasetValidationStatus.BLOCKED for row in rows), "datasets": rows, "decision": "NO_TRAINING_DATASET_APPROVED" if not any(row["validation_status"] == DatasetValidationStatus.VALID for row in rows) else "TRAINING_DATASET_CANDIDATES_AVAILABLE"} self._artifact(training_project, "DATASET_CURATION_REPORT", "Guard dataset curation", summary) return summary def import_spark_guard_datasets(self, training_project: TrainingProject, references: list[str], *, ssh_alias: str = "spark") -> dict[str, Any]: inventory = SparkGuardDatasetInventory(ssh_alias).inspect(references) imported = [] for item in inventory: name = Path(item["reference"]).stem dataset, _ = Dataset.objects.get_or_create(training_project=training_project, name=f"spark-{name}") first = item["first"] tags = ["spark", "imported"] status = DatasetValidationStatus.WARNING reason = "Remote manifest requires record-level provenance and contamination review." lowered = item["reference"].lower() if "/research_only/" in lowered or "holdout" in lowered: tags.extend(["curation_source", "requires_evaluation_resplit"]) status, reason = DatasetValidationStatus.WARNING, "Authorized holdout source requires a newly versioned evaluation split before it may enter training." elif first.get("final_model_eligible") is False: tags.extend(["curation_source", "not_direct_training"]) status, reason = DatasetValidationStatus.WARNING, "Authorized source material requires curation into a new validated DatasetVersion before training." elif first.get("benchmark_source_included") is True: tags.extend(["benchmark_exclusion", "not_training"]) status, reason = DatasetValidationStatus.BLOCKED, "Manifest declares benchmark source inclusion." elif first.get("c4_invalid") is True: tags.extend(["weak_or_invalid_label", "curation_source", "not_direct_training"]) status, reason = DatasetValidationStatus.WARNING, "Authorized source material has weak/invalid labels and requires repair or exclusion before training." DatasetVersion.objects.update_or_create(dataset=dataset, version=item["content_hash"][:12], defaults={"manifest_reference": item["reference"], "content_hash": item["content_hash"], "record_count": item["record_count"], "source_metadata": {"spark_inventory": first, "curation_reason": reason}, "tags": tags, "validation_status": status, "contamination_status": DatasetValidationStatus.UNKNOWN}) imported.append({"reference": item["reference"], "record_count": item["record_count"], "validation_status": status, "reason": reason}) report = {"references": references, "manifest_count": len(imported), "record_count": sum(item["record_count"] or 0 for item in imported), "blocked": sum(item["validation_status"] == DatasetValidationStatus.BLOCKED for item in imported), "warning": sum(item["validation_status"] == DatasetValidationStatus.WARNING for item in imported), "manifests": imported} self._artifact(training_project, "SPARK_DATASET_INVENTORY", "Spark Guard dataset inventory", report) return report def purge_malformed_spark_inventory(self, training_project: TrainingProject, *, reference_prefix: str) -> int: stale = DatasetVersion.objects.filter(dataset__training_project=training_project, dataset__name__startswith="spark-", manifest_reference__startswith=reference_prefix, record_count__isnull=True) count = stale.count() while ids := list(stale.values_list("id", flat=True)[:500]): DatasetVersion.objects.filter(id__in=ids).delete() Dataset.objects.filter(training_project=training_project, name__startswith="spark-", versions__isnull=True).delete() self._artifact(training_project, "SPARK_INVENTORY_PURGE", "Malformed Spark inventory purge", {"reference_prefix": reference_prefix, "deleted_dataset_versions": count}) return count def propose_dataset_curation(self, training_project: TrainingProject, versions: list[DatasetVersion]) -> DatasetCurationProposal: if not versions: raise ValueError("Dataset curation requires at least one source DatasetVersion.") inventory = [{"reference": item.manifest_reference, "records": item.record_count, "validation": item.validation_status, "contamination": item.contamination_status, "tags": item.tags, "evidence": item.source_metadata.get("curation", item.source_metadata.get("spark_inventory", {}))} for item in versions] prompt = "DATASET_CURATOR V0.1. Analyze existing Guard dataset manifest metadata only. Do not invent source quality, labels, coverage, or benchmark results. Return JSON with title, hypothesis, evidence object, proposed_operations list, expected_capability_effect, expected_risks list, validation_plan object, contamination_plan object. Proposed operations must create a NEW immutable DatasetVersion and may filter/reweight/split/select existing records. All legacy data, including former holdouts, is authorized source material. If a former holdout enters training, explicitly require a new source-disjoint evaluation suite version and retire the old comparison split. Inventory: " + json.dumps(inventory, default=str) review: dict[str, Any] = {} if self.router is not None and self.dataset_curator_model_hint in self.router.providers: try: response = self.router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint=self.dataset_curator_model_hint, prompt=prompt)) parsed = extract_json_object(response.content) review = parsed if isinstance(parsed, dict) else {} except Exception as exc: review = {"error": str(exc)} if not review: review = {"title": "Manual provenance and coverage curation required", "hypothesis": "A source-disjoint, schema-valid subset may improve Guard without benchmark leakage.", "evidence": {"inventory": inventory}, "proposed_operations": [{"operation": "REVIEW_ONLY", "reason": "No model-backed curation response available."}], "expected_capability_effect": "Unknown until reviewed records are validated.", "expected_risks": ["provenance gaps", "benchmark contamination", "weak labels"], "validation_plan": {"required": ["schema", "source provenance", "exact and normalized benchmark overlap", "source-group split", "coverage matrix"]}, "contamination_plan": {"required": ["exact source hash", "normalized text", "repository", "benchmark identifier"]}} required = ["title", "hypothesis", "evidence", "proposed_operations", "validation_plan", "contamination_plan"] missing = [field for field in required if review.get(field) in (None, "", {}, [])] if missing: raise ValueError("Dataset curator response incomplete: " + ", ".join(missing)) proposal = DatasetCurationProposal.objects.create(title=str(review["title"])[:255], hypothesis=str(review["hypothesis"]), evidence=review["evidence"], proposed_operations=review["proposed_operations"], expected_capability_effect=str(review.get("expected_capability_effect", "")), expected_risks=review.get("expected_risks", []), validation_plan=review["validation_plan"], contamination_plan=review["contamination_plan"], model_evidence=review) proposal.source_versions.set(versions) self._artifact(training_project, "DATASET_CURATION_PROPOSAL", proposal.title, {"proposal_id": str(proposal.id), "source_versions": [str(item.id) for item in versions], **review}) return proposal def materialize_spark_curation(self, training_project: TrainingProject, proposal: DatasetCurationProposal, *, output_directory: str, ssh_alias: str = "spark", strict: bool = False) -> DatasetVersion: sources = [item.manifest_reference for item in proposal.source_versions.all() if item.manifest_reference.startswith("/")] report = SparkGuardDatasetMaterializer(ssh_alias).materialize(sources, output_directory, strict=strict) dataset, _ = Dataset.objects.get_or_create(training_project=training_project, name=f"curated-{proposal.id.hex[:12]}") version = DatasetVersion.objects.create(dataset=dataset, version=report["train_sha256"][:12], manifest_reference=output_directory + "/train.json", content_hash=report["train_sha256"], record_count=report["train"], split_metadata={"train": report["train"], "validation": report["validation"], "regression": report["regression"], "validation_reference": output_directory + "/validation.json", "regression_reference": output_directory + "/regression.json", "source_group_split": "sha256(source_sha256 or input hash) mod 10"}, source_metadata={"source_manifests": sources, "curation_proposal": str(proposal.id), "materialization_report": report}, generation_metadata={"operations": proposal.proposed_operations, "authorized_source_mandate": True, "old_holdouts_resplit": True, "strict_schema_repair": strict}, tags=["curated", "train", "spark", "requires_new_evaluation_suite", *( ["strict_schema_repaired"] if strict else [])], validation_status=DatasetValidationStatus.VALID, contamination_status=DatasetValidationStatus.WARNING) proposal.materialized_version = version proposal.status = "MATERIALIZED" proposal.save(update_fields=["materialized_version", "status", "updated_at"]) self._artifact(training_project, "CURATED_DATASET_VERSION", dataset.name, {"dataset_version": str(version.id), **report}) return version def audit_curated_dataset_with_qwen(self, training_project: TrainingProject, version: DatasetVersion, *, ssh_alias: str = "spark", sample_count: int = 8) -> dict[str, Any]: samples = SparkGuardDatasetMaterializer(ssh_alias).sample(version.manifest_reference, count=sample_count) prompt = "DATASET_CURATOR QUALITY AUDIT V0.1. Review these bounded Guard SFT samples. Return JSON only with overall_assessment, schema_issues list, label_risks list, provenance_risks list, leakage_risks list, recommended_operations list, and confidence. Do not invent evidence beyond samples. Do not modify data; recommendations must create a new DatasetVersion. Samples: " + json.dumps(samples, default=str) review: dict[str, Any] = {"overall_assessment": "MODEL_UNAVAILABLE", "schema_issues": [], "label_risks": [], "provenance_risks": [], "leakage_risks": [], "recommended_operations": [], "confidence": "LOW"} if self.router is not None and self.dataset_curator_model_hint in self.router.providers: try: response = self.router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint=self.dataset_curator_model_hint, prompt=prompt)) parsed = extract_json_object(response.content) if isinstance(parsed, dict): review = parsed except Exception as exc: review["error"] = str(exc) artifact = self._artifact(training_project, "DATASET_QUALITY_AUDIT", f"Qwen audit {version.version}", {"dataset_version": str(version.id), "sample_count": len(samples), "samples": samples, "review": review}) return {"artifact_id": str(artifact.id), "review": review} def establish_champion(self, training_project: TrainingProject, checkpoint: ModelCheckpoint) -> ModelCheckpoint: suite = self.validate_benchmark(training_project) if suite.integrity_status != DatasetValidationStatus.VALID: raise ValueError("Benchmark integrity is not established; refusing Champion selection.") if not checkpoint.load_verified: raise ValueError("Candidate checkpoint has not passed load verification.") evaluation = self.evaluate(training_project, checkpoint, suite, metrics={"primary": 0.0}, synthetic=False) if evaluation.status != EvaluationRunStatus.SUCCEEDED: raise ValueError("Champion evaluation failed.") checkpoint.checkpoint_type = CheckpointType.CHAMPION checkpoint.evaluation_status = "EVALUATED" checkpoint.save(update_fields=["checkpoint_type", "evaluation_status", "updated_at"]) training_project.current_champion = checkpoint training_project.baseline_evaluation = evaluation training_project.status = TrainingProjectStatus.READY training_project.save(update_fields=["current_champion", "baseline_evaluation", "status", "updated_at"]) self._event("CHAMPION_SELECTED", training_project, {"checkpoint": str(checkpoint.id), "evaluation": str(evaluation.id)}) return checkpoint def evaluate(self, training_project: TrainingProject, checkpoint: ModelCheckpoint, suite: EvaluationSuiteVersion, *, metrics: dict[str, float] | None = None, synthetic: bool = False) -> EvaluationRun: if suite.integrity_status != DatasetValidationStatus.VALID and not synthetic: raise ValueError("Evaluation suite integrity is not valid.") evaluation = EvaluationRun.objects.create(training_project=training_project, checkpoint=checkpoint, suite_version=suite, status=EvaluationRunStatus.RUNNING, integrity_evidence=suite.integrity_evidence) for metric, value in (metrics or {}).items(): BenchmarkResult.objects.create(evaluation_run=evaluation, metric=metric, value=value, unit="score") evaluation.status = EvaluationRunStatus.SUCCEEDED evaluation.completed_at = timezone.now() evaluation.summary = metrics or {} evaluation.save(update_fields=["status", "completed_at", "summary", "updated_at"]) self._event("EVALUATION_COMPLETED", training_project, {"checkpoint": str(checkpoint.id), "evaluation": str(evaluation.id)}) return evaluation def propose_experiment(self, training_project: TrainingProject, contract: dict[str, Any]) -> TrainingExperiment: required = ["hypothesis", "reasoning", "intervention", "controls", "expected_result", "primary_success_metric", "success_threshold", "regression_constraints", "rejection_condition", "ambiguity_policy", "maximum_runtime_seconds", "compute_budget"] missing = [field for field in required if contract.get(field) in (None, "", {}, [])] if missing: raise ValueError("Incomplete scientific contract: " + ", ".join(missing)) fingerprint = self._fingerprint({"intervention": contract["intervention"], "controls": contract["controls"], "input_checkpoint": str(training_project.current_champion_id)}) duplicate = TrainingExperiment.objects.filter(training_project=training_project, fingerprint=fingerprint).exclude(status=ExperimentStatus.CANCELLED).first() if duplicate: raise ValueError(f"Duplicate experiment: {duplicate.experiment_id}") value = self._experiment_value(contract) experiment = TrainingExperiment.objects.create(experiment_id=contract.get("experiment_id", f"EXP-{training_project.slug.upper()}-{TrainingExperiment.objects.filter(training_project=training_project).count() + 1:03d}"), training_project=training_project, title=contract.get("title", contract["hypothesis"][:255]), hypothesis=contract["hypothesis"], reasoning=contract["reasoning"], intervention=contract["intervention"], controls=contract["controls"], expected_result=contract["expected_result"], primary_success_metric=contract["primary_success_metric"], success_threshold=contract["success_threshold"], regression_constraints=contract["regression_constraints"], rejection_condition=contract["rejection_condition"], ambiguity_policy=contract["ambiguity_policy"], estimated_runtime_seconds=int(contract.get("estimated_runtime_seconds", 0)), maximum_runtime_seconds=int(contract["maximum_runtime_seconds"]), compute_budget=contract["compute_budget"], priority=float(contract.get("priority", value)), expected_information_gain=float(contract.get("expected_information_gain", 0)), expected_improvement=float(contract.get("expected_improvement", 0)), estimated_compute_cost=float(contract.get("estimated_compute_cost", 1)), experiment_value_score=value, fingerprint=fingerprint, approved_by_model_director=True) self._event("EXPERIMENT_PROPOSED", training_project, {"experiment": experiment.experiment_id, "value": value}) return experiment @transaction.atomic def run_experiment(self, experiment: TrainingExperiment, recipe_configuration: dict[str, Any]) -> TrainingRun: project = experiment.training_project champion = project.current_champion if champion is None: raise ValueError("Cannot train without an immutable Champion.") if type(self.backend).__name__ == "SparkGuardBackend" and not project.metadata.get("spark_working_directory_verified"): raise ValueError("Spark Guard working directory is not verified; refusing remote training.") if experiment.status not in {ExperimentStatus.PROPOSED, ExperimentStatus.QUEUED}: raise ValueError("Experiment is not runnable.") recipe_hash = self._fingerprint(recipe_configuration) recipe, _ = TrainingRecipe.objects.get_or_create(training_project=project, recipe_hash=recipe_hash, defaults={"name": experiment.experiment_id, "configuration": recipe_configuration}) recipe.immutable = True recipe.save(update_fields=["immutable", "updated_at"]) for dataset_version in DatasetVersion.objects.filter(dataset__training_project=project, manifest_reference__in=recipe_configuration.get("dataset_references", [])): dataset_version.immutable = True dataset_version.save(update_fields=["immutable", "updated_at"]) run = TrainingRun.objects.create(experiment=experiment, recipe=recipe, input_checkpoint=champion, status=TrainingRunStatus.STARTING, working_directory=project.working_directory, command=[]) experiment.status = ExperimentStatus.RUNNING experiment.save(update_fields=["status", "updated_at"]) self._event("TRAINING_STARTED", project, {"experiment": experiment.experiment_id, "run": str(run.id)}) output_directory = str(Path(project.working_directory) / "artifex_runs" / experiment.experiment_id) command = self.profile.training_command(recipe_configuration, output_directory) if recipe_configuration.get("training_data") and hasattr(self.profile, "training_command") else [] run.command = command run.save(update_fields=["command", "updated_at"]) outcome = self.backend.launch(command=command, working_directory=project.working_directory, timeout_seconds=experiment.maximum_runtime_seconds) self._apply_training_outcome(run, outcome) return run def _apply_training_outcome(self, run: TrainingRun, outcome: BackendResult) -> None: experiment = run.experiment project = experiment.training_project if outcome.status != "SUCCEEDED": run.status = getattr(TrainingRunStatus, outcome.status, TrainingRunStatus.FAILED) run.failure_category = outcome.failure_category or outcome.status run.failure_details = outcome.failure_details run.completed_at = timezone.now() run.save(update_fields=["status", "failure_category", "failure_details", "completed_at", "updated_at"]) experiment.status = ExperimentStatus.FAILED experiment.conclusion = Conclusion.EXECUTION_FAILED experiment.result_summary = run.failure_details experiment.save(update_fields=["status", "conclusion", "result_summary", "updated_at"]) self._event("TRAINING_FAILED", project, {"experiment": experiment.experiment_id, "failure": run.failure_category}) return checkpoint = ModelCheckpoint.objects.create(training_project=project, name=f"{experiment.experiment_id}-challenger", checkpoint_type=CheckpointType.CHALLENGER, reference=outcome.checkpoint_reference, content_hash=outcome.checkpoint_hash, base_checkpoint=run.input_checkpoint, training_run=run, recipe=run.recipe, validity_status=CheckpointValidityStatus.VALID if self.backend.validate_checkpoint(outcome.checkpoint_reference) else CheckpointValidityStatus.CORRUPT, load_verified=self.backend.validate_checkpoint(outcome.checkpoint_reference)) run.output_checkpoint = checkpoint run.status = TrainingRunStatus.SUCCEEDED run.completed_at = timezone.now() run.save(update_fields=["output_checkpoint", "status", "completed_at", "updated_at"]) experiment.status = ExperimentStatus.EVALUATING experiment.save(update_fields=["status", "updated_at"]) self._event("TRAINING_COMPLETED", project, {"experiment": experiment.experiment_id, "checkpoint": str(checkpoint.id)}) def decide_promotion(self, experiment: TrainingExperiment, evaluation: EvaluationRun, policy: ModelPromotionPolicy) -> ModelPromotionDecision: project = experiment.training_project champion = project.current_champion candidate = evaluation.checkpoint if champion is None or candidate.training_run_id is None: raise ValueError("Promotion requires a Challenger and starting Champion.") baseline = project.baseline_evaluation if baseline is None or baseline.suite_version_id != evaluation.suite_version_id: raise ValueError("Champion and Challenger must use the same evaluation version.") primary = policy.criteria.get("primary_metric", "primary") minimum_delta = float(policy.criteria.get("minimum_delta", 0)) candidate_value = float(evaluation.summary.get(primary, 0)) baseline_value = float(baseline.summary.get(primary, 0)) regression_ok = all(float(evaluation.summary.get(metric, 0)) >= float(baseline.summary.get(metric, 0)) - float(limit) for metric, limit in policy.criteria.get("max_regression", {}).items()) eligible = candidate.load_verified and evaluation.status == EvaluationRunStatus.SUCCEEDED and candidate_value - baseline_value >= minimum_delta and regression_ok decision = PromotionDecision.PROMOTE if eligible else PromotionDecision.REJECT rationale = "Objective promotion criteria satisfied." if eligible else "Objective promotion criteria not satisfied." record = ModelPromotionDecision.objects.create(training_project=project, from_champion=champion, candidate=candidate, experiment=experiment, decision=decision, policy=policy, evaluation_evidence={"baseline": baseline.summary, "candidate": evaluation.summary, "delta": candidate_value - baseline_value, "regression_ok": regression_ok}, judge_result={"actor": "MODEL_JUDGE", "deterministic_gate": eligible}, reason=rationale) if eligible: champion.checkpoint_type = CheckpointType.IMPORTED champion.save(update_fields=["checkpoint_type", "updated_at"]) candidate.checkpoint_type = CheckpointType.CHAMPION candidate.save(update_fields=["checkpoint_type", "updated_at"]) project.current_champion = candidate project.save(update_fields=["current_champion", "updated_at"]) experiment.status, experiment.conclusion = ExperimentStatus.PROMOTED, Conclusion.SUPPORTED self._event("CHALLENGER_PROMOTED", project, {"experiment": experiment.experiment_id, "checkpoint": str(candidate.id)}) else: experiment.status, experiment.conclusion = ExperimentStatus.REJECTED, Conclusion.REFUTED experiment.result_summary = rationale experiment.save(update_fields=["status", "conclusion", "result_summary", "updated_at"]) return record def create_program(self, training_project: TrainingProject, policy: ModelPromotionPolicy, *, wall_seconds: int = 8 * 3600, max_runs: int = 8, max_failed: int = 3, max_single_run: int = 150 * 60, evaluation_reserve: int = 90 * 60) -> OvernightTrainingProgram: if training_project.current_champion is None: raise ValueError("A verified Champion is required before starting an overnight program.") now = timezone.now() return OvernightTrainingProgram.objects.create(training_project=training_project, starting_champion=training_project.current_champion, deadline=now + timedelta(seconds=wall_seconds), maximum_wall_seconds=wall_seconds, maximum_training_runs=max_runs, maximum_failed_runs=max_failed, maximum_single_run_seconds=max_single_run, evaluation_reserve_seconds=evaluation_reserve, allowed_experiment_types=["DATASET_MIXTURE", "RECIPE", "CHECKPOINT"], promotion_policy=policy) def can_start(self, program: OvernightTrainingProgram, experiment: TrainingExperiment, estimated_evaluation_seconds: int) -> tuple[bool, str]: remaining = max(0, int((program.deadline - timezone.now()).total_seconds())) runs = TrainingRun.objects.filter(experiment__training_project=program.training_project).count() failures = TrainingRun.objects.filter(experiment__training_project=program.training_project, status__in=[TrainingRunStatus.FAILED, TrainingRunStatus.OOM, TrainingRunStatus.TIMEOUT]).count() needed = experiment.estimated_runtime_seconds + estimated_evaluation_seconds + program.evaluation_reserve_seconds if runs >= program.maximum_training_runs: return False, "RUN_LIMIT" if failures >= program.maximum_failed_runs: return False, "FAILURE_LIMIT" if needed > remaining: return False, "FINAL_EVALUATION_RESERVE" return True, "READY" def morning_report(self, program: OvernightTrainingProgram) -> OvernightResearchReport: project = program.training_project experiments = list(project.experiments.order_by("created_at")) payload = {"program_id": str(program.id), "starting_champion": str(program.starting_champion_id), "ending_champion": str(project.current_champion_id), "status": "NO_CHAMPION_CHANGE" if project.current_champion_id == program.starting_champion_id else "CHAMPION_CHANGED", "experiments": [{"id": item.experiment_id, "hypothesis": item.hypothesis, "status": item.status, "conclusion": item.conclusion, "learning": item.result_summary} for item in experiments], "provenance": {"repository": project.repository_path, "baseline_evaluation": str(project.baseline_evaluation_id or "")}} markdown = "# GUARD OVERNIGHT RESEARCH REPORT\n\n" + json.dumps(payload, indent=2, default=str) program.ending_champion = project.current_champion program.status = ProgramStatus.COMPLETED program.completed_at = timezone.now() program.termination_reason = program.termination_reason or "FINALIZED" program.save(update_fields=["ending_champion", "status", "completed_at", "termination_reason", "updated_at"]) report, _ = OvernightResearchReport.objects.update_or_create(program=program, defaults={"markdown": markdown, "payload": payload}) self._artifact(project, "OVERNIGHT_RESEARCH_REPORT", "Morning research report", payload, markdown) self._event("OVERNIGHT_COMPLETED", project, {"program": str(program.id), "status": payload["status"]}) return report def _artifact(self, training_project: TrainingProject, artifact_type: str, name: str, content: dict[str, Any], readable: str = "") -> ModelStudioArtifact: return ModelStudioArtifact.objects.create(training_project=training_project, artifact_type=artifact_type, name=name, content=content, readable=readable) def _event(self, event_type: str, training_project: TrainingProject, payload: dict[str, Any]) -> None: self.bus.publish(event_type, project=training_project.project, actor="MODEL_STUDIO", payload={"training_project": str(training_project.id), **payload}) @staticmethod def _hash_text(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() @staticmethod def _read_json(path: Path) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} @staticmethod def _fingerprint(value: dict[str, Any]) -> str: return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest() @staticmethod def _experiment_value(contract: dict[str, Any]) -> float: return round(float(contract.get("expected_improvement", 0)) * float(contract.get("confidence", 1)) * float(contract.get("expected_information_gain", 1)) * float(contract.get("novelty", 1)) / max(1.0, float(contract.get("estimated_compute_cost", 1))), 4)