docs(milestone): merge v0.3-mastery-scoring → main

v0.3 milestone merged to main. Mastery scoring + competency rubrics + verifiable credentials (formative-tier) shipped. 13/13 REQ-IDs covered. Next milestone: v0.4 (operator tier — cohort dashboard + auth + Postgres). ---ci--- project: praxis phase: 2 milestone: v0.3 status: complete milestone_complete: true milestone_merged_to_main: true ---/ci---
2026-08-04 00:14:59 +00:00
parent bea2af13d4
commit 813bd586d6
67 changed files with 10365 additions and 1697 deletions
@@ -0,0 +1,214 @@
+"""W3C VC 2.0 issuance — Ed25519 + JCS + eddsa-jcs-2022 proof (SLICE-09 TASK-09-02).
+
+Builds a Verifiable Credential per VC-DM 2.0, secures it with a Data Integrity
+`eddsa-jcs-2022` proof (JCS canonicalization, Ed25519 signature), and persists
+it to SQLite. The `issue_credential` coroutine is the entry point wired into
+SessionRecorder.run_mastery_flow (grill Axis 8 MUST).
+
+Credential tier is `formative` (grill Axis 4 MUST #1) — the v0.3 credential is
+a formative mastery signal, not a high-stakes summative credential.
+"""
+
+from __future__ import annotations
+
+import base64
+import datetime as _dt
+import hashlib
+import json
+import os
+import uuid
+from typing import Any
+
+import canonicaljson
+import nacl.signing
+from db.store import PraxisStore
+
+from server.vc.issuer_keys import KeyPair, get_active_signing_key
+from server.vc.status_list import BitstringStatusList
+
+ISSUER_URL_DEFAULT = "https://praxis.example/issuers/v0.3"
+CONTEXTS = [
+    "https://www.w3.org/ns/credentials/v2",
+    "https://praxis.example/contexts/mastery/v1",
+]
+CREDENTIAL_TIER = "formative"
+
+
+def _issuer_url() -> str:
+    return os.environ.get("PRAXIS_ISSUER_URL", ISSUER_URL_DEFAULT).rstrip("/")
+
+
+def _now_iso() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _valid_until(issuance_iso: str, years: int = 3) -> str:
+    dt = _dt.datetime.strptime(issuance_iso, "%Y-%m-%dT%H:%M:%SZ").replace(
+        tzinfo=_dt.timezone.utc
+    )
+    return (dt + _dt.timedelta(days=365 * years)).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def build_vc_payload(
+    learner_ref: str,
+    path: str,
+    scenarios_passed: list[str],
+    rubric_score: float,
+    completed_weeks: int,
+    evidence: list[dict[str, Any]] | None,
+    credential_id: str | None = None,
+    status_list_index: int | None = None,
+) -> dict[str, Any]:
+    issuance = _now_iso()
+    issuer = _issuer_url()
+    cid = credential_id or f"vc-{uuid.uuid4().hex[:16]}"
+    payload: dict[str, Any] = {
+        "@context": list(CONTEXTS),
+        "id": f"{issuer}/vc/{cid}",
+        "type": ["VerifiableCredential", "MasteryCredential"],
+        "issuer": issuer,
+        "validFrom": issuance,
+        "validUntil": _valid_until(issuance, 3),
+        "name": f"Mastery of {path.replace('-', ' ').title()}",
+        "description": (
+            "Praxis v0.3 formative mastery credential — the holder demonstrated "
+            "competency across varied scenarios, scored against a 5-level rubric."
+        ),
+        "credentialTier": CREDENTIAL_TIER,
+        "credentialSubject": {
+            "id": f"urn:uuid:{learner_ref}",
+            "type": "Person",
+            "skill": path,
+            "level": "mastery",
+            "path": path,
+            "completedWeeks": completed_weeks,
+            "rubricScore": round(float(rubric_score), 3),
+            "rubricMax": 5.0,
+            "rubricThreshold": 3.5,
+            "scenariosPassed": list(scenarios_passed),
+            "credentialTier": CREDENTIAL_TIER,
+            "evidence": evidence or [],
+        },
+    }
+    if status_list_index is not None:
+        payload["credentialStatus"] = {
+            "type": "BitstringStatusListEntry",
+            "statusPurpose": "revocation",
+            "statusListIndex": str(status_list_index),
+            "statusListCredential": f"{issuer}/status/default",
+        }
+    return payload
+
+
+def canonicalize(payload: dict[str, Any]) -> bytes:
+    return canonicaljson.encode_canonical_json(payload)
+
+
+def _build_proof_config(key_id: str) -> dict[str, Any]:
+    issuer = _issuer_url()
+    return {
+        "type": "DataIntegrityProof",
+        "cryptosuite": "eddsa-jcs-2022",
+        "created": _now_iso(),
+        "verificationMethod": f"{issuer}/keys/{key_id}",
+        "proofPurpose": "assertionMethod",
+    }
+
+
+def _compute_hash_data(
+    unsecured_doc: dict[str, Any], proof_options: dict[str, Any]
+) -> bytes:
+    canonical_doc = canonicalize(unsecured_doc)
+    canonical_proof = canonicalize(proof_options)
+    return hashlib.sha256(canonical_proof).digest() + hashlib.sha256(
+        canonical_doc
+    ).digest()
+
+
+def sign(payload: dict[str, Any], signing_key: nacl.signing.SigningKey, key_id: str) -> tuple[dict[str, Any], str]:
+    proof_options = _build_proof_config(key_id)
+    hash_data = _compute_hash_data(payload, proof_options)
+    signed = signing_key.sign(hash_data)
+    signature_bytes = signed.signature
+    signature_b64 = base64.b64encode(signature_bytes).decode("ascii")
+    proof = dict(proof_options)
+    proof["proofValue"] = signature_b64
+    secured = dict(payload)
+    secured["proof"] = proof
+    return secured, signature_b64
+
+
+def verify_proof(
+    secured_doc: dict[str, Any],
+    verify_key: nacl.signing.VerifyKey,
+) -> bool:
+    if "proof" not in secured_doc:
+        return False
+    proof = secured_doc["proof"]
+    proof_value_b64 = proof.get("proofValue")
+    if not proof_value_b64:
+        return False
+    proof_options = {k: v for k, v in proof.items() if k != "proofValue"}
+    unsecured = {k: v for k, v in secured_doc.items() if k != "proof"}
+    hash_data = _compute_hash_data(unsecured, proof_options)
+    try:
+        sig = base64.b64decode(proof_value_b64)
+        verify_key.verify(hash_data, sig)
+        return True
+    except Exception:
+        return False
+
+
+def extract_key_id(secured_doc: dict[str, Any]) -> str | None:
+    proof = secured_doc.get("proof") or {}
+    vm = proof.get("verificationMethod") or ""
+    if "/" in vm:
+        return vm.rsplit("/", 1)[-1]
+    return None
+
+
+async def issue_credential(
+    store: PraxisStore,
+    signing_key: nacl.signing.SigningKey | None = None,
+    learner_id: str = "",
+    path: str = "",
+    scenarios_passed: list[str] | None = None,
+    rubric_score: float = 0.0,
+    completed_weeks: int = 6,
+    evidence: list[dict[str, Any]] | None = None,
+    key_id: str | None = None,
+) -> str:
+    if signing_key is None or key_id is None:
+        kp, _enc = await get_active_signing_key(store)
+        signing_key = kp.signing_key
+        key_id = kp.key_id
+    scenarios = list(scenarios_passed or [])
+    ev = list(evidence or [])
+    status_list = BitstringStatusList(store, "default")
+    slot = await status_list.allocate_slot()
+    cred_id = f"vc-{uuid.uuid4().hex[:16]}"
+    payload = build_vc_payload(
+        learner_ref=learner_id,
+        path=path,
+        scenarios_passed=scenarios,
+        rubric_score=rubric_score,
+        completed_weeks=completed_weeks,
+        evidence=ev,
+        credential_id=cred_id,
+        status_list_index=slot,
+    )
+    secured, signature_b64 = sign(payload, signing_key, key_id)
+    payload_json = json.dumps(secured, sort_keys=True, separators=(",", ":"))
+    await store.insert_credential(cred_id, learner_id, payload_json, signature_b64)
+    return cred_id
+
+
+__all__ = [
+    "build_vc_payload",
+    "canonicalize",
+    "sign",
+    "verify_proof",
+    "extract_key_id",
+    "issue_credential",
+    "CREDENTIAL_TIER",
+]