diff --git a/openspace/skill_engine/evolution/formatting.py b/openspace/skill_engine/evolution/formatting.py new file mode 100644 index 00000000..3480baac --- /dev/null +++ b/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) diff --git a/openspace/skill_engine/evolver.py b/openspace/skill_engine/evolver.py index 0204a791..5e97541d 100644 --- a/openspace/skill_engine/evolver.py +++ b/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,7 +55,6 @@ 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, @@ -59,11 +62,7 @@ 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) diff --git a/tests/test_evolution_formatting.py b/tests/test_evolution_formatting.py new file mode 100644 index 00000000..a45ce744 --- /dev/null +++ b/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)"