Revert "Add unit tests and knowledge files for BC domain context"

This reverts commit 7fbb121c24.
This commit is contained in:
Jesper Schulz-Wedde 2026-04-22 14:03:09 +02:00
parent 7fbb121c24
commit 9dad34f48a
18 changed files with 1 additions and 1170 deletions

View file

@ -1,373 +0,0 @@
#!/usr/bin/env python3
"""
Reference implementation of the bc-domain-context action skill.
Executes the skill's Source -> Relevance -> Worklist -> Action pipeline against
a BCQuality repository root and returns a findings-report dict conforming to
the DO output contract.
This module is the *spec by example* for consumers that reimplement the skill
in other languages (for example, the Triage agent's Node.js client). Tests in
.github/scripts/tests/test_bc_domain_context.py exercise this implementation
against fixture knowledge trees.
Usage (from Python):
from bc_domain_context import run_bc_domain_context
report = run_bc_domain_context(repo_root, task_context)
Skill behaviour is defined in microsoft/skills/bc-domain-context.md; READ's
frontmatter-matching semantics live in skills/read.md.
"""
from __future__ import annotations
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable
try:
import yaml
except ImportError:
sys.stderr.write("ERROR: PyYAML is required. Install with: pip install pyyaml\n")
sys.exit(2)
LAYERS = ("custom", "microsoft", "community") # highest precedence first
SKILL_ID = "bc-domain-context"
SKILL_VERSION = 1
@dataclass
class KnowledgeFile:
path: str # repo-relative, forward slashes
layer: str # "microsoft" | "community" | "custom"
domain: str # folder name
slug: str # filename stem
frontmatter: dict[str, Any]
title: str # first H1 in body, or slug if absent
description: str # prose after "## Description", trimmed
# --- Frontmatter + body parsing ---------------------------------------------
def _parse_markdown(text: str) -> tuple[dict[str, Any] | None, str]:
lines = text.splitlines()
if not lines or lines[0].rstrip() != "---":
return None, text
for i in range(1, len(lines)):
if lines[i].rstrip() == "---":
try:
fm = yaml.safe_load("\n".join(lines[1:i])) or {}
except yaml.YAMLError:
return None, text
if not isinstance(fm, dict):
return None, text
return fm, "\n".join(lines[i + 1:])
return None, text
def _extract_title(body: str, fallback: str) -> str:
for line in body.splitlines():
m = re.match(r"^#\s+(.+?)\s*$", line)
if m:
return m.group(1).strip()
return fallback
def _extract_description(body: str) -> str:
lines = body.splitlines()
for i, line in enumerate(lines):
if re.match(r"^##\s+Description\s*$", line):
buf: list[str] = []
for j in range(i + 1, len(lines)):
if re.match(r"^##\s+", lines[j]):
break
buf.append(lines[j])
return "\n".join(buf).strip()
return ""
# --- bc-version expansion (matches validate_frontmatter.expand_bc_version) --
_RANGE = re.compile(r"^(\d+)\.\.(\d+)$")
def _expand_bc_version(value: Any) -> list[int] | None:
if not isinstance(value, list) or not value:
return None
if all(isinstance(v, int) and not isinstance(v, bool) and v > 0 for v in value):
return sorted(set(value))
if len(value) == 1 and isinstance(value[0], str):
m = _RANGE.match(value[0].strip())
if m:
start, end = int(m.group(1)), int(m.group(2))
if start <= end:
return list(range(start, end + 1))
return None
# --- READ's frontmatter matching rules --------------------------------------
def _matches(
kf: KnowledgeFile, task_context: dict[str, Any]
) -> tuple[bool, list[str]]:
"""Return (applicable, unknown_dimensions) per READ's semantics.
A file is applicable when every rule matches. A rule is "unknown" when the
task context omits a dimension and the file does not declare a universal
sentinel for it; unknown rules do not disqualify the file but force a
medium-confidence ceiling on derived findings.
"""
unknown: list[str] = []
fm = kf.frontmatter
# bc-version
file_versions = _expand_bc_version(fm.get("bc-version"))
if file_versions is None:
return False, unknown
task_version = task_context.get("bc-version")
if task_version is None:
unknown.append("bc-version")
elif task_version not in file_versions:
return False, unknown
# technologies (no sentinel)
file_techs = set(fm.get("technologies") or [])
task_techs = set(task_context.get("technologies") or [])
if not task_techs:
unknown.append("technologies")
elif not (file_techs & task_techs):
return False, unknown
# countries (sentinel: w1)
file_countries = set(fm.get("countries") or [])
task_countries = set(task_context.get("countries") or [])
if "w1" in file_countries:
pass
elif not task_countries:
unknown.append("countries")
elif not (file_countries & task_countries):
return False, unknown
# application-area (sentinel: all)
file_areas = set(fm.get("application-area") or [])
task_areas = set(task_context.get("application-area") or [])
if "all" in file_areas:
pass
elif not task_areas or "all" in task_areas:
# Empty or [all] in task means "any area"; file-level specific area
# still matches against any-area when sourced from the area folder.
pass
elif not (file_areas & task_areas):
return False, unknown
return True, unknown
# --- Knowledge corpus load --------------------------------------------------
def _load_knowledge_file(root: Path, path: Path) -> KnowledgeFile | None:
rel = path.relative_to(root).as_posix()
parts = rel.split("/")
if len(parts) != 4 or parts[1] != "knowledge":
return None
layer, _, domain, filename = parts
slug = Path(filename).stem
try:
text = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
return None
fm, body = _parse_markdown(text)
if fm is None:
return None
return KnowledgeFile(
path=rel,
layer=layer,
domain=domain,
slug=slug,
frontmatter=fm,
title=_extract_title(body, slug),
description=_extract_description(body),
)
def _load_corpus(
root: Path, enabled_layers: Iterable[str], areas: Iterable[str] | None
) -> list[KnowledgeFile]:
"""Walk <layer>/knowledge/<area>/*.md for enabled layers.
If `areas` is None or contains "all", walk every area folder. Otherwise,
walk only the named area folders.
"""
enabled = set(enabled_layers) & set(LAYERS)
area_set = None
if areas is not None:
areas_list = list(areas)
if areas_list and "all" not in areas_list:
area_set = set(areas_list)
corpus: list[KnowledgeFile] = []
for layer in enabled:
knowledge_root = root / layer / "knowledge"
if not knowledge_root.is_dir():
continue
for domain_dir in knowledge_root.iterdir():
if not domain_dir.is_dir():
continue
if area_set is not None and domain_dir.name not in area_set:
continue
for md_path in domain_dir.glob("*.md"):
kf = _load_knowledge_file(root, md_path)
if kf is not None:
corpus.append(kf)
return corpus
# --- Layer precedence -------------------------------------------------------
def _resolve_precedence(
files: list[KnowledgeFile],
) -> tuple[list[KnowledgeFile], list[dict[str, Any]]]:
"""Keep the highest-precedence file per (domain, slug). Return (kept, suppressed).
suppressed entries have shape { "reference": { "path": ... }, "reason": "layer-precedence" }.
"""
precedence = {layer: i for i, layer in enumerate(LAYERS)} # lower index = higher precedence
groups: dict[tuple[str, str], list[KnowledgeFile]] = {}
for f in files:
groups.setdefault((f.domain, f.slug), []).append(f)
kept: list[KnowledgeFile] = []
suppressed: list[dict[str, Any]] = []
for group in groups.values():
group.sort(key=lambda f: precedence.get(f.layer, 99))
kept.append(group[0])
for loser in group[1:]:
suppressed.append({
"reference": {"path": loser.path},
"reason": "layer-precedence",
})
return kept, suppressed
# --- Worklist narrowing -----------------------------------------------------
_GOAL_PREFIX = re.compile(r"^bc-domain-context(?:\s+for\s+[a-z0-9,\- ]+)?\s*", re.IGNORECASE)
_TOKEN = re.compile(r"[a-z0-9]+")
def _extract_goal_tokens(goal: str) -> list[str]:
"""Strip the 'bc-domain-context for <area>' prefix and return significant tokens."""
if not goal:
return []
stripped = _GOAL_PREFIX.sub("", goal).strip()
if not stripped:
return []
return [t for t in _TOKEN.findall(stripped.lower()) if len(t) >= 3]
def _score(kf: KnowledgeFile, tokens: list[str]) -> int:
if not tokens:
return 0
kw = {k.lower() for k in (kf.frontmatter.get("keywords") or [])}
text = f"{kf.slug} {kf.title} {kf.description}".lower()
score = 0
for t in tokens:
if t in kw:
score += 3
if t in text:
score += 1
return score
def _narrow(
kept: list[KnowledgeFile], task_context: dict[str, Any], max_top: int = 15
) -> list[KnowledgeFile]:
tokens = _extract_goal_tokens(task_context.get("goal") or "")
if not tokens:
return kept
scored = [(kf, _score(kf, tokens)) for kf in kept]
scored.sort(key=lambda p: (-p[1], p[0].path))
# Keep only files with positive score; fall back to full set if none score.
positive = [kf for kf, s in scored if s > 0]
if not positive:
return kept
return positive[:max_top]
# --- Message construction ---------------------------------------------------
def _message(kf: KnowledgeFile, unknown: list[str]) -> str:
first_sentences = re.split(r"(?<=[.!?])\s+", kf.description, maxsplit=3)
lead = " ".join(first_sentences[:3]).strip() if first_sentences else ""
msg = f"{kf.title}. {lead}" if lead else kf.title
if unknown:
msg += f" (conditional on: {', '.join(sorted(unknown))})"
return msg
# --- Main entrypoint --------------------------------------------------------
def run_bc_domain_context(
root: Path | str, task_context: dict[str, Any]
) -> dict[str, Any]:
"""Execute the skill against the repository at `root`.
Returns a findings-report dict conforming to the DO output contract.
"""
root = Path(root)
enabled = task_context.get("enabled-layers") or list(LAYERS)
areas = task_context.get("application-area")
corpus = _load_corpus(root, enabled, areas)
if not corpus:
return _report("no-knowledge", [], [])
applicable: list[tuple[KnowledgeFile, list[str]]] = []
for kf in corpus:
ok, unknown = _matches(kf, task_context)
if ok:
applicable.append((kf, unknown))
if not applicable:
return _report("not-applicable", [], [])
kept, suppressed = _resolve_precedence([kf for kf, _ in applicable])
kept_set = {kf.path for kf in kept}
unknown_by_path = {kf.path: u for kf, u in applicable if kf.path in kept_set}
worklist = _narrow(kept, task_context)
findings: list[dict[str, Any]] = []
for kf in worklist:
unknown = unknown_by_path.get(kf.path, [])
findings.append({
"id": kf.path,
"severity": "info",
"message": _message(kf, unknown),
"references": [{"path": kf.path}],
"confidence": "medium" if unknown else "high",
})
return _report("completed", findings, suppressed)
def _report(
outcome: str, findings: list[dict[str, Any]], suppressed: list[dict[str, Any]]
) -> dict[str, Any]:
return {
"skill": {"id": SKILL_ID, "version": SKILL_VERSION},
"outcome": outcome,
"summary": {
"counts": {
"blocker": 0,
"major": 0,
"minor": 0,
"info": len(findings),
},
"coverage": {
"worklist-size": len(findings),
"items-evaluated": len(findings),
},
},
"findings": findings,
"suppressed": suppressed,
}

