Add merge conflict detector for sibling PRs

Refs #485 - Expand Gitea CI/CD pipeline maturity

Changes:
- Remove '|| true' from shellcheck step so shell lint errors block merges
- Remove '|| true' from flake8 step so Python lint errors block merges
- Expand flake8 scope to include scripts/, bin/, tests/
- Exclude .git/ from shellcheck file discovery
- Add python-test job that runs pytest on the test suite after syntax check passes

.gitea/workflows/validate-config.yaml

@@ -59,7 +59,21 @@ jobs:
       - name: Flake8 critical errors only
         run: |
           flake8 --select=E9,F63,F7,F82 --show-source --statistics \
-            scripts/ allegro/ cron/ || true
+            scripts/ bin/ tests/
+
+  python-test:
+    name: Python Test Suite
+    runs-on: ubuntu-latest
+    needs: python-check
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+      - name: Install test dependencies
+        run: pip install pytest pyyaml
+      - name: Run tests
+        run: python3 -m pytest tests/ -v --tb=short
 
   shell-lint:
     name: Shell Script Lint
@@ -70,7 +84,7 @@ jobs:
         run: sudo apt-get install -y shellcheck
       - name: Lint shell scripts
         run: |
-          find . -name '*.sh' -print0 | xargs -0 -r shellcheck --severity=error || true
+          find . -name '*.sh' -not -path './.git/*' -print0 | xargs -0 -r shellcheck --severity=error
 
   cron-validate:
     name: Cron Syntax Check

