Epic 5.6: Evolver.py Facade — P5a Capstone (80% reduction)

openspace/skill_engine/evolution/formatting.py

@@ -0,0 +1,77 @@
+"""Formatting helpers for evolution prompts.
+
+Pure functions that transform skill snapshots and execution analyses
+into text blocks suitable for LLM prompt inclusion.
+"""
+
+from __future__ import annotations
+
+from typing import List
+
+from ..patch import SKILL_FILENAME, collect_skill_snapshot
+from ..types import ExecutionAnalysis
+from .triggers import _ANALYSIS_CONTEXT_MAX
+
+
+def format_skill_dir_content(skill_dir) -> str:
+    """Format all text files in a skill directory for prompt inclusion.
+
+    Returns a multi-file listing if there are auxiliary files beyond
+    SKILL.md, or just the SKILL.md content for single-file skills.
+    """
+    files = collect_skill_snapshot(skill_dir)
+    if not files:
+        return ""
+
+    # Single-file skill: return just the content
+    if len(files) == 1 and SKILL_FILENAME in files:
+        return files[SKILL_FILENAME]
+
+    # Multi-file: format as directory listing
+    parts: list[str] = []
+    # SKILL.md first
+    if SKILL_FILENAME in files:
+        parts.append(f"### File: {SKILL_FILENAME}\n```markdown\n{files[SKILL_FILENAME]}\n```")
+    for name, content in sorted(files.items()):
+        if name == SKILL_FILENAME:
+            continue
+        parts.append(f"### File: {name}\n```\n{content}\n```")
+
+    return "\n\n".join(parts)
+
+
+def format_analysis_context(analyses: List[ExecutionAnalysis]) -> str:
+    """Format recent analyses into a concise context block for prompts."""
+    _ANALYSIS_NOTE_MAX_CHARS = 500  # Per-analysis note truncation
+
+    if not analyses:
+        return "(no execution history available)"
+
+    parts: List[str] = []
+    for a in analyses[:_ANALYSIS_CONTEXT_MAX]:
+        completed = "completed" if a.task_completed else "failed"
+
+        # Per-skill notes
+        skill_notes = []
+        for j in a.skill_judgments:
+            applied = "applied" if j.skill_applied else "NOT applied"
+            note = f"  - {j.skill_id}: {applied}"
+            if j.note:
+                note += f" — {j.note[:_ANALYSIS_NOTE_MAX_CHARS]}"
+            skill_notes.append(note)
+
+        # Tool issues
+        tool_lines = []
+        for issue in a.tool_issues[:3]:
+            tool_lines.append(f"  - {issue[:200]}")
+
+        block = f"### Task: {a.task_id} ({completed})\n"
+        if a.execution_note:
+            block += f"{a.execution_note[:_ANALYSIS_NOTE_MAX_CHARS]}\n"
+        if skill_notes:
+            block += "Skills:\n" + "\n".join(skill_notes) + "\n"
+        if tool_lines:
+            block += "Tool issues:\n" + "\n".join(tool_lines) + "\n"
+        parts.append(block)
+
+    return "\n".join(parts)

openspace/skill_engine/evolver.py

@@ -18,7 +18,7 @@
 
 import asyncio
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set
+from typing import TYPE_CHECKING, Dict, List, Optional, Set
 
 from openspace.utils.logging import Logger
 
@@ -36,6 +36,10 @@
     llm_confirm_evolution as _llm_confirm_evolution_impl,
     parse_confirmation as _parse_confirmation_impl,
 )
+from .evolution.formatting import (
+    format_analysis_context as _format_analysis_context_impl,
+    format_skill_dir_content as _format_skill_dir_content_impl,
+)
 from .evolution.loop import (
     EVOLUTION_COMPLETE,
     EVOLUTION_FAILED,
@@ -51,19 +55,14 @@
     evolve_fix as _evolve_fix_impl,
 )
 from .evolution.triggers import (
-    _ANALYSIS_CONTEXT_MAX,
     build_context_from_analysis as _build_context_from_analysis_impl,
     diagnose_skill_health as _diagnose_skill_health_impl,
     load_skill_content as _load_skill_content_impl,
     process_analysis as _process_analysis_impl,
     process_metric_check as _process_metric_check_impl,
     process_tool_degradation as _process_tool_degradation_impl,
 )
