W3.5 implementation: enable Codex/Claude to resolve changed-files in a PR diff to the spec module(s) and ADRs they belong to, before falling back to text retrieval. - platform/contracts/knowledge-module-map.schema.yaml: JSON Schema defining the new knowledge.module-map.yaml file business repos add at their root. Required fields: schema_version, service, modules. Each module entry has paths (fnmatch globs), excluded_paths, spec pointer, and adrs list. Optional fallback_adrs at the top level covers cross-cutting PRs. - scripts/knowledge/validate_module_map.py: validates the yaml against schema + checks every spec/adrs path exists in the working tree + warns on duplicate path patterns + warns on dead module entries. CLI takes --module-map / --repo-root / --deploy-root and exits 1 on any error. - scripts/knowledge/context_pack.py: cap the Dify retrieve query at 240 chars (Dify rejects >250 with HTTP 400 invalid_param). Also switched from full paths to basenames in the query string for higher retrieval signal density. W7 baseline on 4 recent ljwx-platform PRs after this change: - PR #830 (12 care-elder files): 15 sections, 3 SSOT chunks, 2 from module-map (care-elder/module.yml + ADR-008), 13 from Dify - 2/4 PRs returned >=1 SSOT chunk; the 0-SSOT PRs were Dependabot / CI script changes (correctly no spec to retrieve) - 0 Cloudflare 1010 errors, 0 query-too-long errors Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
212 lines
7.0 KiB
Python
212 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate knowledge.module-map.yaml against schema + working tree.
|
|
|
|
Two layers:
|
|
1. JSON Schema validation against
|
|
platform/contracts/knowledge-module-map.schema.yaml.
|
|
2. Working-tree checks:
|
|
- every `spec` path exists on disk
|
|
- every `adrs` glob expands to at least one file
|
|
- no two modules have identical `paths` (would silently produce
|
|
tied resolution warnings forever)
|
|
- `fallback_adrs` files exist
|
|
|
|
Exits non-zero on any validation error; prints warnings (non-fatal) on
|
|
glob-overlap with sibling modules so authors can audit ambiguous
|
|
classifications.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fnmatch
|
|
import glob as glob_lib
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from .common import (
|
|
KnowledgeContractError,
|
|
load_validator,
|
|
load_yaml_mapping,
|
|
validate_payload,
|
|
)
|
|
|
|
|
|
SCHEMA_RELATIVE = Path("platform/contracts/knowledge-module-map.schema.yaml")
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Validate knowledge.module-map.yaml")
|
|
parser.add_argument(
|
|
"--module-map", type=Path, required=True, help="path to knowledge.module-map.yaml"
|
|
)
|
|
parser.add_argument(
|
|
"--repo-root",
|
|
type=Path,
|
|
required=True,
|
|
help="business repo root (paths in the map are relative to this)",
|
|
)
|
|
parser.add_argument(
|
|
"--deploy-root",
|
|
type=Path,
|
|
required=True,
|
|
help="ljwx-deploy root (for loading the schema)",
|
|
)
|
|
parser.add_argument(
|
|
"--strict-overlap",
|
|
action="store_true",
|
|
help="treat path-glob overlap between modules as error not warning",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def _expand_globs(repo_root: Path, patterns: list[str]) -> list[str]:
|
|
"""Expand a list of fnmatch patterns into actual relative paths."""
|
|
|
|
matches: set[str] = set()
|
|
for pattern in patterns:
|
|
# glob doesn't fully match fnmatch semantics for `**`; iterate the
|
|
# whole tree once and filter to mirror what export_artifact does.
|
|
for absolute in repo_root.glob(pattern):
|
|
if absolute.is_file():
|
|
matches.add(str(absolute.relative_to(repo_root)))
|
|
# fnmatch as fallback for patterns glob can't expand (e.g.
|
|
# patterns containing `[]` quirks)
|
|
if not matches:
|
|
for absolute in repo_root.rglob("*"):
|
|
if not absolute.is_file():
|
|
continue
|
|
rel = str(absolute.relative_to(repo_root))
|
|
if fnmatch.fnmatch(rel, pattern):
|
|
matches.add(rel)
|
|
return sorted(matches)
|
|
|
|
|
|
def _check_path_overlap(modules: dict) -> list[tuple[str, str, list[str]]]:
|
|
"""Return list of (module_a, module_b, shared_patterns) for any pair
|
|
whose `paths` lists share an identical pattern.
|
|
|
|
Detects only literal pattern equality, not semantic overlap (that's
|
|
what the fnmatch resolver handles at lookup time with longest-glob).
|
|
"""
|
|
|
|
overlaps: list[tuple[str, str, list[str]]] = []
|
|
items = list(modules.items())
|
|
for i, (name_a, entry_a) in enumerate(items):
|
|
paths_a = set(entry_a.get("paths", []) or [])
|
|
for name_b, entry_b in items[i + 1 :]:
|
|
paths_b = set(entry_b.get("paths", []) or [])
|
|
shared = sorted(paths_a & paths_b)
|
|
if shared:
|
|
overlaps.append((name_a, name_b, shared))
|
|
return overlaps
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
if not args.module_map.exists():
|
|
print(f"ERROR: module map not found: {args.module_map}", file=sys.stderr)
|
|
return 2
|
|
|
|
schema_path = args.deploy_root / SCHEMA_RELATIVE
|
|
if not schema_path.exists():
|
|
print(f"ERROR: schema not found: {schema_path}", file=sys.stderr)
|
|
return 2
|
|
|
|
validator = load_validator(schema_path)
|
|
payload = load_yaml_mapping(args.module_map)
|
|
try:
|
|
validate_payload(validator, payload, args.module_map.as_posix())
|
|
except KnowledgeContractError as exc:
|
|
print(f"SCHEMA ERROR: {exc}", file=sys.stderr)
|
|
return 3
|
|
|
|
repo_root = args.repo_root.resolve()
|
|
if not repo_root.exists():
|
|
print(f"ERROR: repo root not found: {repo_root}", file=sys.stderr)
|
|
return 2
|
|
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
|
|
modules_raw = payload.get("modules", {})
|
|
if not isinstance(modules_raw, dict):
|
|
print("ERROR: modules must be a mapping", file=sys.stderr)
|
|
return 3
|
|
modules: dict[str, dict] = {
|
|
str(k): v for k, v in modules_raw.items() if isinstance(v, dict)
|
|
}
|
|
|
|
# 1. Each module's spec file must exist
|
|
for name, entry in modules.items():
|
|
spec = entry.get("spec")
|
|
if isinstance(spec, str):
|
|
if not (repo_root / spec).exists():
|
|
errors.append(f"modules.{name}.spec: file not found: {spec}")
|
|
|
|
# 2. Each module's adrs glob must match at least one file
|
|
for name, entry in modules.items():
|
|
adrs = entry.get("adrs", []) or []
|
|
for adr_pattern in adrs:
|
|
if not isinstance(adr_pattern, str):
|
|
continue
|
|
hits = _expand_globs(repo_root, [adr_pattern])
|
|
if not hits:
|
|
warnings.append(
|
|
f"modules.{name}.adrs[{adr_pattern!r}]: glob matched 0 files"
|
|
)
|
|
|
|
# 3. fallback_adrs files must exist
|
|
for adr_pattern in payload.get("fallback_adrs", []) or []:
|
|
if not isinstance(adr_pattern, str):
|
|
continue
|
|
hits = _expand_globs(repo_root, [adr_pattern])
|
|
if not hits:
|
|
errors.append(f"fallback_adrs[{adr_pattern!r}]: glob matched 0 files")
|
|
|
|
# 4. Sibling modules sharing identical path patterns -> ambiguous
|
|
overlap_pairs = _check_path_overlap(modules)
|
|
for name_a, name_b, shared in overlap_pairs:
|
|
msg = (
|
|
f"modules.{name_a} and modules.{name_b} both declare paths: "
|
|
+ ", ".join(repr(p) for p in shared)
|
|
)
|
|
if args.strict_overlap:
|
|
errors.append(msg)
|
|
else:
|
|
warnings.append(msg)
|
|
|
|
# 5. Each module must classify at least one real file (catches dead modules)
|
|
for name, entry in modules.items():
|
|
paths = entry.get("paths", []) or []
|
|
excluded = entry.get("excluded_paths", []) or []
|
|
included = set(_expand_globs(repo_root, paths))
|
|
excluded_set = set(_expand_globs(repo_root, excluded))
|
|
net = included - excluded_set
|
|
if not net:
|
|
warnings.append(
|
|
f"modules.{name}: paths globs match 0 files in the repo "
|
|
f"(dead module entry?)"
|
|
)
|
|
|
|
for w in warnings:
|
|
print(f"WARN: {w}", file=sys.stderr)
|
|
for e in errors:
|
|
print(f"ERROR: {e}", file=sys.stderr)
|
|
|
|
if errors:
|
|
print(
|
|
f"validate_module_map: {len(errors)} error(s), {len(warnings)} warning(s)",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
print(
|
|
f"validate_module_map: OK ({len(modules)} modules, {len(warnings)} warning(s))"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|