feat: Add advanced Decision Engine and unit tests

SteveBlackbeard · SteveBlackbeard · commit 0d7da252bebc · 2026-04-01T23:04:31.000-06:00
diff --git a/CONTINUITY LEGACY Pro/tests/test_logic.py b/CONTINUITY LEGACY Pro/tests/test_logic.py
@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""
+test_logic.py — CONTINUITY LEGACY Pro Tests
+===========================================
+Advanced Unit Tests for the Continuity Legacy Framework.
+
+Covers:
+- Decision Engine momentum parsing
+- Continuity Cycle edge cases
+- State JSON validations
+
+Uses Python's standard `unittest` framework.
+"""
+
+import unittest
+import json
+import tempfile
+import os
+import datetime
+from pathlib import Path
+
+# Import the modules to test (assuming they are in path or relative)
+import sys
+# Add tools/continuity_legacy to python path for testing
+current_dir = Path(__file__).parent
+tools_dir = current_dir.parent / "tools" / "continuity_legacy"
+sys.path.append(str(tools_dir))
+
+# Attempt to import decision_engine. If it fails, tests will error appropriately.
+try:
+    import decision_engine
+except ImportError:
+    decision_engine = None
+
+class TestDecisionEngine(unittest.TestCase):
+    
+    def setUp(self):
+        """Create a temporary directory simulating a project root."""
+        self.test_dir = tempfile.TemporaryDirectory()
+        self.root_path = Path(self.test_dir.name)
+        
+    def tearDown(self):
+        self.test_dir.cleanup()
+        
+    def test_momentum_high(self):
+        """Test high momentum scoring for recent updates."""
+        if not decision_engine:
+            self.skipTest("decision_engine.py module not found.")
+            
+        recent_time = datetime.datetime.now(datetime.timezone.utc).isoformat()
+        state_data = {"last_update": recent_time}
+        
+        result = decision_engine.evaluate_momentum(state_data)
+        self.assertEqual(result["score"], 95, "Recent update should score 95.")
+        
+    def test_momentum_stale(self):
+        """Test stale momentum scoring for 30+ day old updates."""
+        if not decision_engine:
+            self.skipTest("decision_engine.py module not found.")
+            
+        old_time = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=35)).isoformat()
+        state_data = {"last_update": old_time}
+        
+        result = decision_engine.evaluate_momentum(state_data)
+        self.assertEqual(result["score"], 20, "Stale update should score 20.")
+        
+    def test_tactical_directive_extraction(self):
+        """Test parsing of the LIVE_HANDOFF.md file for next actions."""
+        if not decision_engine:
+            self.skipTest("decision_engine.py module not found.")
+            
+        continuity_dir = self.root_path / ".continuity"
+        continuity_dir.mkdir(parents=True)
+        
+        handoff_file = continuity_dir / "LIVE_HANDOFF.md"
+        handoff_content = """# LIVE HANDOFF
+        
+## Next Exact Action
+Fix the unit tests in the tests folder.
+"""
+        handoff_file.write_text(handoff_content, encoding="utf-8")
+        
+        result = decision_engine.extract_tactical_directive(self.root_path)
+        self.assertIn("Fix the unit tests", result, "Engine failed to extract the exact action.")
+
+    def test_missing_files_handling(self):
+        """Test graceful degradation when canonical files are missing."""
+        if not decision_engine:
+            self.skipTest("decision_engine.py module not found.")
+            
+        # Empty repo root, STATE.json is missing
+        state = decision_engine.parse_state(self.root_path)
+        self.assertEqual(state["status"], "error")
+        self.assertIn("not found", state["message"])
+        
+        # LIVE_HANDOFF missing
+        tactical = decision_engine.extract_tactical_directive(self.root_path)
+        self.assertIn("CRITICAL", tactical)
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/CONTINUITY LEGACY Pro/tools/continuity_legacy/decision_engine.py b/CONTINUITY LEGACY Pro/tools/continuity_legacy/decision_engine.py
@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""
+decision_engine.py — CONTINUITY LEGACY Pro
+==========================================
+Advanced Strategic Decision Engine (Alpha).
+
+This module analyzes the canonical memory files (STATE.json, ROADMAP.md, LIVE_HANDOFF.md)
+to identify technical debt, project momentum stalls, and missing contexts.
+It outputs actionable strategic recommendations for the human / AI developer team.
+
+Features:
+- State staleness detection
+- Roadmap alignment validation
+- Tactical extraction from LIVE_HANDOFF.md
+"""
+
+import json
+import datetime
+from pathlib import Path
+
+def parse_state(repo_root: Path) -> dict:
+    """Safely loads and parses STATE.json."""
+    state_file = repo_root / "STATE.json"
+    if not state_file.exists():
+        return {"status": "error", "message": "STATE.json not found."}
+    try:
+        with open(state_file, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception as e:
+        return {"status": "error", "message": f"Malformed STATE.json: {e}"}
+
+def evaluate_momentum(state_data: dict) -> dict:
+    """Evaluates project momentum based on the last update timestamp."""
+    try:
+        last_update_str = state_data.get("last_update", state_data.get("generated_at", ""))
+        if not last_update_str:
+            return {"score": 50, "insight": "No timestamp found in STATE.json."}
+            
+        # Parse ISO format handling potential timezone strings (basic implementation)
+        clean_str = last_update_str.replace("Z", "+00:00")
+        last_update = datetime.datetime.fromisoformat(clean_str)
+        
+        # Calculate delta
+        now = datetime.datetime.now(datetime.timezone.utc)
+        delta = now - last_update
+        
+        if delta.days > 30:
+            return {"score": 20, "insight": "Project stale. Over 30 days without updates."}
+        elif delta.days > 7:
+            return {"score": 60, "insight": "Momentum slowing. Over a week since last update."}
+        else:
+            return {"score": 95, "insight": "High momentum. Active development detected."}
+    except Exception:
+        return {"score": 50, "insight": "Could not parse last update time."}
+
+def extract_tactical_directive(repo_root: Path) -> str:
+    """Extracts the immediate next action from LIVE_HANDOFF.md."""
+    handoff_file = repo_root / ".continuity" / "LIVE_HANDOFF.md"
+    if not handoff_file.exists():
+        return "CRITICAL: LIVE_HANDOFF.md missing. Next exact action is undefined."
+    
+    try:
+        content = handoff_file.read_text(encoding="utf-8")
+        lines = content.splitlines()
+        for idx, line in enumerate(lines):
+            if "Next Exact Action" in line:
+                # Capture the next non-empty line as the directive
+                for next_line in lines[idx+1:idx+5]:
+                    stripped = next_line.strip()
+                    if stripped and not stripped.startswith("#"):
+                        return f"Directive found: {stripped}"
+        return "Warning: Next Exact Action header found but no actionable step defined."
+    except Exception:
+        return "Error reading LIVE_HANDOFF.md."
+
+def generate_strategy(repo_root: Path) -> dict:
+    """Orchestrates the decision engine analysis."""
+    print("[*] Engine Initializing: Strategic Analysis...")
+    
+    state = parse_state(repo_root)
+    if state.get("status") == "error":
+        print(f"[!] {state['message']}")
+        return state
+        
+    momentum = evaluate_momentum(state)
+    tactical = extract_tactical_directive(repo_root)
+    
+    # Formulate recommendation
+    recommendation = "Continue execution of the roadmap."
+    if momentum["score"] < 50:
+        recommendation = "PRIORITY: Review STATE.json. Update current context to regain momentum."
+    elif "missing" in tactical.lower() or "error" in tactical.lower():
+        recommendation = "PRIORITY: Re-establish the LIVE_HANDOFF.md to define the next exact action for the AI agent."
+        
+    report = {
+        "status": "success",
+        "generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
+        "momentum_score": momentum["score"],
+        "momentum_insight": momentum["insight"],
+        "tactical_directive": tactical,
+        "strategic_recommendation": recommendation
+    }
+    
+    # Write output
+    out_dir = repo_root / "outputs" / "continuity"
+    out_dir.mkdir(parents=True, exist_ok=True)
+    out_file = out_dir / "strategic_decision.json"
+    
+    with open(out_file, "w", encoding="utf-8") as f:
+        json.dump(report, f, indent=2)
+        
+    print(f"[+] Advanced Strategy Report generated at: {out_file.relative_to(repo_root)}")
+    print(f"    -> Momentum: {momentum['score']}/100")
+    print(f"    -> Strategy: {recommendation}")
+    
+    return report
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser(description="CONTINUITY LEGACY Pro - Decision Engine")
+    parser.add_argument("--repo-root", default=".", help="Root directory of the project")
+    args = parser.parse_args()
+    
+    root_path = Path(args.repo_root).resolve()
+    generate_strategy(root_path)
