"""Step definitions for ChangeSet models and invocation tracking.""" from __future__ import annotations from datetime import UTC, datetime from typing import Any from behave import given, then, when from cleveragents.domain.models.core.change import ( ChangeEntry, ChangeOperation, ChangeType, InMemoryInvocationTracker, SpecChangeSet, ToolInvocation, normalize_change_path, ) __all__: list[str] = [] # --------------------------------------------------------------------------- # ChangeType alias # --------------------------------------------------------------------------- @given("I import ChangeType and ChangeOperation") def step_import_change_type(context: Any) -> None: context.change_type = ChangeType context.change_operation = ChangeOperation @then("ChangeType should be the same class as ChangeOperation") def step_changetype_same(context: Any) -> None: assert context.change_type is context.change_operation @then('ChangeType CREATE should equal "{value}"') def step_changetype_create_value(context: Any, value: str) -> None: assert value == ChangeType.CREATE # --------------------------------------------------------------------------- # ChangeEntry validators # --------------------------------------------------------------------------- def _base_entry_kwargs() -> dict[str, Any]: return { "plan_id": "plan-test", "resource_id": "res-test", "tool_name": "builtin/file-write", "path": "src/test.py", } @when('I create a ChangeEntry with CREATE and before_hash "{h}"') def step_create_entry_create_before(context: Any, h: str) -> None: try: ChangeEntry( **_base_entry_kwargs(), operation=ChangeOperation.CREATE, before_hash=h, after_hash="xyz", ) context.validation_error = None except Exception as exc: context.validation_error = str(exc) @when('I create a ChangeEntry with DELETE and after_hash "{h}"') def step_create_entry_delete_after(context: Any, h: str) -> None: try: ChangeEntry( **_base_entry_kwargs(), operation=ChangeOperation.DELETE, before_hash="xyz", after_hash=h, ) context.validation_error = None except Exception as exc: context.validation_error = str(exc) @then('a validation error should be raised mentioning "{text}"') def step_validation_error(context: Any, text: str) -> None: assert context.validation_error is not None, "No error was raised" assert text in context.validation_error, ( f"Expected '{text}' in: {context.validation_error}" ) @when('I create a valid CREATE ChangeEntry with after_hash "{h}"') def step_valid_create_entry(context: Any, h: str) -> None: context.entry = ChangeEntry( **_base_entry_kwargs(), operation=ChangeOperation.CREATE, after_hash=h, ) @when("I create a valid MODIFY ChangeEntry with both hashes") def step_valid_modify_entry(context: Any) -> None: context.entry = ChangeEntry( **_base_entry_kwargs(), operation=ChangeOperation.MODIFY, before_hash="hash-a", after_hash="hash-b", ) @when("I create a RENAME ChangeEntry without hashes") def step_rename_no_hashes(context: Any) -> None: context.entry = ChangeEntry( **_base_entry_kwargs(), operation=ChangeOperation.RENAME, ) @when("I create a CREATE ChangeEntry without any hashes") def step_create_no_hashes(context: Any) -> None: context.entry = ChangeEntry( **_base_entry_kwargs(), operation=ChangeOperation.CREATE, ) @then('the change entry operation should be "{op}"') def step_change_entry_operation(context: Any, op: str) -> None: assert context.entry.operation == op @then('the change entry after_hash should be "{h}"') def step_change_entry_after_hash(context: Any, h: str) -> None: assert context.entry.after_hash == h @then("the change entry has_integrity_hashes should be true") def step_change_entry_integrity_true(context: Any) -> None: assert context.entry.has_integrity_hashes is True @then("the change entry has_integrity_hashes should be false") def step_change_entry_integrity_false(context: Any) -> None: assert context.entry.has_integrity_hashes is False # --------------------------------------------------------------------------- # Path normalization # --------------------------------------------------------------------------- @given('a repo root "{root}"') def step_repo_root(context: Any, root: str) -> None: context.repo_root = root @given("an empty repo root") def step_empty_repo_root(context: Any) -> None: context.repo_root = "" @when('I normalize the path "{path}"') def step_normalize_path(context: Any, path: str) -> None: context.normalized = normalize_change_path(path, context.repo_root) @then('the normalized path should be "{expected}"') def step_normalized_result(context: Any, expected: str) -> None: assert context.normalized == expected, ( f"Expected '{expected}', got '{context.normalized}'" ) # --------------------------------------------------------------------------- # SpecChangeSet ordering # --------------------------------------------------------------------------- def _make_entry(resource_id: str, path: str, ts_offset: int = 0) -> ChangeEntry: """Create a ChangeEntry with specific ordering fields.""" return ChangeEntry( plan_id="plan-order", resource_id=resource_id, tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path=path, after_hash="hash-new", timestamp=datetime(2026, 1, 1, 0, 0, ts_offset, tzinfo=UTC), ) @given("a SpecChangeSet with entries in random order") def step_changeset_random_order(context: Any) -> None: entries = [ _make_entry("res-B", "z.py", 3), _make_entry("res-A", "b.py", 1), _make_entry("res-B", "a.py", 2), _make_entry("res-A", "a.py", 0), ] context.changeset = SpecChangeSet(plan_id="plan-order", entries=entries) @when("I call sorted_entries") def step_call_sorted(context: Any) -> None: context.sorted = context.changeset.sorted_entries() @then("entries should be sorted by resource_id then path then timestamp") def step_verify_sorted(context: Any) -> None: keys = [(e.resource_id, e.path, e.timestamp) for e in context.sorted] assert keys == sorted(keys), f"Not sorted: {keys}" @given("a SpecChangeSet with entries for multiple resources") def step_changeset_multi_resource(context: Any) -> None: entries = [ _make_entry("res-B", "y.py", 1), _make_entry("res-A", "x.py", 0), _make_entry("res-B", "w.py", 2), _make_entry("res-A", "z.py", 3), ] context.changeset = SpecChangeSet(plan_id="plan-grp", entries=entries) @when("I call grouped_by_resource") def step_call_grouped(context: Any) -> None: context.groups = context.changeset.grouped_by_resource() @then("entries should be grouped by resource_id") def step_verify_grouped(context: Any) -> None: assert set(context.groups.keys()) == {"res-A", "res-B"} @then("each group should be sorted by path and timestamp") def step_verify_group_sort(context: Any) -> None: for entries in context.groups.values(): keys = [(e.path, e.timestamp) for e in entries] assert keys == sorted(keys), f"Not sorted: {keys}" @given("an empty SpecChangeSet") def step_empty_changeset(context: Any) -> None: context.changeset = SpecChangeSet(plan_id="plan-empty") @when("I add a change entry to it") def step_add_change(context: Any) -> None: entry = ChangeEntry( plan_id="plan-empty", resource_id="res-1", tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path="new.py", after_hash="hash-new", ) context.changeset.add_change(entry) @then("the spec changeset should contain {count:d} entry") def step_spec_changeset_count(context: Any, count: int) -> None: assert len(context.changeset.entries) == count # --------------------------------------------------------------------------- # ToolInvocation # --------------------------------------------------------------------------- @when('I create a ToolInvocation for plan "{pid}" and tool "{tool}"') def step_create_invocation(context: Any, pid: str, tool: str) -> None: context.invocation = ToolInvocation(plan_id=pid, tool_name=tool) @then("the invocation should have a ULID invocation_id") def step_invocation_ulid(context: Any) -> None: assert len(context.invocation.invocation_id) == 26 @then('the invocation plan_id should be "{pid}"') def step_invocation_plan(context: Any, pid: str) -> None: assert context.invocation.plan_id == pid @then('the invocation tool_name should be "{tool}"') def step_invocation_tool(context: Any, tool: str) -> None: assert context.invocation.tool_name == tool @then("the invocation success should be true") def step_invocation_success_true(context: Any) -> None: assert context.invocation.success is True @when('I create a failed ToolInvocation with error "{msg}"') def step_failed_invocation(context: Any, msg: str) -> None: context.invocation = ToolInvocation( plan_id="plan-fail", tool_name="builtin/file-read", success=False, error=msg, ) @then("the invocation success should be false") def step_invocation_success_false(context: Any) -> None: assert context.invocation.success is False @then('the invocation error should be "{msg}"') def step_invocation_error(context: Any, msg: str) -> None: assert context.invocation.error == msg @when("I create a ToolInvocation with change_ids") def step_invocation_change_ids(context: Any) -> None: context.invocation = ToolInvocation( plan_id="plan-chg", tool_name="builtin/file-write", change_ids=["chg-1", "chg-2"], ) @then("the invocation should have {count:d} change_ids") def step_invocation_change_count(context: Any, count: int) -> None: assert len(context.invocation.change_ids) == count @when("I create a ToolInvocation with provider metadata") def step_invocation_provider(context: Any) -> None: context.invocation = ToolInvocation( plan_id="plan-prov", tool_name="builtin/file-write", provider_metadata={ "model_name": "gpt-4", "provider_id": "openai", "latency_ms": 150.0, }, ) @then("the invocation provider_metadata should contain model name") def step_invocation_provider_meta(context: Any) -> None: meta = context.invocation.provider_metadata assert meta is not None assert "model_name" in meta # --------------------------------------------------------------------------- # InMemoryInvocationTracker # --------------------------------------------------------------------------- @given("an InMemoryInvocationTracker") def step_tracker(context: Any) -> None: context.tracker = InMemoryInvocationTracker() @when('I track {count:d} invocations for plan "{pid}"') def step_track_n(context: Any, count: int, pid: str) -> None: for i in range(count): context.tracker.track( ToolInvocation( plan_id=pid, tool_name=f"tool-{i}", sequence_number=i, ) ) @then('get_invocations for "{pid}" should return {count:d}') def step_get_invocations_count(context: Any, pid: str, count: int) -> None: result = context.tracker.get_invocations(pid) assert len(result) == count, f"Expected {count}, got {len(result)}" @when("I track invocations for different skills") def step_track_skills(context: Any) -> None: context.tracker.track( ToolInvocation( plan_id="plan-sk", tool_name="t1", skill_name="skill-A", sequence_number=0, ) ) context.tracker.track( ToolInvocation( plan_id="plan-sk", tool_name="t2", skill_name="skill-B", sequence_number=1, ) ) context.tracker.track( ToolInvocation( plan_id="plan-sk", tool_name="t3", skill_name="skill-A", sequence_number=2, ) ) @then("get_invocations_for_skill should filter correctly") def step_filter_skill(context: Any) -> None: a = context.tracker.get_invocations_for_skill("plan-sk", "skill-A") assert len(a) == 2 b = context.tracker.get_invocations_for_skill("plan-sk", "skill-B") assert len(b) == 1 @when("I track invocations with change_ids") def step_track_with_changes(context: Any) -> None: context.tracker.track( ToolInvocation( plan_id="plan-chg", tool_name="t1", change_ids=["c1", "c2"], sequence_number=0, ) ) context.tracker.track( ToolInvocation( plan_id="plan-chg", tool_name="t2", change_ids=["c3"], sequence_number=1, ) ) @then("get_changes should return all change_ids for the plan") def step_get_all_changes(context: Any) -> None: changes = context.tracker.get_changes("plan-chg") assert changes == ["c1", "c2", "c3"] @when("I track invocations out of order") def step_track_out_of_order(context: Any) -> None: context.tracker.track( ToolInvocation( plan_id="plan-ord", tool_name="t3", sequence_number=2, ) ) context.tracker.track( ToolInvocation( plan_id="plan-ord", tool_name="t1", sequence_number=0, ) ) context.tracker.track( ToolInvocation( plan_id="plan-ord", tool_name="t2", sequence_number=1, ) ) @then("get_invocations should return them in sequence order") def step_verify_sequence_order(context: Any) -> None: invs = context.tracker.get_invocations("plan-ord") seqs = [i.sequence_number for i in invs] assert seqs == [0, 1, 2], f"Expected [0, 1, 2], got {seqs}" # --------------------------------------------------------------------------- # SpecChangeSet summary # --------------------------------------------------------------------------- @given("a SpecChangeSet with mixed operation entries") def step_changeset_mixed_ops(context: Any) -> None: entries = [ ChangeEntry( plan_id="plan-mix", resource_id="res-A", tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path="new.py", after_hash="h1", ), ChangeEntry( plan_id="plan-mix", resource_id="res-A", tool_name="builtin/file-edit", operation=ChangeOperation.MODIFY, path="edit.py", before_hash="h2a", after_hash="h2b", ), ChangeEntry( plan_id="plan-mix", resource_id="res-B", tool_name="builtin/file-delete", operation=ChangeOperation.DELETE, path="old.py", before_hash="h3", ), ChangeEntry( plan_id="plan-mix", resource_id="res-B", tool_name="builtin/file-rename", operation=ChangeOperation.RENAME, path="moved.py", ), ] context.changeset = SpecChangeSet(plan_id="plan-mix", entries=entries) @when("I call summary on the changeset") def step_call_summary(context: Any) -> None: context.summary_result = context.changeset.summary() @then("the summary total should be {count:d}") def step_summary_total(context: Any, count: int) -> None: assert context.summary_result["total"] == count, ( f"Expected total {count}, got {context.summary_result['total']}" ) @then("the summary creates should be {count:d}") def step_summary_creates(context: Any, count: int) -> None: assert context.summary_result["creates"] == count @then("the summary modifies should be {count:d}") def step_summary_modifies(context: Any, count: int) -> None: assert context.summary_result["modifies"] == count @then("the summary deletes should be {count:d}") def step_summary_deletes(context: Any, count: int) -> None: assert context.summary_result["deletes"] == count @then("the summary renames should be {count:d}") def step_summary_renames(context: Any, count: int) -> None: assert context.summary_result["renames"] == count @then("the summary paths_changed should be {count:d}") def step_summary_paths(context: Any, count: int) -> None: assert context.summary_result["paths_changed"] == count @then("the summary resources_involved should be {count:d}") def step_summary_resources(context: Any, count: int) -> None: assert context.summary_result["resources_involved"] == count # --------------------------------------------------------------------------- # add_change plan_id validation # --------------------------------------------------------------------------- @when("I add a change entry with a different plan_id") def step_add_mismatched_plan(context: Any) -> None: entry = ChangeEntry( plan_id="wrong-plan", resource_id="res-1", tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path="file.py", after_hash="hash-new", ) try: context.changeset.add_change(entry) context.validation_error = None except Exception as exc: context.validation_error = str(exc) # --------------------------------------------------------------------------- # Empty tracker queries # --------------------------------------------------------------------------- @then('get_changes for "{pid}" should return {count:d} ids') def step_get_changes_empty(context: Any, pid: str, count: int) -> None: changes = context.tracker.get_changes(pid) assert len(changes) == count, f"Expected {count}, got {len(changes)}" @then('get_invocations_for_skill "{pid}" "{skill}" should return {count:d}') def step_get_skill_empty(context: Any, pid: str, skill: str, count: int) -> None: result = context.tracker.get_invocations_for_skill(pid, skill) assert len(result) == count, f"Expected {count}, got {len(result)}" # --------------------------------------------------------------------------- # get_changes ordering proof # --------------------------------------------------------------------------- @when("I track change invocations in reverse sequence order") def step_track_changes_reverse(context: Any) -> None: context.tracker.track( ToolInvocation( plan_id="plan-rev", tool_name="t2", sequence_number=1, change_ids=["c-second"], ) ) context.tracker.track( ToolInvocation( plan_id="plan-rev", tool_name="t1", sequence_number=0, change_ids=["c-first"], ) ) @then("get_changes should return change_ids in sequence order") def step_verify_changes_sequence(context: Any) -> None: changes = context.tracker.get_changes("plan-rev") assert changes == ["c-first", "c-second"], ( f"Expected ['c-first', 'c-second'], got {changes}" )