-from .patch import (
-    SKILL_FILENAME,
-    SkillEditResult,
-    collect_skill_snapshot,
-)
+from .patch import SkillEditResult
 from .store import SkillStore
 from .types import (
     EvolutionSuggestion,
@@ -297,68 +296,8 @@ def _load_skill_content(self, record: SkillRecord) -> str:
         """Load SKILL.md content from disk via registry or direct read."""
         return _load_skill_content_impl(self, record)
 
-    @staticmethod
-    def _format_skill_dir_content(skill_dir: Path) -> str:
-        """Format all text files in a skill directory for prompt inclusion.
+    _format_skill_dir_content = staticmethod(_format_skill_dir_content_impl)
 
-        Returns a multi-file listing if there are auxiliary files beyond
-        SKILL.md, or just the SKILL.md content for single-file skills.
-        """
-        files = collect_skill_snapshot(skill_dir)
-        if not files:
-            return ""
-
-        # Single-file skill: return just the content
-        if len(files) == 1 and SKILL_FILENAME in files:
-            return files[SKILL_FILENAME]
-
-        # Multi-file: format as directory listing
-        parts: list[str] = []
-        # SKILL.md first
-        if SKILL_FILENAME in files:
-            parts.append(f"### File: {SKILL_FILENAME}\n```markdown\n{files[SKILL_FILENAME]}\n```")
-        for name, content in sorted(files.items()):
-            if name == SKILL_FILENAME:
-                continue
-            parts.append(f"### File: {name}\n```\n{content}\n```")
-
-        return "\n\n".join(parts)
-
-    @staticmethod
-    def _format_analysis_context(analyses: List[ExecutionAnalysis]) -> str:
-        """Format recent analyses into a concise context block for prompts."""
-        _ANALYSIS_NOTE_MAX_CHARS = 500  # Per-analysis note truncation
-
-        if not analyses:
-            return "(no execution history available)"
-
-        parts: List[str] = []
-        for a in analyses[:_ANALYSIS_CONTEXT_MAX]:
-            completed = "completed" if a.task_completed else "failed"
-
-            # Per-skill notes
-            skill_notes = []
-            for j in a.skill_judgments:
-                applied = "applied" if j.skill_applied else "NOT applied"
-                note = f"  - {j.skill_id}: {applied}"
-                if j.note:
-                    note += f" — {j.note[:_ANALYSIS_NOTE_MAX_CHARS]}"
-                skill_notes.append(note)
-
-            # Tool issues
-            tool_lines = []
-            for issue in a.tool_issues[:3]:
-                tool_lines.append(f"  - {issue[:200]}")
-
-            block = f"### Task: {a.task_id} ({completed})\n"
-            if a.execution_note:
-                block += f"{a.execution_note[:_ANALYSIS_NOTE_MAX_CHARS]}\n"
-            if skill_notes:
-                block += "Skills:\n" + "\n".join(skill_notes) + "\n"
-            if tool_lines:
-                block += "Tool issues:\n" + "\n".join(tool_lines) + "\n"
-            parts.append(block)
-
-        return "\n".join(parts)
+    _format_analysis_context = staticmethod(_format_analysis_context_impl)
 
     _diagnose_skill_health = staticmethod(_diagnose_skill_health_impl)

tests/test_evolution_formatting.py

@@ -0,0 +1,190 @@
+"""Tests for openspace.skill_engine.evolution.formatting (Epic 5.6).
+
+Verifies:
+  1. format_skill_dir_content — single-file, multi-file, empty
+  2. format_analysis_context — with analyses, empty, truncation
+  3. Backward compat: SkillEvolver delegates to formatting functions
+  4. Logger namespace
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from openspace.skill_engine.evolution.formatting import (
+    format_analysis_context,
+    format_skill_dir_content,
+)
+
+
+# ---------------------------------------------------------------------------
+# format_skill_dir_content
+# ---------------------------------------------------------------------------
+
+class TestFormatSkillDirContent:
+    def test_empty_dir(self):
+        """Empty directory → empty string."""
+        with patch("openspace.skill_engine.evolution.formatting.collect_skill_snapshot", return_value={}):
+            result = format_skill_dir_content(Path("/fake"))
+        assert result == ""
+
+    def test_single_file_skill(self):
+        """Single SKILL.md → return content directly (no formatting)."""
+        with patch(
+            "openspace.skill_engine.evolution.formatting.collect_skill_snapshot",
+            return_value={"SKILL.md": "---\nname: test\n---\ncontent"},
+        ):
+            result = format_skill_dir_content(Path("/fake"))
+        assert result == "---\nname: test\n---\ncontent"
+
+    def test_multi_file_skill(self):
+        """Multiple files → formatted listing with SKILL.md first."""
+        files = {
+            "SKILL.md": "skill content",
+            "helper.py": "def foo(): pass",
+            "README.md": "readme text",
+        }
+        with patch(
+            "openspace.skill_engine.evolution.formatting.collect_skill_snapshot",
+            return_value=files,
+        ):
+            result = format_skill_dir_content(Path("/fake"))
+
+        # SKILL.md appears first
+        assert result.index("### File: SKILL.md") < result.index("### File: README.md")
+        assert result.index("### File: SKILL.md") < result.index("### File: helper.py")
+        # All files present
+        assert "skill content" in result
+        assert "def foo(): pass" in result
+        assert "readme text" in result
+
+    def test_multi_file_without_skill_md(self):
+        """Multi-file with no SKILL.md → still formats all files."""
+        files = {"helper.py": "code", "config.yaml": "key: val"}
+        with patch(
+            "openspace.skill_engine.evolution.formatting.collect_skill_snapshot",
+            return_value=files,
+        ):
+            result = format_skill_dir_content(Path("/fake"))
+        assert "### File: config.yaml" in result
+        assert "### File: helper.py" in result
+
+
+# ---------------------------------------------------------------------------
+# format_analysis_context
+# ---------------------------------------------------------------------------
+
+class _FakeSkillJudgment:
+    def __init__(self, skill_id, applied=True, note=""):
+        self.skill_id = skill_id
+        self.skill_applied = applied
+        self.note = note
+
+
+class _FakeAnalysis:
+    def __init__(self, task_id, completed=True, note="", judgments=None, tool_issues=None):
+        self.task_id = task_id
+        self.task_completed = completed
+        self.execution_note = note
+        self.skill_judgments = judgments or []
+        self.tool_issues = tool_issues or []
+
+
+class TestFormatAnalysisContext:
+    def test_empty_analyses(self):
+        """No analyses → fallback message."""
+        result = format_analysis_context([])
+        assert "no execution history" in result.lower()
+
+    def test_single_analysis(self):
+        """Single analysis → formatted block."""
+        analysis = _FakeAnalysis(
+            task_id="task-1",
+            completed=True,
+            note="Skill worked well",
+            judgments=[_FakeSkillJudgment("skill-a", applied=True, note="Good result")],
+        )
+        result = format_analysis_context([analysis])
+        assert "task-1" in result
+        assert "completed" in result
+        assert "skill-a" in result
+        assert "applied" in result
+        assert "Good result" in result
+
+    def test_failed_task(self):
+        """Failed task shows 'failed' label."""
+        analysis = _FakeAnalysis(task_id="task-2", completed=False)
+        result = format_analysis_context([analysis])
+        assert "failed" in result
+
+    def test_tool_issues_included(self):
+        """Tool issues appear in output."""
+        analysis = _FakeAnalysis(
+            task_id="task-3",
+            tool_issues=["Tool X timed out", "Tool Y returned error"],
+        )
+        result = format_analysis_context([analysis])
+        assert "Tool X timed out" in result
+        assert "Tool Y returned error" in result
+
+    def test_note_truncation(self):
+        """Execution notes are truncated to 500 chars."""
+        long_note = "x" * 1000
+        analysis = _FakeAnalysis(task_id="task-4", note=long_note)
+        result = format_analysis_context([analysis])
+        # The note should be truncated — full 1000 chars should NOT appear
+        assert "x" * 501 not in result
+
+    def test_not_applied_skill(self):
+        """Skill not applied shows 'NOT applied'."""
+        analysis = _FakeAnalysis(
+            task_id="task-5",
+            judgments=[_FakeSkillJudgment("skill-b", applied=False, note="Didn't match")],
+        )
+        result = format_analysis_context([analysis])
+        assert "NOT applied" in result
+
+
+# ---------------------------------------------------------------------------
+# Backward compat
+# ---------------------------------------------------------------------------
+
+class TestDelegationSeam:
+    def test_format_skill_dir_is_staticmethod(self):
+        from openspace.skill_engine.evolver import SkillEvolver
+        assert isinstance(
+            SkillEvolver.__dict__["_format_skill_dir_content"],
+            staticmethod,
+        )
+
+    def test_format_analysis_is_staticmethod(self):
+        from openspace.skill_engine.evolver import SkillEvolver
+        assert isinstance(
+            SkillEvolver.__dict__["_format_analysis_context"],
+            staticmethod,
+        )
+
+    def test_format_skill_dir_callable(self):
+        from openspace.skill_engine.evolver import SkillEvolver
+        with patch(
+            "openspace.skill_engine.evolution.formatting.collect_skill_snapshot",
+            return_value={"SKILL.md": "test"},
+        ):
+            result = SkillEvolver._format_skill_dir_content(Path("/fake"))
+        assert result == "test"
+
+
+# ---------------------------------------------------------------------------
+# Constants / logger
+# ---------------------------------------------------------------------------
+
+class TestMeta:
+    def test_module_size(self):
+        """formatting.py should stay under 120 lines."""
+        import openspace.skill_engine.evolution.formatting as mod
+        src = Path(mod.__file__)
+        lines = src.read_text(encoding="utf-8").splitlines()
+        assert len(lines) <= 120, f"formatting.py has {len(lines)} lines (limit 120)"