feat: add engine overlay module (resolve/discover, ast scan, tombstone)

engine/overlay.py

@@ -0,0 +1,110 @@
+# engine/overlay.py
+"""Resolve action names across commons + proprietaries.
+
+proprietaries first (override), commons second. Listing uses static ast scanning
+(no import side effects). Dispatch loads exactly one module by path.
+"""
+from __future__ import annotations
+
+import ast
+import importlib.util
+import os
+from dataclasses import dataclass
+from typing import List, Optional
+
+ACTION_EXT = ".py"
+
+
+@dataclass
+class Resolved:
+    name: str
+    path: str
+    shadowed: Optional[str]   # commons path when a proprietaries action overrides it
+
+
+@dataclass
+class ActionInfo:
+    name: str
+    source: str               # "commons" | "local" | "override"
+    summary: str
+    disabled: object          # None | True | "reason"
+
+
+def _rel_names(root: Optional[str]):
+    if not root or not os.path.isdir(root):
+        return {}
+    out = {}
+    for dirpath, _dirs, files in os.walk(root):
+        for fn in files:
+            if not fn.endswith(ACTION_EXT):
+                continue
+            full = os.path.join(dirpath, fn)
+            rel = os.path.relpath(full, root)[: -len(ACTION_EXT)]
+            out[rel.replace(os.sep, "/")] = full
+    return out
+
+
+def resolve(name: str, commons_root: str, prop_root: Optional[str]) -> Optional[Resolved]:
+    rel = name + ACTION_EXT
+    commons_path = os.path.join(commons_root, rel)
+    c = os.path.isfile(commons_path)
+    prop_path = os.path.join(prop_root, rel) if prop_root else None
+    p = bool(prop_path) and os.path.isfile(prop_path)
+    if p and c:
+        return Resolved(name, prop_path, commons_path)
+    if p:
+        return Resolved(name, prop_path, None)
+    if c:
+        return Resolved(name, commons_path, None)
+    return None
+
+
+def discover(commons_root: str, prop_root: Optional[str]) -> List[ActionInfo]:
+    commons = _rel_names(commons_root)
+    prop = _rel_names(prop_root)
+    infos = []
+    for name in sorted(set(commons) | set(prop)):
+        if name in prop and name in commons:
+            source, path = "override", prop[name]
+        elif name in prop:
+            source, path = "local", prop[name]
+        else:
+            source, path = "commons", commons[name]
+        summary, disabled = _scan(path)
+        infos.append(ActionInfo(name, source, summary, disabled))
+    return infos
+
+
+def _scan(path: str):
+    """Statically extract (summary, DISABLED) without importing the module."""
+    try:
+        with open(path) as f:
+            tree = ast.parse(f.read(), path)
+    except (SyntaxError, OSError):
+        return "(unparseable)", None
+    summary, disabled = "", None
+    for node in tree.body:
+        if not isinstance(node, ast.Assign):
+            continue
+        targets = [t.id for t in node.targets if isinstance(t, ast.Name)]
+        if "METADATA" in targets and isinstance(node.value, ast.Call):
+            summary = _summary_of(node.value)
+        if "DISABLED" in targets and isinstance(node.value, ast.Constant):
+            disabled = node.value.value if node.value.value else None
+    return summary, disabled
+
+
+def _summary_of(call: ast.Call) -> str:
+    for kw in call.keywords:
+        if kw.arg == "summary" and isinstance(kw.value, ast.Constant):
+            return kw.value.value
+    if call.args and isinstance(call.args[0], ast.Constant):
+        return call.args[0].value
+    return ""
+
+
+def load_module(path: str, modname: str):
+    spec = importlib.util.spec_from_file_location(modname, path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod

tests/test_overlay.py

@@ -0,0 +1,54 @@
+# tests/test_overlay.py
+import os, tempfile, unittest
+from engine import overlay
+
+def _write(root, rel, body):
+    p = os.path.join(root, rel)
+    os.makedirs(os.path.dirname(p), exist_ok=True)
+    with open(p, "w") as f:
+        f.write(body)
+    return p
+
+ACT = 'from engine.descriptor import Meta\nMETADATA = Meta(summary="{s}")\ndef run(ctx, args):\n    return "{r}"\n'
+
+class TestOverlay(unittest.TestCase):
+    def setUp(self):
+        self.commons = tempfile.mkdtemp()
+        self.prop = tempfile.mkdtemp()
+
+    def test_commons_only(self):
+        _write(self.commons, "configure/ufw.py", ACT.format(s="commons ufw", r="C"))
+        r = overlay.resolve("configure/ufw", self.commons, self.prop)
+        self.assertIsNotNone(r); self.assertIsNone(r.shadowed)
+        self.assertTrue(r.path.startswith(self.commons))
+
+    def test_override_exposes_shadowed(self):
+        _write(self.commons, "configure/ufw.py", ACT.format(s="commons", r="C"))
+        _write(self.prop, "configure/ufw.py", ACT.format(s="local", r="L"))
+        r = overlay.resolve("configure/ufw", self.commons, self.prop)
+        self.assertTrue(r.path.startswith(self.prop))     # local wins
+        self.assertTrue(r.shadowed.startswith(self.commons))
+
+    def test_unknown_returns_none(self):
+        self.assertIsNone(overlay.resolve("no/such", self.commons, self.prop))
+
+    def test_discover_marks_sources_and_disabled(self):
+        _write(self.commons, "configure/ufw.py", ACT.format(s="commons ufw", r="C"))
+        _write(self.commons, "sync/repository.py", ACT.format(s="sync repo", r="C"))
+        _write(self.prop, "configure/ufw.py", ACT.format(s="local ufw", r="L"))
+        _write(self.prop, "sync/minecraft.py",
+               'from engine.descriptor import Meta\nMETADATA = Meta(summary="mc")\nDISABLED = "wip"\ndef run(ctx, args): pass\n')
+        infos = {i.name: i for i in overlay.discover(self.commons, self.prop)}
+        self.assertEqual(infos["configure/ufw"].source, "override")
+        self.assertEqual(infos["sync/repository"].source, "commons")
+        self.assertEqual(infos["sync/minecraft"].source, "local")
+        self.assertEqual(infos["sync/minecraft"].disabled, "wip")
+        self.assertEqual(infos["configure/ufw"].summary, "local ufw")
+
+    def test_load_module_runs(self):
+        p = _write(self.commons, "x/y.py", ACT.format(s="s", r="Z"))
+        mod = overlay.load_module(p, "apex_action_x_y")
+        self.assertEqual(mod.run(None, None), "Z")
+
+if __name__ == "__main__":
+    unittest.main()