refactor(paths): own the goal-id traversal rule in one module

loopx/control_plane/quota/slot_accounting.py

@@ -22,6 +22,7 @@
 )
 from .monitor_poll import QUOTA_MONITOR_POLL_CLASSIFICATION
 from .scheduler_ack import QUOTA_SCHEDULER_ACK_CLASSIFICATION
+from ...paths import validate_goal_id_path_segment as _validate_goal_id_path_segment
 from ..effect_program import SettlementBindingKind
 from .settlement import (
     SettlementFailureKind,
@@ -297,17 +298,6 @@ def _repair_settlement_workspace_causality(
     return repaired or causality
 
 
-def _validate_goal_id_path_segment(goal_id: str) -> str:
-    value = goal_id.strip()
-    if not value:
-        raise ValueError("goal id is required")
-    if value in {".", ".."} or "/" in value or "\\" in value:
-        raise ValueError("goal id must be a single path segment")
-    if Path(value).name != value:
-        raise ValueError("goal id must not include path traversal")
-    return value
-
-
 def _queue_item_for_goal(status_payload: dict[str, Any], *, goal_id: str) -> dict[str, Any]:
     queue = status_payload.get("attention_queue") if isinstance(status_payload.get("attention_queue"), dict) else {}
     queue_items = queue.get("items") if isinstance(queue.get("items"), list) else []

loopx/history.py

@@ -58,7 +58,11 @@
 from .doctor import PROMOTION_READINESS_CLASSIFICATIONS
 from .execution_profile import compact_execution_profile
 from .explore_graph import compact_explore_graph_policy
-from .paths import registry_project_root, resolve_runtime_root
+from .paths import (
+    registry_project_root,
+    resolve_runtime_root,
+    validate_goal_id_path_segment as validate_goal_id_path_segment,
+)
 from .presentation.markdown import (
     markdown_code,
     markdown_table_row,
@@ -263,17 +267,6 @@ def write_reserved_run_artifacts(
         append_run_index_record(index_path, index_record, allow_nan=False)
 
 
-def validate_goal_id_path_segment(goal_id: str) -> str:
-    value = str(goal_id or "").strip()
-    if not value:
-        raise ValueError("goal id is required")
-    if value in {".", ".."} or "/" in value or "\\" in value:
-        raise ValueError("goal id must be a single path segment")
-    if Path(value).name != value:
-        raise ValueError("goal id must not include path traversal")
-    return value
-
-
 def discover_goal_ids(
     runtime_root: Path,
     registry: dict[str, Any],

loopx/paths.py

@@ -34,6 +34,20 @@ def legacy_goal_state_file(project: Path, goal_id: str) -> Path:
     return project / LEGACY_PROJECT_GOALS / goal_id / "ACTIVE_GOAL_STATE.md"
 
 
+# Single owner for the goal-id traversal rule; callers keep their own fields and errors.
+def validate_goal_id_path_segment(goal_id: str) -> str:
+    """Return a goal id that is safe to use as one path segment."""
+
+    value = str(goal_id or "").strip()
+    if not value:
+        raise ValueError("goal id is required")
+    if value in {".", ".."} or "/" in value or "\\" in value:
+        raise ValueError("goal id must be a single path segment")
+    if Path(value).name != value:
+        raise ValueError("goal id must not include path traversal")
+    return value
+
+
 def registered_goal_state_file(
     project: Path, goal_id: str, registry: dict[str, object] | None = None
 ) -> Path:

loopx/runtime.py

@@ -8,21 +8,13 @@
 from typing import Any
 
 from .history import load_registry
-from .paths import resolve_runtime_root
+from .paths import (
+    resolve_runtime_root,
+    validate_goal_id_path_segment as validate_goal_id_path_segment,
+)
 from .registry import registry_goals
 
 
-def validate_goal_id_path_segment(goal_id: str) -> str:
-    value = goal_id.strip()
-    if not value:
-        raise ValueError("goal id is required")
-    if value in {".", ".."} or "/" in value or "\\" in value:
-        raise ValueError("goal id must be a single path segment")
-    if Path(value).name != value:
-        raise ValueError("goal id must not include path traversal")
-    return value
-
-
 def unique_archive_path(archive_root: Path, goal_id: str, timestamp: str) -> Path:
     base = archive_root / f"{goal_id}-{timestamp}"
     candidate = base

tests/architecture/test_goal_id_path_segment_owner.py

@@ -0,0 +1,119 @@
+"""The goal-id traversal rule is decided in exactly one module.
+
+Four modules used to restate the same three checks, and one normalized a
+missing id while the others raised ``AttributeError``. ``loopx/paths.py`` owns
+the rule: both public entrypoints already import path helpers from it, and it
+imports no project module, so the direction is a leaf rather than a preference.
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+import pytest
+
+from loopx import history, paths, runtime
+from loopx.control_plane.quota import slot_accounting
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+OWNER_MODULE = "loopx/paths.py"
+RULE_ERRORS = frozenset(
+    {
+        "goal id is required",
+        "goal id must be a single path segment",
+        "goal id must not include path traversal",
+    }
+)
+# quota.py still restates the rule while #6875 and #5677 rewrite that module.
+# Converting it deletes this entry; an entry whose site has already been
+# converted fails the census too.
+DECLARED_SITES = {"loopx/quota.py": 1}
+
+ENTRYPOINTS = (
+    paths.validate_goal_id_path_segment,
+    runtime.validate_goal_id_path_segment,
+    history.validate_goal_id_path_segment,
+    slot_accounting._validate_goal_id_path_segment,
+)
+
+
+def _states_traversal_rule(node: ast.AST) -> bool:
+    texts = {
+        child.value
+        for child in ast.walk(node)
+        if isinstance(child, ast.Constant) and isinstance(child.value, str)
+    }
+    return RULE_ERRORS <= texts
+
+
+def _sites(root: Path) -> dict[str, int]:
+    found: dict[str, int] = {}
+    for path in sorted((root / "loopx").rglob("*.py")):
+        relative = path.relative_to(root).as_posix()
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+        count = sum(
+            1
+            for node in ast.walk(tree)
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+            and _states_traversal_rule(node)
+        )
+        if count:
+            found[relative] = count
+    return found
+
+
+def test_every_entrypoint_holds_the_owners_object() -> None:
+    owner = paths.validate_goal_id_path_segment
+    for entrypoint in ENTRYPOINTS:
+        assert entrypoint is owner
+
+
+def test_only_the_owner_and_declared_sites_state_the_rule() -> None:
+    assert _sites(REPO_ROOT) == {OWNER_MODULE: 1, **DECLARED_SITES}
+
+
+def test_the_census_sees_a_restated_rule(tmp_path: Path) -> None:
+    package = tmp_path / "loopx"
+    package.mkdir()
+    (package / "restated.py").write_text(
+        "def validate_goal_id_path_segment(goal_id: str) -> str:\n"
+        "    if not goal_id:\n"
+        "        raise ValueError('goal id is required')\n"
+        "    if '/' in goal_id:\n"
+        "        raise ValueError('goal id must be a single path segment')\n"
+        "    if '..' in goal_id:\n"
+        "        raise ValueError('goal id must not include path traversal')\n"
+        "    return goal_id\n",
+        encoding="utf-8",
+    )
+    (package / "consumer.py").write_text(
+        "from loopx.paths import validate_goal_id_path_segment\n",
+        encoding="utf-8",
+    )
+    assert _sites(tmp_path) == {"loopx/restated.py": 1}
+
+
+def test_declared_sites_cannot_go_stale_silently() -> None:
+    still_declared = {
+        relative: count
+        for relative, count in _sites(REPO_ROOT).items()
+        if relative in DECLARED_SITES
+    }
+    assert still_declared == DECLARED_SITES
+
+
+@pytest.mark.parametrize(
+    "value",
+    [".", "..", "a/b", "a\\b", " a/b ", "", "   ", None],
+)
+def test_every_entrypoint_rejects_the_same_invalid_values(value: object) -> None:
+    for entrypoint in ENTRYPOINTS:
+        with pytest.raises(ValueError):
+            entrypoint(value)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize("value", ["loopx-meta", "goal-shared", "Goal_X", "a.b:c"])
+def test_every_entrypoint_accepts_a_safe_single_segment(value: str) -> None:
+    for entrypoint in ENTRYPOINTS:
+        assert entrypoint(value) == value