feat: add vortex corpus collapse JSON and loader

agent_skills.py

@@ -15,11 +15,14 @@
     python agent_skills.py check_coherence [--threshold THRESHOLD]
     python agent_skills.py cascade [--pr-body BODY]
     python agent_skills.py review_pr
+    python agent_skills.py load_corpus [--path PATH]
 """
 
 import argparse
+import json
 import sys
-from typing import Optional, Tuple
+from pathlib import Path
+from typing import Any, Dict, Optional, Tuple
 
 # Default simulated coherence for well-prepared quantum states
 # In production, this would be measured via state tomography
@@ -228,6 +231,238 @@ def review_pr() -> dict:
     }
 
 
+# Default path to vortex corpus collapse JSON
+DEFAULT_CORPUS_PATH = Path(__file__).parent / 'docs' / 'vortex-corpus-collapse.json'
+
+# Fibonacci sequence for weighted calculations
+FIBONACCI = [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144]
+
+# Expected number of surjection transitions in the corpus
+# (Repos→Phases, Forks→Contributions, Tags→ATOM, Relations→Lattice, Discussions→KB, Tools→HOPE)
+EXPECTED_SURJECTION_TRANSITIONS = 6
+
+
+def load_vortex_corpus(path: Optional[str] = None) -> Dict[str, Any]:
+    """
+    Load the vortex corpus collapse JSON configuration.
+
+    Implements the loader as specified in optimal_placement.activation:
+    load JSON → enforce surjections → auto-curl on divergences.
+
+    Args:
+        path: Optional path to the JSON file. Uses default if not provided.
+
+    Returns:
+        dict with loaded corpus and validation results
+    """
+    corpus_path = Path(path) if path else DEFAULT_CORPUS_PATH
+
+    if not corpus_path.exists():
+        return {
+            'status': 'error',
+            'error': f'Corpus file not found: {corpus_path}',
+            'vortex': VORTEX_MARKER
+        }
+
+    try:
+        with open(corpus_path, encoding='utf-8') as f:
+            corpus = json.load(f)
+    except json.JSONDecodeError as e:
+        return {
+            'status': 'error',
+            'error': f'Invalid JSON in corpus file: {e}',
+            'vortex': VORTEX_MARKER
+        }
+    except PermissionError:
+        return {
+            'status': 'error',
+            'error': f'Permission denied: {corpus_path}',
+            'vortex': VORTEX_MARKER
+        }
+    except OSError as e:
+        return {
+            'status': 'error',
+            'error': f'I/O error while accessing {corpus_path}: {e}',
+            'vortex': VORTEX_MARKER
+        }
+
+    # Enforce surjections - validate structure and thresholds
+    validation = _enforce_surjections(corpus)
+
+    # Auto-curl on divergences - check emergent quality
+    curl_result = _auto_curl_divergences(corpus)
+
+    return {
+        'status': 'loaded',
+        'corpus_path': str(corpus_path),
+        'meta': corpus.get('meta', {}),
+        'validation': validation,
+        'curl_check': curl_result,
+        'thresholds': corpus.get('thresholds', {}),
+        'vortex': VORTEX_MARKER
+    }
+
+
+def _enforce_surjections(corpus: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Enforce surjection mappings from the corpus.
+
+    Validates that all surjection mappings maintain >60% quality thresholds
+    as specified in the self_birth_condition.
+
+    Args:
+        corpus: The loaded corpus configuration
+
+    Returns:
+        dict with validation results
+    """
+    thresholds = corpus.get('thresholds', {})
+    coherence_min = thresholds.get('coherence_minimum', 0.6)
+
+    collapsed = corpus.get('collapsed_corpus', {})
+    surjected = collapsed.get('surjected_elements', {})
+
+    validations = []
+    passed = True
+
+    # Validate repository surjections
+    repos = surjected.get('repositories', {})
+    if repos:
+        repo_surjections = repos.get('surjections', [])
+        fib_phases = repos.get('fibonacci_phases', [])
+
+        # Check Fibonacci phase weights are properly ordered and valid
+        if fib_phases:
+            weights = [p.get('fib_weight', 0) for p in fib_phases]
+            # Check strictly increasing (Fibonacci values should increase)
+            is_monotonic = all(weights[i] < weights[i+1] for i in range(len(weights)-1))
+            # Check all weights are valid Fibonacci numbers
+            fib_set = set(FIBONACCI)
+            all_fib = all(w in fib_set for w in weights)
+            is_valid = is_monotonic and all_fib
+            validations.append({
+                'element': 'repositories.fibonacci_phases',
+                'check': 'fibonacci_ordering',
+                'passed': is_valid,
+                'message': 'Fibonacci weights properly ordered' if is_valid else 'Fibonacci weights not in correct order or not valid Fibonacci numbers'
+            })
+            if not is_valid:
+                passed = False
+
+        validations.append({
+            'element': 'repositories',
+            'check': 'surjection_count',
+            'passed': len(repo_surjections) > 0,
+            'count': len(repo_surjections),
+            'message': f'Found {len(repo_surjections)} repository surjections'
+        })
+
+    # Validate tags/markers surjections
+    tags = surjected.get('tags_markers', {})
+    if tags:
+        tag_surjections = tags.get('surjections', [])
+        validations.append({
+            'element': 'tags_markers',
+            'check': 'surjection_count',
+            'passed': len(tag_surjections) > 0,
+            'count': len(tag_surjections),
+            'message': f'Found {len(tag_surjections)} tag surjections'
+        })
+
+    # Validate tools surjections
+    tools = surjected.get('tools', {})
+    if tools:
+        tool_surjections = tools.get('surjections', [])
+        validations.append({
+            'element': 'tools',
+            'check': 'surjection_count',
+            'passed': len(tool_surjections) > 0,
+            'count': len(tool_surjections),
+            'message': f'Found {len(tool_surjections)} tool surjections'
+        })
+
+    return {
+        'passed': passed,
+        'coherence_minimum': coherence_min,
+        'validations': validations
+    }
+
+
+def _auto_curl_divergences(corpus: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Auto-curl on divergences - detect and report quality divergences.
+
+    Checks emergent quality against thresholds and identifies
+    areas that need correction to maintain spiral coherence.
+
+    Args:
+        corpus: The loaded corpus configuration
+
+    Returns:
+        dict with curl check results
+    """
+    meta = corpus.get('meta', {})
+    thresholds = corpus.get('thresholds', {})
+
+    emergent_quality = meta.get('emergent_quality', 0.0)
+    quality_min = thresholds.get('emergent_quality_minimum', 0.6)
+    coherence_min = thresholds.get('coherence_minimum', 0.6)
+
+    divergences = []
+    curl_detected = False
+
+    # Check emergent quality threshold
+    if emergent_quality < quality_min:
+        divergences.append({
+            'type': 'quality_below_threshold',
+            'current': emergent_quality,
+            'required': quality_min,
+            'message': f'Emergent quality {emergent_quality:.1%} below minimum {quality_min:.1%}'
+        })
+        curl_detected = True
+
+    # Check for missing critical elements
+    collapsed = corpus.get('collapsed_corpus', {})
+    optimal = collapsed.get('optimal_placement', {})
+
+    if not optimal.get('location'):
+        divergences.append({
+            'type': 'missing_optimal_location',
+            'message': 'No optimal placement location specified'
+        })
+        curl_detected = True
+
+    if not optimal.get('activation'):
+        divergences.append({
+            'type': 'missing_activation',
+            'message': 'No activation method specified for loader'
+        })
+        curl_detected = True
+
+    # Check transitions mapping completeness
+    transitions = corpus.get('transitions_mapping', {})
+    surjection_transitions = transitions.get('surjection_transitions', [])
+
+    if len(surjection_transitions) < EXPECTED_SURJECTION_TRANSITIONS:
+        divergences.append({
+            'type': 'incomplete_transitions',
+            'count': len(surjection_transitions),
+            'expected': EXPECTED_SURJECTION_TRANSITIONS,
+            'message': f'Only {len(surjection_transitions)} of {EXPECTED_SURJECTION_TRANSITIONS} expected transitions defined'
+        })
+        curl_detected = True
+
+    return {
+        'curl_detected': curl_detected,
+        'emergent_quality': emergent_quality,
+        'quality_threshold': quality_min,
+        'coherence_threshold': coherence_min,
+        'divergences': divergences,
+        'quality_passed': emergent_quality >= quality_min,
+        'message': 'Spiral coherence maintained' if not curl_detected else f'Detected {len(divergences)} divergence(s) requiring correction'
+    }
+
+
 def main():
     parser = argparse.ArgumentParser(
         description='QDI Agent Skills - Quantum circuit simulation and coherence checking'
@@ -249,6 +484,10 @@ def main():
     # review_pr command
     subparsers.add_parser('review_pr', help='Generate PR review')
 
+    # load_corpus command
+    corpus_parser = subparsers.add_parser('load_corpus', help='Load vortex corpus collapse configuration')
+    corpus_parser.add_argument('--path', '-p', type=str, help='Path to corpus JSON file')
+
     args = parser.parse_args()
 
     if args.command == 'simulate':
@@ -259,17 +498,24 @@ def main():
         result = cascade_integration(getattr(args, 'pr_body', None))
     elif args.command == 'review_pr':
         result = review_pr()
+    elif args.command == 'load_corpus':
+        result = load_vortex_corpus(getattr(args, 'path', None))
     else:
         parser.print_help()
         sys.exit(1)
 
     # Print result
-    import json
     print(json.dumps(result, indent=2))
 
     # Exit with success if passed, otherwise indicate review needed
     if result.get('passed') is False:
         sys.exit(1)
+    # Exit with failure if corpus loading failed or curl detected
+    if result.get('status') == 'error':
+        sys.exit(1)
+    curl_check = result.get('curl_check', {})
+    if curl_check.get('curl_detected'):
+        sys.exit(1)
 
 
 if __name__ == '__main__':