q792602257
a279441aee
- divergence_tolerance was loaded from config but diff_flow_versions never consulted it, so structural_divergence was always plain sequence-equality regardless of the configured tolerance. Now actually applied per design.md D4. - The skill-catalog-subscription isolation guard test asserted 'skills.catalog' not in sys.modules, which is vacuously true since that module doesn't exist anywhere yet. Replaced with a real check against store.py's actual imports. openspec: skill-versioning capability, archived change skill-learning-runtime
133 lines
4.6 KiB
Python
133 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from skills_learning.config import SkillAuthoringConfig
|
|
from skills_learning.models import FlowStep, FlowTemplateSkill, SkillMetadata
|
|
from skills_learning.store import SkillStore
|
|
from skills_learning.versioning import diff_flow_versions, store_synthesized_skill
|
|
|
|
|
|
def _skill(
|
|
*,
|
|
name: str = "search",
|
|
steps: list[FlowStep] | None = None,
|
|
parameters: dict[str, dict[str, object]] | None = None,
|
|
) -> FlowTemplateSkill:
|
|
return FlowTemplateSkill(
|
|
metadata=SkillMetadata(
|
|
name=name,
|
|
description="Learned search",
|
|
originating_goal="search coffee",
|
|
),
|
|
steps=steps or [FlowStep("input_text", {"text": "coffee"})],
|
|
parameters=parameters or {},
|
|
)
|
|
|
|
|
|
def test_diff_flow_versions_detects_extra_missing_and_reordered_steps() -> None:
|
|
stored = [FlowStep("tap"), FlowStep("input_text")]
|
|
|
|
assert diff_flow_versions(stored, [FlowStep("tap")]).structural_divergence
|
|
assert diff_flow_versions(
|
|
stored,
|
|
[FlowStep("tap"), FlowStep("input_text"), FlowStep("tap")],
|
|
).structural_divergence
|
|
assert diff_flow_versions(
|
|
stored,
|
|
[FlowStep("input_text"), FlowStep("tap")],
|
|
).structural_divergence
|
|
|
|
|
|
def test_argument_only_difference_updates_existing_version_without_bump() -> None:
|
|
store = SkillStore()
|
|
first = store.create_version(_skill())
|
|
candidate = _skill(
|
|
steps=[FlowStep("input_text", {"text": "{search_query}"})],
|
|
parameters={"search_query": {"type": "string"}},
|
|
)
|
|
|
|
result = store_synthesized_skill(store, candidate)
|
|
|
|
assert result.created_new_version is False
|
|
assert result.skill.version == first.version
|
|
assert result.skill.id == first.id
|
|
assert result.skill.parameters == {"search_query": {"type": "string"}}
|
|
assert store.get_latest_by_name("search") == result.skill
|
|
|
|
|
|
def test_structural_divergence_creates_new_version_and_preserves_parent() -> None:
|
|
store = SkillStore()
|
|
first = store.create_version(_skill())
|
|
candidate = _skill(
|
|
steps=[
|
|
FlowStep("tap", {"x": 1}),
|
|
FlowStep("input_text", {"text": "coffee"}),
|
|
]
|
|
)
|
|
|
|
result = store_synthesized_skill(store, candidate)
|
|
|
|
assert result.created_new_version is True
|
|
assert result.skill.version == 2
|
|
assert result.skill.parent_version_id == first.id
|
|
assert store.get_by_id(first.id) == first
|
|
assert store.get_latest_by_name("search") == result.skill
|
|
|
|
|
|
def _two_step_skill(*, text: str) -> FlowTemplateSkill:
|
|
return _skill(
|
|
steps=[
|
|
FlowStep("tap", {"x": 1}),
|
|
FlowStep("input_text", {"text": text}),
|
|
]
|
|
)
|
|
|
|
|
|
def test_argument_divergence_within_configured_tolerance_does_not_bump_version() -> None:
|
|
store = SkillStore()
|
|
first = store.create_version(_two_step_skill(text="coffee"))
|
|
candidate = _two_step_skill(text="tea")
|
|
# Half (1 of 2) step positions diverge; tolerance allows up to half.
|
|
config = SkillAuthoringConfig(divergence_tolerance=0.5)
|
|
|
|
diff = diff_flow_versions(first.steps, candidate.steps, config=config)
|
|
assert diff.structural_divergence is False
|
|
assert diff.argument_divergence_fraction == 0.5
|
|
|
|
result = store_synthesized_skill(store, candidate, config=config)
|
|
|
|
assert result.created_new_version is False
|
|
assert result.skill.version == first.version
|
|
assert result.skill.id == first.id
|
|
|
|
|
|
def test_argument_divergence_beyond_configured_tolerance_bumps_version() -> None:
|
|
store = SkillStore()
|
|
first = store.create_version(_two_step_skill(text="coffee"))
|
|
candidate = _two_step_skill(text="tea")
|
|
# Half (1 of 2) step positions diverge; tolerance only allows less than that.
|
|
config = SkillAuthoringConfig(divergence_tolerance=0.3)
|
|
|
|
diff = diff_flow_versions(first.steps, candidate.steps, config=config)
|
|
assert diff.structural_divergence is True
|
|
assert diff.argument_divergence_fraction == 0.5
|
|
|
|
result = store_synthesized_skill(store, candidate, config=config)
|
|
|
|
assert result.created_new_version is True
|
|
assert result.skill.version == 2
|
|
assert result.skill.parent_version_id == first.id
|
|
|
|
|
|
def test_argument_divergence_without_config_never_bumps_version() -> None:
|
|
# Backward-compatible default: no config supplied means unlimited
|
|
# tolerance, matching pre-existing behavior of absorbing any argument
|
|
# divergence as a parameter update rather than a new version.
|
|
store = SkillStore()
|
|
first = store.create_version(_two_step_skill(text="coffee"))
|
|
candidate = _two_step_skill(text="tea")
|
|
|
|
result = store_synthesized_skill(store, candidate)
|
|
|
|
assert result.created_new_version is False
|
|
assert result.skill.version == first.version
|