feat(terminology): add fleet scan allowlist and prose gate (CUST-WP-0055 T08)
Some checks failed
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Has been cancelled

Extend scan_workstream_terminology.py with allowlist loading, --apply-allowlist,
and --check-prose-gate for regression detection. Commit the T08 exclusion
config and unit tests; mark T01/T03/T08 done and activate the fleet workplan.
This commit is contained in:
codex 2026-07-08 16:08:32 +02:00
parent 0f6b5cec52
commit 20173e4270
4 changed files with 246 additions and 12 deletions

View file

@ -0,0 +1,49 @@
from __future__ import annotations
from pathlib import Path
from tools.scan_workstream_terminology import (
DEFAULT_ALLOWLIST_PATH,
load_allowlist,
path_is_excluded,
prose_gate_violations,
)
def test_load_allowlist_reads_fleet_config() -> None:
config = load_allowlist(DEFAULT_ALLOWLIST_PATH)
assert config["version"] == 1
assert "the-custodian" in config["per_repo"]
def test_path_is_excluded_for_archived_workplans() -> None:
config = load_allowlist(DEFAULT_ALLOWLIST_PATH)
assert path_is_excluded("any-repo", "workplans/archived/foo.md", config)
assert not path_is_excluded("any-repo", "workplans/CUST-WP-0001.md", config)
def test_path_is_excluded_for_state_hub_compat_router() -> None:
config = load_allowlist(DEFAULT_ALLOWLIST_PATH)
assert path_is_excluded(
"state-hub",
"api/routers/workstreams.py",
config,
)
assert not path_is_excluded(
"state-hub",
"dashboard/src/index.md",
config,
)
def test_path_is_excluded_for_whole_repo() -> None:
config = load_allowlist(DEFAULT_ALLOWLIST_PATH)
assert path_is_excluded("agentic-resources", "README.md", config)
def test_prose_gate_violations_detect_user_facing_workstream() -> None:
root = Path(__file__).parent.parent
config = load_allowlist(DEFAULT_ALLOWLIST_PATH)
violations = prose_gate_violations(root, "the-custodian", config)
prose_paths = {path for path, _ in violations}
assert "canon/standards/workplan-terminology-fleet_v0.1.md" not in prose_paths

View file

@ -0,0 +1,60 @@
# Fleet scan exclusions for CUST-WP-0055 T08 prose gate.
# Paths are relative to each repo root. Globs use fnmatch (*, ?).
version: 1
global:
path_prefixes:
- workplans/archived/
- memory/
per_repo:
the-custodian:
path_prefixes:
- canon/standards/workplan-terminology-fleet_v0.1.md
- tools/scan_workstream_terminology.py
- tools/scan_workstream_allowlist.yaml
- docs/evidence/workstream-terminology-baseline-20260708.json
- workplans/CUST-WP-0055-workplan-terminology-fleet-refactor.md
- workplans/CUST-WP-0010-workstream-lifecycle-docs.md
- wiki/
- roadmap/
state-hub:
path_prefixes:
- docs/workplan-terminology-transition.md
- docs/workplan-terminology-legacy-retirement-backlog.md
- docs/nats-event-subjects.md
- api/routers/workstreams.py
- api/routers/workstream_dependencies.py
- api/routers/legacy_meter.py
- api/services/legacy_meter.py
- api/models/legacy_meter.py
- mcp_server/
- migrations/
- tests/test_legacy_meter.py
- tests/test_routers_core.py
- custodian_cli.py
- scripts/consistency_check.py
path_globs:
- workplans/STATE-WP-0054*
- workplans/STATE-WP-0069*
activity-core:
path_globs:
- event-types/org.workstream.completed.md
- event-types/org.statehub.workstream.completed.md
agentic-resources:
exclude_repo: true
# Wire-compat pattern buckets — always excluded from prose gate counts.
patterns_always_allowed:
- frontmatter:state_hub_workstream_id
- api:workstream_id
- mcp:create_workstream
- mcp:update_workstream
- mcp:list_workstreams
- api:open_workstreams
- path:workstreams/
- term:workstreams

View file