View file

@ -1,297 +0,0 @@
#!/usr/bin/env python3
"""
Tests for the bc-domain-context reference implementation.
Run with: python -m unittest .github.scripts.tests.test_bc_domain_context
Or: cd .github/scripts/tests && python -m unittest test_bc_domain_context
The tests build fixture knowledge trees in tempfile directories and exercise
the skill's Source / Relevance / Worklist / Action pipeline end to end.
"""
from __future__ import annotations
import os
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path
# Add the scripts directory (parent of tests/) to sys.path so we can import
# bc_domain_context.py.
_SCRIPTS_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_SCRIPTS_DIR))
from bc_domain_context import run_bc_domain_context # noqa: E402
def _write(
root: Path,
layer: str,
domain: str,
slug: str,
*,
bc_version: str = "[26..28]",
technologies: str = "[al]",
countries: str = "[w1]",
application_area: str | None = None,
keywords: str = "[sample]",
domain_value: str | None = None,
description: str = "Short description. Another sentence.",
title: str | None = None,
) -> Path:
"""Write a fixture knowledge file and return its path."""
if application_area is None:
application_area = f"[{domain}]"
if domain_value is None:
domain_value = domain
if title is None:
title = slug.replace("-", " ").capitalize()
target = root / layer / "knowledge" / domain / f"{slug}.md"
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(
textwrap.dedent(
f"""\
---
bc-version: {bc_version}
domain: {domain_value}
keywords: {keywords}
technologies: {technologies}
countries: {countries}
application-area: {application_area}
---
# {title}
## Description
{description}
"""
),
encoding="utf-8",
)
return target
class BcDomainContextTests(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.root = Path(self._tmp.name)
self.addCleanup(self._tmp.cleanup)
# --- Core filtering ---
def test_filter_by_area_returns_only_requested_area(self):
_write(self.root, "microsoft", "finance", "chart-of-accounts")
_write(self.root, "microsoft", "finance", "general-ledger-entries")
_write(self.root, "microsoft", "finance", "dimensions")
_write(self.root, "microsoft", "sales", "order-to-cash")
_write(self.root, "microsoft", "sales", "pricing")
report = run_bc_domain_context(self.root, {
"application-area": ["finance"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
self.assertEqual(report["outcome"], "completed")
paths = sorted(f["id"] for f in report["findings"])
self.assertEqual(paths, [
"microsoft/knowledge/finance/chart-of-accounts.md",
"microsoft/knowledge/finance/dimensions.md",
"microsoft/knowledge/finance/general-ledger-entries.md",
])
for finding in report["findings"]:
self.assertEqual(finding["severity"], "info")
self.assertEqual(finding["confidence"], "high")
def test_application_area_all_returns_union(self):
_write(self.root, "microsoft", "finance", "chart-of-accounts")
_write(self.root, "microsoft", "sales", "order-to-cash")
_write(self.root, "microsoft", "manufacturing", "bom")
report = run_bc_domain_context(self.root, {
"application-area": ["all"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
self.assertEqual(report["outcome"], "completed")
self.assertEqual(len(report["findings"]), 3)
paths = {f["id"] for f in report["findings"]}
self.assertIn("microsoft/knowledge/finance/chart-of-accounts.md", paths)
self.assertIn("microsoft/knowledge/sales/order-to-cash.md", paths)
self.assertIn("microsoft/knowledge/manufacturing/bom.md", paths)
def test_technologies_mismatch_drops_file(self):
_write(self.root, "microsoft", "finance", "al-concept")
_write(
self.root,
"microsoft",
"finance",
"kql-only-concept",
technologies="[kql]",
)
report = run_bc_domain_context(self.root, {
"application-area": ["finance"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
paths = {f["id"] for f in report["findings"]}
self.assertIn("microsoft/knowledge/finance/al-concept.md", paths)
self.assertNotIn("microsoft/knowledge/finance/kql-only-concept.md", paths)
def test_no_matching_knowledge_returns_not_applicable(self):
_write(self.root, "microsoft", "finance", "chart-of-accounts")
report = run_bc_domain_context(self.root, {
"application-area": ["manufacturing"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
self.assertEqual(report["outcome"], "no-knowledge")
self.assertEqual(report["findings"], [])
# --- Layer precedence ---
def test_layer_precedence_microsoft_wins_over_community(self):
_write(self.root, "community", "finance", "vat-on-prepayment")
_write(self.root, "microsoft", "finance", "vat-on-prepayment")
report = run_bc_domain_context(self.root, {
"application-area": ["finance"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
paths = [f["id"] for f in report["findings"]]
self.assertEqual(
paths, ["microsoft/knowledge/finance/vat-on-prepayment.md"]
)
suppressed_paths = [s["reference"]["path"] for s in report["suppressed"]]
self.assertEqual(
suppressed_paths,
["community/knowledge/finance/vat-on-prepayment.md"],
)
self.assertEqual(report["suppressed"][0]["reason"], "layer-precedence")
def test_layer_precedence_custom_wins_over_microsoft(self):
_write(self.root, "microsoft", "finance", "vat-on-prepayment")
_write(self.root, "custom", "finance", "vat-on-prepayment")
report = run_bc_domain_context(self.root, {
"application-area": ["finance"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
paths = [f["id"] for f in report["findings"]]
self.assertEqual(
paths, ["custom/knowledge/finance/vat-on-prepayment.md"]
)
# --- Conditional applicability (unknown dimensions) ---
def test_unknown_bc_version_caps_confidence_at_medium(self):
_write(self.root, "microsoft", "finance", "chart-of-accounts")
_write(self.root, "microsoft", "finance", "dimensions")
# bc-version omitted from task-context.
report = run_bc_domain_context(self.root, {
"application-area": ["finance"],
"technologies": ["al"],
"countries": ["w1"],
})
self.assertTrue(report["findings"])
for finding in report["findings"]:
self.assertEqual(finding["confidence"], "medium")
self.assertIn("bc-version", finding["message"])
# --- Goal-directed narrowing ---
def test_goal_tokens_narrow_worklist(self):
_write(
self.root,
"microsoft",
"finance",
"vat-on-prepayment-chains",
keywords="[vat, prepayment, credit-memo]",
)
_write(
self.root,
"microsoft",
"finance",
"dimensions",
keywords="[dimensions, default-priority]",
)
_write(
self.root,
"microsoft",
"finance",
"chart-of-accounts",
keywords="[chart, account]",
)
report = run_bc_domain_context(self.root, {
"goal": "bc-domain-context for finance — VAT wrong on prepayment credit memo",
"application-area": ["finance"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
paths = [f["id"] for f in report["findings"]]
self.assertIn(
"microsoft/knowledge/finance/vat-on-prepayment-chains.md",
paths,
)
# The highest-scoring file should come first.
self.assertEqual(
paths[0],
"microsoft/knowledge/finance/vat-on-prepayment-chains.md",
)
def test_generic_goal_keeps_full_area(self):
_write(self.root, "microsoft", "finance", "chart-of-accounts")
_write(self.root, "microsoft", "finance", "dimensions")
report = run_bc_domain_context(self.root, {
"goal": "bc-domain-context for finance",
"application-area": ["finance"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
})
self.assertEqual(len(report["findings"]), 2)
# --- Layer disabling ---
def test_disabled_layer_is_invisible(self):
_write(self.root, "community", "finance", "only-in-community")
_write(self.root, "microsoft", "finance", "only-in-microsoft")
report = run_bc_domain_context(self.root, {
"application-area": ["finance"],
"technologies": ["al"],
"bc-version": 28,
"countries": ["w1"],
"enabled-layers": ["microsoft"],
})
paths = [f["id"] for f in report["findings"]]
self.assertEqual(paths, ["microsoft/knowledge/finance/only-in-microsoft.md"])
if __name__ == "__main__":
unittest.main()