tinymind-native-colab-handoff/bundle/evaluation/rule_evolution.py

"""Report builder for TinyMind self-rule evolution."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from model.rule_evolution import RuleEvolutionGovernor, RuleMetricSnapshot, SelfRuleCandidate


def _candidate_rules() -> list[SelfRuleCandidate]:
    base = RuleMetricSnapshot(quality=0.64, purity=0.82, grounding=0.58, instruction=0.55, latency_ms=900, vram_mb=15_700)
    return [
        SelfRuleCandidate(
            name="retrieval-first-current-knowledge",
            scope="retrieval",
            trigger="question contains current, web, date-sensitive, or source-needed facts",
            action="retrieve_first",
            rule_text="Before answering unstable factual questions, retrieve or attach evidence, then answer only from verified sources.",
            before=base,
            after=RuleMetricSnapshot(quality=0.69, purity=0.84, grounding=0.72, instruction=0.56, latency_ms=980, vram_mb=15_700),
            evidence_paths=("reports/current_knowledge_autopilot/current_knowledge_report.json",),
            rollback_plan="disable retrieval_first_current_knowledge in rule manifest and return to base chat route",
            risk_score=0.05,
        ),
        SelfRuleCandidate(
            name="completion-only-loss-mask",
            scope="training",
            trigger="SFT records contain prompt plus assistant completion",
            action="mask_loss",
            rule_text="Train only on assistant completion tokens and ignore prompt/context tokens to prevent fixed prompt memorization.",
            before=base,
            after=RuleMetricSnapshot(quality=0.68, purity=0.87, grounding=0.60, instruction=0.68, latency_ms=900, vram_mb=15_700),
            evidence_paths=("reports/train_12b_qlora_safety/completion_mask_audit.json",),
            rollback_plan="disable completion_only_loss and restore previous collator config",
            risk_score=0.04,
        ),
        SelfRuleCandidate(
            name="unsafe-direct-promotion",
            scope="inference",
            trigger="local probe reaches 100 percent once",
            action="promote",
            rule_text="Immediately promote a rule after one local perfect score.",
            before=base,
            after=RuleMetricSnapshot(quality=0.70, purity=0.70, grounding=0.52, instruction=0.56, latency_ms=900, vram_mb=15_700),
            evidence_paths=("reports/probe_lock_train/local_probe.json",),
            rollback_plan="disable local perfect score promotion",
            risk_score=0.65,
        ),
    ]


def build_rule_evolution_report(out_dir: str | Path) -> dict[str, Any]:
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    report = RuleEvolutionGovernor(min_quality_gain=0.02).decide(_candidate_rules())
    report["report_path"] = str(out / "rule_evolution_report.json")
    report["active_manifest_path"] = str(out / "active_self_rules.json")
    active_manifest = {
        "schema_version": "tinymind-active-self-rules-v1",
        "rules": [
            {
                "rule_id": rule["rule_id"],
                "name": rule["name"],
                "scope": rule["scope"],
                "trigger": rule["trigger"],
                "action": rule["action"],
                "rule_text": rule["rule_text"],
                "rollback_plan": rule["rollback_plan"],
            }
            for rule in report["accepted_rules"]
        ],
        "claim_boundary": {
            "self_evolution_rules_active": report["self_evolution_ready"],
            "world_best_claim_allowed": False,
        },
    }
    (out / "rule_evolution_report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    (out / "active_self_rules.json").write_text(json.dumps(active_manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    return report