@ -13,6 +13,7 @@ terminology refactor planning (CUST-WP-0055).
from __future__ import annotations
import argparse
import fnmatch
import json
import re
import sys
@ -20,8 +21,13 @@ import urllib.request
from collections import defaultdict
from pathlib import Path
import yaml
DEFAULT_API_BASE = "http://127.0.0.1:8000"
HOME = Path("/home/worsch")
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_ALLOWLIST_PATH = REPO_ROOT / "tools" / "scan_workstream_allowlist.yaml"
PROSE_PATTERN = "workstream "
EXCLUDE_DIRS = {
".git",
"node_modules",
@ -84,6 +90,65 @@ PATTERN_LABELS = [
NAMED_FILES = {"Makefile", "Dockerfile", "AGENTS.md", "CLAUDE.md", "SCOPE.md", "INTENT.md", "README"}
def load_allowlist(path: Path = DEFAULT_ALLOWLIST_PATH) -> dict:
if not path.is_file():
return {"version": 0, "global": {}, "per_repo": {}, "patterns_always_allowed": []}
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
data.setdefault("global", {})
data.setdefault("per_repo", {})
data.setdefault("patterns_always_allowed", [])
return data
def _matches_any(rel_path: str, prefixes: list[str], globs: list[str]) -> bool:
for prefix in prefixes:
if rel_path == prefix or rel_path.startswith(prefix):
return True
for pattern in globs:
if fnmatch.fnmatch(rel_path, pattern):
return True
return False
def path_is_excluded(repo_slug: str, rel_path: str, allowlist: dict) -> bool:
repo_rules = allowlist.get("per_repo", {}).get(repo_slug, {})
if repo_rules.get("exclude_repo"):
return True
global_rules = allowlist.get("global", {})
if _matches_any(
rel_path,
global_rules.get("path_prefixes", []),
global_rules.get("path_globs", []),
):
return True
return _matches_any(
rel_path,
repo_rules.get("path_prefixes", []),
repo_rules.get("path_globs", []),
)
def prose_gate_violations(root: Path, repo_slug: str, allowlist: dict) -> list[tuple[str, int]]:
violations: list[tuple[str, int]] = []
for path in root.rglob("*"):
if not path.is_file() or any(part in EXCLUDE_DIRS for part in path.parts):
continue
if path.suffix.lower() not in TEXT_EXTENSIONS and path.name not in NAMED_FILES:
continue
rel = str(path.relative_to(root))
if path_is_excluded(repo_slug, rel, allowlist):
continue
try:
text = path.read_text(encoding="utf-8", errors="ignore")
except OSError:
continue
hits = len(re.findall(re.escape(PROSE_PATTERN), text, flags=re.I))
if hits:
violations.append((rel, hits))
violations.sort(key=lambda item: (-item[1], item[0]))
return violations
def bucket(path: str) -> str:
p = path.lower()
if p.startswith("workplans/"):
@ -110,17 +175,28 @@ def fetch_repos(api_base: str) -> list[dict]:
return json.load(resp)
def scan_repo(root: Path) -> dict:
def scan_repo(
root: Path,
*,
repo_slug: str | None = None,
allowlist: dict | None = None,
apply_allowlist: bool = False,
) -> dict:
patterns: dict[str, int] = defaultdict(int)
buckets: dict[str, int] = defaultdict(int)
total = 0
files = 0
excluded_files = 0
top_files: list[tuple[int, str]] = []
for path in root.rglob("*"):
if not path.is_file() or any(part in EXCLUDE_DIRS for part in path.parts):
continue
if path.suffix.lower() not in TEXT_EXTENSIONS and path.name not in NAMED_FILES:
continue
rel = str(path.relative_to(root))
if apply_allowlist and allowlist is not None and repo_slug and path_is_excluded(repo_slug, rel, allowlist):
excluded_files += 1
continue
try:
text = path.read_text(encoding="utf-8", errors="ignore")
except OSError:
@ -131,7 +207,6 @@ def scan_repo(root: Path) -> dict:
count = len(matches)
total += count
files += 1
rel = str(path.relative_to(root))
top_files.append((count, rel))
buckets[bucket(rel)] += count
low = text.lower()
@ -140,13 +215,16 @@ def scan_repo(root: Path) -> dict:
if hits:
patterns[label] += hits
top_files.sort(reverse=True)
return {
result = {
"occurrences": total,
"files": files,
"patterns": dict(patterns),
"buckets": dict(buckets),
"top_files": top_files[:10],
}
if apply_allowlist:
result["excluded_files"] = excluded_files
return result
def main() -> int:
@ -154,8 +232,24 @@ def main() -> int:
parser.add_argument("--api-base", default=DEFAULT_API_BASE)
parser.add_argument("--repo", help="Scan a single repo slug")
parser.add_argument("--json", action="store_true", help="Emit JSON report")
parser.add_argument(
"--apply-allowlist",
action="store_true",
help="Exclude allowlisted paths from counts (CUST-WP-0055 T08)",
)
parser.add_argument(
"--allowlist",
default=str(DEFAULT_ALLOWLIST_PATH),
help="Path to scan allowlist YAML",
)
parser.add_argument(
"--check-prose-gate",
action="store_true",
help="Exit 1 when prose 'workstream ' hits remain outside the allowlist",
)
args = parser.parse_args()
allowlist = load_allowlist(Path(args.allowlist))
repos = fetch_repos(args.api_base)
if args.repo:
repos = [r for r in repos if r["slug"] == args.repo]
@ -163,8 +257,13 @@ def main() -> int:
print(f"Repo '{args.repo}' not registered", file=sys.stderr)
return 1
report: dict = {"repos": {}, "totals": {"occurrences": 0, "files": 0, "repos_with_hits": 0}}
report: dict = {
"repos": {},
"totals": {"occurrences": 0, "files": 0, "repos_with_hits": 0},
"allowlist_applied": args.apply_allowlist or args.check_prose_gate,
}
pattern_totals: dict[str, int] = defaultdict(int)
prose_violations: dict[str, list[dict[str, int | str]]] = {}
for repo in sorted(repos, key=lambda r: r["slug"]):
slug = repo["slug"]
@ -172,7 +271,21 @@ def main() -> int:
if not root.is_dir():
report["repos"][slug] = {"missing_checkout": True}
continue
data = scan_repo(root)
if allowlist.get("per_repo", {}).get(slug, {}).get("exclude_repo"):
report["repos"][slug] = {"excluded_repo": True}
continue
data = scan_repo(
root,
repo_slug=slug,
allowlist=allowlist,
apply_allowlist=args.apply_allowlist,
)
if args.check_prose_gate:
violations = prose_gate_violations(root, slug, allowlist)
if violations:
prose_violations[slug] = [
{"path": path, "hits": hits} for path, hits in violations
]
if data["occurrences"]:
report["repos"][slug] = data
report["totals"]["occurrences"] += data["occurrences"]
@ -182,6 +295,18 @@ def main() -> int:
pattern_totals[label] += count
report["pattern_totals"] = dict(sorted(pattern_totals.items(), key=lambda x: -x[1]))
if prose_violations:
report["prose_gate_violations"] = prose_violations
if args.check_prose_gate and prose_violations:
if args.json:
print(json.dumps(report, indent=2))
else:
print("Prose gate failed — 'workstream ' hits outside allowlist:", file=sys.stderr)
for slug, rows in sorted(prose_violations.items()):
for row in rows[:10]:
print(f" {slug}: {row['hits']:3d} {row['path']}", file=sys.stderr)
return 1
if args.json:
print(json.dumps(report, indent=2))

View file

@ -4,7 +4,7 @@ type: workplan
title: "Fleet-wide workplan terminology refactor (workstream → workplan)"
domain: infotech
repo: the-custodian
status: proposed
status: active
owner: codex
topic_slug: custodian
planning_priority: medium
@ -112,7 +112,7 @@ Promote this workplan when all of the following are true:
```task
id: CUST-WP-0055-T01
status: todo
status: done
priority: high
state_hub_task_id: "9db434bd-55c5-4499-a365-8ac6a47726c8"
```
@ -129,14 +129,14 @@ only mentions `workstream` in an explicit compatibility footnote.
Progress 2026-07-08: canon addendum drafted at
`canon/standards/workplan-terminology-fleet_v0.1.md` (fleet term, legacy
bridges, event subjects, agent rules, retirement rule). Baseline JSON at
`docs/evidence/workstream-terminology-baseline-20260708.json`. Template
regeneration (`state-hub/scripts/project_rules/*.template`) remains open.
`docs/evidence/workstream-terminology-baseline-20260708.json`. Template regeneration (`state-hub/scripts/project_rules/*.template`) verified
workplan-first with explicit legacy footnotes only.
## Task: State Hub and hub-core legacy surface retirement plan
```task
id: CUST-WP-0055-T02
status: todo
status: progress
priority: high
state_hub_task_id: "2bb01721-a86b-43a0-ab4c-e5966743d295"
```
@ -159,7 +159,7 @@ until those interfaces retire — not in user-facing prose.
```task
id: CUST-WP-0055-T03
status: todo
status: done
priority: high
state_hub_task_id: "72c2ecf3-c0c1-4241-b0f2-339a97ccf949"
```
@ -260,7 +260,7 @@ non-rename policy.
```task
id: CUST-WP-0055-T08
status: todo
status: done
priority: high
state_hub_task_id: "5b6596ef-fd04-4469-b421-f548d292cb0d"
```