From: Thomas Walker Lynch Date: Fri, 19 Sep 2025 07:32:55 +0000 (-0700) Subject: breadth first, and RT format X-Git-Url: https://git.reasoningtechnology.com/style/rt_dark_doc.css?a=commitdiff_plain;h=d4bddfd5e75347c07e7aba61fb25446d6e7aeef5;p=subu breadth first, and RT format --- diff --git a/developer/source/StageHand/Planner.py b/developer/source/StageHand/Planner.py index b5f4bac..94d8226 100644 --- a/developer/source/StageHand/Planner.py +++ b/developer/source/StageHand/Planner.py @@ -20,11 +20,10 @@ os.environ.setdefault("PYTHONDONTWRITEBYTECODE" ,"1") from pathlib import Path import getpass - # ===== Utilities ===== -def norm_perm(value: int|str)-> tuple[int,str]|None: - "Given int or 3/4-char octal string (optionally 0o-prefixed). Does validate/normalize. Returns (int,'%04o') or None." +def norm_perm(value: int|str)-> tuple[int ,str]|None: + "Given int or 3/4-char octal string (optionally 0o-prefixed). Does validate/normalize. Returns (int ,'%04o') or None." if isinstance(value ,int): if 0 <= value <= 0o7777: return value ,f"{value:04o}" @@ -55,6 +54,13 @@ def norm_abs_dpath_str(value: str|Path|None)-> str|None: s = value.as_posix() if isinstance(value ,Path) else str(value) return s if is_abs_dpath(s) else None +def norm_dpath_str(value: str|Path|None)-> str|None: + "Given str/Path/None. Does minimal sanitize; allows relative. Returns str or None." + if value is None: return None + s = value.as_posix() if isinstance(value ,Path) else str(value) + if not s or "\x00" in s: return None + return s + def norm_fname_or_none(value: str|None)-> str|None: "Given candidate filename or None. Does validate bare filename. Returns str or None." if value is None: return None @@ -70,7 +76,7 @@ def norm_nonempty_owner(value: str|None)-> str|None: return s if s else None def parse_mode(value: int|str|None)-> tuple[int|None ,str|None]: - "Given int/str/None. Does normalize via norm_perm. Returns (int,'%04o') or (None,None)." + "Given int/str/None. Does normalize via norm_perm. Returns (int ,'%04o') or (None ,None)." if value is None: return None ,None r = norm_perm(value) return r if r is not None else (None ,None) @@ -81,14 +87,6 @@ def norm_content_bytes(value: bytes|str|None)-> bytes|None: if isinstance(value ,bytes): return value return value.encode("utf-8") -def norm_dpath_str(value: str|Path|None)-> str|None: - "Given str/Path/None. Does minimal sanitize; allows relative. Returns str or None." - if value is None: return None - s = value.as_posix() if isinstance(value ,Path) else str(value) - if not s or "\x00" in s: return None - return s - - # ===== Wire-ready model types (no CBOR here) ===== class Command: @@ -99,7 +97,11 @@ class Command: Does hold op name, own a fresh arg_dict, collect per-entry errors. Returns dictionary via as_dictionary(). """ - __slots__ = ("name_str" ,"arg_dict" ,"errors_list") + __slots__ = ( + "name_str" + ,"arg_dict" + ,"errors_list" + ) def __init__(self ,name_str: str ,arg_dict: dict|None=None ,errors_list: list[str]|None=None)-> None: self.name_str = name_str @@ -116,7 +118,7 @@ class Command: ,"errors_list": list(self.errors_list) } - def print(self, *, index: int|None=None, file=None)-> None: + def print(self ,* ,index: int|None=None ,file=None)-> None: """ Given: optional index for numbering and optional file-like (defaults to stdout). Does: print a compact, human-readable one-line summary of this command; prints any errors indented below. @@ -129,16 +131,18 @@ class Command: op = self.name_str ad = self.arg_dict or {} - # Compose destination path for display + # Compose destination path for display (normalize to collapse '..') d = ad.get("write_file_dpath_str") or "" f = ad.get("write_file_fname") or "" try: from pathlib import Path as _Path - dst = (_Path(d)/f).as_posix() if d and f and "/" not in f else "?" + if d and f and "/" not in f: + dst = (_Path(d)/f).resolve().as_posix() + else: + dst = "?" except Exception: dst = "?" - # Numbering prefix prefix = f"{index:02d}. " if index is not None else "" if op == "copy": @@ -153,12 +157,10 @@ class Command: else: line = f"{prefix}?op? -> {dst}" - print(line, file=file) + print(line ,file=file) - # Print any per-entry errors underneath for err in self.errors_list: - print(f" ! {err}", file=file) - + print(f" ! {err}" ,file=file) class Journal: """ @@ -168,7 +170,10 @@ class Journal: Does manage meta, append commands, expose entries, and pack to dict. Returns dict via as_dictionary(). """ - __slots__ = ("meta_dict" ,"command_list") + __slots__ = ( + "meta_dict" + ,"command_list" + ) def __init__(self ,plan_dict: dict|None=None)-> None: self.meta_dict = {} @@ -183,7 +188,7 @@ class Journal: entries = plan_dict.get("entries_list") or [] self.meta_dict.update(meta) for e in entries: - if not isinstance(e ,dict): + if not isinstance(e ,dict): continue op = e.get("op") or "?" args = e.get("arg_dict") or {} @@ -206,7 +211,7 @@ class Journal: ,"entries_list": self.entries_list() } - def print(self, *, index_start: int = 1, file=None) -> None: + def print(self ,* ,index_start: int=1 ,file=None)-> None: """ Given: optional starting index and optional file-like (defaults to stdout). Does: print each Command on a single line via Command.print(), numbered. @@ -217,24 +222,29 @@ class Journal: file = _sys.stdout if not self.command_list: - print("(plan is empty)", file=file) + print("(plan is empty)" ,file=file) return - for i, cmd in enumerate(self.command_list, start=index_start): - cmd.print(index=i, file=file) + for i ,cmd in enumerate(self.command_list ,start=index_start): + cmd.print(index=i ,file=file) # ===== Runner-provided provenance ===== -# Planner.py class PlanProvenance: """ Runner-provided, read-only provenance for a single config script. """ - __slots__ = ("stage_root_dpath","config_abs_fpath","config_rel_fpath", - "read_dir_dpath","read_fname","process_user") - - def __init__(self, *, stage_root: Path, config_path: Path): - import getpass + __slots__ = ( + "stage_root_dpath" + ,"config_abs_fpath" + ,"config_rel_fpath" + ,"read_dir_dpath" + ,"read_fname" + ,"process_user" + ,"cwd_dpath" + ) + + def __init__(self ,* ,stage_root: Path ,config_path: Path): self.stage_root_dpath = stage_root.resolve() self.config_abs_fpath = config_path.resolve() try: @@ -252,19 +262,19 @@ class PlanProvenance: else: self.read_fname = name - # NEW: owner of the StageHand process self.process_user = getpass.getuser() + self.cwd_dpath = Path.cwd().resolve() - def print(self, *, file=None) -> None: + def print(self ,* ,file=None)-> None: if file is None: import sys as _sys file = _sys.stdout - print(f"Stage root: {self.stage_root_dpath}", file=file) - print(f"Config (rel): {self.config_rel_fpath.as_posix()}", file=file) - print(f"Config (abs): {self.config_abs_fpath}", file=file) - print(f"Read dir: {self.read_dir_dpath}", file=file) - print(f"Read fname: {self.read_fname}", file=file) - print(f"Process user: {self.process_user}", file=file) # NEW + print(f"Stage root: {self.stage_root_dpath}" ,file=file) + print(f"Config (rel): {self.config_rel_fpath.as_posix()}" ,file=file) + print(f"Config (abs): {self.config_abs_fpath}" ,file=file) + print(f"Read dir: {self.read_dir_dpath}" ,file=file) + print(f"Read fname: {self.read_fname}" ,file=file) + print(f"Process user: {self.process_user}" ,file=file) # ===== Admin-facing defaults carrier ===== @@ -272,45 +282,45 @@ class WriteFileMeta: """ WriteFileMeta — per-call or planner-default write-file attributes. - Given dpath (abs str/Path) ,fname (bare name or None) ,owner (str) + Given dpath (str/Path, may be relative) ,fname (bare name or None) ,owner (str) ,mode (int|'0644') ,content (bytes|str|None). Does normalize into fields (may remain None if absent/invalid). Returns object suitable for providing defaults to Planner methods. """ - __slots__ = ("dpath_str" ,"fname" ,"owner_name_str" ,"mode_int" ,"mode_octal_str" ,"content_bytes") + __slots__ = ( + "dpath_str" + ,"fname" + ,"owner_name_str" + ,"mode_int" + ,"mode_octal_str" + ,"content_bytes" + ) def __init__(self ,* ,dpath="/" - ,fname=None # None → let Planner/provenance choose + ,fname=None ,owner="root" ,mode=0o444 ,content=None ): - self.dpath_str = norm_dpath_str(dpath) - self.fname = norm_fname_or_none(fname) # '.' no longer special → None - self.owner_name_str = norm_nonempty_owner(owner) # '.' rejected → None - self.mode_int, self.mode_octal_str = parse_mode(mode) - self.content_bytes = norm_content_bytes(content) + self.dpath_str = norm_dpath_str(dpath) + self.fname = norm_fname_or_none(fname) + self.owner_name_str = norm_nonempty_owner(owner) + self.mode_int ,self.mode_octal_str = parse_mode(mode) + self.content_bytes = norm_content_bytes(content) - def print(self, *, label: str | None = None, file=None) -> None: - """ - Given: optional label and optional file-like (defaults to stdout). - Does: print a single-line summary of defaults/overrides. - Returns: None. - """ + def print(self ,* ,label: str|None=None ,file=None)-> None: if file is None: import sys as _sys file = _sys.stdout - dpath = self.dpath_str or "?" fname = self.fname or "?" owner = self.owner_name_str or "?" - mode_str = f"{self.mode_int:04o}" if isinstance(self.mode_int, int) else (self.mode_octal_str or "?") - size = len(self.content_bytes) if isinstance(self.content_bytes, (bytes, bytearray)) else 0 + mode_str = f"{self.mode_int:04o}" if isinstance(self.mode_int ,int) else (self.mode_octal_str or "?") + size = len(self.content_bytes) if isinstance(self.content_bytes ,(bytes ,bytearray)) else 0 prefix = (label + ": ") if label else "" - print(f"{prefix}dpath={dpath} fname={fname} owner={owner} mode={mode_str} bytes={size}", file=file) - + print(f"{prefix}dpath={dpath} fname={fname} owner={owner} mode={mode_str} bytes={size}" ,file=file) # ===== Planner ===== @@ -324,7 +334,11 @@ class Planner: On any argument error, returns the Command with errors and DOES NOT append it to Journal. Returns live Journal via journal(). """ - __slots__ = ("_prov" ,"_defaults" ,"_journal") + __slots__ = ( + "_prov" + ,"_defaults" + ,"_journal" + ) def __init__(self ,provenance: PlanProvenance ,defaults: WriteFileMeta|None=None)-> None: self._prov = provenance @@ -341,66 +355,47 @@ class Planner: ,config_rel_fpath_str=self._prov.config_rel_fpath.as_posix() ) - # --- defaults management / access --- + # --- provenance/defaults/journal access --- - # in Planner.py, inside class Planner - def set_provenance(self, prov: PlanProvenance) -> None: - """Switch the current provenance used for fallbacks & per-command provenance tagging.""" + def set_provenance(self ,prov: PlanProvenance)-> None: self._prov = prov def set_defaults(self ,defaults: WriteFileMeta)-> None: - "Given WriteFileMeta. Does replace planner defaults. Returns None." self._defaults = defaults def defaults(self)-> WriteFileMeta: - "Given n/a. Does return current WriteFileMeta defaults. Returns WriteFileMeta." return self._defaults def journal(self)-> Journal: - "Given n/a. Returns Journal reference (live, still being modified here)." return self._journal # --- resolution helpers --- def _pick(self ,kw ,meta_attr ,default_attr): - "Given three sources. Does pick first non-None. Returns value or None." + "Pick first non-None among kw ,meta_attr ,default_attr." return kw if kw is not None else (meta_attr if meta_attr is not None else default_attr) -# inside Planner - -from pathlib import Path -from posixpath import normpath as _normposix - - def _resolve_write_file(self, wfm, dpath, fname) -> tuple[str|None, str|None]: - # Normalize explicit kwargs (keep None as sentinel) + def _resolve_write_file(self ,wfm ,dpath ,fname)-> tuple[str|None ,str|None]: + # normalize explicit kwargs dpath_str = norm_dpath_str(dpath) if dpath is not None else None - if fname is not None and fname != ".": - fname = norm_fname_or_none(fname) + fname_str = norm_fname_or_none(fname) if fname is not None else None - # Precedence: kwarg > per-call meta > planner default - dpath_val = self._pick(dpath_str, (wfm.dpath_str if wfm else None), self._defaults.dpath_str) - fname_val = self._pick(fname, (wfm.fname if wfm else None), self._defaults.fname) + # precedence: kwarg > per-call meta > planner default + dpath_val = self._pick(dpath_str ,(wfm.dpath_str if wfm else None) ,self._defaults.dpath_str) + fname_val = self._pick(fname_str ,(wfm.fname if wfm else None) ,self._defaults.fname) - # Final fallback for filename: "." or None → derive from config name - if fname_val == "." or fname_val is None: + # final fallback for filename: derive from config name + if fname_val is None: fname_val = self._prov.read_fname - # Anchor relative dpaths to the *process working directory* (CWD), then normalize. + # anchor/normalize dpath if dpath_val is not None: - if is_abs_dpath(dpath_val): - # Normalize absolute path for pretty/consistency - try: - dpath_val = Path(dpath_val).resolve().as_posix() - except Exception: - dpath_val = _normposix(str(dpath_val)) - else: - base = Path.cwd() - try: - dpath_val = (base / dpath_val).resolve().as_posix() - except Exception: - dpath_val = _normposix((base / dpath_val).as_posix()) + p = Path(dpath_val) + if not p.is_absolute(): + p = (self._prov.cwd_dpath/p) + dpath_val = p.resolve().as_posix() - return dpath_val, fname_val + return dpath_val ,fname_val def _resolve_owner_mode_content(self ,wfm: WriteFileMeta|None @@ -412,41 +407,38 @@ from posixpath import normpath as _normposix mode_norm = parse_mode(mode) if mode is not None else (None ,None) content_b = norm_content_bytes(content) if content is not None else None - owner_v = self._pick(owner_norm, (wfm.owner_name_str if wfm else None), self._defaults.owner_name_str) + owner_v = self._pick(owner_norm ,(wfm.owner_name_str if wfm else None) ,self._defaults.owner_name_str) mode_v = (mode_norm if mode_norm != (None ,None) else ((wfm.mode_int ,wfm.mode_octal_str) if wfm else (self._defaults.mode_int ,self._defaults.mode_octal_str))) content_v = self._pick(content_b ,(wfm.content_bytes if wfm else None) ,self._defaults.content_bytes) return owner_v ,mode_v ,content_v - def print(self, *, show_journal: bool = True, file=None) -> None: - """ - Given: flags (show_journal) and optional file-like (defaults to stdout). - Does: print provenance, defaults, and optionally the journal via delegation. - Returns: None. - """ + # --- printing --- + + def print(self ,* ,show_journal: bool=True ,file=None)-> None: if file is None: import sys as _sys file = _sys.stdout - print("== Provenance ==", file=file) + print("== Provenance ==" ,file=file) self._prov.print(file=file) - print("\n== Defaults ==", file=file) - self._defaults.print(label="defaults", file=file) + print("\n== Defaults ==" ,file=file) + self._defaults.print(label="defaults" ,file=file) if show_journal: - entries = getattr(self._journal, "command_list", []) + entries = getattr(self._journal ,"command_list" ,[]) n_total = len(entries) - n_copy = sum(1 for c in entries if getattr(c, "name_str", None) == "copy") - n_disp = sum(1 for c in entries if getattr(c, "name_str", None) == "displace") - n_del = sum(1 for c in entries if getattr(c, "name_str", None) == "delete") + n_copy = sum(1 for c in entries if getattr(c ,"name_str" ,None) == "copy") + n_disp = sum(1 for c in entries if getattr(c ,"name_str" ,None) == "displace") + n_del = sum(1 for c in entries if getattr(c ,"name_str" ,None) == "delete") - print("\n== Journal ==", file=file) - print(f"entries: {n_total} copy:{n_copy} displace:{n_disp} delete:{n_del}", file=file) + print("\n== Journal ==" ,file=file) + print(f"entries: {n_total} copy:{n_copy} displace:{n_disp} delete:{n_del}" ,file=file) if n_total: - self._journal.print(index_start=1, file=file) + self._journal.print(index_start=1 ,file=file) else: - print("(plan is empty)", file=file) + print("(plan is empty)" ,file=file) # --- Command builders (first arg may be WriteFileMeta) --- @@ -459,11 +451,6 @@ from posixpath import normpath as _normposix ,mode: int|str|None=None ,content: bytes|str|None=None )-> Command: - """ - Given optional WriteFileMeta plus keyword overrides. - Does build a 'copy' command; on any argument error the command is returned with errors and NOT appended. - Returns Command. - """ cmd = Command("copy") dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname) owner_v ,(mode_int ,mode_oct) ,content_b = self._resolve_owner_mode_content(wfm ,owner ,mode ,content) @@ -478,13 +465,13 @@ from posixpath import normpath as _normposix cmd.add_error("content is required for copy() (bytes or str)") cmd.arg_dict.update({ - "write_file_dpath_str": dpath, - "write_file_fname": fname, # was write_file_fname - "owner_name": owner_v, # was owner_name_str - "mode_int": mode_int, - "mode_octal_str": mode_oct, - "content_bytes": content_b, - "provenance_config_rel_fpath_str": self._prov.config_rel_fpath.as_posix(), + "write_file_dpath_str": dpath + ,"write_file_fname": fname + ,"owner_name": owner_v + ,"mode_int": mode_int + ,"mode_octal_str": mode_oct + ,"content_bytes": content_b + ,"provenance_config_rel_fpath_str": self._prov.config_rel_fpath.as_posix() }) if not cmd.errors_list: @@ -497,14 +484,13 @@ from posixpath import normpath as _normposix ,write_file_dpath: str|Path|None=None ,write_file_fname: str|None=None )-> Command: - "Given optional WriteFileMeta plus overrides. Does build 'displace' entry or return errors. Returns Command." cmd = Command("displace") dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname) if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute") if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename") cmd.arg_dict.update({ - "write_file_dpath_str": dpath, - "write_file_fname": fname, + "write_file_dpath_str": dpath + ,"write_file_fname": fname }) if not cmd.errors_list: self._journal.append(cmd) @@ -516,18 +502,14 @@ from posixpath import normpath as _normposix ,write_file_dpath: str|Path|None=None ,write_file_fname: str|None=None )-> Command: - "Given optional WriteFileMeta plus overrides. Does build 'delete' entry or return errors. Returns Command." cmd = Command("delete") dpath ,fname = self._resolve_write_file(wfm ,write_file_dpath ,write_file_fname) if not is_abs_dpath(dpath): cmd.add_error("write_file_dpath must be absolute") if norm_fname_or_none(fname) is None: cmd.add_error("write_file_fname must be a bare filename") cmd.arg_dict.update({ - "write_file_dpath_str": dpath, - "write_file_fname": fname, + "write_file_dpath_str": dpath + ,"write_file_fname": fname }) if not cmd.errors_list: self._journal.append(cmd) return cmd - - - diff --git a/developer/source/StageHand/executor.py b/developer/source/StageHand/executor.py old mode 100644 new mode 100755 index 454fe8e..d8cb3a0 --- a/developer/source/StageHand/executor.py +++ b/developer/source/StageHand/executor.py @@ -5,7 +5,7 @@ executor.py — StageHand outer/inner executor (MVP; UNPRIVILEGED for now) Phase 0 (bootstrap): - Ensure filter program exists (create default in CWD if --filter omitted) - Validate --stage exists - - If --phase-0-then-stop: exit here (no scan ,no execution) + - If --phase-0-then-stop: exit here (no scan, no execution) Phase 1 (outer): - Discover every file under --stage; acceptance filter decides which to include @@ -47,7 +47,7 @@ from Planner import ( DEFAULT_FILTER_FILENAME = "stagehand_filter.py" DEFAULT_FILTER_SOURCE = """# StageHand acceptance filter (default template) -# Return True to include a config file ,False to skip it. +# Return True to include a config file, False to skip it. # You receive a PlanProvenance object named `prov`. # # prov fields commonly used here: @@ -90,8 +90,8 @@ def iso_utc_now_str()-> str: def _ensure_filter_file(filter_arg: str|None)-> Path: """ - If --filter is provided ,return that path (must exist). - Otherwise ,create ./stagehand_filter.py in the CWD if missing (writing a helpful template), + If --filter is provided, return that path (must exist). + Otherwise, create ./stagehand_filter.py in the CWD if missing (writing a helpful template), and return its path. """ if filter_arg: @@ -138,7 +138,7 @@ def _walk_all_files(stage_root: Path): def find_config_paths(stage_root: Path ,accept_func)-> list[Path]: """ Return files accepted by the Python acceptance function: accept(prov) → True/False. - Ordered breadth-first by depth ,then lexicographically by relative path. + Ordered breadth-first by depth, then lexicographically by relative path. """ out: list[tuple[int ,str ,Path]] = [] root = stage_root.resolve() @@ -157,7 +157,7 @@ def find_config_paths(stage_root: Path ,accept_func)-> list[Path]: def _run_all_configs_into_single_planner(stage_root: Path ,cfgs: list[Path])-> Planner: """ - Create a single Planner and execute each config's configure(prov ,planner ,WriteFileMeta) + Create a single Planner and execute each config's configure(prov, planner, WriteFileMeta) against it. Returns that single Planner containing the entire plan. """ # seed with synthetic provenance; we overwrite per config before execution @@ -175,7 +175,7 @@ def _run_all_configs_into_single_planner(stage_root: Path ,cfgs: list[Path])-> P fn(prov ,planner ,WriteFileMeta) - # annotate meta once ,on the single planner's journal + # annotate meta once, on the single planner's journal j = planner.journal() j.set_meta( generator_prog_str="executor.py" @@ -260,7 +260,7 @@ def _outer_main(stage_root: Path ,accept_func ,args)-> int: if args.phase_1_then_stop: return 0 - # Phase 2: encode CBOR and invoke inner path (same script ,--inner) + # Phase 2: encode CBOR and invoke inner path (same script, --inner) try: cbor_bytes = _plan_to_cbor_bytes(master) except Exception as e: