cleveragents-core/features/steps/tool_builtins_steps.py

"""Step definitions for the Built-in File Tools feature."""

from __future__ import annotations

import fnmatch
import shutil
import tempfile
from pathlib import Path
from typing import Any

from behave import given, then, when

from cleveragents.tool.builtins import (
    register_file_tools,
    register_file_tools_with_changeset,
)
from cleveragents.tool.builtins.changeset import (
    ChangeSetCapture,
    ChangeSetEntry,
    _detect_operation,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner

__all__: list[str] = []


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _run_tool(context: Any, tool_name: str, inputs: dict[str, Any]) -> None:
    """Execute a built-in file tool through the runner."""
    registry = ToolRegistry()
    if hasattr(context, "changeset_capture") and context.changeset_capture is not None:
        register_file_tools_with_changeset(registry, context.changeset_capture)
    else:
        register_file_tools(registry)
    runner = ToolRunner(registry)
    inputs["sandbox_root"] = context.sandbox_dir
    context.tool_result = runner.execute(tool_name, inputs)


# ---------------------------------------------------------------------------
# Givens
# ---------------------------------------------------------------------------


@given("a temporary sandbox directory")
def step_given_sandbox(context: Any) -> None:
    context.sandbox_dir = tempfile.mkdtemp()
    context._cleanup_handlers.append(
        lambda: shutil.rmtree(context.sandbox_dir, ignore_errors=True)
    )


@given('a file "{name}" with content "{content}"')
def step_given_file_with_content(context: Any, name: str, content: str) -> None:
    path = Path(context.sandbox_dir) / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)


@given('a file "{name}" with lines "{csv_lines}"')
def step_given_file_with_lines(context: Any, name: str, csv_lines: str) -> None:
    lines = csv_lines.split(",")
    path = Path(context.sandbox_dir) / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines) + "\n")


@given('a changeset capture with plan_id "{plan_id}"')
def step_given_changeset_capture(context: Any, plan_id: str) -> None:
    context.changeset_capture = ChangeSetCapture(plan_id=plan_id)


@given('a subdirectory "{name}" in the sandbox')
def step_given_subdirectory(context: Any, name: str) -> None:
    path = Path(context.sandbox_dir) / name
    path.mkdir(parents=True, exist_ok=True)


@given("a sibling directory with a name that is a prefix of the sandbox name")
def step_given_sibling_prefix_dir(context: Any) -> None:
    """Create a sibling directory whose name is a string prefix of the sandbox dir.

    For example, if the sandbox is /tmp/abc123, this creates /tmp/abc123-escape.
    This exercises the path traversal bypass where str.startswith() would
    incorrectly allow /tmp/abc123-escape to pass a /tmp/abc123 sandbox check.
    The step stores the relative escape path on context for use in When steps.
    """
    sandbox_path = Path(context.sandbox_dir)
    sibling_name = sandbox_path.name + "-escape"
    sibling_path = sandbox_path.parent / sibling_name
    sibling_path.mkdir(parents=True, exist_ok=True)
    context._cleanup_handlers.append(
        lambda: shutil.rmtree(str(sibling_path), ignore_errors=True)
    )
    context.sibling_escape_dir = str(sibling_path)
    # Relative path from sandbox root that resolves into the sibling directory
    context.sibling_escape_rel_path = f"../{sibling_name}/secret.txt"


# ---------------------------------------------------------------------------
# Whens
# ---------------------------------------------------------------------------


@when('I execute the "builtin/file-read" tool with path "{path}"')
def step_when_file_read(context: Any, path: str) -> None:
    _run_tool(context, "builtin/file-read", {"path": path})


@when("I attempt to read a file in the sibling escape directory")
def step_when_read_sibling_escape(context: Any) -> None:
    """Attempt to read a file using the dynamic sibling-escape relative path.

    Uses the path stored by the 'sibling directory with a name that is a prefix'
    Given step to exercise the prefix-collision path traversal bypass.
    """
    _run_tool(context, "builtin/file-read", {"path": context.sibling_escape_rel_path})


@when(
    'I execute the "builtin/file-read" tool with path "{path}" offset {offset:d} limit {limit:d}'
)
def step_when_file_read_offset_limit(
    context: Any, path: str, offset: int, limit: int
) -> None:
    _run_tool(
        context, "builtin/file-read", {"path": path, "offset": offset, "limit": limit}
    )


@when(
    'I execute the "builtin/file-write" tool with path "{path}" and content "{content}"'
)
def step_when_file_write(context: Any, path: str, content: str) -> None:
    _run_tool(context, "builtin/file-write", {"path": path, "content": content})


@when('I execute the "builtin/file-edit" tool replacing "{old}" with "{new}"')
def step_when_file_edit(context: Any, old: str, new: str) -> None:
    _run_tool(
        context,
        "builtin/file-edit",
        {"path": "edit.txt", "old_text": old, "new_text": new},
    )


@when('I execute the "builtin/file-edit" tool replacing all "{old}" with "{new}"')
def step_when_file_edit_all(context: Any, old: str, new: str) -> None:
    _run_tool(
        context,
        "builtin/file-edit",
        {"path": "edit.txt", "old_text": old, "new_text": new, "replace_all": True},
    )


@when('I execute the "builtin/file-delete" tool with path "{path}"')
def step_when_file_delete(context: Any, path: str) -> None:
    _run_tool(context, "builtin/file-delete", {"path": path})


@when('I list directory "{path}" with the file-list tool')
def step_when_file_list(context: Any, path: str) -> None:
    _run_tool(context, "builtin/file-list", {"path": path})


@when('I list directory "{path}" filtering by pattern "{pattern}"')
def step_when_file_list_pattern(context: Any, path: str, pattern: str) -> None:
    _run_tool(context, "builtin/file-list", {"path": path, "pattern": pattern})


@when('I list directory "{path}" recursively with the file-list tool')
def step_when_file_list_recursive(context: Any, path: str) -> None:
    _run_tool(context, "builtin/file-list", {"path": path, "recursive": True})


@when('I search directory "{path}" for pattern "{pattern}"')
def step_when_file_search(context: Any, path: str, pattern: str) -> None:
    _run_tool(context, "builtin/file-search", {"path": path, "pattern": pattern})


@when(
    'I search directory "{path}" for pattern "{pattern}" with max_results {max_results:d}'
)
def step_when_file_search_max(
    context: Any, path: str, pattern: str, max_results: int
) -> None:
    _run_tool(
        context,
        "builtin/file-search",
        {"path": path, "pattern": pattern, "max_results": max_results},
    )


@when("I write a file through changeset-wrapped tools")
def step_when_write_changeset(context: Any) -> None:
    _run_tool(
        context, "builtin/file-write", {"path": "cs-new.txt", "content": "changeset"}
    )


@when("I edit a file through changeset-wrapped tools")
def step_when_edit_changeset(context: Any) -> None:
    _run_tool(
        context,
        "builtin/file-edit",
        {"path": "cs-edit.txt", "old_text": "hello", "new_text": "goodbye"},
    )


@when("I delete a file through changeset-wrapped tools")
def step_when_delete_changeset(context: Any) -> None:
    _run_tool(context, "builtin/file-delete", {"path": "cs-delete.txt"})


@when("I perform multiple changeset-tracked operations")
def step_when_multiple_changeset_ops(context: Any) -> None:
    _run_tool(context, "builtin/file-write", {"path": "multi1.txt", "content": "a"})
    _run_tool(context, "builtin/file-write", {"path": "multi2.txt", "content": "b"})
    # Edit multi1.txt
    _run_tool(
        context,
        "builtin/file-edit",
        {"path": "multi1.txt", "old_text": "a", "new_text": "c"},
    )


@when("I clear the changeset")
def step_when_clear_changeset(context: Any) -> None:
    context.changeset_capture.clear()


@when("I register all file tools")
def step_when_register_all(context: Any) -> None:
    register_file_tools(context.registry)


@when("I register file tools with changeset")
def step_when_register_with_changeset(context: Any) -> None:
    register_file_tools_with_changeset(context.registry, context.changeset_capture)


@when('I edit "{filename}" through changeset-wrapped tools')
def step_when_edit_specific_changeset(context: Any, filename: str) -> None:
    _run_tool(
        context,
        "builtin/file-edit",
        {"path": filename, "old_text": "original", "new_text": "modified"},
    )


@when("I write a new file through raw changeset capture")
def step_when_raw_capture_write_new(context: Any) -> None:
    """Test _detect_operation fallback when before is None and after exists."""
    # Simulate: before=None, after=hash, output has no 'created' key
    op = _detect_operation("builtin/file-write", None, "abc123hash", {})
    context.changeset_capture._entries.append(
        ChangeSetEntry(
            operation=op,
            path="new-fallback.txt",
            before_hash=None,
            after_hash="abc123hash",
            metadata={"tool": "builtin/file-write"},
        )
    )


@when("I overwrite a file through raw changeset capture")
def step_when_raw_capture_overwrite(context: Any) -> None:
    """Test _detect_operation fallback when before and after both exist."""
    # Simulate: before=hash, after=different_hash, output has no 'created' key
    op = _detect_operation("builtin/file-write", "hash1", "hash2", {})
    context.changeset_capture._entries.append(
        ChangeSetEntry(
            operation=op,
            path="existing-cs.txt",
            before_hash="hash1",
            after_hash="hash2",
            metadata={"tool": "builtin/file-write"},
        )
    )


# ---------------------------------------------------------------------------
# Thens
# ---------------------------------------------------------------------------


@then('the output "{key}" should be "{value}"')
def step_then_output_str(context: Any, key: str, value: str) -> None:
    assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
    assert str(context.tool_result.output[key]) == value, (
        f"Expected {key}={value}, got {context.tool_result.output[key]}"
    )


@then('the output "{key}" should be greater than {value:d}')
def step_then_output_gt(context: Any, key: str, value: int) -> None:
    assert context.tool_result.output[key] > value


@then('the output "{key}" should be boolean true')
def step_then_output_true(context: Any, key: str) -> None:
    assert context.tool_result.output[key] is True


@then('the output "{key}" should be boolean false')
def step_then_output_false(context: Any, key: str) -> None:
    assert context.tool_result.output[key] is False


@then('the output "{key}" should equal {value:d}')
def step_then_output_int(context: Any, key: str, value: int) -> None:
    assert context.tool_result.output[key] == value, (
        f"Expected {key}={value}, got {context.tool_result.output[key]}"
    )


@then('the output "{key}" should contain "{text}"')
def step_then_output_contains(context: Any, key: str, text: str) -> None:
    assert text in context.tool_result.output[key], (
        f"Expected '{text}' in {key}: {context.tool_result.output[key]}"
    )


@then('the output "{key}" should not contain "{text}"')
def step_then_output_not_contains(context: Any, key: str, text: str) -> None:
    assert text not in context.tool_result.output[key]


@then('the file "{name}" should exist in the sandbox')
def step_then_file_exists(context: Any, name: str) -> None:
    path = Path(context.sandbox_dir) / name
    assert path.exists(), f"File {path} does not exist"


@then('the file "{name}" should not exist in the sandbox')
def step_then_file_not_exists(context: Any, name: str) -> None:
    path = Path(context.sandbox_dir) / name
    assert not path.exists(), f"File {path} still exists"


@then("the output files list should contain at least {count:d} entries")
def step_then_files_count(context: Any, count: int) -> None:
    assert context.tool_result.success
    files = context.tool_result.output["files"]
    assert len(files) >= count, f"Expected >= {count} files, got {len(files)}"


@then('all listed files should match pattern "{pattern}"')
def step_then_files_match_pattern(context: Any, pattern: str) -> None:
    assert context.tool_result.success
    files = context.tool_result.output["files"]
    for f in files:
        assert fnmatch.fnmatch(f["name"], pattern), (
            f"File {f['name']} does not match pattern {pattern}"
        )


@then("the search matches should contain at least {count:d} result")
def step_then_search_min(context: Any, count: int) -> None:
    assert context.tool_result.success
    matches = context.tool_result.output["matches"]
    assert len(matches) >= count, f"Expected >= {count} matches, got {len(matches)}"


@then("the search matches should contain exactly {count:d} results")
def step_then_search_exact(context: Any, count: int) -> None:
    assert context.tool_result.success
    matches = context.tool_result.output["matches"]
    assert len(matches) == count, f"Expected {count} matches, got {len(matches)}"


@then("the changeset should have {count:d} entry")
def step_then_changeset_count_singular(context: Any, count: int) -> None:
    cs = context.changeset_capture.get_changeset()
    assert len(cs.entries) == count, f"Expected {count} entries, got {len(cs.entries)}"


@then("the changeset should have {count:d} entries")
def step_then_changeset_count_plural(context: Any, count: int) -> None:
    cs = context.changeset_capture.get_changeset()
    assert len(cs.entries) == count, f"Expected {count} entries, got {len(cs.entries)}"


@then('the changeset entry operation should be "{operation}"')
def step_then_changeset_operation(context: Any, operation: str) -> None:
    cs = context.changeset_capture.get_changeset()
    assert len(cs.entries) > 0, "No changeset entries"
    assert cs.entries[-1].operation == operation, (
        f"Expected operation '{operation}', got '{cs.entries[-1].operation}'"
    )


@then("the changeset summary should mention the total count")
def step_then_summary_count(context: Any) -> None:
    cs = context.changeset_capture.get_changeset()
    summary = cs.summary()
    count = len(cs.entries)
    assert str(count) in summary, f"Expected count {count} in summary: {summary}"


@then("the changeset summary should mention operation types")
def step_then_summary_ops(context: Any) -> None:
    cs = context.changeset_capture.get_changeset()
    summary = cs.summary()
    assert "create" in summary or "modify" in summary or "delete" in summary, (
        f"No operation types in summary: {summary}"
    )


@then('the changeset summary should be "{expected}"')
def step_then_summary_exact(context: Any, expected: str) -> None:
    cs = context.changeset_capture.get_changeset()
    assert cs.summary() == expected, f"Expected '{expected}', got '{cs.summary()}'"


@then("the registry should contain {count:d} tools")
def step_then_registry_count(context: Any, count: int) -> None:
    tools = context.registry.list_tools()
    assert len(tools) == count, f"Expected {count} tools, got {len(tools)}"


@then('the registry should contain tool "{name}"')
def step_then_registry_has_tool(context: Any, name: str) -> None:
    assert context.registry.get(name) is not None, (
        f"Tool '{name}' not found in registry"
    )


@then("the changeset entry should have a before_hash")
def step_then_has_before_hash(context: Any) -> None:
    cs = context.changeset_capture.get_changeset()
    assert cs.entries[-1].before_hash is not None


@then("the changeset entry should have an after_hash")
def step_then_has_after_hash(context: Any) -> None:
    cs = context.changeset_capture.get_changeset()
    assert cs.entries[-1].after_hash is not None


@then("the before_hash and after_hash should differ")
def step_then_hashes_differ(context: Any) -> None:
    cs = context.changeset_capture.get_changeset()
    entry = cs.entries[-1]
    assert entry.before_hash != entry.after_hash, "Hashes should differ after edit"