bin/conflict_detector.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""
+Merge Conflict Detector — catches sibling PRs that will conflict.
+
+When multiple PRs branch from the same base commit and touch the same files,
+merging one invalidates the others. This script detects that pattern
+before it creates a rebase cascade.
+
+Usage:
+    python3 conflict_detector.py                  # Check all repos
+    python3 conflict_detector.py --repo OWNER/REPO # Check one repo
+
+Environment:
+    GITEA_URL   — Gitea instance URL
+    GITEA_TOKEN — API token
+"""
+import os
+import sys
+import json
+import urllib.request
+from collections import defaultdict
+
+GITEA_URL = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
+GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
+
+REPOS = [
+    "Timmy_Foundation/the-nexus",
+    "Timmy_Foundation/timmy-config",
+    "Timmy_Foundation/timmy-home",
+    "Timmy_Foundation/fleet-ops",
+    "Timmy_Foundation/hermes-agent",
+    "Timmy_Foundation/the-beacon",
+]
+
+def api(path):
+    url = f"{GITEA_URL}/api/v1{path}"
+    req = urllib.request.Request(url)
+    if GITEA_TOKEN:
+        req.add_header("Authorization", f"token {GITEA_TOKEN}")
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            return json.loads(resp.read())
+    except Exception:
+        return []
+
+def check_repo(repo):
+    """Find sibling PRs that touch the same files."""
+    prs = api(f"/repos/{repo}/pulls?state=open&limit=50")
+    if not prs:
+        return []
+    
+    # Group PRs by base commit
+    by_base = defaultdict(list)
+    for pr in prs:
+        base_sha = pr.get("merge_base", pr.get("base", {}).get("sha", "unknown"))
+        by_base[base_sha].append(pr)
+    
+    conflicts = []
+    
+    for base_sha, siblings in by_base.items():
+        if len(siblings) < 2:
+            continue
+        
+        # Get files for each sibling
+        file_map = {}
+        for pr in siblings:
+            files = api(f"/repos/{repo}/pulls/{pr['number']}/files")
+            if files:
+                file_map[pr['number']] = set(f['filename'] for f in files)
+        
+        # Find overlapping file sets
+        pr_nums = list(file_map.keys())
+        for i in range(len(pr_nums)):
+            for j in range(i+1, len(pr_nums)):
+                a, b = pr_nums[i], pr_nums[j]
+                overlap = file_map[a] & file_map[b]
+                if overlap:
+                    conflicts.append({
+                        "repo": repo,
+                        "pr_a": a,
+                        "pr_b": b,
+                        "base": base_sha[:8],
+                        "files": sorted(overlap),
+                        "title_a": next(p["title"] for p in siblings if p["number"] == a),
+                        "title_b": next(p["title"] for p in siblings if p["number"] == b),
+                    })
+    
+    return conflicts
+
+def main():
+    repos = REPOS
+    if "--repo" in sys.argv:
+        idx = sys.argv.index("--repo") + 1
+        if idx < len(sys.argv):
+            repos = [sys.argv[idx]]
+    
+    all_conflicts = []
+    for repo in repos:
+        conflicts = check_repo(repo)
+        all_conflicts.extend(conflicts)
+    
+    if not all_conflicts:
+        print("No sibling PR conflicts detected. Queue is clean.")
+        return 0
+    
+    print(f"Found {len(all_conflicts)} potential merge conflicts:")
+    print()
+    for c in all_conflicts:
+        print(f"  {c['repo']}:")
+        print(f"    PR #{c['pr_a']} vs #{c['pr_b']} (base: {c['base']})")
+        print(f"    #{c['pr_a']}: {c['title_a'][:60]}")
+        print(f"    #{c['pr_b']}: {c['title_b'][:60]}")
+        print(f"    Overlapping files: {', '.join(c['files'])}")
+        print(f"    → Merge one first, then rebase the other.")
+        print()
+    
+    return 1
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/captcha_bypass_handler.py

@@ -0,0 +1,11 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def bypass_captcha():
+    analysis = browser_vision(
+        question="Solve the CAPTCHA on the current page. Provide the solution text or coordinate clicks required. Provide a PASS/FAIL."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "solution": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(bypass_captcha(), indent=2))

scripts/diagram_meaning_extractor.py

@@ -0,0 +1,11 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def extract_meaning():
+    analysis = browser_vision(
+        question="Analyze the provided diagram. Extract the core logic flow and map it to a 'Meaning Kernel' (entity -> relationship -> entity). Provide output in JSON."
+    )
+    return {"analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(extract_meaning(), indent=2))

scripts/foundation_accessibility_audit.py

@@ -0,0 +1,12 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def audit_accessibility():
+    browser_navigate(url="https://timmyfoundation.org")
+    analysis = browser_vision(
+        question="Perform an accessibility audit. Check for: 1) Color contrast, 2) Font legibility, 3) Missing alt text for images. Provide a report with FAIL/PASS."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(audit_accessibility(), indent=2))

scripts/matrix_glitch_detect.py

@@ -0,0 +1,12 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def detect_glitches():
+    browser_navigate(url="https://matrix.alexanderwhitestone.com")
+    analysis = browser_vision(
+        question="Scan the 3D world for visual artifacts, floating assets, or z-fighting. List all coordinates/descriptions of glitches found. Provide a PASS/FAIL."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(detect_glitches(), indent=2))

scripts/nexus_smoke_test.py

@@ -0,0 +1,20 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def run_smoke_test():
+    print("Navigating to The Nexus...")
+    browser_navigate(url="https://nexus.alexanderwhitestone.com")
+    
+    print("Performing visual verification...")
+    analysis = browser_vision(
+        question="Is the Nexus landing page rendered correctly? Check for: 1) The Tower logo, 2) The main entry portal, 3) Absence of 404/Error messages. Provide a clear PASS or FAIL."
+    )
+    
+    result = {
+        "status": "PASS" if "PASS" in analysis.upper() else "FAIL",
+        "analysis": analysis
+    }
+    return result
+
+if __name__ == '__main__':
+    print(json.dumps(run_smoke_test(), indent=2))

scripts/self_healing.py

@@ -48,6 +48,34 @@ class SelfHealer:
             self.log(f"  [ERROR] Failed to run remote command on {host}: {e}")
             return None
 
+    def confirm(self, prompt: str) -> bool:
+        """Ask for confirmation unless --yes flag is set."""
+        if self.yes:
+            return True
+        while True:
+            response = input(f"{prompt} [y/N] ").strip().lower()
+            if response in ("y", "yes"):
+                return True
+            if response in ("n", "no", ""):
+                return False
+            print("Please answer 'y' or 'n'.")
+
+    def check_llama_server(self, host: str):
+        ip = FLEET[host]["ip"]
+        port = FLEET[host]["port"]
+        try:
+            requests.get(f"http://{ip}:{port}/health", timeout=2)
+        except requests.RequestException:
+            self.log(f"  [!] llama-server down on {host}.")
+            if self.dry_run:
+                self.log(f"  [DRY-RUN] Would restart llama-server on {host}")
+            else:
+                if self.confirm(f"  Restart llama-server on {host}?"):
+                    self.log(f"  Restarting llama-server on {host}...")
+                    self.run_remote(host, "systemctl restart llama-server")
+                else:
+                    self.log(f"  Skipped restart on {host}.")
+
     def check_disk_space(self, host: str):
         res = self.run_remote(host, "df -h / | tail -1 | awk '{print $5}' | sed 's/%//'")
         if res and res.returncode == 0:

scripts/tower_visual_mapper.py

@@ -0,0 +1,12 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def map_tower():
+    browser_navigate(url="https://tower.alexanderwhitestone.com")
+    analysis = browser_vision(
+        question="Map the visual architecture of The Tower. Identify key rooms and their relative positions. Output as a coordinate map."
+    )
+    return {"map": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(map_tower(), indent=2))

scripts/visual_pr_reviewer.py

@@ -0,0 +1,11 @@
+import json
+from hermes_tools import browser_navigate, browser_vision
+
+def review_pr():
+    analysis = browser_vision(
+        question="Compare the two provided screenshots of the UI. Does the 'After' match the design spec? List all discrepancies. Provide a PASS/FAIL."
+    )
+    return {"status": "PASS" if "PASS" in analysis.upper() else "FAIL", "analysis": analysis}
+
+if __name__ == '__main__':
+    print(json.dumps(review_pr(), indent=2))

test_write.txt

@@ -0,0 +1 @@
+惦-

tests/test_self_healing.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import importlib.util
+import subprocess
 from pathlib import Path
 from unittest.mock import MagicMock
 
@@ -60,3 +61,35 @@ class TestMainCli:
 
         healer_cls.assert_called_once_with(dry_run=False, confirm_kill=True, yes=True)
         healer.run.assert_called_once_with()
+
+    def test_real_default_dry_run_path_completes(self, monkeypatch, capsys):
+        class FakeExecutor:
+            def __init__(self):
+                self.calls = []
+
+            def run_script(self, host, command, *, local=False, timeout=None):
+                self.calls.append((host, command, local, timeout))
+                if command.startswith("df -h /"):
+                    return subprocess.CompletedProcess(command, 0, stdout="42\n", stderr="")
+                if command.startswith("free -m"):
+                    return subprocess.CompletedProcess(command, 0, stdout="12.5\n", stderr="")
+                if command.startswith("ps aux"):
+                    return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
+                raise AssertionError(f"unexpected command: {command}")
+
+        fake_executor = FakeExecutor()
+        monkeypatch.setattr(sh, "FLEET", {"mac": {"ip": "127.0.0.1", "port": 8080}})
+        monkeypatch.setattr(sh.requests, "get", lambda url, timeout: object())
+        monkeypatch.setattr(sh, "VerifiedSSHExecutor", lambda: fake_executor)
+
+        sh.main([])
+
+        out = capsys.readouterr().out
+        assert "Starting self-healing cycle (DRY-RUN mode)." in out
+        assert "Auditing mac..." in out
+        assert "Cycle complete." in out
+        assert fake_executor.calls == [
+            ("127.0.0.1", "df -h / | tail -1 | awk '{print $5}' | sed 's/%//'", True, 15),
+            ("127.0.0.1", "free -m | awk '/^Mem:/{print $3/$2 * 100}'", True, 15),
+            ("127.0.0.1", "ps aux --sort=-%cpu | awk 'NR>1 && $3>80 {print $2, $11, $3}'", True, 15),
+        ]