refactor(docs): code analysis engine

changes:
  - file: file_analyzer.py
    area: analyzer
    modified: [analyze_directory, FileAnalyzer]
  - file: generator.py
    area: analyzer
    modified: [_make_serializable, PlanfileGenerator]
  - file: yaml_parser.py
    area: analyzer
    modified: [extract_from_yaml_structure]
  - file: models.py
    area: model
    modified: [to_yaml, export, Strategy]
  - file: store.py
    area: core
    modified: [_write_yaml, PlanfileStore]
  - file: state.py
    area: core
    modified: [SyncState, save_sync]

stats:
  lines: "+413/-330 (net +83)"
  files: 19
  complexity: "Large structural change (normalized)"

Author: tom-sapletta-com

CHANGELOG.md

@@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.1.49] - 2026-03-27
+
+### Docs
+- Update project/context.md
+
+### Other
+- Update planfile/analysis/file_analyzer.py
+- Update planfile/analysis/generator.py
+- Update planfile/analysis/parsers/yaml_parser.py
+- Update planfile/core/models.py
+- Update planfile/core/store.py
+- Update planfile/sync/state.py
+- Update project.sh
+- Update project/analysis.toon.yaml
+- Update project/calls.mmd
+- Update project/calls.png
+- ... and 8 more files
+
 ## [0.1.48] - 2026-03-27
 
 ### Docs

VERSION

@@ -1 +1 @@
-0.1.48
+0.1.49

planfile/__init__.py

