zemion
2ffdb23f69
Verified with the coordinated workspace changes by devkit full run 2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed). This shared UI pass does not mark the individual module reviews complete.
567 lines
21 KiB
Python
Executable File
567 lines
21 KiB
Python
Executable File
"""Test planning over existing checks; planning never executes a test or server."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from copy import deepcopy
|
||
from itertools import islice
|
||
import re
|
||
|
||
from .common import read_json
|
||
from .package_tests import declared_tests, discovered_sources, source_stage_id
|
||
|
||
|
||
PROFILES = ("quick", "ui", "backend", "full")
|
||
|
||
|
||
def stage(
|
||
identity: str,
|
||
title: str,
|
||
argv: list[str],
|
||
cwd: Path,
|
||
*,
|
||
reason: str,
|
||
deps: list[str] | None = None,
|
||
after: list[str] | None = None,
|
||
resources: list[str] | None = None,
|
||
timeout_seconds: int = 300,
|
||
) -> dict:
|
||
return {
|
||
"id": identity,
|
||
"title": title,
|
||
"argv": argv,
|
||
"cwd": str(cwd),
|
||
"deps": deps or [],
|
||
"after": after or [],
|
||
"resources": resources or [],
|
||
"timeout_seconds": timeout_seconds,
|
||
"reason": reason,
|
||
}
|
||
|
||
|
||
def _custom_stages(
|
||
project, workspace_root: Path, profile: str, selected, filtered: bool
|
||
) -> list[dict]:
|
||
# load_project validates every nested declaration before selection.
|
||
records = project.config.get("checks", [])
|
||
checks = {item["id"]: item for item in records}
|
||
profiles = project.config.get("profiles", {})
|
||
if profile not in profiles:
|
||
raise ValueError(
|
||
f"Project {project.name!r} does not declare profile {profile!r}"
|
||
)
|
||
selected_names = {repo.name for repo in selected}
|
||
wanted = set()
|
||
|
||
def include(identity: str, visiting: frozenset[str] = frozenset()) -> None:
|
||
if identity not in checks:
|
||
raise ValueError(f"Unknown check dependency: {identity}")
|
||
if identity in visiting:
|
||
raise ValueError(f"Cyclic check dependency: {identity}")
|
||
if identity in wanted:
|
||
return
|
||
for dependency in [
|
||
*checks[identity].get("deps", []),
|
||
*checks[identity].get("after", []),
|
||
]:
|
||
include(dependency, visiting | {identity})
|
||
wanted.add(identity)
|
||
|
||
for identity in profiles[profile]:
|
||
if identity not in checks:
|
||
raise ValueError(f"Unknown profile check: {identity}")
|
||
owned = set(checks[identity].get("repos", []))
|
||
if not filtered or not owned or selected_names & owned:
|
||
include(identity)
|
||
result = []
|
||
# A dependency may precede/follow its consumer in the configuration: the
|
||
# execution engine owns scheduling, not JSON declaration order.
|
||
for identity, item in checks.items():
|
||
if identity in wanted:
|
||
result.append(
|
||
stage(
|
||
identity,
|
||
item.get("title", identity),
|
||
list(item["argv"]),
|
||
workspace_root / item.get("cwd", "."),
|
||
deps=list(item.get("deps", [])),
|
||
after=list(item.get("after", [])),
|
||
resources=list(item.get("resources", [])),
|
||
timeout_seconds=item.get("timeout_seconds", 300),
|
||
reason=f"Project profile {profile}",
|
||
)
|
||
)
|
||
for field in ("inputs", "reuse"):
|
||
if field in item:
|
||
result[-1][field] = deepcopy(item[field])
|
||
return result
|
||
|
||
|
||
def focused_phases(meta: Path) -> list[dict]:
|
||
"""Read the same bounded, ordered phase metadata as the standalone gate."""
|
||
value = read_json(meta / "tools/checks/focused-phases.json", max_bytes=1024 * 1024)
|
||
if (
|
||
not isinstance(value, dict)
|
||
or set(value) != {"schema_version", "phases"}
|
||
or type(value["schema_version"]) is not int
|
||
or value["schema_version"] != 1
|
||
):
|
||
raise ValueError("Unsupported focused phase metadata schema")
|
||
phases = value["phases"]
|
||
if not isinstance(phases, list) or not 1 <= len(phases) <= 32:
|
||
raise ValueError("Focused phase catalog requires 1–32 definitions")
|
||
seen = set()
|
||
fields = {
|
||
"id",
|
||
"title",
|
||
"cwd",
|
||
"order_after",
|
||
"depends_on",
|
||
"resources",
|
||
"outputs",
|
||
"notes",
|
||
}
|
||
for phase in phases:
|
||
if not isinstance(phase, dict) or set(phase) != fields:
|
||
raise ValueError("Invalid focused phase metadata fields")
|
||
identity = phase["id"]
|
||
if (
|
||
not isinstance(identity, str)
|
||
or not re.fullmatch(r"[a-z][a-z0-9-]{0,63}", identity)
|
||
or identity in seen
|
||
):
|
||
raise ValueError("Invalid or duplicate focused phase ID")
|
||
if (
|
||
not isinstance(phase["title"], str)
|
||
or not phase["title"].strip()
|
||
or len(phase["title"]) > 256
|
||
or any(ord(char) < 32 for char in phase["title"])
|
||
):
|
||
raise ValueError("Invalid focused phase title")
|
||
if not isinstance(phase["cwd"], str) or phase["cwd"] not in {
|
||
"core",
|
||
"meta",
|
||
"core-webui",
|
||
"access-webui",
|
||
}:
|
||
raise ValueError("Unsupported focused phase working directory")
|
||
for field in ("order_after", "depends_on", "resources", "outputs", "notes"):
|
||
values = phase[field]
|
||
if (
|
||
not isinstance(values, list)
|
||
or len(values) > 64
|
||
or any(
|
||
not isinstance(item, str)
|
||
or not item
|
||
or len(item) > 4096
|
||
or "\0" in item
|
||
for item in values
|
||
)
|
||
or len(set(values)) != len(values)
|
||
):
|
||
raise ValueError("Invalid focused phase list: " + field)
|
||
if (set(phase["order_after"]) | set(phase["depends_on"])) - seen:
|
||
raise ValueError("Focused prerequisites must precede their consumer")
|
||
seen.add(identity)
|
||
return phases
|
||
|
||
|
||
def focused_phase_bodies(text: str, phases: list[dict]) -> dict[str, str]:
|
||
"""Only exact top-level registered wrappers are authoritative phase bodies.
|
||
|
||
Do not extract lookalike markers from heredocs, conditionals or unrelated
|
||
shell functions. This recognizes the maintained wrapper convention, not
|
||
arbitrary executable shell semantics.
|
||
"""
|
||
registered = {
|
||
"focused_phase_" + phase["id"].replace("-", "_"): phase["id"]
|
||
for phase in phases
|
||
}
|
||
lines, bodies, index, depth, heredoc = text.splitlines(), {}, 0, 0, None
|
||
while index < len(lines):
|
||
line = lines[index]
|
||
stripped = line.strip()
|
||
if heredoc is not None:
|
||
if stripped == heredoc:
|
||
heredoc = None
|
||
index += 1
|
||
continue
|
||
match = re.search(r"<<-?\s*['\"]?([A-Za-z_][A-Za-z0-9_]*)['\"]?", line)
|
||
if match:
|
||
heredoc = match[1]
|
||
index += 1
|
||
continue
|
||
function = re.fullmatch(r"(focused_phase_[a-z0-9_]+)\(\) \{", line)
|
||
if depth == 0 and function and function[1] in registered:
|
||
identity = registered[function[1]]
|
||
if (
|
||
identity in bodies
|
||
or index + 1 >= len(lines)
|
||
or lines[index + 1] != f"# devkit-phase: {identity} begin"
|
||
):
|
||
raise ValueError("Invalid or duplicate focused phase wrapper")
|
||
end = index + 2
|
||
while end < len(lines) and lines[end] != f"# devkit-phase: {identity} end":
|
||
end += 1
|
||
if end + 1 >= len(lines) or lines[end + 1] != "}":
|
||
raise ValueError("Unclosed focused phase wrapper")
|
||
bodies[identity] = "\n".join(lines[index + 2 : end]) + "\n"
|
||
index = end + 2
|
||
continue
|
||
if re.match(
|
||
r"(?:if|for|while|until|case|select)\b|(?:function\s+\w+|\w+\s*\(\s*\))",
|
||
stripped,
|
||
):
|
||
depth += 1
|
||
elif re.match(r"(?:fi|done|esac)\b|^}\s*;?$", stripped):
|
||
depth = max(0, depth - 1)
|
||
index += 1
|
||
if set(bodies) != {phase["id"] for phase in phases}:
|
||
raise ValueError("Focused phase metadata and marked implementations differ")
|
||
return {phase["id"]: bodies[phase["id"]] for phase in phases}
|
||
|
||
|
||
def _undeclared_source_note(project, workspace_root: Path) -> str | None:
|
||
"""Directory discovery is wider than registered Git input ownership."""
|
||
registered_paths = {repo.path.resolve() for repo in project.repositories}
|
||
children = list(islice(workspace_root.iterdir(), 4097))
|
||
if len(children) > 4096:
|
||
return "Workspace discovery exceeds its bounded ownership audit; native reuse is disabled."
|
||
for child in children:
|
||
if not child.name.startswith("govoplan") or child.resolve() in registered_paths:
|
||
continue
|
||
if any(
|
||
(child / name).exists() or (child / name).is_symlink()
|
||
for name in ("src", "webui")
|
||
):
|
||
return "Unregistered sibling src/WebUI inputs may be consumed by workspace discovery or PYTHONPATH; native reuse is disabled until their repository ownership is declared."
|
||
return None
|
||
|
||
|
||
def _apply_undeclared_source_note(checks: list[dict], note: str | None) -> None:
|
||
if note:
|
||
for check in checks:
|
||
check["reuse"] = "never"
|
||
check.setdefault("coverage_notes", []).append(note)
|
||
|
||
|
||
def _focused_ui_inputs(project) -> tuple[list[str] | None, str]:
|
||
names = {repo.name for repo in project.repositories}
|
||
if not {"govoplan", "govoplan-core"} <= names:
|
||
return None, "Missing Core/Meta ownership; input scope stays workspace-wide."
|
||
selected = {"govoplan", "govoplan-core"}
|
||
for repo in project.repositories:
|
||
root, webui = repo.path, repo.path / "webui"
|
||
if (
|
||
not root.is_dir()
|
||
or root.is_symlink()
|
||
or webui.is_symlink()
|
||
or webui.exists()
|
||
and not webui.is_dir()
|
||
):
|
||
return (
|
||
None,
|
||
"Missing or ambiguous repository/WebUI layout; input scope stays workspace-wide.",
|
||
)
|
||
if webui.is_dir():
|
||
selected.add(repo.name)
|
||
return (
|
||
sorted(selected),
|
||
"UI input scope includes whole Core/Meta and every registered WebUI repository, including helpers/configuration; it is not per-file dependency inference.",
|
||
)
|
||
|
||
|
||
def _apply_ui_inputs(check: dict, scope: tuple[list[str] | None, str]) -> None:
|
||
names, note = scope
|
||
if names is not None:
|
||
check["inputs"] = {"repos": list(names)}
|
||
check.setdefault("coverage_notes", []).append(note)
|
||
|
||
|
||
def _full_stages(project, workspace_root: Path, meta: Path, core: Path) -> list[dict]:
|
||
phases = focused_phases(meta)
|
||
ui_scope = _focused_ui_inputs(project)
|
||
directories = {
|
||
"core": core,
|
||
"meta": meta,
|
||
"core-webui": core / "webui",
|
||
"access-webui": workspace_root / "govoplan-access/webui",
|
||
}
|
||
checks = []
|
||
for phase in phases:
|
||
check = stage(
|
||
"focused." + phase["id"],
|
||
phase["title"],
|
||
[
|
||
"bash",
|
||
str(meta / "tools/checks/check-focused.sh"),
|
||
"--phase",
|
||
phase["id"],
|
||
],
|
||
directories[phase["cwd"]],
|
||
reason="Full retains every canonical phase in order; repository filters never narrow the required gate.",
|
||
deps=["focused." + value for value in phase["depends_on"]],
|
||
after=["focused." + value for value in phase["order_after"]],
|
||
resources=list(dict.fromkeys(["workspace:focused", *phase["resources"]])),
|
||
timeout_seconds=14400,
|
||
)
|
||
check["coverage_notes"] = list(phase["notes"])
|
||
check["phase_outputs"] = list(phase["outputs"])
|
||
if phase["id"] in {"core-ui", "module-builds", "browser", "module-ui"}:
|
||
check["resources"] = list(
|
||
dict.fromkeys(
|
||
[
|
||
*check["resources"],
|
||
*[f"webui:{repo.name}" for repo in project.repositories],
|
||
]
|
||
)
|
||
)
|
||
_apply_ui_inputs(check, ui_scope)
|
||
else:
|
||
check["coverage_notes"].append(
|
||
"Cross-module backend/tooling checks retain conservative whole-workspace inputs."
|
||
)
|
||
checks.append(check)
|
||
_apply_undeclared_source_note(
|
||
checks, _undeclared_source_note(project, workspace_root)
|
||
)
|
||
return checks
|
||
|
||
|
||
def _expanded_repositories(project, selected, *, changed: bool):
|
||
"""Conservative shared changes; declared interface consumers otherwise.
|
||
|
||
This is a selection aid, not a claim of exhaustive runtime dependency
|
||
analysis. The full profile always retains the canonical workspace gate.
|
||
"""
|
||
if not changed or not selected:
|
||
return selected, "explicit selection" if selected else "no changed repositories"
|
||
names = {repo.name for repo in selected}
|
||
if names & {"govoplan", "govoplan-core"}:
|
||
return list(
|
||
project.repositories
|
||
), "Core/Meta changed; all registered consumers conservatively selected"
|
||
# Reuse the release contract parser without importing module application
|
||
# code. If available declarations cannot be parsed, broaden selection.
|
||
import sys
|
||
|
||
meta = next(
|
||
(repo.path for repo in project.repositories if repo.name == "govoplan"), None
|
||
)
|
||
if meta is None:
|
||
return selected, "changed repositories; no GovOPlaN contract catalog"
|
||
release = meta / "tools" / "release"
|
||
if not (release / "govoplan_release" / "contracts.py").is_file():
|
||
return selected, "changed repositories; contract parser unavailable"
|
||
sys.path.insert(0, str(release))
|
||
try:
|
||
from govoplan_release.contracts import parse_manifest_contract
|
||
|
||
contracts = []
|
||
for repo in project.repositories:
|
||
for manifest in sorted((repo.path / "src").glob("*/backend/manifest.py")):
|
||
parsed = parse_manifest_contract(manifest, repo_name=repo.name)
|
||
if parsed is None:
|
||
return list(
|
||
project.repositories
|
||
), "unresolved manifest contract; conservative workspace selection"
|
||
contracts.append(parsed)
|
||
while True:
|
||
providers = {
|
||
item.name
|
||
for contract in contracts
|
||
if contract.repo in names
|
||
for item in contract.provides_interfaces
|
||
}
|
||
consumers = {
|
||
contract.repo
|
||
for contract in contracts
|
||
if any(item.name in providers for item in contract.requires_interfaces)
|
||
}
|
||
added = consumers - names
|
||
if not added:
|
||
break
|
||
names.update(added)
|
||
except (ImportError, AttributeError, OSError, SyntaxError, ValueError):
|
||
return list(
|
||
project.repositories
|
||
), "contract analysis unavailable; conservative workspace selection"
|
||
finally:
|
||
sys.path.remove(str(release))
|
||
return [
|
||
repo for repo in project.repositories if repo.name in names
|
||
], "changed repositories plus declared interface consumers"
|
||
|
||
|
||
def _module_ui_plan(repo, *, reason: str) -> tuple[list[dict], list[str]]:
|
||
"""Use package-owned direct Node test metadata, excluding shell/build chains.
|
||
|
||
Source structural scripts are also discoverable by the existing established
|
||
names. Unknown shell commands are deliberately not guessed or rewritten.
|
||
"""
|
||
webui = repo.path / "webui"
|
||
package_path = webui / "package.json"
|
||
if not package_path.is_file():
|
||
return [], []
|
||
scripts: dict[tuple[str, ...], list[str]] = {}
|
||
omitted = []
|
||
for item in [*declared_tests(repo, package_path), *discovered_sources(repo)]:
|
||
if item["component_suite"] is not None:
|
||
omitted.append(
|
||
f"{repo.name} {item['name']}: only covered by the explicit UI component batch; quick does not compile components"
|
||
)
|
||
elif item["name"] in {"test:module-permutations", "test:vite-cache-isolation"}:
|
||
omitted.append(
|
||
f"{repo.name} {item['name']}: separate environment/permutation verification, not run by this scoped profile"
|
||
)
|
||
elif item["_argv"]:
|
||
scripts[tuple(item["_argv"])] = item["_argv"]
|
||
else:
|
||
omitted.append(f"{repo.name} {item['name']}: {item['reason']}")
|
||
return [
|
||
stage(
|
||
source_stage_id(repo.name, argv),
|
||
f"{repo.name}: {Path(argv[-1]).stem}",
|
||
argv,
|
||
webui,
|
||
reason=reason,
|
||
resources=[f"webui:{repo.name}"],
|
||
)
|
||
for _, argv in sorted(scripts.items())
|
||
], omitted
|
||
|
||
|
||
def module_ui_stages(repo, *, reason: str) -> list[dict]:
|
||
return _module_ui_plan(repo, reason=reason)[0]
|
||
|
||
|
||
def build_stages(
|
||
workspace_root: Path,
|
||
profile: str,
|
||
repos: list[str],
|
||
changed: bool,
|
||
project: Path | None = None,
|
||
) -> list[dict]:
|
||
from .workspace import load_project, selected_repositories
|
||
|
||
if profile not in PROFILES:
|
||
raise ValueError(f"Unknown check profile: {profile}")
|
||
workspace_root = workspace_root.resolve()
|
||
loaded = load_project(workspace_root, project)
|
||
selected = selected_repositories(loaded, repos, changed=changed)
|
||
if project is not None:
|
||
return _custom_stages(
|
||
loaded, workspace_root, profile, selected, bool(repos or changed)
|
||
)
|
||
selected, reason = _expanded_repositories(loaded, selected, changed=changed)
|
||
mapping = {repo.name: repo.path for repo in loaded.repositories}
|
||
meta = mapping.get("govoplan", workspace_root / "govoplan")
|
||
core = mapping.get("govoplan-core", workspace_root / "govoplan-core")
|
||
if profile == "full":
|
||
return _full_stages(loaded, workspace_root, meta, core)
|
||
if changed and not selected:
|
||
return []
|
||
checks = []
|
||
for identity, command in (
|
||
(
|
||
"contracts",
|
||
[
|
||
"{python}",
|
||
str(meta / "tools/checks/check-contracts.py"),
|
||
"--workspace-root",
|
||
str(workspace_root),
|
||
"--no-impact",
|
||
],
|
||
),
|
||
(
|
||
"manifests",
|
||
[
|
||
"{python}",
|
||
str(meta / "tools/checks/check-manifest-shapes.py"),
|
||
"--workspace-root",
|
||
str(workspace_root),
|
||
"--require-architecture",
|
||
],
|
||
),
|
||
):
|
||
checks.append(
|
||
stage(
|
||
identity,
|
||
f"Workspace {identity}",
|
||
command,
|
||
meta,
|
||
reason="Shared manifest/interface invariants",
|
||
timeout_seconds=600,
|
||
)
|
||
)
|
||
if profile in {"quick", "ui"}:
|
||
ui_scope = _focused_ui_inputs(loaded)
|
||
coverage_notes = [
|
||
"Scoped source/component checks are not a complete module review or the full focused gate."
|
||
]
|
||
for identity in ("jsx-value-imports", "heading-help"):
|
||
checks.append(
|
||
stage(
|
||
identity,
|
||
f"Shared {identity} contract",
|
||
["{node}", str(meta / f"tools/checks/check-{identity}.mjs")],
|
||
meta,
|
||
reason="Existing source-only cross-module guard",
|
||
)
|
||
)
|
||
_apply_ui_inputs(checks[-1], ui_scope)
|
||
for repo in selected:
|
||
stages, omissions = _module_ui_plan(repo, reason=reason)
|
||
checks.extend(stages)
|
||
coverage_notes.extend(omissions)
|
||
checks[0]["coverage_notes"] = coverage_notes
|
||
if profile == "ui":
|
||
checks.append(
|
||
stage(
|
||
"core.component-batch",
|
||
"Core component suites (compile once)",
|
||
["{node}", str(core / "webui/scripts/run-component-tests.mjs")],
|
||
core / "webui",
|
||
reason="Shared components affect every UI consumer",
|
||
timeout_seconds=900,
|
||
)
|
||
)
|
||
_apply_ui_inputs(checks[-1], ui_scope)
|
||
if profile == "backend":
|
||
for repo in selected:
|
||
if (repo.path / "tests").is_dir():
|
||
checks.append(
|
||
stage(
|
||
f"{repo.name}.backend",
|
||
f"{repo.name} backend tests",
|
||
["{python}", "-m", "pytest", "-q", str(repo.path / "tests")],
|
||
repo.path,
|
||
reason=reason,
|
||
resources=["backend:test-state"],
|
||
timeout_seconds=1800,
|
||
)
|
||
)
|
||
_apply_undeclared_source_note(
|
||
checks, _undeclared_source_note(loaded, workspace_root)
|
||
)
|
||
return checks
|
||
|
||
|
||
def build_coverage(
|
||
workspace_root: Path,
|
||
profile: str,
|
||
repos: list[str],
|
||
changed: bool,
|
||
project: Path | None = None,
|
||
*,
|
||
stages: list[dict] | None = None,
|
||
) -> dict:
|
||
"""Read-only suite inventory; prebuilt stages avoid repeating selection queries."""
|
||
from .coverage import coverage_inventory
|
||
|
||
if profile not in PROFILES:
|
||
raise ValueError(f"Unknown check profile: {profile}")
|
||
if stages is None:
|
||
stages = build_stages(workspace_root, profile, repos, changed, project)
|
||
return coverage_inventory(workspace_root, profile, project, stages)
|