diff --git a/core/competences/__init__.py b/core/competences/__init__.py index 0c114b719..f5c697c74 100644 --- a/core/competences/__init__.py +++ b/core/competences/__init__.py @@ -15,15 +15,25 @@ from .verdicts import ( iter_competence_verdicts, store_competence_verdict, ) +from .promotions import ( + CompetencePromotionError, + iter_competence_promotions, + promote_competence_from_verdicts, + summarize_competence_promotions, +) __all__ = [ "CompetenceSummary", + "CompetencePromotionError", "CompetenceVerdictError", "build_competence_replay_actions", "build_competence_replay_payload", "find_competence", + "iter_competence_promotions", "iter_competence_verdicts", "load_competence_catalog_actions", "load_competences", + "promote_competence_from_verdicts", + "summarize_competence_promotions", "store_competence_verdict", ] diff --git a/core/competences/promotions.py b/core/competences/promotions.py new file mode 100644 index 000000000..5e76bd85d --- /dev/null +++ b/core/competences/promotions.py @@ -0,0 +1,666 @@ +"""Promote Lea competences from supervised verdict evidence.""" + +from __future__ import annotations + +import difflib +import hashlib +import json +import shutil +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Iterable, Optional + +import yaml + +from .catalog import ( + DEFAULT_COMPETENCE_ROOT, + KNOWN_STATES, + REPO_ROOT, + load_competence_file, +) +from .replay import find_competence +from .verdicts import DEFAULT_VERDICT_LOG, iter_competence_verdicts + + +DEFAULT_PROMOTION_LOG = REPO_ROOT / "data" / "competences" / "promotions.jsonl" +PROMOTION_SCHEMA_VERSION = "lea_competence_promotion.v1" +PROMOTABLE_STATES = {"candidate", "stable"} + + +class CompetencePromotionError(ValueError): + """Raised when a competence promotion request is invalid.""" + + +def promote_competence_from_verdicts( + competence_id: str, + payload: Dict[str, Any], + *, + competence_root: Path | str = DEFAULT_COMPETENCE_ROOT, + verdict_log_path: Path | str = DEFAULT_VERDICT_LOG, + promotion_log_path: Path | str = DEFAULT_PROMOTION_LOG, + states: Optional[Iterable[str]] = None, + now: Optional[datetime] = None, +) -> Dict[str, Any]: + """Dry-run or apply a dashboard-controlled competence promotion. + + ``dry_run=True`` never writes. A real write requires the exact + ``dry_run_token`` returned by a prior dry-run for the same evidence. + """ + + if not isinstance(payload, dict): + raise CompetencePromotionError("Payload promotion invalide") + + dry_run = bool(payload.get("dry_run", True)) + promotion_id = _promotion_id(payload, dry_run=dry_run) + target_state = _target_state(payload) + confirmed_by = _text(payload.get("confirmed_by") or "human:dom", "confirmed_by") + verdict_ids = _verdict_ids(payload.get("verdict_ids")) + timestamp = _timestamp(now) + + root = Path(competence_root) + promotion_log = Path(promotion_log_path) + + existing = _find_existing_promotion(promotion_id, log_path=promotion_log) + if existing: + duplicate = dict(existing) + duplicate["duplicate"] = True + duplicate["dry_run"] = dry_run + return duplicate + + plan = _build_promotion_plan( + competence_id=competence_id, + target_state=target_state, + verdict_ids=verdict_ids, + promotion_id=promotion_id, + confirmed_by=confirmed_by, + timestamp=timestamp, + competence_root=root, + verdict_log_path=verdict_log_path, + states=states, + ) + + if dry_run: + return { + **plan, + "dry_run": True, + "write_applied": False, + "duplicate": False, + } + + provided_token = _text(payload.get("dry_run_token"), "dry_run_token") + if provided_token != plan["dry_run_token"]: + raise CompetencePromotionError("dry_run_token invalide ou absent") + if not plan["eligible"]: + raise CompetencePromotionError( + "Promotion refusee: " + "; ".join(plan["blocking_reasons"]) + ) + + record = { + "schema_version": PROMOTION_SCHEMA_VERSION, + "promotion_id": promotion_id, + "competence_id": competence_id, + "from_state": plan["from_state"], + "to_state": target_state, + "triggered_by": confirmed_by, + "promoted_at": timestamp, + "evidence_verdict_ids": verdict_ids, + "evidence_summary": plan["evidence_summary"], + "yaml_path_before": plan["yaml_path_before"], + "yaml_path_after": plan["yaml_path_after"], + "backup_path": "", + "dry_run_token": plan["dry_run_token"], + "write_back_enabled": True, + "yaml_write": True, + "duplicate": False, + } + + backup_path = _apply_yaml_plan(plan, root=root, timestamp=timestamp) + record["backup_path"] = _relative_path(backup_path) + _append_jsonl(promotion_log, record) + + return { + **plan, + "dry_run": False, + "write_applied": True, + "promotion": record, + "backup_path": record["backup_path"], + "promotions_log_path": _relative_path(promotion_log), + "duplicate": False, + } + + +def summarize_competence_promotions( + *, + competence_root: Path | str = DEFAULT_COMPETENCE_ROOT, + verdict_log_path: Path | str = DEFAULT_VERDICT_LOG, + states: Optional[Iterable[str]] = None, +) -> list[Dict[str, Any]]: + """Return dashboard-safe promotion state for all known competences.""" + + root = Path(competence_root) + summaries: list[Dict[str, Any]] = [] + for state in KNOWN_STATES: + if states and state not in set(states): + continue + state_dir = root / state + if not state_dir.exists(): + continue + for path in sorted(state_dir.glob("*.yaml")): + competence = load_competence_file(path, repo_root=REPO_ROOT) + verdicts = iter_competence_verdicts( + log_path=verdict_log_path, + competence_id=competence.id, + ) + counts = _verdict_counts(verdicts) + valid_ids = [ + str(verdict.get("verdict_id")) + for verdict in verdicts + if verdict.get("verdict_kind") == "valid" and verdict.get("verdict_id") + ] + targets = {} + for target in _available_targets(competence.learning_state): + try: + plan = _build_promotion_plan( + competence_id=competence.id, + target_state=target, + verdict_ids=valid_ids, + promotion_id=str(uuid.uuid4()), + confirmed_by="dashboard:summary", + timestamp=_timestamp(None), + competence_root=root, + verdict_log_path=verdict_log_path, + states=states, + ) + targets[target] = { + "eligible": plan["eligible"], + "blocking_reasons": plan["blocking_reasons"], + "recommended_verdict_ids": valid_ids, + } + except (CompetencePromotionError, KeyError) as exc: + targets[target] = { + "eligible": False, + "blocking_reasons": [str(exc)], + "recommended_verdict_ids": valid_ids, + } + + summaries.append({ + "id": competence.id, + "name": competence.name, + "intent_fr": competence.intent_fr, + "learning_state": competence.learning_state, + "source_path": competence.source_path, + "verdict_counts": counts, + "distinct_contexts": len(_distinct_contexts([ + verdict for verdict in verdicts + if verdict.get("verdict_kind") == "valid" + ])), + "latest_verdict_at": _latest_verdict_at(verdicts), + "eligible_targets": targets, + "regression_suspected": _regression_suspected(verdicts), + }) + + return sorted(summaries, key=lambda item: (item["learning_state"], item["id"])) + + +def iter_competence_promotions( + *, + log_path: Path | str = DEFAULT_PROMOTION_LOG, + competence_id: Optional[str] = None, +) -> list[Dict[str, Any]]: + log = Path(log_path) + if not log.exists(): + return [] + + records: list[Dict[str, Any]] = [] + with log.open("r", encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(record, dict): + continue + if competence_id and record.get("competence_id") != competence_id: + continue + records.append(record) + return records + + +def _build_promotion_plan( + *, + competence_id: str, + target_state: str, + verdict_ids: list[str], + promotion_id: str, + confirmed_by: str, + timestamp: str, + competence_root: Path, + verdict_log_path: Path | str, + states: Optional[Iterable[str]], +) -> Dict[str, Any]: + competence = find_competence(competence_id, root=competence_root, states=states) + if target_state == competence.learning_state: + raise CompetencePromotionError("target_state identique a l'etat courant") + if target_state not in _available_targets(competence.learning_state): + raise CompetencePromotionError( + f"Promotion {competence.learning_state} -> {target_state} interdite" + ) + + source_path = _absolute_source_path(competence.source_path) + data = _load_yaml_mapping(source_path) + verdicts = _selected_verdicts( + competence_id=competence_id, + verdict_ids=verdict_ids, + verdict_log_path=verdict_log_path, + ) + + evidence_summary = _evidence_summary(verdicts) + blocking_reasons = _blocking_reasons( + current_state=competence.learning_state, + target_state=target_state, + verdicts=verdicts, + all_verdicts=iter_competence_verdicts( + log_path=verdict_log_path, + competence_id=competence_id, + ), + ) + eligible = not blocking_reasons + + updated = _updated_yaml_data( + data=data, + competence_id=competence_id, + current_state=competence.learning_state, + target_state=target_state, + verdicts=verdicts, + promotion_id=promotion_id, + confirmed_by=confirmed_by, + timestamp=timestamp, + ) + current_text = source_path.read_text(encoding="utf-8") + updated_text = yaml.safe_dump( + updated, + allow_unicode=True, + sort_keys=False, + default_flow_style=False, + ) + target_path = competence_root / target_state / f"{competence_id}.yaml" + yaml_diff = "\n".join(difflib.unified_diff( + current_text.splitlines(), + updated_text.splitlines(), + fromfile=_relative_path(source_path), + tofile=_relative_path(target_path), + lineterm="", + )) + dry_run_token = _dry_run_token( + promotion_id=promotion_id, + competence_id=competence_id, + target_state=target_state, + verdict_ids=verdict_ids, + source_text=current_text, + updated_text=updated_text, + ) + + return { + "schema_version": PROMOTION_SCHEMA_VERSION, + "promotion_id": promotion_id, + "competence_id": competence_id, + "from_state": competence.learning_state, + "to_state": target_state, + "target_state": target_state, + "confirmed_by": confirmed_by, + "eligible": eligible, + "blocking_reasons": blocking_reasons, + "evidence_summary": evidence_summary, + "verdict_ids": verdict_ids, + "yaml_path_before": _relative_path(source_path), + "yaml_path_after": _relative_path(target_path), + "yaml_diff": yaml_diff, + "dry_run_token": dry_run_token, + "_source_path": source_path, + "_target_path": target_path, + "_updated_text": updated_text, + } + + +def _blocking_reasons( + *, + current_state: str, + target_state: str, + verdicts: list[Dict[str, Any]], + all_verdicts: list[Dict[str, Any]], +) -> list[str]: + valid = [verdict for verdict in verdicts if verdict.get("verdict_kind") == "valid"] + reasons: list[str] = [] + if len(valid) != len(verdicts): + reasons.append("Tous les verdict_ids selectionnes doivent etre valid") + if not valid: + reasons.append("Au moins un verdict valid est requis") + missing_evidence = [ + str(verdict.get("verdict_id")) + for verdict in valid + if not verdict.get("workflow_id") or not verdict.get("step_results") + ] + if missing_evidence: + reasons.append( + "Evidence workflow_id/step_results manquante: " + + ", ".join(missing_evidence) + ) + + if current_state == "candidate" and target_state == "stable": + contexts = _distinct_contexts(valid) + if len(valid) < 3: + reasons.append(f"3 verdicts valid requis pour stable ({len(valid)}/3)") + if len(contexts) < 3: + reasons.append(f"3 contextes distincts requis pour stable ({len(contexts)}/3)") + invalid_unexplained = [ + verdict for verdict in all_verdicts + if verdict.get("verdict_kind") == "invalid" and not _is_explained(verdict) + ] + if invalid_unexplained: + reasons.append( + "Invalid non explique present: " + + ", ".join(str(v.get("verdict_id")) for v in invalid_unexplained) + ) + return reasons + + +def _updated_yaml_data( + *, + data: Dict[str, Any], + competence_id: str, + current_state: str, + target_state: str, + verdicts: list[Dict[str, Any]], + promotion_id: str, + confirmed_by: str, + timestamp: str, +) -> Dict[str, Any]: + updated = json.loads(json.dumps(data, ensure_ascii=False)) + updated["learning_state"] = target_state + updated["last_updated_at"] = timestamp + + promotion = updated.setdefault("promotion", {}) + history = promotion.setdefault("history", []) + if isinstance(history, list): + history.append({ + "at": timestamp, + "from": current_state, + "to": target_state, + "by": confirmed_by, + "reason": "Promotion dashboard supervisee par verdicts humains", + "promotion_id": promotion_id, + "evidence_verdict_ids": [ + verdict.get("verdict_id") for verdict in verdicts + ], + }) + + generalisation = updated.setdefault("generalisation", {}) + seen_contexts = generalisation.setdefault("seen_contexts", []) + if isinstance(seen_contexts, list): + existing_ids = { + context.get("verdict_id") + for context in seen_contexts + if isinstance(context, dict) + } + for verdict in verdicts: + verdict_id = verdict.get("verdict_id") + if verdict_id in existing_ids: + continue + context = verdict.get("context_signature") or {} + seen_contexts.append({ + "at": timestamp, + "verdict_id": verdict_id, + "promotion_id": promotion_id, + "machine_id": context.get("machine_id", ""), + "workflow_id": verdict.get("workflow_id", ""), + "screen_state_initial": context.get("screen_state_initial", ""), + "screen_state_after_action": context.get("screen_state_after_action", ""), + "verdict_at": verdict.get("verdict_at", ""), + }) + + return updated + + +def _apply_yaml_plan(plan: Dict[str, Any], *, root: Path, timestamp: str) -> Path: + source_path = Path(plan["_source_path"]) + target_path = Path(plan["_target_path"]) + updated_text = str(plan["_updated_text"]) + + backup_path = source_path.with_name( + f"{source_path.name}.{timestamp.replace(':', '').replace('+', '_')}.bak" + ) + shutil.copy2(source_path, backup_path) + + target_path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = target_path.with_suffix(target_path.suffix + ".tmp") + tmp_path.write_text(updated_text, encoding="utf-8") + + try: + load_competence_file(tmp_path, repo_root=REPO_ROOT) + tmp_path.replace(target_path) + load_competence_file(target_path, repo_root=REPO_ROOT) + if source_path != target_path and source_path.exists(): + source_path.unlink() + except Exception: + if tmp_path.exists(): + tmp_path.unlink() + if source_path.exists(): + shutil.copy2(backup_path, source_path) + raise + + return backup_path + + +def _selected_verdicts( + *, + competence_id: str, + verdict_ids: list[str], + verdict_log_path: Path | str, +) -> list[Dict[str, Any]]: + all_records = iter_competence_verdicts( + log_path=verdict_log_path, + competence_id=competence_id, + ) + by_id = {str(record.get("verdict_id")): record for record in all_records} + missing = [verdict_id for verdict_id in verdict_ids if verdict_id not in by_id] + if missing: + raise CompetencePromotionError( + "Verdicts introuvables: " + ", ".join(missing) + ) + return [by_id[verdict_id] for verdict_id in verdict_ids] + + +def _evidence_summary(verdicts: list[Dict[str, Any]]) -> Dict[str, Any]: + return { + "counts": _verdict_counts(verdicts), + "distinct_contexts": len(_distinct_contexts([ + verdict for verdict in verdicts + if verdict.get("verdict_kind") == "valid" + ])), + "verdicts": [ + { + "verdict_id": verdict.get("verdict_id"), + "verdict_kind": verdict.get("verdict_kind"), + "verdict_at": verdict.get("verdict_at"), + "workflow_id": verdict.get("workflow_id", ""), + "machine_id": (verdict.get("context_signature") or {}).get("machine_id", ""), + "step_results_count": len(verdict.get("step_results") or []), + } + for verdict in verdicts + ], + } + + +def _verdict_counts(verdicts: list[Dict[str, Any]]) -> Dict[str, int]: + return { + "valid": sum(1 for item in verdicts if item.get("verdict_kind") == "valid"), + "invalid": sum(1 for item in verdicts if item.get("verdict_kind") == "invalid"), + "inconclusive": sum( + 1 for item in verdicts if item.get("verdict_kind") == "inconclusive" + ), + } + + +def _distinct_contexts(verdicts: list[Dict[str, Any]]) -> set[str]: + contexts: set[str] = set() + for verdict in verdicts: + context = verdict.get("context_signature") or {} + parts = [ + str(context.get("machine_id") or ""), + str(context.get("os_name") or ""), + str(context.get("os_version") or ""), + str(context.get("keyboard_layout") or ""), + str(context.get("screen_resolution") or ""), + str(context.get("scaling") or ""), + str(context.get("app_name") or ""), + str(context.get("app_version") or ""), + str(context.get("screen_state_initial") or ""), + str(context.get("screen_state_after_action") or ""), + ] + contexts.add("|".join(parts)) + return contexts + + +def _regression_suspected(verdicts: list[Dict[str, Any]]) -> bool: + latest = sorted( + verdicts, + key=lambda item: str(item.get("verdict_at") or ""), + reverse=True, + )[:3] + return len(latest) == 3 and all( + item.get("verdict_kind") == "invalid" for item in latest + ) + + +def _is_explained(verdict: Dict[str, Any]) -> bool: + evidence = verdict.get("evidence") if isinstance(verdict.get("evidence"), dict) else {} + if evidence.get("explained") is True: + return True + return bool(str(verdict.get("comments") or "").strip()) + + +def _available_targets(current_state: str) -> list[str]: + if current_state == "observed": + return ["candidate"] + if current_state == "candidate": + return ["stable"] + return [] + + +def _target_state(payload: Dict[str, Any]) -> str: + target = _text(payload.get("target_state"), "target_state") + if target not in PROMOTABLE_STATES: + raise CompetencePromotionError("target_state doit etre candidate ou stable") + return target + + +def _promotion_id(payload: Dict[str, Any], *, dry_run: bool) -> str: + value = payload.get("promotion_id") + if value is None and dry_run: + return str(uuid.uuid4()) + text = _text(value, "promotion_id") + _validate_uuid(text, field="promotion_id") + return text + + +def _verdict_ids(value: Any) -> list[str]: + if not isinstance(value, list) or not value: + raise CompetencePromotionError("verdict_ids doit etre une liste non vide") + verdict_ids: list[str] = [] + for item in value: + text = _text(item, "verdict_id") + _validate_uuid(text, field="verdict_id") + verdict_ids.append(text) + return verdict_ids + + +def _text(value: Any, field: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise CompetencePromotionError(f"{field} requis") + return value.strip() + + +def _validate_uuid(value: str, *, field: str) -> None: + try: + parsed = uuid.UUID(value, version=4) + except ValueError as exc: + raise CompetencePromotionError(f"{field} doit etre un UUID v4") from exc + if str(parsed) != value.lower(): + raise CompetencePromotionError(f"{field} UUID v4 invalide") + + +def _timestamp(now: Optional[datetime]) -> str: + timestamp = now or datetime.now(timezone.utc) + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=timezone.utc) + return timestamp.astimezone(timezone.utc).isoformat() + + +def _dry_run_token( + *, + promotion_id: str, + competence_id: str, + target_state: str, + verdict_ids: list[str], + source_text: str, + updated_text: str, +) -> str: + payload = { + "promotion_id": promotion_id, + "competence_id": competence_id, + "target_state": target_state, + "verdict_ids": verdict_ids, + "source_hash": hashlib.sha256(source_text.encode("utf-8")).hexdigest(), + "updated_hash": hashlib.sha256(updated_text.encode("utf-8")).hexdigest(), + } + raw = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8") + return hashlib.sha256(raw).hexdigest() + + +def _find_existing_promotion( + promotion_id: str, + *, + log_path: Path, +) -> Optional[Dict[str, Any]]: + for record in iter_competence_promotions(log_path=log_path): + if record.get("promotion_id") == promotion_id: + return record + return None + + +def _load_yaml_mapping(path: Path) -> Dict[str, Any]: + with path.open("r", encoding="utf-8") as handle: + data = yaml.safe_load(handle) or {} + if not isinstance(data, dict): + raise CompetencePromotionError(f"{path} doit contenir un objet YAML") + return data + + +def _absolute_source_path(source_path: str) -> Path: + path = Path(source_path) + if path.is_absolute(): + return path + return REPO_ROOT / path + + +def _relative_path(path: Path) -> str: + try: + return str(path.resolve().relative_to(REPO_ROOT.resolve())) + except ValueError: + return str(path) + + +def _latest_verdict_at(verdicts: list[Dict[str, Any]]) -> str: + values = [str(item.get("verdict_at") or "") for item in verdicts] + return max(values) if values else "" + + +def _append_jsonl(log_path: Path, record: Dict[str, Any]) -> None: + log_path.parent.mkdir(parents=True, exist_ok=True) + with log_path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True)) + handle.write("\n") diff --git a/tests/unit/test_competence_promotions.py b/tests/unit/test_competence_promotions.py new file mode 100644 index 000000000..4a2197bb8 --- /dev/null +++ b/tests/unit/test_competence_promotions.py @@ -0,0 +1,213 @@ +from datetime import datetime, timezone +from uuid import uuid4 + +import pytest + +from core.competences.promotions import ( + CompetencePromotionError, + iter_competence_promotions, + promote_competence_from_verdicts, + summarize_competence_promotions, +) +from core.competences.verdicts import store_competence_verdict + + +def _write_competence(root, state="observed", competence_id="demo_competence"): + state_dir = root / state + state_dir.mkdir(parents=True, exist_ok=True) + path = state_dir / f"{competence_id}.yaml" + path.write_text( + "\n".join([ + "schema_version: 1", + f"id: {competence_id}", + "name: Demo competence", + f"learning_state: {state}", + "intent:", + " fr: demo competence", + "methods: []", + "promotion:", + " candidate_requires:", + " - cleaned_segment_validated", + " stable_requires:", + " min_successes: 3", + " distinct_contexts: 3", + " max_unexplained_failures: 0", + "generalisation:", + " seen_contexts: []", + "failure_log: []", + "created_at: '2026-05-29T00:00:00+00:00'", + "last_updated_at: '2026-05-29T00:00:00+00:00'", + ]), + encoding="utf-8", + ) + return path + + +def _valid_verdict(root, log_path, competence_id="demo_competence", machine="machine_a"): + verdict_id = str(uuid4()) + return store_competence_verdict( + competence_id, + { + "verdict_id": verdict_id, + "verdict_kind": "valid", + "verdict_by": "human:dom", + "workflow_id": f"workflow_{machine}", + "step_results": [ + {"step_id": "step_1", "action_type": "keyboard_shortcut", "status": "success"} + ], + "context_signature": { + "machine_id": machine, + "screen_state_initial": f"before_{machine}", + "screen_state_after_action": f"after_{machine}", + }, + }, + competence_root=root, + log_path=log_path, + now=datetime(2026, 5, 29, 17, 0, tzinfo=timezone.utc), + ) + + +def test_promote_competence_dry_run_does_not_write_yaml(tmp_path): + root = tmp_path / "competences" + source = _write_competence(root, state="observed") + verdict_log = tmp_path / "verdicts.jsonl" + verdict = _valid_verdict(root, verdict_log) + before = source.read_text(encoding="utf-8") + + result = promote_competence_from_verdicts( + "demo_competence", + { + "promotion_id": str(uuid4()), + "target_state": "candidate", + "verdict_ids": [verdict["verdict_id"]], + "confirmed_by": "human:dom", + "dry_run": True, + }, + competence_root=root, + verdict_log_path=verdict_log, + promotion_log_path=tmp_path / "promotions.jsonl", + now=datetime(2026, 5, 29, 18, 0, tzinfo=timezone.utc), + ) + + assert result["dry_run"] is True + assert result["eligible"] is True + assert result["write_applied"] is False + assert "learning_state: candidate" in result["yaml_diff"] + assert source.read_text(encoding="utf-8") == before + + +def test_promote_competence_confirm_writes_backup_and_audit(tmp_path): + root = tmp_path / "competences" + source = _write_competence(root, state="observed") + verdict_log = tmp_path / "verdicts.jsonl" + promotion_log = tmp_path / "promotions.jsonl" + verdict = _valid_verdict(root, verdict_log) + promotion_id = str(uuid4()) + + dry_run = promote_competence_from_verdicts( + "demo_competence", + { + "promotion_id": promotion_id, + "target_state": "candidate", + "verdict_ids": [verdict["verdict_id"]], + "confirmed_by": "human:dom", + "dry_run": True, + }, + competence_root=root, + verdict_log_path=verdict_log, + promotion_log_path=promotion_log, + now=datetime(2026, 5, 29, 18, 0, tzinfo=timezone.utc), + ) + + result = promote_competence_from_verdicts( + "demo_competence", + { + "promotion_id": promotion_id, + "target_state": "candidate", + "verdict_ids": [verdict["verdict_id"]], + "confirmed_by": "human:dom", + "dry_run": False, + "dry_run_token": dry_run["dry_run_token"], + }, + competence_root=root, + verdict_log_path=verdict_log, + promotion_log_path=promotion_log, + now=datetime(2026, 5, 29, 18, 0, tzinfo=timezone.utc), + ) + + target = root / "candidate" / "demo_competence.yaml" + assert result["write_applied"] is True + assert target.exists() + assert not source.exists() + written = target.read_text(encoding="utf-8") + assert "learning_state: candidate" in written + assert verdict["verdict_id"] in written + assert list(source.parent.glob("demo_competence.yaml.*.bak")) + promotions = iter_competence_promotions(log_path=promotion_log) + assert promotions[0]["promotion_id"] == promotion_id + assert promotions[0]["yaml_write"] is True + + +def test_promote_competence_requires_prior_dry_run_token(tmp_path): + root = tmp_path / "competences" + _write_competence(root, state="observed") + verdict_log = tmp_path / "verdicts.jsonl" + verdict = _valid_verdict(root, verdict_log) + + with pytest.raises(CompetencePromotionError, match="dry_run_token"): + promote_competence_from_verdicts( + "demo_competence", + { + "promotion_id": str(uuid4()), + "target_state": "candidate", + "verdict_ids": [verdict["verdict_id"]], + "confirmed_by": "human:dom", + "dry_run": False, + }, + competence_root=root, + verdict_log_path=verdict_log, + promotion_log_path=tmp_path / "promotions.jsonl", + ) + + +def test_stable_promotion_requires_three_distinct_contexts(tmp_path): + root = tmp_path / "competences" + _write_competence(root, state="candidate") + verdict_log = tmp_path / "verdicts.jsonl" + verdict = _valid_verdict(root, verdict_log, machine="machine_a") + + result = promote_competence_from_verdicts( + "demo_competence", + { + "promotion_id": str(uuid4()), + "target_state": "stable", + "verdict_ids": [verdict["verdict_id"]], + "confirmed_by": "human:dom", + "dry_run": True, + }, + competence_root=root, + verdict_log_path=verdict_log, + promotion_log_path=tmp_path / "promotions.jsonl", + ) + + assert result["eligible"] is False + assert any("3 verdicts valid requis" in reason for reason in result["blocking_reasons"]) + assert any("3 contextes distincts requis" in reason for reason in result["blocking_reasons"]) + + +def test_summarize_competence_promotions_reports_eligible_candidate(tmp_path): + root = tmp_path / "competences" + _write_competence(root, state="observed") + verdict_log = tmp_path / "verdicts.jsonl" + verdict = _valid_verdict(root, verdict_log) + + summary = summarize_competence_promotions( + competence_root=root, + verdict_log_path=verdict_log, + ) + + assert summary[0]["id"] == "demo_competence" + assert summary[0]["verdict_counts"]["valid"] == 1 + candidate = summary[0]["eligible_targets"]["candidate"] + assert candidate["eligible"] is True + assert candidate["recommended_verdict_ids"] == [verdict["verdict_id"]] diff --git a/tests/unit/test_dashboard_routes.py b/tests/unit/test_dashboard_routes.py index 813cafee4..2c042f036 100644 --- a/tests/unit/test_dashboard_routes.py +++ b/tests/unit/test_dashboard_routes.py @@ -54,6 +54,14 @@ class TestDashboardRoutes: data = resp.get_json() assert 'faiss' in data + def test_knowledge_base_stats_include_competences(self, client): + """La base de connaissances expose les competences supervisees.""" + resp = client.get('/api/knowledge-base/stats') + assert resp.status_code == 200 + data = resp.get_json() + assert 'competences' in data + assert 'items' in data['competences'] + def test_version(self, client): """L'API version retourne la version actuelle.""" resp = client.get('/api/version') diff --git a/tests/unit/test_lea_competence_verdict_api.py b/tests/unit/test_lea_competence_verdict_api.py index f7cbd8870..a3834fdc5 100644 --- a/tests/unit/test_lea_competence_verdict_api.py +++ b/tests/unit/test_lea_competence_verdict_api.py @@ -107,3 +107,42 @@ def test_list_competence_verdicts_endpoint(monkeypatch): assert data["verdicts"] == [ {"competence_id": "key_win_r_wait_explorer_exe", "verdict_kind": "valid"} ] + + +def test_promote_competence_endpoint_dry_run(monkeypatch): + def fake_promote(competence_id, payload): + assert competence_id == "key_win_r_wait_explorer_exe" + assert payload["dry_run"] is True + return { + "promotion_id": payload["promotion_id"], + "competence_id": competence_id, + "target_state": "candidate", + "dry_run": True, + "eligible": True, + "dry_run_token": "token", + "write_applied": False, + } + + monkeypatch.setattr( + lea_competences_module, + "promote_competence_from_verdicts", + fake_promote, + ) + + with _app().test_client() as client: + response = client.post( + "/api/v1/lea/competences/key_win_r_wait_explorer_exe/promote", + json={ + "promotion_id": "123e4567-e89b-42d3-a456-426614174000", + "target_state": "candidate", + "verdict_ids": ["123e4567-e89b-42d3-a456-426614174001"], + "confirmed_by": "human:dom", + "dry_run": True, + }, + ) + + data = response.get_json() + assert response.status_code == 200 + assert data["success"] is True + assert data["dry_run"] is True + assert data["yaml_write"] is False diff --git a/visual_workflow_builder/backend/api/lea_competences.py b/visual_workflow_builder/backend/api/lea_competences.py index 952ad9151..c7cb84b16 100644 --- a/visual_workflow_builder/backend/api/lea_competences.py +++ b/visual_workflow_builder/backend/api/lea_competences.py @@ -9,6 +9,10 @@ from core.competences.verdicts import ( iter_competence_verdicts, store_competence_verdict, ) +from core.competences.promotions import ( + CompetencePromotionError, + promote_competence_from_verdicts, +) lea_competences_bp = Blueprint( @@ -59,3 +63,41 @@ def list_competence_verdicts(competence_id: str): "write_back_enabled": False, "yaml_write": False, }) + + +@lea_competences_bp.route("//promote", methods=["POST"]) +def promote_competence(competence_id: str): + """Dry-run or apply a supervised dashboard promotion.""" + + payload = request.get_json(silent=True) or {} + try: + promotion = promote_competence_from_verdicts(competence_id, payload) + except KeyError: + return jsonify({ + "success": False, + "error": f"Competence '{competence_id}' introuvable", + }), 404 + except CompetencePromotionError as exc: + return jsonify({ + "success": False, + "error": str(exc), + "write_back_enabled": False, + "yaml_write": False, + }), 400 + + status = 200 if promotion.get("dry_run") or promotion.get("duplicate") else 201 + return jsonify({ + "success": True, + "competence_id": competence_id, + "promotion": _public_promotion_payload(promotion), + "dry_run": promotion.get("dry_run", False), + "write_back_enabled": not promotion.get("dry_run", False), + "yaml_write": bool(promotion.get("write_applied", False)), + }), status + + +def _public_promotion_payload(promotion: dict): + return { + key: value for key, value in promotion.items() + if not str(key).startswith("_") + } diff --git a/web_dashboard/app.py b/web_dashboard/app.py index ed738a54c..64f6f8af4 100644 --- a/web_dashboard/app.py +++ b/web_dashboard/app.py @@ -2465,10 +2465,56 @@ def knowledge_base_stats(): 'sessions': _kb_sessions_stats(), 'patterns': _kb_patterns_stats(), 'workflows': _kb_workflows_stats(), + 'competences': _kb_competences_stats(), } return jsonify(result) +@app.route('/api/v1/lea/competences//promote', methods=['POST']) +def dashboard_promote_competence(competence_id): + """Dry-run or apply a supervised competence promotion from the dashboard.""" + + try: + from core.competences.promotions import ( + CompetencePromotionError, + promote_competence_from_verdicts, + ) + payload = request.get_json(silent=True) or {} + promotion = promote_competence_from_verdicts(competence_id, payload) + except KeyError: + return jsonify({ + 'success': False, + 'error': f"Competence '{competence_id}' introuvable", + }), 404 + except CompetencePromotionError as exc: + return jsonify({ + 'success': False, + 'error': str(exc), + 'write_back_enabled': False, + 'yaml_write': False, + }), 400 + except Exception as exc: + return jsonify({ + 'success': False, + 'error': str(exc), + 'write_back_enabled': False, + 'yaml_write': False, + }), 500 + + status = 200 if promotion.get('dry_run') or promotion.get('duplicate') else 201 + return jsonify({ + 'success': True, + 'competence_id': competence_id, + 'promotion': { + key: value for key, value in promotion.items() + if not str(key).startswith('_') + }, + 'dry_run': promotion.get('dry_run', False), + 'write_back_enabled': not promotion.get('dry_run', False), + 'yaml_write': bool(promotion.get('write_applied', False)), + }), status + + def _kb_faiss_stats() -> dict: """Statistiques de l'index FAISS.""" faiss_index_path = DATA_PATH / "faiss_index" / "main.index" @@ -2597,6 +2643,29 @@ def _kb_workflows_stats() -> dict: return {'total': total} +def _kb_competences_stats() -> dict: + """Statistiques des compétences Lea YAML et verdicts supervisés.""" + try: + from core.competences.promotions import summarize_competence_promotions + competences = summarize_competence_promotions() + by_state = {} + for item in competences: + state = item.get('learning_state', 'unknown') + by_state[state] = by_state.get(state, 0) + 1 + return { + 'total': len(competences), + 'by_state': by_state, + 'items': competences, + } + except Exception as exc: + return { + 'total': 0, + 'by_state': {}, + 'items': [], + 'error': str(exc), + } + + def _dir_size(path: Path) -> int: """Calcule la taille totale d'un dossier (non récursif profond pour la perf).""" total = 0 diff --git a/web_dashboard/templates/knowledge_base.html b/web_dashboard/templates/knowledge_base.html index 5abeadfe5..d01bd47eb 100644 --- a/web_dashboard/templates/knowledge_base.html +++ b/web_dashboard/templates/knowledge_base.html @@ -89,6 +89,74 @@ justify-content: center; font-size: 11px; font-weight: 700; } + .state-badge { + display: inline-flex; align-items: center; justify-content: center; + min-width: 78px; padding: 4px 8px; border-radius: 6px; + border: 1px solid #334155; background: #0f172a; + color: #cbd5e1; font-size: 11px; font-weight: 700; + text-transform: uppercase; + } + .state-badge.observed { color: #fbbf24; border-color: #92400e; } + .state-badge.candidate { color: #93c5fd; border-color: #1d4ed8; } + .state-badge.stable { color: #6ee7b7; border-color: #047857; } + .verdict-strip { display: flex; flex-wrap: wrap; gap: 6px; } + .verdict-pill { + display: inline-flex; gap: 5px; align-items: center; + padding: 4px 7px; border-radius: 6px; background: #0f172a; + border: 1px solid #334155; font-size: 12px; + } + .verdict-pill.valid { color: #6ee7b7; } + .verdict-pill.invalid { color: #fca5a5; } + .verdict-pill.inconclusive { color: #fcd34d; } + .action-row { display: flex; flex-wrap: wrap; gap: 8px; } + .action-btn { + border: 1px solid #2563eb; background: #1d4ed8; color: #fff; + border-radius: 6px; padding: 7px 10px; font-size: 12px; + font-weight: 700; cursor: pointer; min-height: 32px; + } + .action-btn:hover:not(:disabled) { background: #2563eb; } + .action-btn:disabled { + cursor: not-allowed; opacity: 0.45; border-color: #475569; + background: #1e293b; color: #94a3b8; + } + .modal-backdrop { + display: none; position: fixed; inset: 0; z-index: 50; + background: rgba(15,23,42,0.82); padding: 24px; + align-items: center; justify-content: center; + } + .modal-backdrop.visible { display: flex; } + .modal { + width: min(980px, 96vw); max-height: 90vh; overflow: auto; + background: #1e293b; border: 1px solid #475569; + border-radius: 8px; box-shadow: 0 24px 70px rgba(0,0,0,0.35); + } + .modal-header { + display: flex; justify-content: space-between; align-items: center; + padding: 16px 18px; border-bottom: 1px solid #334155; + } + .modal-title { font-size: 17px; font-weight: 700; color: #f8fafc; } + .modal-close { + border: 1px solid #475569; background: #0f172a; color: #cbd5e1; + border-radius: 6px; padding: 5px 8px; cursor: pointer; + } + .modal-body { padding: 18px; } + .diff-box { + white-space: pre-wrap; overflow-x: auto; background: #020617; + border: 1px solid #334155; border-radius: 6px; padding: 12px; + color: #cbd5e1; font-size: 12px; line-height: 1.4; + max-height: 320px; + } + .evidence-list { margin: 10px 0 16px; color: #cbd5e1; font-size: 13px; } + .modal-actions { + display: flex; justify-content: flex-end; gap: 10px; + padding: 14px 18px; border-top: 1px solid #334155; + } + .secondary-btn { + border: 1px solid #475569; background: #0f172a; color: #cbd5e1; + border-radius: 6px; padding: 8px 12px; cursor: pointer; + } + .danger-note { color: #fcd34d; font-size: 12px; margin-top: 8px; } + /* === Loading === */ .loading { text-align: center; padding: 40px; color: #64748b; font-size: 14px; @@ -193,7 +261,48 @@ - + +
🧪 Compétences apprises par supervision
+ +
+
+
--
+
Compétences YAML
+
+
+
--
+
Candidates
+
+
+
--
+
Observées
+
+
+ +
+

📌 Promotions supervisées

+
+ Les boutons ci-dessous font un dry-run, affichent le diff YAML, puis demandent confirmation. Aucun write-back silencieux. +
+
+ + + + + + + + + + + + + +
CompétenceÉtatVerdictsContextesPromotion
Chargement...
+
+
+ +
🔄 Workflows
@@ -205,9 +314,33 @@
+ +