feat: load external coach knowledge packs

data/knowledge_cards/grounded_exercise_expansion.json

@@ -0,0 +1,67 @@
+{
+  "cards": [
+    {
+      "card_id": "exercise:squat",
+      "card_type": "exercise",
+      "labels": ["squat"],
+      "title": "Squat",
+      "summary": "A squat summary should stay grounded in rep evidence while using standardized execution cues for depth, balance, torso position, and stance context.",
+      "evidence_rules": [
+        "Use rep analysis and issue markers instead of inferring directly from the video.",
+        "Describe squat depth, torso control, and balance only when supported by structured evidence.",
+        "Treat stance-width context as a valid variation unless issue markers show a separate problem."
+      ],
+      "coaching_points": [
+        "Reference depth, torso control, and rep-to-rep consistency using the measured evidence.",
+        "Keep top fixes practical, such as controlling the descent or sitting slightly deeper when `shallow_depth` is present.",
+        "Keep language descriptive rather than diagnostic."
+      ],
+      "allowed_interpretations": [
+        "Target muscles commonly include quads and glutes, but coaching should remain focused on the observed movement evidence."
+      ],
+      "related_cards": ["issue:shallow_depth", "issue:excessive_torso_lean"]
+    },
+    {
+      "card_id": "exercise:push_up",
+      "card_type": "exercise",
+      "labels": ["push_up"],
+      "title": "Push-up",
+      "summary": "A push-up summary should explain body-line control, bottom depth, and rep consistency from structured evidence, while recognizing valid regressions and grip variations.",
+      "evidence_rules": [
+        "Treat hand placement or knee support as variation context when the variation detector marks them as not-issues.",
+        "Do not infer pain, fatigue causes, or joint pathology from the movement.",
+        "Use only the detected issue labels for correction language."
+      ],
+      "coaching_points": [
+        "Explain whether the body line stayed organized across the set before suggesting fixes.",
+        "If `hip_sag` or `incomplete_depth` appears, keep corrections specific to those labels.",
+        "Keep next-session advice simple and repeatable for the current goal."
+      ],
+      "allowed_interpretations": [
+        "Primary effort often involves chest, shoulders, and triceps, but the summary should still stay grounded in the measured rep evidence."
+      ],
+      "related_cards": ["variation:wide_grip_push_up", "variation:knee_push_up", "issue:hip_sag", "issue:incomplete_depth"]
+    },
+    {
+      "card_id": "exercise:shoulder_press",
+      "card_type": "exercise",
+      "labels": ["shoulder_press"],
+      "title": "Shoulder Press",
+      "summary": "A shoulder press summary should focus on top position, left-right symmetry, and rep consistency using structured evidence and standardized overhead pressing cues.",
+      "evidence_rules": [
+        "Use rep analysis and issue markers instead of diagnosing shoulder limitations.",
+        "Partial range can be valid variation context and should not be overcorrected by default.",
+        "Only describe asymmetry or incomplete lockout when those labels are present in JSON."
+      ],
+      "coaching_points": [
+        "Separate partial-range context from `incomplete_lockout` issue language.",
+        "Call out rep-to-rep symmetry drift only when the evidence shows it.",
+        "Keep the next-session plan limited to one or two repeatable priorities."
+      ],
+      "allowed_interpretations": [
+        "Primary effort often involves shoulders and triceps, but the coach summary must stay focused on observed rep evidence."
+      ],
+      "related_cards": ["issue:incomplete_lockout", "issue:asymmetry"]
+    }
+  ]
+}

src/pozify/knowledge_cards.py

@@ -1,6 +1,10 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import json
+import os
+from functools import lru_cache
+from pathlib import Path
 
 from pozify.contracts import ExerciseClassification, GOALS, IssueMarkers, UserProfile, Variation
 
@@ -26,6 +30,24 @@ class KnowledgeCard:
     allowed_interpretations: tuple[str, ...] = ()
     forbidden_claims: tuple[str, ...] = ()
     related_cards: tuple[str, ...] = ()
+    source_kind: str = "builtin"
+    source_path: str | None = None
+
+
+@dataclass(frozen=True)
+class KnowledgeCatalog:
+    cards: tuple[KnowledgeCard, ...]
+    cards_by_id: dict[str, KnowledgeCard]
+    cards_by_label: dict[str, KnowledgeCard]
+    loaded_pack_paths: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class KnowledgeRetrieval:
+    cards: list[KnowledgeCard]
+    loaded_pack_paths: tuple[str, ...]
+    external_cards_loaded: int
+    external_cards_retrieved: int
 
 
 def _card(
@@ -40,6 +62,8 @@ def _card(
     allowed_interpretations: tuple[str, ...] = (),
     forbidden_claims: tuple[str, ...] = (),
     related_cards: tuple[str, ...] = (),
+    source_kind: str = "builtin",
+    source_path: str | None = None,
 ) -> KnowledgeCard:
     return KnowledgeCard(
         card_id=card_id,
@@ -52,6 +76,8 @@ def _card(
         allowed_interpretations=allowed_interpretations,
         forbidden_claims=forbidden_claims,
         related_cards=related_cards,
+        source_kind=source_kind,
+        source_path=source_path,
     )
 
 
@@ -315,22 +341,202 @@ CARD_REGISTRY: tuple[KnowledgeCard, ...] = (
 )
 
 
-CARDS_BY_ID = {card.card_id: card for card in CARD_REGISTRY}
-CARDS_BY_LABEL = {
-    label: card
-    for card in CARD_REGISTRY
-    for label in card.labels
-}
-KNOWN_ISSUE_LABELS = frozenset(
-    label for card in CARD_REGISTRY if card.card_type == "issue" for label in card.labels
-)
-KNOWN_VARIATION_LABELS = frozenset(
-    label for card in CARD_REGISTRY if card.card_type == "variation" for label in card.labels
+DEFAULT_CARD_PACK_PATHS = (
+    Path(__file__).resolve().parents[2] / "data/knowledge_cards/grounded_exercise_expansion.json",
 )
+CARD_PACKS_ENV = "POZIFY_KNOWLEDGE_CARD_PACKS"
+
+
+def _candidate_card_pack_paths() -> tuple[str, ...]:
+    candidates: list[str] = []
+    for path in DEFAULT_CARD_PACK_PATHS:
+        if path.is_file():
+            candidates.append(str(path.resolve()))
+
+    configured = os.getenv(CARD_PACKS_ENV, "").strip()
+    if not configured:
+        return tuple(candidates)
+
+    for raw_path in configured.split(os.pathsep):
+        cleaned = raw_path.strip()
+        if not cleaned:
+            continue
+        resolved = str(Path(cleaned).expanduser().resolve())
+        if resolved not in candidates:
+            candidates.append(resolved)
+    return tuple(candidates)
+
+
+def _normalize_string_tuple(values: object, field_name: str, source_path: str) -> tuple[str, ...]:
+    if not isinstance(values, list):
+        raise ValueError(f"{source_path}: {field_name} must be a list of strings")
+    normalized: list[str] = []
+    for index, value in enumerate(values):
+        if not isinstance(value, str) or not value.strip():
+            raise ValueError(
+                f"{source_path}: {field_name}[{index}] must be a non-empty string"
+            )
+        normalized.append(value.strip())
+    return tuple(normalized)
+
+
+def _load_pack_card(payload: object, source_path: str) -> KnowledgeCard:
+    if not isinstance(payload, dict):
+        raise ValueError(f"{source_path}: each card must be an object")
+
+    required = {
+        "card_id",
+        "card_type",
+        "labels",
+        "title",
+        "summary",
+        "evidence_rules",
+        "coaching_points",
+    }
+    missing = sorted(required - payload.keys())
+    if missing:
+        raise ValueError(
+            f"{source_path}: card missing required field(s): {', '.join(missing)}"
+        )
+
+    card_id = payload["card_id"]
+    card_type = payload["card_type"]
+    title = payload["title"]
+    summary = payload["summary"]
+    if not isinstance(card_id, str) or not card_id.strip():
+        raise ValueError(f"{source_path}: card_id must be a non-empty string")
+    if not isinstance(card_type, str) or not card_type.strip():
+        raise ValueError(f"{source_path}: card_type must be a non-empty string")
+    if not isinstance(title, str) or not title.strip():
+        raise ValueError(f"{source_path}: title must be a non-empty string")
+    if not isinstance(summary, str) or not summary.strip():
+        raise ValueError(f"{source_path}: summary must be a non-empty string")
+
+    return _card(
+        card_id=card_id.strip(),
+        card_type=card_type.strip(),
+        labels=_normalize_string_tuple(payload["labels"], "labels", source_path),
+        title=title.strip(),
+        summary=summary.strip(),
+        evidence_rules=_normalize_string_tuple(
+            payload["evidence_rules"], "evidence_rules", source_path
+        ),
+        coaching_points=_normalize_string_tuple(
+            payload["coaching_points"], "coaching_points", source_path
+        ),
+        allowed_interpretations=_normalize_string_tuple(
+            payload.get("allowed_interpretations", []),
+            "allowed_interpretations",
+            source_path,
+        ),
+        forbidden_claims=_normalize_string_tuple(
+            payload.get("forbidden_claims", []),
+            "forbidden_claims",
+            source_path,
+        ),
+        related_cards=_normalize_string_tuple(
+            payload.get("related_cards", []),
+            "related_cards",
+            source_path,
+        ),
+        source_kind="external",
+        source_path=source_path,
+    )
+
+
+def _load_external_cards(path: str) -> tuple[KnowledgeCard, ...]:
+    resolved = Path(path).expanduser().resolve()
+    if not resolved.is_file():
+        raise FileNotFoundError(f"Knowledge card pack not found: {resolved}")
+
+    payload = json.loads(resolved.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict):
+        raise ValueError(f"{resolved}: card pack must be a JSON object")
+    cards_payload = payload.get("cards")
+    if not isinstance(cards_payload, list):
+        raise ValueError(f"{resolved}: `cards` must be a list")
+
+    cards = [_load_pack_card(card_payload, str(resolved)) for card_payload in cards_payload]
+    seen_ids: set[str] = set()
+    for card in cards:
+        if card.card_id in seen_ids:
+            raise ValueError(f"{resolved}: duplicate card_id {card.card_id!r}")
+        seen_ids.add(card.card_id)
+    return tuple(cards)
+
+
+def _build_cards_by_label(cards: tuple[KnowledgeCard, ...]) -> dict[str, KnowledgeCard]:
+    cards_by_label: dict[str, KnowledgeCard] = {}
+    for card in cards:
+        for label in card.labels:
+            existing = cards_by_label.get(label)
+            if existing is not None and existing.card_id != card.card_id:
+                raise ValueError(
+                    f"Knowledge label conflict for {label!r}: "
+                    f"{existing.card_id!r} vs {card.card_id!r}"
+                )
+            cards_by_label[label] = card
+    return cards_by_label
+
+
+@lru_cache(maxsize=8)
+def _catalog_for_paths(pack_paths: tuple[str, ...]) -> KnowledgeCatalog:
+    cards_by_id = {card.card_id: card for card in CARD_REGISTRY}
+    for path in pack_paths:
+        for card in _load_external_cards(path):
+            cards_by_id[card.card_id] = card
+
+    cards = tuple(
+        sorted(
+            cards_by_id.values(),
+            key=lambda card: (CARD_TYPE_ORDER.get(card.card_type, 99), card.card_id),
+        )
+    )
+    return KnowledgeCatalog(
+        cards=cards,
+        cards_by_id={card.card_id: card for card in cards},
+        cards_by_label=_build_cards_by_label(cards),
+        loaded_pack_paths=pack_paths,
+    )
+
+
+def get_catalog(pack_paths: tuple[str, ...] | None = None) -> KnowledgeCatalog:
+    selected_paths = _candidate_card_pack_paths() if pack_paths is None else pack_paths
+    return _catalog_for_paths(tuple(selected_paths))
+
+
+def configured_card_pack_paths() -> tuple[str, ...]:
+    return get_catalog().loaded_pack_paths
+
+
+def clear_catalog_cache() -> None:
+    _catalog_for_paths.cache_clear()
+
+
+def _labels_for_card_type(card_type: str, *, pack_paths: tuple[str, ...] | None = None) -> frozenset[str]:
+    catalog = get_catalog(pack_paths)
+    return frozenset(
+        label
+        for card in catalog.cards
+        if card.card_type == card_type
+        for label in card.labels
+    )
+
+
+KNOWN_ISSUE_LABELS = _labels_for_card_type("issue")
+KNOWN_VARIATION_LABELS = _labels_for_card_type("variation")
+
+
+def known_issue_labels() -> frozenset[str]:
+    return _labels_for_card_type("issue")
+
+
+def known_variation_labels() -> frozenset[str]:
+    return _labels_for_card_type("variation")
 
 
 def get_card_by_label(label: str) -> KnowledgeCard | None:
-    return CARDS_BY_LABEL.get(label)
+    return get_catalog().cards_by_label.get(label)
 
 
 def cards_for_labels(labels: list[str] | tuple[str, ...]) -> list[KnowledgeCard]:
@@ -346,6 +552,32 @@ def cards_for_labels(labels: list[str] | tuple[str, ...]) -> list[KnowledgeCard]
     )
 
 
+def retrieve_cards_with_metadata(
+    *,
+    profile: UserProfile,
+    classification: ExerciseClassification,
+    variation: Variation,
+    issues: IssueMarkers,
+) -> KnowledgeRetrieval:
+    cards = retrieve_cards(
+        profile=profile,
+        classification=classification,
+        variation=variation,
+        issues=issues,
+    )
+    catalog = get_catalog()
+    return KnowledgeRetrieval(
+        cards=cards,
+        loaded_pack_paths=catalog.loaded_pack_paths,
+        external_cards_loaded=sum(
+            1 for card in catalog.cards if card.source_kind == "external"
+        ),
+        external_cards_retrieved=sum(
+            1 for card in cards if card.source_kind == "external"
+        ),
+    )
+
+
 def retrieve_cards(
     *,
     profile: UserProfile,

src/pozify/pipeline.py

@@ -11,7 +11,7 @@ from pozify.artifacts import write_json
 from pozify.contracts import UserProfile, to_dict
 from pozify.env import env_truthy, load_local_env
 from pozify.exercises import create_exercise_strategy
-from pozify.knowledge_cards import retrieve_cards
+from pozify.knowledge_cards import retrieve_cards_with_metadata
 from pozify.steps import (
     annotated_renderer,
     coach_summary,
@@ -232,12 +232,13 @@ def run_pipeline(
     if mock_mode:
         mock_steps.insert(0, "exercise_classifier")
 
-    summary_cards = retrieve_cards(
+    knowledge_retrieval = retrieve_cards_with_metadata(
         profile=profile,
         classification=classification,
         variation=variation,
         issues=issues,
     )
+    summary_cards = knowledge_retrieval.cards
     coach_result = coach_summary.run_with_metadata(
         profile,
         classification,
@@ -319,6 +320,9 @@ def run_pipeline(
             "coach_summary_model": coach_summary_model,
             "coach_summary_verifier_bypassed": coach_summary_verifier_bypassed,
             "coach_summary_verifier_bypass_requested": bypass_verifier_enabled,
+            "knowledge_card_pack_paths": list(knowledge_retrieval.loaded_pack_paths),
+            "knowledge_external_cards_loaded": knowledge_retrieval.external_cards_loaded,
+            "knowledge_external_cards_retrieved": knowledge_retrieval.external_cards_retrieved,
         },
     }
     write_artifact("final_report.json", final_report)

src/pozify/steps/verifier.py

@@ -11,7 +11,7 @@ from pozify.contracts import (
     Variation,
     Verification,
 )
-from pozify.knowledge_cards import KNOWN_ISSUE_LABELS
+from pozify.knowledge_cards import known_issue_labels
 
 
 DIAGNOSIS_PATTERNS = (
@@ -80,7 +80,7 @@ def _mentioned_labels(summary: CoachSummary) -> set[str]:
     text = " ".join(_summary_sections(summary))
     labels = set(re.findall(r"`([a-z0-9_]+)`", text))
     lowered = text.lower()
-    for label in KNOWN_ISSUE_LABELS:
+    for label in known_issue_labels():
         if label in lowered:
             labels.add(label)
     return labels
@@ -117,7 +117,7 @@ def run(
 ) -> Verification:
     allowed_issues = {issue.issue for issue in issues.issues}
     mentioned_labels = _mentioned_labels(summary)
-    mentioned_issues = mentioned_labels & KNOWN_ISSUE_LABELS
+    mentioned_issues = mentioned_labels & known_issue_labels()
 
     no_issue_outside_json = mentioned_issues <= allowed_issues