@@ -8,7 +8,7 @@
 - CLI and API for applying and reviewing strategies
 """
 
-__version__ = "0.1.48"
+__version__ = "0.1.49"
 __author__ = "Tom Sapletta"
 __email__ = "tom@sapletta.com"
 

planfile/analysis/file_analyzer.py

@@ -82,7 +82,11 @@ def analyze_directory(self, directory: Path, patterns: List[str] = None) -> Dict
         for pattern in patterns:
             for file_path in directory.rglob(pattern):
                 # Skip hidden files and common exclusions
-                if file_path.name.startswith('.') or any(skip in str(file_path) for skip in ['__pycache__', '.git', 'node_modules', '.pytest_cache']):
+                if file_path.name.startswith('.') or any(skip in str(file_path) for skip in ['__pycache__', '.git', 'node_modules', '.pytest_cache', '.planfile_analysis']):
+                    continue
+                
+                # Skip analysis files to prevent recursive analysis
+                if 'analysis_summary.json' in file_path.name or 'local-strategy.yaml' in file_path.name:
                     continue
 
                 issues, metrics, tasks = self.analyze_file(file_path)

planfile/analysis/generator.py

@@ -261,16 +261,46 @@ def _generate_success_criteria(self, metrics: Dict[str, Any]) -> List[str]:
     def _create_strategy_object(self, strategy_data: Dict[str, Any]) -> Strategy:
         return strategy_data
 
-    def _make_serializable(self, obj: Any) -> Any:
+    def _make_serializable(self, obj: Any, visited: set = None) -> Any:
+        """Convert object to serializable format with cycle detection."""
+        if visited is None:
+            visited = set()
+        
+        # Prevent infinite recursion
+        obj_id = id(obj)
+        if obj_id in visited:
+            return f"<circular_reference_{obj_id}>"
+        visited.add(obj_id)
+        
         if hasattr(obj, '__dict__'):
-            return {k: self._make_serializable(v) for k, v in obj.__dict__.items()}
+            result = {}
+            for k, v in obj.__dict__.items():
+                # Skip private attributes and large data structures
+                if k.startswith('_') or k in ['content', 'file_contents', 'raw_data']:
+                    continue
+                result[k] = self._make_serializable(v, visited)
+            return result
         elif isinstance(obj, dict):
-            return {k: self._make_serializable(v) for k, v in obj.items()}
+            result = {}
+            for k, v in obj.items():
+                # Skip large values
+                if isinstance(v, str) and len(v) > 1000:
+                    result[k] = f"<string_length_{len(v)}>"
+                else:
+                    result[k] = self._make_serializable(v, visited)
+            return result
         elif isinstance(obj, list):
-            return [self._make_serializable(item) for item in obj]
+            # Limit list size to prevent bloat
+            if len(obj) > 100:
+                return f"<list_length_{len(obj)}>"
+            return [self._make_serializable(item, visited) for item in obj]
+        elif isinstance(obj, (str, int, float, bool)) or obj is None:
+            if isinstance(obj, str) and len(obj) > 1000:
+                return f"<string_length_{len(obj)}>"
+            return obj
         elif hasattr(obj, '__name__'):
             return str(obj)
         else:
-            return obj
+            return f"<type_{type(obj).__name__}>"
 
 generator = PlanfileGenerator()

planfile/analysis/parsers/yaml_parser.py

@@ -7,31 +7,45 @@
 from planfile.analysis.parsers.text_parser import analyze_text
 from planfile.analysis.parsers.toon_parser import analyze_toon
 
-def extract_from_yaml_structure(data: Any, path: str, parent_key: str = "") -> List[ExtractedIssue]:
-    """Extract issues from YAML structure."""
+def extract_from_yaml_structure(data: Any, path: str, parent_key: str = "", visited: set = None) -> List[ExtractedIssue]:
+    """Extract issues from YAML structure with recursion protection."""
+    if visited is None:
+        visited = set()
+    
     issues = []
 
+    # Prevent infinite recursion
+    if id(data) in visited:
+        return issues
+    visited.add(id(data))
+    
     if isinstance(data, dict):
         for key, value in data.items():
             full_key = f"{parent_key}.{key}" if parent_key else key
 
-            # Look for common issue indicators
-            if isinstance(value, str):
+            # Skip if we're already processing issues (prevent self-reference)
+            if 'issues' in full_key.lower():
+                continue
+            
+            # Look for common issue indicators, but not in our own generated content
+            if isinstance(value, str) and len(value) < 500:  # Limit string length
                 if any(keyword in value.lower() for keyword in ['error', 'fail', 'bug', 'issue']):
-                    issues.append(ExtractedIssue(
-                        title=f"Issue in {full_key}",
-                        description=value,
-                        priority="medium",
-                        category="bug",
-                        file_path=path
-                    ))
+                    # Skip if this looks like our own generated issue
+                    if not any(skip in value.lower() for skip in ['extractedissue', 'file_path', 'priority:', 'category:']):
+                        issues.append(ExtractedIssue(
+                            title=f"Issue in {full_key}",
+                            description=value[:200],  # Limit description length
+                            priority="medium",
+                            category="bug",
+                            file_path=path
+                        ))
 
-            # Recurse
-            issues.extend(extract_from_yaml_structure(value, path, full_key))
+            # Recurse with protection
+            issues.extend(extract_from_yaml_structure(value, path, full_key, visited))
 
     elif isinstance(data, list):
         for i, item in enumerate(data):
-            issues.extend(extract_from_yaml_structure(item, path, f"{parent_key}[{i}]"))
+            issues.extend(extract_from_yaml_structure(item, path, f"{parent_key}[{i}]", visited))
 
     return issues
 

planfile/core/models.py

@@ -354,7 +354,12 @@ def merge(self, others: List['Strategy'], name: str = None) -> 'Strategy':
     def export(self, format: str = 'yaml') -> str:
         """Export strategy to specified format."""
         if format.lower() == 'yaml':
-            return yaml.dump(self.model_dump(), default_flow_style=False, sort_keys=False)
+            try:
+                return yaml.safe_dump(self.model_dump(), default_flow_style=False, sort_keys=False)
+            except Exception as e:
+                # Fallback to regular dump if safe_dump fails
+                print(f"Warning: safe_dump failed in export, using regular dump: {e}")
+                return yaml.dump(self.model_dump(), default_flow_style=False, sort_keys=False)
         elif format.lower() == 'json':
             import json
             return json.dumps(self.model_dump(), indent=2, default=str)
@@ -445,7 +450,12 @@ def _convert_old_format(data: Dict) -> Dict:
 
     def to_yaml(self) -> str:
         """Export to YAML with clean formatting."""
-        return yaml.dump(self.model_dump(exclude_none=True), default_flow_style=False, sort_keys=False)
+        try:
+            return yaml.safe_dump(self.model_dump(exclude_none=True), default_flow_style=False, sort_keys=False)
+        except Exception as e:
+            # Fallback to regular dump if safe_dump fails
+            print(f"Warning: safe_dump failed in to_yaml, using regular dump: {e}")
+            return yaml.dump(self.model_dump(exclude_none=True), default_flow_style=False, sort_keys=False)
 
 
 # ─── Ticket types (new in Sprint 3) ───

planfile/core/store.py

@@ -308,11 +308,16 @@ def _write_yaml(self, path: Path, data: dict):
         path.parent.mkdir(parents=True, exist_ok=True)
         lock = FileLock(str(path) + ".lock", timeout=5)
         with lock:
-            path.write_text(
-                yaml.dump(data, default_flow_style=False,
-                          allow_unicode=True, sort_keys=False),
-                encoding="utf-8",
-            )
+            try:
+                # Use safe_dump to prevent circular reference issues
+                content = yaml.safe_dump(data, default_flow_style=False,
+                                         allow_unicode=True, sort_keys=False)
+            except Exception as e:
+                # Fallback to regular dump if safe_dump fails
+                print(f"Warning: safe_dump failed, using regular dump: {e}")
+                content = yaml.dump(data, default_flow_style=False,
+                                   allow_unicode=True, sort_keys=False)
+            path.write_text(content, encoding="utf-8")
 
     def load_sprint(self, sprint: str = "current") -> dict:
         """Load sprint data as dictionary."""

planfile/sync/state.py

@@ -22,10 +22,14 @@ def save_sync(self, ticket_map: dict):
         state["ticket_map"] = {**state.get("ticket_map", {}), **ticket_map}
         state["synced_at"] = datetime.utcnow().isoformat()
         self.state_file.parent.mkdir(parents=True, exist_ok=True)
-        self.state_file.write_text(
-            yaml.dump(state, default_flow_style=False, sort_keys=False),
-            encoding="utf-8",
-        )
+        try:
+            # Use safe_dump to prevent circular reference issues
+            content = yaml.safe_dump(state, default_flow_style=False, sort_keys=False)
+        except Exception as e:
+            # Fallback to regular dump if safe_dump fails
+            print(f"Warning: safe_dump failed in save_sync, using regular dump: {e}")
+            content = yaml.dump(state, default_flow_style=False, sort_keys=False)
+        self.state_file.write_text(content, encoding="utf-8")
 
     def get_remote_id(self, local_id: str) -> str | None:
         """Look up remote ID for a local ticket."""

project.sh

@@ -1,6 +1,7 @@
 #!/usr/bin/env bash
 clear
 pip install -e .
+venv/bin/pip install prefact --upgrade
 venv/bin/pip install vallm --upgrade
 venv/bin/pip install redup --upgrade
 venv/bin/pip install glon --upgrade
@@ -19,4 +20,5 @@ venv/bin/redup scan . --format toon --output ./project
 #venv/bin/redup scan . --functions-only -f toon --output ./project
 #venv/bin/vallm batch ./src --recursive --semantic --model qwen2.5-coder:7b
 #vallm batch --parallel .
-venv/bin/vallm batch . --recursive --format toon --output ./project
+venv/bin/vallm batch . --recursive --format toon --output ./project
+venv/bin/prefact -a