fix(spawn): seedgo remediation on is_protected work — narrowed logged excepts (OSError/ValueError/KeyError + logger.info, fallthrough semantics unchanged) and extracted _check_nested_pollution to cut detect_pollution to depth 4. Audit 100% (43/43), 357 tests, probes re-verified (pollution 0, delete aipass refused). DPLAN-0250 Track A follow-up, dispatched to @spawn

This commit is contained in:
AIOSAI
2026-07-19 01:18:15 -07:00
parent d631fd8d54
commit 8cb3895264
2 changed files with 46 additions and 33 deletions
+4 -4
View File
@@ -73,8 +73,8 @@ def is_protected(branch_name, branch_dir=None, registry_path=None):
if branch_dir is None:
branch_dir = (rp.parent / entry.get("path", "")).resolve()
break
except Exception:
pass
except (OSError, ValueError, KeyError) as exc:
logger.info("[is_protected] Registry lookup failed for %s: %s", branch_name, exc)
if branch_dir is not None:
passport_path = Path(branch_dir) / ".trinity" / "passport.json"
@@ -83,8 +83,8 @@ def is_protected(branch_name, branch_dir=None, registry_path=None):
passport = json_handler.read_json(passport_path)
if passport and passport.get("citizenship", {}).get("registered") is True:
return True, "active citizen"
except Exception:
pass
except (OSError, ValueError, KeyError) as exc:
logger.info("[is_protected] Passport read failed for %s: %s", branch_name, exc)
return False, ""
+42 -29
View File
@@ -319,6 +319,28 @@ def move_branch(branch_name, new_path, registry_path=None, dry_run=False, reloca
# =============================================================================
def _check_nested_pollution(nested_dir, label, project_root, registry_path):
"""Check if a nested duplicate directory is real pollution.
Returns an issue dict if pollution, or None if the directory is a
protected branch (active passport, registry owner, infrastructure floor).
"""
protected, _reason = is_protected(
nested_dir.name,
branch_dir=nested_dir,
registry_path=registry_path,
)
if protected:
return None
rel = nested_dir.relative_to(project_root).as_posix()
return {
"type": "duplicate_nested_dir",
"path": rel,
"description": f"Duplicate nested directory: {label}/",
}
def detect_pollution(project_root):
"""Detect init pollution — duplicate nested directories.
@@ -343,40 +365,31 @@ def detect_pollution(project_root):
nested = project_root / project_name
if nested.is_dir():
protected, _reason = is_protected(
project_name,
branch_dir=nested,
registry_path=registry_path,
issue = _check_nested_pollution(
nested,
f"{project_name}/{project_name}",
project_root,
registry_path,
)
if not protected:
issues.append(
{
"type": "duplicate_nested_dir",
"path": project_name,
"description": f"Duplicate nested directory: {project_name}/{project_name}/",
}
)
if issue:
issues.append(issue)
src_dir = project_root / "src"
if src_dir.is_dir():
for child in sorted(src_dir.iterdir()):
if child.is_dir() and not child.name.startswith(".") and not child.name.startswith("__"):
nested_dup = child / child.name
if nested_dup.is_dir():
protected, _reason = is_protected(
child.name,
branch_dir=nested_dup,
registry_path=registry_path,
)
if not protected:
rel = nested_dup.relative_to(project_root).as_posix()
issues.append(
{
"type": "duplicate_nested_dir",
"path": rel,
"description": f"Duplicate nested directory: src/{child.name}/{child.name}/",
}
)
if not child.is_dir() or child.name.startswith(".") or child.name.startswith("__"):
continue
nested_dup = child / child.name
if not nested_dup.is_dir():
continue
issue = _check_nested_pollution(
nested_dup,
f"src/{child.name}/{child.name}",
project_root,
registry_path,
)
if issue:
issues.append(issue)
return issues