diff --git a/Courseforge/scripts/generate_course.py b/Courseforge/scripts/generate_course.py
index ad7d47652..52404dcf4 100644
--- a/Courseforge/scripts/generate_course.py
+++ b/Courseforge/scripts/generate_course.py
@@ -12,7 +12,7 @@
- discussion.html (forum prompt with guidelines)
Usage:
- python generate_course.py DIGPED_101_course_data.json output_dir/
+ python generate_course.py SAMPLE_101_course_data.json output_dir/
"""
import html as html_mod
@@ -155,14 +155,18 @@ def _slugify(text: str) -> str:
def _wrap_page(title: str, course_code: str, week_num: int, body_html: str,
extra_js: str = "",
- page_metadata: Optional[Dict[str, Any]] = None) -> str:
+ page_metadata: Optional[Dict[str, Any]] = None,
+ course_title: Optional[str] = None) -> str:
"""Wrap body content in a full HTML page with Courseforge styling.
Args:
page_metadata: Optional structured metadata dict rendered as JSON-LD
in for downstream Trainforge extraction.
+ course_title: Human-readable course title used in the header/footer.
+ Falls back to ``course_code`` if not supplied.
"""
safe_title = html_mod.escape(title)
+ header_label = html_mod.escape(course_title) if course_title else course_code
json_ld = ""
if page_metadata:
json_ld = (
@@ -181,14 +185,14 @@ def _wrap_page(title: str, course_code: str, week_num: int, body_html: str,
Skip to main content
- {course_code}: Foundations of Digital Pedagogy — Week {week_num}
+ {course_code}: {header_label} — Week {week_num}
{safe_title}
{body_html}
{extra_js}
@@ -488,7 +492,8 @@ def _build_page_metadata(
return meta
-def generate_week(week_data: Dict, output_dir: Path, course_code: str):
+def generate_week(week_data: Dict, output_dir: Path, course_code: str,
+ course_title: Optional[str] = None):
"""Generate all files for a single week."""
week_num = week_data["week_number"]
week_dir = output_dir / f"week_{week_num:02d}"
@@ -523,6 +528,7 @@ def generate_week(week_data: Dict, output_dir: Path, course_code: str):
f"Week {week_num} Overview: {week_data['title']}",
course_code, week_num, overview_body,
page_metadata=overview_meta,
+ course_title=course_title,
)
(week_dir / f"week_{week_num:02d}_overview.html").write_text(overview_html, encoding="utf-8")
@@ -544,6 +550,7 @@ def generate_week(week_data: Dict, output_dir: Path, course_code: str):
f"Week {week_num}: {content['title']}",
course_code, week_num, content_body, extra_js,
page_metadata=content_meta,
+ course_title=course_title,
)
filename = f"{page_id}.html"
(week_dir / filename).write_text(content_html, encoding="utf-8")
@@ -562,6 +569,7 @@ def generate_week(week_data: Dict, output_dir: Path, course_code: str):
f"Week {week_num}: Application & Activities",
course_code, week_num, app_body,
page_metadata=app_meta,
+ course_title=course_title,
)
(week_dir / f"week_{week_num:02d}_application.html").write_text(app_html, encoding="utf-8")
@@ -580,6 +588,7 @@ def generate_week(week_data: Dict, output_dir: Path, course_code: str):
f"Week {week_num}: Self-Check Quiz",
course_code, week_num, sc_body, SELF_CHECK_JS,
page_metadata=sc_meta,
+ course_title=course_title,
)
(week_dir / f"week_{week_num:02d}_self_check.html").write_text(sc_html, encoding="utf-8")
@@ -604,6 +613,7 @@ def generate_week(week_data: Dict, output_dir: Path, course_code: str):
f"Week {week_num}: Summary & Reflection",
course_code, week_num, summary_body,
page_metadata=summary_meta,
+ course_title=course_title,
)
(week_dir / f"week_{week_num:02d}_summary.html").write_text(summary_html, encoding="utf-8")
@@ -630,6 +640,7 @@ def generate_week(week_data: Dict, output_dir: Path, course_code: str):
f"Week {week_num}: Discussion",
course_code, week_num, disc_body,
page_metadata=disc_meta,
+ course_title=course_title,
)
(week_dir / f"week_{week_num:02d}_discussion.html").write_text(disc_html, encoding="utf-8")
@@ -643,10 +654,11 @@ def generate_course(course_data_path: str, output_dir: str):
data = json.loads(Path(course_data_path).read_text())
out = Path(output_dir)
course_code = data.get("course_code", "COURSE_101")
+ course_title = data.get("course_title")
total_files = 0
for week in data["weeks"]:
- count, files = generate_week(week, out, course_code)
+ count, files = generate_week(week, out, course_code, course_title)
total_files += count
print(f" Week {week['week_number']:2d}: {count} files - {', '.join(files)}")
diff --git a/Courseforge/scripts/package_multifile_imscc.py b/Courseforge/scripts/package_multifile_imscc.py
index 5aee5ecad..31559c984 100644
--- a/Courseforge/scripts/package_multifile_imscc.py
+++ b/Courseforge/scripts/package_multifile_imscc.py
@@ -137,8 +137,8 @@ def package_imscc(content_dir: Path, output_path: Path, course_code: str, course
content_dir = Path(sys.argv[1])
output_path = Path(sys.argv[2])
- course_code = sys.argv[3] if len(sys.argv) > 3 else "DIGPED_101"
- course_title = sys.argv[4] if len(sys.argv) > 4 else "Foundations of Digital Pedagogy"
+ course_code = sys.argv[3] if len(sys.argv) > 3 else "SAMPLE_101"
+ course_title = sys.argv[4] if len(sys.argv) > 4 else "Sample Course"
output_path.parent.mkdir(parents=True, exist_ok=True)
package_imscc(content_dir, output_path, course_code, course_title)
diff --git a/Trainforge/align_chunks.py b/Trainforge/align_chunks.py
index d7c2b64ca..cc2a6571f 100644
--- a/Trainforge/align_chunks.py
+++ b/Trainforge/align_chunks.py
@@ -8,8 +8,8 @@
Usage:
python -m Trainforge.align_chunks \
- --corpus Trainforge/output/digped_101 \
- --objectives Courseforge/inputs/exam-objectives/DIGPED_101_objectives.json \
+ --corpus Trainforge/output/sample_101 \
+ --objectives Courseforge/inputs/exam-objectives/SAMPLE_101_objectives.json \
--llm-provider mock
"""
diff --git a/Trainforge/process_course.py b/Trainforge/process_course.py
index 5615486d8..3e6e281af 100644
--- a/Trainforge/process_course.py
+++ b/Trainforge/process_course.py
@@ -8,17 +8,17 @@
Usage:
python -m Trainforge.process_course \
--imscc path/to/course.imscc \
- --course-code DIGPED_101 \
+ --course-code SAMPLE_101 \
--division ARTS --domain education --subdomain instructional-design \
- --output Trainforge/output/digped_101
+ --output Trainforge/output/sample_101
# With objectives file for Bloom's-based difficulty mapping:
python -m Trainforge.process_course \
--imscc path/to/course.imscc \
--objectives path/to/objectives.json \
- --course-code DIGPED_101 \
+ --course-code SAMPLE_101 \
--division ARTS --domain education \
- --output Trainforge/output/digped_101 \
+ --output Trainforge/output/sample_101 \
--import-to-libv2
"""
@@ -51,7 +51,11 @@
# Bumped whenever the semantics of quality_report.json metrics change.
# v1: field-presence metrics (legacy).
# v2: referential, structural, and content-sanity metrics.
-METRICS_SEMANTIC_VERSION = 2
+# v3: adds outcome_reverse_coverage (metric) + integrity.uncovered_outcomes
+# (list); guaranteed bloom_level on every chunk via verb/default fallback;
+# pedagogy_model.json grows module_sequence, bloom_progression,
+# prerequisite_chain, prerequisite_violations.
+METRICS_SEMANTIC_VERSION = 3
class PipelineIntegrityError(RuntimeError):
@@ -178,9 +182,42 @@ def load_objectives(objectives_path: Path) -> Dict[str, Any]:
"bloom_distribution": data.get("bloom_distribution", {}),
"description": data.get("description", ""),
"course_title": data.get("course_title", ""),
+ # Optional per-course domain concept seeds. Shape:
+ # [{"id": "pour", "aliases": ["POUR", "perceivable operable"]}, ...]
+ # CONCEPT_PATTERNS covers pedagogy terms only, so domain seeds are
+ # the only text-based extraction path for course-specific vocabulary.
+ "domain_concepts": data.get("domain_concepts", []),
}
+def compile_domain_concept_seeds(
+ raw: List[Dict[str, Any]],
+) -> List[Tuple[str, List[re.Pattern]]]:
+ """Compile the domain_concepts block from an objectives file into
+ (canonical_tag, [word-boundary regex]) pairs for fast matching.
+
+ Aliases are matched case-insensitively with \\b word boundaries so that
+ short tokens (``aria``, ``udl``) don't match inside longer words.
+ """
+ seeds: List[Tuple[str, List[re.Pattern]]] = []
+ for entry in raw or []:
+ canonical = normalize_tag(entry.get("id", ""))
+ if not canonical:
+ continue
+ aliases = list(entry.get("aliases") or [])
+ if entry.get("id") and entry["id"] not in aliases:
+ aliases.append(entry["id"])
+ patterns: List[re.Pattern] = []
+ for alias in aliases:
+ alias = str(alias).strip()
+ if not alias:
+ continue
+ patterns.append(re.compile(rf"\b{re.escape(alias)}\b", re.IGNORECASE))
+ if patterns:
+ seeds.append((canonical, patterns))
+ return seeds
+
+
# ---------------------------------------------------------------------------
# Concept tag normalization
# ---------------------------------------------------------------------------
@@ -240,6 +277,12 @@ def normalize_tag(raw: str) -> str:
"correct", "incorrect", "submit", "deadline", "grading",
"readings", "resources", "learning-objectives",
"estimated-time", "time", "minutes", "hours",
+ # "feedback" is legitimate pedagogy vocabulary (formative feedback in
+ # course theory courses). For domain courses it reliably pollutes the
+ # concept graph via boilerplate like "you'll receive immediate feedback"
+ # in quiz intros. Routing it to pedagogy_graph keeps the signal without
+ # polluting the domain graph.
+ "feedback",
}
# Divs carrying these attribute prefixes are atomic — the chunker must not
@@ -444,8 +487,12 @@ def __init__(
# Objectives (optional)
self.objectives: Optional[Dict[str, Any]] = None
+ self.domain_concept_seeds: List[Tuple[str, List[re.Pattern]]] = []
if objectives_path:
self.objectives = load_objectives(Path(objectives_path))
+ self.domain_concept_seeds = compile_domain_concept_seeds(
+ self.objectives.get("domain_concepts", [])
+ )
# Decision capture
self.capture = DecisionCapture(
@@ -519,7 +566,7 @@ def process(self) -> Dict[str, Any]:
# Stage 6
print("[6/6] Writing metadata files...")
self._write_metadata(manifest, corpus_stats, concept_graph, quality_report,
- pedagogy_graph=pedagogy_graph)
+ pedagogy_graph=pedagogy_graph, chunks=chunks)
summary = {
"status": "success",
@@ -862,28 +909,69 @@ def _create_chunk(
"word_count": word_count,
}
- # Enrich from Courseforge metadata (JSON-LD / data-cf-*)
+ # Enrich from Courseforge metadata (JSON-LD / data-cf-*).
+ # Resolution order: section JSON-LD → page JSON-LD → parsed LOs →
+ # text verb heuristic → hardcoded default. Every chunk ends up with
+ # a bloom_level; bloom_level_source records where it came from so
+ # downstream consumers can weight low-confidence sources.
bloom_level, content_type_label, key_terms = self._extract_section_metadata(
item, section_heading
)
- # Fallback: if section metadata didn't provide bloom_level,
- # derive from page-level JSON-LD objectives or parsed objectives
+ bloom_source = "section_jsonld" if bloom_level else None
+
+ # Merge structured JSON-LD keyTerms into concept_tags. These are the
+ # highest-fidelity domain vocabulary Courseforge emits; leaving them
+ # in chunk["key_terms"] only meant the concept graph missed them.
+ for kt in key_terms or []:
+ term = kt.get("term") if isinstance(kt, dict) else kt
+ tag = normalize_tag(term or "")
+ if not tag or len(tag) < 3 or tag in concept_tags:
+ continue
+ if (self.OBJECTIVE_CODE_RE.match(tag)
+ or self.WEEK_PREFIX_RE.match(tag)
+ or tag in self.NON_CONCEPT_TAGS):
+ continue
+ concept_tags.append(tag)
+
if not bloom_level:
cf_meta = item.get("courseforge_metadata")
if cf_meta and cf_meta.get("learningObjectives"):
for lo in cf_meta["learningObjectives"]:
if lo.get("bloomLevel"):
bloom_level = lo["bloomLevel"]
+ bloom_source = "page_jsonld"
break
if not bloom_level:
for lo in item.get("learning_objectives", []):
bl = lo.bloom_level if hasattr(lo, "bloom_level") else lo.get("bloom_level")
if bl:
bloom_level = bl
+ bloom_source = "lo_inherited"
break
-
- if bloom_level:
- chunk["bloom_level"] = bloom_level
+ if not bloom_level:
+ derived = derive_bloom_from_verbs(text)
+ if derived:
+ bloom_level = derived
+ bloom_source = "verbs"
+ if not bloom_level:
+ bloom_level = "understand"
+ bloom_source = "default"
+
+ chunk["bloom_level"] = bloom_level
+ # Only tag the source when it's below lo_inherited confidence;
+ # authoritative chunks stay schema-identical to pre-fallback output.
+ if bloom_source in ("verbs", "default"):
+ chunk["bloom_level_source"] = bloom_source
+ self.capture.log_decision(
+ decision_type="bloom_level_assignment",
+ decision=f"Assigned bloom_level={bloom_level} via {bloom_source}",
+ rationale=(
+ "No JSON-LD, data-cf-*, or parsed learning objective "
+ "supplied a bloom_level for this chunk; fell back to the "
+ "text verb heuristic (or the understand-level default) "
+ "so every chunk carries a level for downstream filters."
+ ),
+ )
if content_type_label:
chunk["content_type_label"] = content_type_label
if key_terms:
@@ -1067,13 +1155,26 @@ def _extract_concept_tags(self, text: str, item: Dict[str, Any]) -> List[str]:
continue
tags.append(tag)
- # Text-based concept detection
+ # Text-based concept detection (pedagogy-only patterns).
text_lower = text.lower()
for tag, patterns in self.CONCEPT_PATTERNS.items():
if tag not in tags and any(p in text_lower for p in patterns):
tags.append(tag)
- return tags[:10]
+ # Per-course domain concept seeds. Pedagogy filter still applies
+ # below since seeds are authored per course; a well-formed seed
+ # list won't collide with NON_CONCEPT_TAGS, but we defend anyway.
+ for canonical, patterns in self.domain_concept_seeds:
+ if canonical in tags:
+ continue
+ if (self.OBJECTIVE_CODE_RE.match(canonical)
+ or self.WEEK_PREFIX_RE.match(canonical)
+ or canonical in self.NON_CONCEPT_TAGS):
+ continue
+ if any(p.search(text) for p in patterns):
+ tags.append(canonical)
+
+ return tags[:20]
def _extract_objective_refs(self, item: Dict[str, Any]) -> List[str]:
"""Extract learning objective reference codes from item.
@@ -1492,6 +1593,18 @@ def _generate_quality_report(self, chunks: List[Dict[str, Any]]) -> Dict[str, An
valid_ids = self._valid_outcome_ids or set()
lo_coverage = self._resolving_lo_coverage(chunks, valid_ids)
broken_refs = self._collect_broken_refs(chunks, valid_ids)
+ # Reverse coverage: which declared outcomes have ZERO resolving chunks?
+ # This is the symmetric complement of learning_outcome_coverage and
+ # catches content-generation gaps that the chunk-ratio metric misses.
+ referenced_ids = {
+ r for c in chunks for r in c.get("learning_outcome_refs", [])
+ if r in valid_ids
+ }
+ uncovered_outcomes = sorted(valid_ids - referenced_ids)
+ outcome_reverse_coverage = (
+ (len(valid_ids) - len(uncovered_outcomes)) / len(valid_ids)
+ if valid_ids else 1.0
+ )
# Content sanity: boilerplate contamination + factual flags + follows_chunk scope.
footer_rate = contamination_rate(chunks, self._boilerplate_spans) if self._boilerplate_spans else 0.0
@@ -1514,6 +1627,11 @@ def _generate_quality_report(self, chunks: List[Dict[str, Any]]) -> Dict[str, An
issues.append(f"Bloom level coverage {bloom_coverage:.0%} — below 90% threshold")
if lo_coverage < 0.8:
issues.append(f"Learning outcome coverage {lo_coverage:.0%} — below 80% threshold")
+ if valid_ids and outcome_reverse_coverage < 0.9:
+ issues.append(
+ f"{len(uncovered_outcomes)} learning outcomes have zero resolving chunks: "
+ + ", ".join(uncovered_outcomes)
+ )
if html_preservation < 1.0:
issues.append(f"HTML balance violations in {len(balance_violations)} chunks")
if footer_rate > 0.05:
@@ -1536,6 +1654,7 @@ def _generate_quality_report(self, chunks: List[Dict[str, Any]]) -> Dict[str, An
"html_preservation_rate": round(html_preservation, 3),
"bloom_level_coverage": round(bloom_coverage, 3),
"learning_outcome_coverage": round(lo_coverage, 3),
+ "outcome_reverse_coverage": round(outcome_reverse_coverage, 3),
"footer_contamination_rate": round(footer_rate, 3),
"follows_chunk_boundary_violations": len(boundary_violations),
"avg_chunk_size_words": round(self.stats["total_words"] / total, 1),
@@ -1550,6 +1669,11 @@ def _generate_quality_report(self, chunks: List[Dict[str, Any]]) -> Dict[str, An
"Fraction of chunks that reference at least one outcome ID that "
"resolves to course.json (referential integrity, not field presence)."
),
+ "outcome_reverse_coverage": (
+ "Fraction of declared course.json outcomes that have at least one "
+ "chunk referencing them (catches content-generation gaps where whole "
+ "outcomes are orphaned, which the chunk-ratio coverage misses)."
+ ),
"footer_contamination_rate": (
"Fraction of chunks whose text still contains a detected corpus-wide "
"repeated n-gram (likely footer/template-chrome that escaped stripping)."
@@ -1563,6 +1687,7 @@ def _generate_quality_report(self, chunks: List[Dict[str, Any]]) -> Dict[str, An
"html_balance_violations": balance_violations,
"follows_chunk_boundary_violations": boundary_violations,
"factual_inconsistency_flags": factual_flags,
+ "uncovered_outcomes": uncovered_outcomes,
},
"validation": {"passed": overall >= 0.75 and not broken_refs, "issues": issues},
"recommendations": recommendations,
@@ -1724,7 +1849,17 @@ def _assert_integrity(self, report: Dict[str, Any]) -> None:
+ ". Disable strict_mode to produce a non-final artifact."
)
- def _build_pedagogy_summary(self) -> Dict[str, Any]:
+ def _build_pedagogy_summary(
+ self, chunks: Optional[List[Dict[str, Any]]] = None
+ ) -> Dict[str, Any]:
+ """Build a pedagogy model grounded in the actual chunk set.
+
+ Emits module_sequence (order + per-module stats), bloom_progression
+ (per-module Bloom distribution), and prerequisite_chain (concepts
+ referenced as prereqs after being first introduced earlier). Falls
+ back to just the top-level keys when chunks aren't provided so
+ older call sites don't break.
+ """
summary: Dict[str, Any] = {
"instructional_approach": "competency-based",
"learning_theory": "constructivism",
@@ -1732,6 +1867,99 @@ def _build_pedagogy_summary(self) -> Dict[str, Any]:
}
if self.objectives and self.objectives.get("bloom_distribution"):
summary["bloom_coverage"] = self.objectives["bloom_distribution"]
+
+ if not chunks:
+ return summary
+
+ # --- module_sequence + bloom_progression --------------------------------
+ module_meta: Dict[str, Dict[str, Any]] = {}
+ module_order: List[str] = []
+ bloom_zero = lambda: {
+ "remember": 0, "understand": 0, "apply": 0,
+ "analyze": 0, "evaluate": 0, "create": 0,
+ }
+
+ for chunk in chunks:
+ src = chunk.get("source") or {}
+ module_id = src.get("module_id")
+ if not module_id:
+ continue
+ if module_id not in module_meta:
+ module_order.append(module_id)
+ week_match = re.search(r"week[_\-\s]?(\d+)", module_id, re.IGNORECASE)
+ module_meta[module_id] = {
+ "module_id": module_id,
+ "module_title": src.get("module_title", ""),
+ "week_num": int(week_match.group(1)) if week_match else 0,
+ "chunk_count": 0,
+ "outcome_refs_covered": set(),
+ "bloom_counts": bloom_zero(),
+ "first_seen": len(module_order),
+ }
+ meta = module_meta[module_id]
+ meta["chunk_count"] += 1
+ meta["outcome_refs_covered"].update(chunk.get("learning_outcome_refs", []))
+ bloom = chunk.get("bloom_level")
+ if bloom in meta["bloom_counts"]:
+ meta["bloom_counts"][bloom] += 1
+
+ # Deterministic order: by week_num, then by first-seen position.
+ module_order.sort(key=lambda m: (module_meta[m]["week_num"], module_meta[m]["first_seen"]))
+
+ module_sequence = []
+ bloom_progression: Dict[str, Dict[str, int]] = {}
+ for mid in module_order:
+ meta = module_meta[mid]
+ module_sequence.append({
+ "module_id": mid,
+ "module_title": meta["module_title"],
+ "week_num": meta["week_num"],
+ "chunk_count": meta["chunk_count"],
+ "outcome_refs_covered": sorted(meta["outcome_refs_covered"]),
+ })
+ bloom_progression[mid] = meta["bloom_counts"]
+
+ summary["module_sequence"] = module_sequence
+ summary["bloom_progression"] = bloom_progression
+
+ # --- prerequisite_chain + prerequisite_violations ------------------------
+ # For each concept tag, record earliest (module_idx, chunk_id) where it
+ # appears in concept_tags (definition site) vs prereq_concepts (use site).
+ # Valid chain: first use in module index > first definition's module index.
+ module_idx = {mid: i for i, mid in enumerate(module_order)}
+ first_def: Dict[str, Tuple[int, str, str]] = {}
+ first_use: Dict[str, Tuple[int, str, str]] = {}
+ for chunk in chunks:
+ src = chunk.get("source") or {}
+ mid = src.get("module_id")
+ if mid not in module_idx:
+ continue
+ idx = module_idx[mid]
+ cid = chunk["id"]
+ for tag in chunk.get("concept_tags", []) or []:
+ if tag not in first_def or idx < first_def[tag][0]:
+ first_def[tag] = (idx, mid, cid)
+ for tag in chunk.get("prereq_concepts", []) or []:
+ if tag not in first_use or idx < first_use[tag][0]:
+ first_use[tag] = (idx, mid, cid)
+
+ prerequisite_chain = []
+ prerequisite_violations = []
+ for tag in sorted(set(first_def) & set(first_use)):
+ def_idx, def_mod, def_chunk = first_def[tag]
+ use_idx, use_mod, use_chunk = first_use[tag]
+ record = {
+ "concept": tag,
+ "defined_in": {"module_id": def_mod, "chunk_id": def_chunk},
+ "first_used_in": {"module_id": use_mod, "chunk_id": use_chunk},
+ }
+ if use_idx > def_idx:
+ prerequisite_chain.append(record)
+ elif use_idx < def_idx:
+ prerequisite_violations.append(record)
+
+ summary["prerequisite_chain"] = prerequisite_chain
+ summary["prerequisite_violations"] = prerequisite_violations
return summary
def _build_course_json(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
@@ -1772,6 +2000,7 @@ def _write_metadata(
concept_graph: Dict[str, Any],
quality_report: Dict[str, Any],
pedagogy_graph: Optional[Dict[str, Any]] = None,
+ chunks: Optional[List[Dict[str, Any]]] = None,
):
# Strict-mode gate: refuse to write an artifact whose quality report
# shows integrity violations. Disabled by default for v0.1.x; flipped
@@ -1795,8 +2024,8 @@ def _write(path: Path, data: Dict[str, Any]):
_write(self.graph_dir / "pedagogy_graph.json", pedagogy_graph)
_write(self.quality_dir / "quality_report.json", quality_report)
- # Pedagogy model
- pedagogy = self._build_pedagogy_summary()
+ # Pedagogy model (full: module sequence, bloom progression, prereq chain)
+ pedagogy = self._build_pedagogy_summary(chunks=chunks)
_write(self.pedagogy_dir / "pedagogy_model.json", pedagogy)
# Training specs
@@ -1884,7 +2113,7 @@ def build_parser() -> argparse.ArgumentParser:
description="Process a Courseforge IMSCC into a Trainforge RAG corpus",
)
p.add_argument("--imscc", required=True, help="Path to .imscc file")
- p.add_argument("--course-code", required=True, help="Course code (e.g. DIGPED_101)")
+ p.add_argument("--course-code", required=True, help="Course code (e.g. SAMPLE_101)")
p.add_argument("--output", required=True, help="Output directory")
p.add_argument("--objectives", help="Path to objectives JSON (optional)")
p.add_argument("--division", default="STEM", choices=["STEM", "ARTS"], help="Division")
diff --git a/Trainforge/tests/fixtures/mini_course_defective/source_html/week_01_overview.html b/Trainforge/tests/fixtures/mini_course_defective/source_html/week_01_overview.html
index 728dcc437..aa8aa24c4 100644
--- a/Trainforge/tests/fixtures/mini_course_defective/source_html/week_01_overview.html
+++ b/Trainforge/tests/fixtures/mini_course_defective/source_html/week_01_overview.html
@@ -17,7 +17,7 @@
- Week 1 Overview: Foundations of Digital Pedagogy
+ Week 1 Overview: Mini Fixture Course
Introduction
diff --git a/Trainforge/tests/test_generator_defects.py b/Trainforge/tests/test_generator_defects.py
index 3ab24143f..031a14716 100644
--- a/Trainforge/tests/test_generator_defects.py
+++ b/Trainforge/tests/test_generator_defects.py
@@ -468,6 +468,340 @@ def test_no_report_when_below_threshold(self):
# Shared fixture sanity checks
# ---------------------------------------------------------------------------
+class _NullCapture:
+ """Stub DecisionCapture — records calls so tests can assert on them."""
+
+ def __init__(self):
+ self.calls = []
+
+ def log_decision(self, **kwargs):
+ self.calls.append(kwargs)
+
+
+def _bare_processor():
+ """A CourseProcessor skipped through __init__, hydrated with only the
+ attributes the methods-under-test read. Keeps these unit tests free of
+ IMSCC fixtures, filesystem, and DecisionCapture side-effects.
+ """
+ from collections import defaultdict
+ from Trainforge.process_course import CourseProcessor
+
+ proc = CourseProcessor.__new__(CourseProcessor)
+ proc.course_code = "MINI_101"
+ proc.capture = _NullCapture()
+ proc.stats = {
+ "total_words": 0,
+ "total_tokens_estimate": 0,
+ "chunk_types": defaultdict(int),
+ "difficulty_distribution": defaultdict(int),
+ }
+ proc._all_concept_tags = set()
+ proc.domain_concept_seeds = []
+ proc.objectives = None
+ proc._valid_outcome_ids = set()
+ proc._boilerplate_spans = []
+ proc._factual_flags = []
+ proc.MIN_CHUNK_SIZE = 50
+ proc.MAX_CHUNK_SIZE = 800
+ return proc
+
+
+# ---------------------------------------------------------------------------
+# Bloom-level fallback (every chunk gets a level)
+# ---------------------------------------------------------------------------
+
+class TestBloomLevelFallback:
+ def _item(self, **kw):
+ base = {
+ "module_id": "m1",
+ "module_title": "Module 1",
+ "item_id": "l1",
+ "title": "Lesson 1",
+ "resource_type": "page",
+ "key_concepts": [],
+ "learning_objectives": [],
+ }
+ base.update(kw)
+ return base
+
+ def test_verb_heuristic_fallback(self):
+ proc = _bare_processor()
+ text = "Evaluate the design, critique the rationale, and justify your reasoning."
+ chunk = proc._create_chunk(
+ chunk_id="c1", text=text, html="" + text + "
",
+ item=self._item(), section_heading="H", chunk_type="explanation",
+ )
+ assert chunk["bloom_level"] == "evaluate"
+ assert chunk["bloom_level_source"] == "verbs"
+
+ def test_default_when_no_signal(self):
+ proc = _bare_processor()
+ chunk = proc._create_chunk(
+ chunk_id="c1", text="A quiet paragraph with no taxonomy verbs.",
+ html="A quiet paragraph with no taxonomy verbs.
",
+ item=self._item(), section_heading="H", chunk_type="explanation",
+ )
+ assert chunk["bloom_level"] == "understand"
+ assert chunk["bloom_level_source"] == "default"
+
+ def test_authoritative_source_does_not_get_source_tag(self):
+ """When JSON-LD supplies a level, schema stays back-compat
+ (no bloom_level_source field)."""
+ proc = _bare_processor()
+ item = self._item(
+ courseforge_metadata={
+ "learningObjectives": [{"id": "co-01", "bloomLevel": "analyze"}],
+ "sections": [],
+ }
+ )
+ chunk = proc._create_chunk(
+ chunk_id="c1", text="Plain text.", html="Plain text.
",
+ item=item, section_heading="Missing", chunk_type="explanation",
+ )
+ assert chunk["bloom_level"] == "analyze"
+ assert "bloom_level_source" not in chunk
+
+
+# ---------------------------------------------------------------------------
+# Concept tag pollution filter (Bloom verbs never leak through)
+# ---------------------------------------------------------------------------
+
+class TestConceptTagPollutionFilter:
+ def test_bloom_verbs_dropped_from_key_concepts(self):
+ proc = _bare_processor()
+ item = {"key_concepts": ["apply", "aria-labelledby", "create", "landmark"]}
+ tags = proc._extract_concept_tags("Plain body text.", item)
+ assert "aria-labelledby" in tags
+ assert "landmark" in tags
+ assert "apply" not in tags
+ assert "create" not in tags
+
+ def test_objective_codes_dropped(self):
+ proc = _bare_processor()
+ item = {"key_concepts": ["co-01", "to-03", "w04-co-02", "accessibility"]}
+ tags = proc._extract_concept_tags("Sample.", item)
+ assert tags == ["accessibility"]
+
+ def test_logistics_scaffolding_dropped(self):
+ proc = _bare_processor()
+ item = {"key_concepts": ["initial-post", "replies", "due", "skip-link"]}
+ tags = proc._extract_concept_tags("Sample.", item)
+ assert tags == ["skip-link"]
+
+
+# ---------------------------------------------------------------------------
+# Domain concept seeds (text-based extraction of per-course vocabulary)
+# ---------------------------------------------------------------------------
+
+class TestDomainConceptSeeds:
+ def test_compile_builds_word_boundary_patterns(self):
+ from Trainforge.process_course import compile_domain_concept_seeds
+
+ seeds = compile_domain_concept_seeds([
+ {"id": "pour", "aliases": ["POUR", "perceivable operable"]},
+ ])
+ assert len(seeds) == 1
+ tag, patterns = seeds[0]
+ assert tag == "pour"
+ assert any(p.search("POUR principles apply everywhere") for p in patterns)
+ # Must not match substring inside longer word.
+ assert not any(p.search("downpour") for p in patterns)
+
+ def test_seed_matched_in_text(self):
+ from Trainforge.process_course import compile_domain_concept_seeds
+
+ proc = _bare_processor()
+ proc.domain_concept_seeds = compile_domain_concept_seeds([
+ {"id": "aria", "aliases": ["ARIA", "WAI-ARIA"]},
+ {"id": "pour", "aliases": ["POUR"]},
+ ])
+ tags = proc._extract_concept_tags(
+ "ARIA roles complement the POUR principles.", {"key_concepts": []}
+ )
+ assert "aria" in tags
+ assert "pour" in tags
+
+ def test_seed_ignored_when_not_present(self):
+ from Trainforge.process_course import compile_domain_concept_seeds
+
+ proc = _bare_processor()
+ proc.domain_concept_seeds = compile_domain_concept_seeds([
+ {"id": "aria", "aliases": ["ARIA"]},
+ ])
+ tags = proc._extract_concept_tags("No special vocabulary here.", {"key_concepts": []})
+ assert "aria" not in tags
+
+
+# ---------------------------------------------------------------------------
+# JSON-LD keyTerms merged into concept_tags
+# ---------------------------------------------------------------------------
+
+class TestKeyTermsMergedIntoConceptTags:
+ def test_key_terms_surface_as_tags(self):
+ proc = _bare_processor()
+ item = {
+ "module_id": "m1", "module_title": "Module 1",
+ "item_id": "l1", "title": "Lesson 1", "resource_type": "page",
+ "key_concepts": [], "learning_objectives": [],
+ "courseforge_metadata": {
+ "sections": [{
+ "heading": "Focus Management",
+ "contentType": "explanation",
+ "bloomRange": ["apply"],
+ "keyTerms": [
+ {"term": "Focus Indicator", "definition": "Visible outline."},
+ {"term": "Skip Link", "definition": "Bypass to main."},
+ ],
+ }],
+ "learningObjectives": [],
+ },
+ }
+ chunk = proc._create_chunk(
+ chunk_id="c1",
+ text="Content about focus management.",
+ html="Content about focus management.
",
+ item=item, section_heading="Focus Management", chunk_type="explanation",
+ )
+ assert "focus-indicator" in chunk["concept_tags"]
+ assert "skip-link" in chunk["concept_tags"]
+
+ def test_key_terms_still_filtered_against_non_concepts(self):
+ proc = _bare_processor()
+ item = {
+ "module_id": "m1", "module_title": "Module 1",
+ "item_id": "l1", "title": "Lesson 1", "resource_type": "page",
+ "key_concepts": [], "learning_objectives": [],
+ "courseforge_metadata": {
+ "sections": [{
+ "heading": "Intro",
+ "contentType": "explanation",
+ "bloomRange": ["apply"],
+ "keyTerms": [
+ {"term": "Apply"}, # Bloom verb
+ {"term": "ARIA role"},
+ ],
+ }],
+ "learningObjectives": [],
+ },
+ }
+ chunk = proc._create_chunk(
+ chunk_id="c1", text="Body.", html="Body.
",
+ item=item, section_heading="Intro", chunk_type="explanation",
+ )
+ assert "aria-role" in chunk["concept_tags"]
+ assert "apply" not in chunk["concept_tags"]
+
+
+# ---------------------------------------------------------------------------
+# Uncovered outcomes surfaced in quality_report
+# ---------------------------------------------------------------------------
+
+class TestUncoveredOutcomesInQualityReport:
+ def test_uncovered_ids_listed_and_issue_emitted(self):
+ proc = _bare_processor()
+ proc.stats["total_words"] = 300
+ proc._valid_outcome_ids = {"co-01", "co-02", "co-03", "co-04"}
+ chunks = [
+ _chunk(id="c1", word_count=120, html="a
",
+ learning_outcome_refs=["co-01"], concept_tags=["aria", "pour"]),
+ _chunk(id="c2", word_count=120, html="b
",
+ learning_outcome_refs=["co-02"], concept_tags=["landmark", "aria"]),
+ ]
+ report = proc._generate_quality_report(chunks)
+ assert report["integrity"]["uncovered_outcomes"] == ["co-03", "co-04"]
+ assert report["metrics"]["outcome_reverse_coverage"] == 0.5
+ assert any("have zero resolving chunks" in i for i in report["validation"]["issues"])
+
+ def test_full_reverse_coverage_passes(self):
+ proc = _bare_processor()
+ proc.stats["total_words"] = 120
+ proc._valid_outcome_ids = {"co-01"}
+ chunks = [
+ _chunk(id="c1", word_count=120, html="a
",
+ learning_outcome_refs=["co-01"], concept_tags=["aria", "pour"]),
+ ]
+ report = proc._generate_quality_report(chunks)
+ assert report["integrity"]["uncovered_outcomes"] == []
+ assert report["metrics"]["outcome_reverse_coverage"] == 1.0
+
+
+# ---------------------------------------------------------------------------
+# Pedagogy model (module sequence, Bloom progression, prereq chain)
+# ---------------------------------------------------------------------------
+
+class TestPedagogyModelRichness:
+ def _mk(self, chunk_id, module_id, module_title, bloom, tags, prereqs,
+ los=(), position=0):
+ return _chunk(
+ id=chunk_id, bloom_level=bloom, concept_tags=list(tags),
+ prereq_concepts=list(prereqs), learning_outcome_refs=list(los),
+ source={
+ "course_id": "MINI_101",
+ "module_id": module_id,
+ "module_title": module_title,
+ "lesson_id": module_id,
+ "lesson_title": module_title,
+ "resource_type": "page",
+ "section_heading": "H",
+ "position_in_module": position,
+ },
+ )
+
+ def test_thin_summary_when_no_chunks(self):
+ proc = _bare_processor()
+ summary = proc._build_pedagogy_summary()
+ assert summary["instructional_approach"] == "competency-based"
+ assert "module_sequence" not in summary
+
+ def test_module_sequence_ordered_by_week(self):
+ proc = _bare_processor()
+ chunks = [
+ self._mk("a", "week_02_foo", "Week 2", "apply", ["aria"], []),
+ self._mk("b", "week_01_foo", "Week 1", "understand", ["pour"], []),
+ self._mk("c", "week_03_foo", "Week 3", "evaluate", ["landmark"], []),
+ ]
+ summary = proc._build_pedagogy_summary(chunks)
+ weeks = [m["week_num"] for m in summary["module_sequence"]]
+ assert weeks == [1, 2, 3]
+
+ def test_bloom_progression_counts_per_module(self):
+ proc = _bare_processor()
+ chunks = [
+ self._mk("a", "week_01_foo", "Week 1", "understand", ["pour"], []),
+ self._mk("b", "week_01_foo", "Week 1", "apply", ["pour"], []),
+ self._mk("c", "week_02_foo", "Week 2", "evaluate", ["aria"], []),
+ ]
+ summary = proc._build_pedagogy_summary(chunks)
+ assert summary["bloom_progression"]["week_01_foo"]["understand"] == 1
+ assert summary["bloom_progression"]["week_01_foo"]["apply"] == 1
+ assert summary["bloom_progression"]["week_02_foo"]["evaluate"] == 1
+
+ def test_prerequisite_chain_valid_order(self):
+ proc = _bare_processor()
+ chunks = [
+ # Week 1 defines 'pour'
+ self._mk("a", "week_01_foo", "Week 1", "understand", ["pour"], []),
+ # Week 2 uses 'pour' as prereq → valid chain
+ self._mk("b", "week_02_foo", "Week 2", "apply", ["aria"], ["pour"]),
+ ]
+ summary = proc._build_pedagogy_summary(chunks)
+ chain_concepts = {e["concept"] for e in summary["prerequisite_chain"]}
+ assert "pour" in chain_concepts
+ assert summary["prerequisite_violations"] == []
+
+ def test_prerequisite_violation_detected(self):
+ proc = _bare_processor()
+ chunks = [
+ # Week 1 uses 'aria' as prereq BEFORE it is defined anywhere visible
+ self._mk("a", "week_01_foo", "Week 1", "understand", [], ["aria"]),
+ # Week 2 finally defines 'aria'
+ self._mk("b", "week_02_foo", "Week 2", "apply", ["aria"], []),
+ ]
+ summary = proc._build_pedagogy_summary(chunks)
+ violations = {v["concept"] for v in summary["prerequisite_violations"]}
+ assert "aria" in violations
+
+
class TestFixtures:
def test_clean_fixture_present(self):
assert (CLEAN_DIR / "course_objectives.json").exists()
diff --git a/Trainforge/tests/test_metadata_extraction.py b/Trainforge/tests/test_metadata_extraction.py
index 462e85f57..3cef6b413 100644
--- a/Trainforge/tests/test_metadata_extraction.py
+++ b/Trainforge/tests/test_metadata_extraction.py
@@ -24,12 +24,12 @@
- Week 2: Constructivism — DIGPED_101
+ Week 2: Constructivism — SAMPLE_101