import os
import tempfile
from pathlib import Path

import pytest
from git import Repo

from openhands_aci.code_search import initialize_code_search, search_code, CodeSearchIndex

@pytest.fixture
def test_repo():
    """Create a temporary git repository with some test files."""
    with tempfile.TemporaryDirectory() as temp_dir:
        # Initialize git repo
        repo = Repo.init(temp_dir)
        
        # Create test files
        files = {
            'main.py': 'def hello():\n    print("Hello, World!")',
            'utils/helper.py': 'def add(a, b):\n    return a + b',
            'README.md': '# Test Repository\nThis is a test.'
        }
        
        for path, content in files.items():
            file_path = Path(temp_dir) / path
            file_path.parent.mkdir(parents=True, exist_ok=True)
            with open(file_path, 'w') as f:
                f.write(content)
        
        # Add and commit files
        repo.index.add('*')
        repo.index.commit('Initial commit')
        
        yield temp_dir

def test_initialize_code_search(test_repo):
    """Test initializing code search for a repository."""
    with tempfile.TemporaryDirectory() as save_dir:
        # Initialize with only Python files
        result = initialize_code_search(
            test_repo,
            save_dir,
            extensions=['.py'],
            embedding_model='BAAI/bge-base-en-v1.5'
        )
        
        assert result['status'] == 'success'
        assert result['num_documents'] == 2  # main.py and helper.py
        
        # Check that index files were created
        assert os.path.exists(os.path.join(save_dir, 'index.faiss'))
        assert os.path.exists(os.path.join(save_dir, 'documents.pkl'))

def test_search_code(test_repo):
    """Test searching code in an indexed repository."""
        # First initialize
        initialize_code_search(
            save_dir,
        )
        
        # Test search
        result = search_code(save_dir, "function that adds two numbers")
        
        assert len(result['results']) > 0
        
        # The add function should be in the results
        found_add = False
        for doc in result['results']:
            if 'add' in doc['content']:
                found_add = True
                break
        assert found_add

def test_code_search_index():
    """Test the CodeSearchIndex class directly."""
    # Create test documents
    docs = [
        {'id': '1', 'content': 'def add(a, b): return a + b'},
        {'id': '2', 'content': 'def subtract(a, b): return a - b'},
        {'id': '3', 'content': 'print("Hello, World!")'}
    ]
    
    # Create index
    index = CodeSearchIndex(embedding_model='BAAI/bge-base-en-v1.5')
    index.add_documents(docs)
    
    # Test search
    results = index.search("function that adds numbers")
    assert len(results) > 0
    assert any('add' in doc['content'] for doc in results)
    
    # Test save and load
        index.save(temp_dir)
        loaded_index = CodeSearchIndex.load(temp_dir)
        
        # Search with loaded index
        loaded_results = loaded_index.search("function that adds numbers")
        assert len(loaded_results) == len(results)

def test_error_handling():
    """Test error handling in the tools."""
    # Test with non-existent repository
    result = initialize_code_search(
        '/path/that/does/not/exist',
        '/tmp/save',
        extensions=['.py']
    )
    assert result['status'] == 'error'
    assert 'Error' in result['message']
    
    # Test search with non-existent index
    result = search_code('/path/that/does/not/exist', "query")

import json
import re

from openhands_aci.editor import file_editor

def temp_file(tmp_path):
    # Set up a temporary directory with test files
    test_file = tmp_path / 'test_file.txt'
    test_file.write_text('This is a test file.\nThis file is for testing purposes.')
    return str(test_file)

def test_file_editor_happy_path(temp_file):
    command = 'str_replace'
    old_str = 'test file'
    new_str = 'sample file'

    # Call the `file_editor` function
    result = file_editor(
        command=command,
        path=temp_file,
        old_str=old_str,
        new_str=new_str,
        enable_linting=False,
    )

    # Extract the JSON content using a regular expression
    match = re.search(
        r'<oh_aci_output_[0-9a-f]{32}>(.*?)</oh_aci_output_[0-9a-f]{32}>',
        result,
        re.DOTALL,
    )
    assert match, 'Output does not contain the expected <oh_aci_output_> tags in the correct format.'
    result_dict = json.loads(match.group(1))

    # Validate the formatted output in the result dictionary
    formatted_output = result_dict['formatted_output_and_error']
    assert (
        formatted_output
        == f"""The file {temp_file} has been edited. Here's the result of running `cat -n` on a snippet of {temp_file}:
     1\tThis is a sample file.
     2\tThis file is for testing purposes.
Review the changes and make sure they are as expected. Edit the file again if necessary."""
    )
    assert result_dict['path'] == temp_file
    assert result_dict['prev_exist'] is True
    assert (
        result_dict['old_content']
        == 'This is a test file.\nThis file is for testing purposes.'
    )
    assert (
        result_dict['new_content']
        == 'This is a sample file.\nThis file is for testing purposes.'
    )

    # Ensure the file content was updated
    with open(temp_file, 'r') as f:
        content = f.read()
    assert 'This is a sample file.' in content

def test_file_editor_with_xml_tag_parsing(temp_file):
    # Create content that includes the XML tag pattern
    xml_content = """This is a file with XML tags parsing logic...
match = re.search(
    r'<oh_aci_output_[0-9a-f]{32}>(.*?)</oh_aci_output_[0-9a-f]{32}>',
    result,
    re.DOTALL,
)
...More text here.
"""

    with open(temp_file, 'w') as f:
        f.write(xml_content)

        command='view',
    )

    # Ensure the content is extracted correctly
        result,
    )

    print(formatted_output)
    assert (
        == f"""Here's the result of running `cat -n` on {temp_file}:
     1\tThis is a file with XML tags parsing logic...
     2\tmatch = re.search(
     3\t    r'<oh_aci_output_[0-9a-f]{{32}}>(.*?)</oh_aci_output_[0-9a-f]{{32}}>',
     4\t    result,
     5\t    re.DOTALL,
     6\t)
     7\t...More text here.
     8\t
"""
    )

from openhands_aci.editor.editor import OHEditor
from openhands_aci.editor.exceptions import (
    EditorToolParameterInvalidError,
    EditorToolParameterMissingError,
    ToolError,
)
from openhands_aci.editor.results import CLIResult, ToolResult

def editor(tmp_path):
    editor = OHEditor()
    test_file = tmp_path / 'test.txt'
    return editor, test_file

def editor_python_file_with_tabs(tmp_path):
    test_file = tmp_path / 'test.py'
    test_file.write_text('def test():\n\tprint("Hello, World!")')

def test_view_file(editor):
    editor, test_file = editor
    result = editor(command='view', path=str(test_file))
    assert isinstance(result, CLIResult)
    assert f"Here's the result of running `cat -n` on {test_file}:" in result.output
    assert '1\tThis is a test file.' in result.output
    assert '2\tThis file is for testing purposes.' in result.output

def test_view_directory(editor):
    result = editor(command='view', path=str(test_file.parent))
    assert str(test_file.parent) in result.output
    assert test_file.name in result.output

def test_create_file(editor):
    new_file = test_file.parent / 'new_file.txt'
    result = editor(command='create', path=str(new_file), file_text='New file content')
    assert isinstance(result, ToolResult)
    assert new_file.exists()
    assert new_file.read_text() == 'New file content'
    assert 'File created successfully' in result.output

def test_str_replace_no_linting(editor):
    result = editor(
        command='str_replace',
        path=str(test_file),
        old_str='test file',
        new_str='sample file',
    )

    # Test str_replace command
    assert (
        result.output
        == f"""The file {test_file} has been edited. Here's the result of running `cat -n` on a snippet of {test_file}:
    )

    # Test that the file content has been updated
    assert 'This is a sample file.' in test_file.read_text()

def test_str_replace_multi_line_no_linting(editor):
        old_str='This is a test file.\nThis file is for testing purposes.',
        new_str='This is a sample file.\nThis file is for testing purposes.',
    )

    assert (
    )

def test_str_replace_multi_line_with_tabs_no_linting(editor_python_file_with_tabs):
    editor, test_file = editor_python_file_with_tabs
        old_str='def test():\n\tprint("Hello, World!")',
        new_str='def test():\n\tprint("Hello, Universe!")',
    )

    assert (
     1\tdef test():
     2\t{'\t'.expandtabs()}print("Hello, Universe!")
    )

def test_str_replace_with_linting(editor):
        enable_linting=True,
    )

    assert (

No linting issues found in the changes.
    )

def test_str_replace_error_multiple_occurrences(editor):
    with pytest.raises(ToolError) as exc_info:
        editor(
            command='str_replace', path=str(test_file), old_str='test', new_str='sample'
        )
    assert 'Multiple occurrences of old_str `test`' in str(exc_info.value.message)
    assert '[1, 2]' in str(exc_info.value.message)  # Should show both line numbers

def test_str_replace_error_multiple_multiline_occurrences(editor):
    # Create a file with two identical multi-line blocks
    multi_block = """def example():
    print("Hello")
    return True"""
    content = f"{multi_block}\n\nprint('separator')\n\n{multi_block}"
    test_file.write_text(content)

        editor(
            command='str_replace',
            path=str(test_file),
            old_str=multi_block,
            new_str='def new():\n    print("World")',
        )
    error_msg = str(exc_info.value.message)
    assert 'Multiple occurrences of old_str' in error_msg
    assert '[1, 7]' in error_msg  # Should show correct starting line numbers

def test_str_replace_nonexistent_string(editor):
        editor(
            old_str='Non-existent Line',
            new_str='New Line',
        )
    assert 'No replacement was performed' in str(exc_info)
    assert f'old_str `Non-existent Line` did not appear verbatim in {test_file}' in str(
        exc_info.value.message
    )

def test_insert_no_linting(editor):
        command='insert', path=str(test_file), insert_line=1, new_str='Inserted line'
    )
    assert 'Inserted line' in test_file.read_text()
    print(result.output)
    assert (
        == f"""The file {test_file} has been edited. Here's the result of running `cat -n` on a snippet of the edited file:
     1\tThis is a test file.
     2\tInserted line
     3\tThis file is for testing purposes.
Review the changes and make sure they are as expected (correct indentation, no duplicate lines, etc). Edit the file again if necessary."""
    )

def test_insert_with_linting(editor):
        command='insert',
        insert_line=1,
        new_str='Inserted line',
    )
    assert (

    )

def test_insert_invalid_line(editor):
    with pytest.raises(EditorToolParameterInvalidError) as exc_info:
        editor(
            command='insert',
            insert_line=10,
            new_str='Invalid Insert',
        )
    assert 'Invalid `insert_line` parameter' in str(exc_info.value.message)
    assert 'It should be within the range of lines of the file' in str(
    )

def test_undo_edit(editor):
    # Make an edit to be undone
    )
    # Undo the edit
    result = editor(command='undo_edit', path=str(test_file))
    assert 'Last edit to' in result.output
    assert 'test file' in test_file.read_text()  # Original content restored

def test_validate_path_invalid(editor):
    invalid_file = test_file.parent / 'nonexistent.txt'
    with pytest.raises(EditorToolParameterInvalidError):
        editor(command='view', path=str(invalid_file))

def test_create_existing_file_error(editor):
        editor(command='create', path=str(test_file), file_text='New content')

def test_str_replace_missing_old_str(editor):
    with pytest.raises(EditorToolParameterMissingError):
        editor(command='str_replace', path=str(test_file), new_str='sample')

def test_str_replace_new_str_and_old_str_same(editor):
        editor(
            old_str='test file',
            new_str='test file',
        )
    assert (
        'No replacement was performed. `new_str` and `old_str` must be different.'
        in str(exc_info.value.message)
    )

def test_insert_missing_line_param(editor):
        editor(command='insert', path=str(test_file), new_str='Missing insert line')

def test_undo_edit_no_history_error(editor):
    empty_file = test_file.parent / 'empty.txt'
    empty_file.write_text('')
    with pytest.raises(ToolError):
        editor(command='undo_edit', path=str(empty_file))

def test_view_symlinked_directory(tmp_path):

    # Create a directory with some test files
    source_dir = tmp_path / 'source_dir'
    source_dir.mkdir()
    (source_dir / 'file1.txt').write_text('content1')
    (source_dir / 'file2.txt').write_text('content2')

    # Create a subdirectory with a file
    subdir = source_dir / 'subdir'
    subdir.mkdir()
    (subdir / 'file3.txt').write_text('content3')

    # Create a symlink to the directory
    symlink_dir = tmp_path / 'symlink_dir'
    symlink_dir.symlink_to(source_dir)

    # View the symlinked directory
    result = editor(command='view', path=str(symlink_dir))

    # Verify that all files are listed through the symlink
    assert str(symlink_dir) in result.output
    assert 'file1.txt' in result.output
    assert 'file2.txt' in result.output
    assert 'subdir' in result.output
    assert 'file3.txt' in result.output

)

def test_tool_error():
    """Test ToolError raises with correct message."""
        raise ToolError('A tool error occurred')
    assert str(exc_info.value) == 'A tool error occurred'

def test_editor_tool_parameter_missing_error():
    """Test EditorToolParameterMissingError for missing parameter error message."""
    parameter = 'old_str'
    with pytest.raises(EditorToolParameterMissingError) as exc_info:
        raise EditorToolParameterMissingError(command, parameter)
    assert exc_info.value.command == command
    assert exc_info.value.parameter == parameter
    assert (
        == f'Parameter `{parameter}` is required for command: {command}.'
    )

def test_editor_tool_parameter_invalid_error_with_hint():
    """Test EditorToolParameterInvalidError with hint."""
    parameter = 'timeout'
    value = -10
    hint = 'Must be a positive integer.'
        raise EditorToolParameterInvalidError(parameter, value, hint)
    assert exc_info.value.value == value
    assert exc_info.value.message == f'Invalid `{parameter}` parameter: {value}. {hint}'

def test_editor_tool_parameter_invalid_error_without_hint():
    """Test EditorToolParameterInvalidError without hint."""
    value = -10
        raise EditorToolParameterInvalidError(parameter, value)
    assert exc_info.value.message == f'Invalid `{parameter}` parameter: {value}.'

from openhands_aci.editor.config import MAX_RESPONSE_LEN_CHAR
from openhands_aci.editor.prompts import CONTENT_TRUNCATED_NOTICE
from openhands_aci.editor.results import ToolResult, maybe_truncate

def test_tool_result_bool():
    """Test the boolean value of ToolResult based on output and error."""
    # Case: Both output and error are None
    result = ToolResult()
    assert not bool(result)

    # Case: Only output is set
    result = ToolResult(output='Some output')
    assert bool(result)

    # Case: Only error is set
    result = ToolResult(error='An error occurred')

    # Case: Both output and error are set
    result = ToolResult(output='Some output', error='An error occurred')

def test_maybe_truncate_no_truncation():
    """Test maybe_truncate when content does not exceed the length limit."""
    content = 'Short content'
    result = maybe_truncate(content, truncate_after=MAX_RESPONSE_LEN_CHAR)
    assert result == content  # Should return content as-is

def test_maybe_truncate_with_truncation():
    """Test maybe_truncate when content exceeds the length limit."""
    content = 'a' * (MAX_RESPONSE_LEN_CHAR + 10)
    assert result == content[:MAX_RESPONSE_LEN_CHAR] + CONTENT_TRUNCATED_NOTICE

def test_maybe_truncate_no_limit():
    """Test maybe_truncate when truncate_after is None."""
    content = 'Content that exceeds the default max length'
    result = maybe_truncate(content, truncate_after=None)
    assert result == content  # No truncation applied when limit is None

import subprocess
from unittest.mock import MagicMock, patch

from openhands_aci.utils.shell import check_tool_installed, run_shell_cmd

def test_run_shell_cmd_success():
    """Test running a successful shell command."""
    cmd = "echo 'Hello, World!'"
    returncode, stdout, stderr = run_shell_cmd(cmd)

    assert returncode == 0
    assert stdout.strip() == 'Hello, World!'
    assert stderr == ''

@patch('subprocess.Popen')
def test_run_shell_cmd_timeout(mock_popen):
    """Test that a TimeoutError is raised if command times out."""
    mock_process = MagicMock()
    mock_process.communicate.side_effect = subprocess.TimeoutExpired(
        cmd='sleep 2', timeout=1
    )
    mock_popen.return_value = mock_process

    with pytest.raises(TimeoutError, match="Command 'sleep 2' timed out"):
        run_shell_cmd('sleep 2', timeout=1)

def test_run_shell_cmd_truncation(mock_popen):
    """Test that stdout and stderr are truncated correctly."""
    long_output = 'a' * (MAX_RESPONSE_LEN_CHAR + 10)
    mock_process.communicate.return_value = (long_output, long_output)
    mock_process.returncode = 0

    returncode, stdout, stderr = run_shell_cmd('echo long_output')

    assert len(stdout) <= MAX_RESPONSE_LEN_CHAR + len(CONTENT_TRUNCATED_NOTICE)
    assert len(stderr) <= MAX_RESPONSE_LEN_CHAR + len(CONTENT_TRUNCATED_NOTICE)

def test_check_tool_installed_whoami():
    """Test check_tool_installed returns True for an installed tool (whoami)."""
    # 'python' is usually available if Python is installed
    assert check_tool_installed('whoami') is True

def test_check_tool_installed_nonexistent_tool():
    """Test check_tool_installed returns False for a nonexistent tool."""
    # Use a made-up tool name that is very unlikely to exist
    assert check_tool_installed('nonexistent_tool_xyz') is False

from unittest.mock import mock_open, patch

from openhands_aci.linter.base import LintResult

def mock_file_content():
    return '\n'.join([f'Line {i}' for i in range(1, 21)])

def test_visualize_standard_case(mock_file_content):
    lint_result = LintResult(
        file='test_file.py', line=10, column=5, message='Test error message'
    )

    with patch('builtins.open', mock_open(read_data=mock_file_content)):
        result = lint_result.visualize(half_window=3)

    expected_output = (
        " 7|Line 7\n"
        " 8|Line 8\n"
        " 9|Line 9\n"
        "\033[91m10|Line 10\033[0m\n"
        f"  {' ' * lint_result.column}^ ERROR HERE: Test error message\n"
        "11|Line 11\n"
        "12|Line 12\n"
        "13|Line 13"
    )

    assert result == expected_output

def test_visualize_small_window(mock_file_content):
    )

        result = lint_result.visualize(half_window=1)

        "11|Line 11"
    )

def test_visualize_error_at_start(mock_file_content):
        file='test_file.py', line=1, column=3, message='Start error'
    )

        result = lint_result.visualize(half_window=2)

        "\033[91m 1|Line 1\033[0m\n"
        f"  {' ' * lint_result.column}^ ERROR HERE: Start error\n"
        " 2|Line 2\n"
        " 3|Line 3"
    )

def test_visualize_error_at_end(mock_file_content):
        file='test_file.py', line=20, column=1, message='End error'
    )

        "18|Line 18\n"
        "19|Line 19\n"
        "\033[91m20|Line 20\033[0m\n"
        f"  {' ' * lint_result.column}^ ERROR HERE: End error"
    )

from openhands_aci.linter import DefaultLinter, LintResult
from openhands_aci.linter.impl.python import (
    PythonLinter,
    flake_lint,
    python_compile_lint,
)

def test_wrongly_indented_py_file(wrongly_indented_py_file):
    # Test Python linter
    linter = PythonLinter()
    assert '.py' in linter.supported_extensions
    result = linter.lint(wrongly_indented_py_file)
    print(result)
    assert isinstance(result, list) and len(result) == 1
    assert result[0] == LintResult(
        file=wrongly_indented_py_file,
        line=2,
        column=5,
        message='E999 IndentationError: unexpected indent',
    )
    print(result[0].visualize())
    assert result[0].visualize() == (
        '1|\n'
        '\033[91m2|    def foo():\033[0m\n'
        '      ^ ERROR HERE: E999 IndentationError: unexpected indent\n'
        '3|            print("Hello, World!")\n'
        '4|'
    )

    # General linter should have same result as Python linter
    # bc it uses PythonLinter under the hood
    general_linter = DefaultLinter()
    assert '.py' in general_linter.supported_extensions
    result = general_linter.lint(wrongly_indented_py_file)
    assert result == linter.lint(wrongly_indented_py_file)

    # Test flake8_lint
    assert result == flake_lint(wrongly_indented_py_file)

    # Test python_compile_lint
    compile_result = python_compile_lint(wrongly_indented_py_file)
    assert isinstance(compile_result, list) and len(compile_result) == 1
    assert compile_result[0] == LintResult(
        file=wrongly_indented_py_file, line=2, column=4, message='unexpected indent'
    )

def test_simple_correct_py_file(simple_correct_py_file):
    result = linter.lint(simple_correct_py_file)
    assert result == []

    result = general_linter.lint(simple_correct_py_file)
    assert result == linter.lint(simple_correct_py_file)

    compile_result = python_compile_lint(simple_correct_py_file)
    assert compile_result == []

    # Test flake_lint
    flake_result = flake_lint(simple_correct_py_file)
    assert flake_result == []

def test_simple_correct_py_func_def(simple_correct_py_func_def):
    result = linter.lint(simple_correct_py_func_def)

    result = general_linter.lint(simple_correct_py_func_def)
    assert result == linter.lint(simple_correct_py_func_def)

    assert result == flake_lint(simple_correct_py_func_def)

    compile_result = python_compile_lint(simple_correct_py_func_def)

from openhands_aci.utils.diff import get_diff, parse_diff

OLD_CONTENT = """
def foo():
    print("Hello, World!")
    x = UNDEFINED_VARIABLE
foo()
"""

NEW_CONTENT_V1 = (
    OLD_CONTENT
    + """
def new_function_that_causes_error():
    y = ANOTHER_UNDEFINED_VARIABLE
"""
)

NEW_CONTENT_V2 = """
def foo():
foo()
"""

def test_get_and_parse_diff(tmp_path):
    diff = get_diff(OLD_CONTENT, NEW_CONTENT_V1, 'test.py')
    print(diff)
    assert (
        diff
        == """
--- test.py
+++ test.py
@@ -6,0 +7,3 @@
+def new_function_that_causes_error():
+    y = ANOTHER_UNDEFINED_VARIABLE
+
""".strip()
    )

    print(
        '\n'.join(
            [f'{i+1}|{line}' for i, line in enumerate(NEW_CONTENT_V1.splitlines())]
        )
    )
    changes = parse_diff(diff)
    assert len(changes) == 3
    assert (
        changes[0].old is None
        and changes[0].new == 7
        and changes[0].line == 'def new_function_that_causes_error():'
    )
    assert (
        changes[1].old is None
        and changes[1].new == 8
        and changes[1].line == '    y = ANOTHER_UNDEFINED_VARIABLE'
    )
    assert changes[2].old is None and changes[2].new == 9 and changes[2].line == ''

def test_lint_with_diff_append(tmp_path):
    with open(tmp_path / 'old.py', 'w') as f:
        f.write(OLD_CONTENT)
    with open(tmp_path / 'new.py', 'w') as f:
        f.write(NEW_CONTENT_V1)

    linter = DefaultLinter()
    result: list[LintResult] = linter.lint_file_diff(
        str(tmp_path / 'old.py'),
        str(tmp_path / 'new.py'),
    )
    assert len(result) == 1
    assert (
        result[0].line == 8
        and result[0].column == 9
        and result[0].message == "F821 undefined name 'ANOTHER_UNDEFINED_VARIABLE'"
    )

def test_lint_with_diff_insert(tmp_path):
        f.write(NEW_CONTENT_V2)

    )
    assert (
        result[0].line == 5
    )

def test_lint_with_multiple_changes_and_errors(tmp_path):
    old_content = """
def foo():
    x = 10
foo()
"""
    new_content = """
def foo():
    y = 20

def bar():
    z = ANOTHER_UNDEFINED_VARIABLE
    return z + 1

foo()
bar()
"""
        f.write(old_content)
        f.write(new_content)

    )
    assert len(result) == 2
    assert (
        result[0].line == 4
        and result[0].message == "F821 undefined name 'UNDEFINED_VARIABLE'"
    )
    assert (
        result[1].line == 8
        and result[1].column == 9
        and result[1].message == "F821 undefined name 'ANOTHER_UNDEFINED_VARIABLE'"
    )

def test_lint_with_introduced_and_fixed_errors(tmp_path):
x = UNDEFINED_VARIABLE
y = 10
"""
x = 5
y = ANOTHER_UNDEFINED_VARIABLE
z = UNDEFINED_VARIABLE
"""

    )
    assert (
        result[0].line == 3
        and result[0].column == 5
    )
    assert (
        result[1].line == 4
        and result[1].column == 5
        and result[1].message == "F821 undefined name 'UNDEFINED_VARIABLE'"
    )

def test_lint_with_multiline_changes(tmp_path):
def complex_function(a, b, c):
    return (a +
            b +
            c)
"""
    return (a +
            UNDEFINED_VARIABLE +
            b +
            c)
"""

    )
    assert (
        and result[0].column == 13
    )

def test_lint_with_syntax_error(tmp_path):
def foo():
"""
def foo():
    print("Hello, World!"
"""

    )
    assert (
        and result[0].column == 11
        and result[0].message == "E999 SyntaxError: '(' was never closed"
    )

def test_lint_with_docstring_changes(tmp_path):
    old_content = '''
def foo():
    """This is a function."""
'''
    new_content = '''
def foo():
    """
    This is a function.
    It now has a multi-line docstring with an UNDEFINED_VARIABLE.
    """
'''

    )
    assert len(result) == 0  # Linter should ignore changes in docstrings

def test_lint_with_multiple_errors_on_same_line(tmp_path):
def foo():
    x = 10
foo()
"""
def foo():
    x = UNDEFINED_VARIABLE + ANOTHER_UNDEFINED_VARIABLE
foo()
"""

    )
    assert (
    )
    assert (
        and result[1].column == 30
    )

def test_parse_diff_with_empty_patch():
    diff_patch = ''
    changes = parse_diff(diff_patch)
    assert len(changes) == 0

def test_lint_file_diff_ignore_existing_errors(tmp_path):
    """
    Make sure we allow edits as long as it does not introduce new errors. In other
    words, we don't care about existing linting errors. Although they might be
    real syntax issues, sometimes they are just false positives, or errors that
    we don't care about.
    """
    content = """def some_valid_but_weird_function():
    # this function is legitimate, yet static analysis tools like flake8
    # reports 'F821 undefined name'
    if 'variable' in locals():
        print(variable)
def some_wrong_but_unused_function():
    # this function has a linting error, but it is not modified by us, and
    # who knows, this function might be completely dead code
    x = 1
def sum(a, b):
    return a - b
"""
    new_content = content.replace('    return a - b', '    return a + b')
    temp_file_old_path = tmp_path / 'problematic-file-test.py'
    temp_file_old_path.write_text(content)
    temp_file_new_path = tmp_path / 'problematic-file-test-new.py'
    temp_file_new_path.write_text(new_content)

        str(temp_file_old_path),
        str(temp_file_new_path),
    )
    assert len(result) == 0  # no new errors introduced

def test_lint_file_diff_catch_new_errors_in_edits(tmp_path):
    """
    Make sure we catch new linting errors in our edit chunk, and at the same
    time, ignore old linting errors (in this case, the old linting error is
    a false positive)
    """
    return a - b
"""

    new_content = content.replace('    return a - b', '    return a + variable')

    )
    assert (
        result[0].line == 7
        and result[0].column == 16
        and result[0].message == "F821 undefined name 'variable'"
    )

def test_lint_file_diff_catch_new_errors_outside_edits(tmp_path):
    """
    Make sure we catch new linting errors induced by our edits, even
    though the error itself is not in the edit chunk
    """
    content = """def valid_func1():
    print(my_sum(1, 2))
def my_sum(a, b):
    return a - b
def valid_func2():
    print(my_sum(0, 0))
"""
    # Add 100 lines of invalid code, which linter shall ignore
    # because they are not being edited. For testing purpose, we
    # must add these existing linting errors, otherwise the pre-edit
    # linting would pass, and thus there won't be any comparison
    # between pre-edit and post-edit linting.
    for _ in range(100):
        content += '\ninvalid_func()'

    new_content = content.replace('def my_sum(a, b):', 'def my_sum2(a, b):')

    )
    assert (
        result[0].line == 2
        and result[0].message == "F821 undefined name 'my_sum'"
    )
    assert (
        result[1].line == 6
        and result[1].column == 11
        and result[1].message == "F821 undefined name 'my_sum'"
    )

def syntax_error_py_file(tmp_path):
    file_content = """
    def foo():
        print("Hello, World!")
    print("Wrong indent")
    foo(
    """
    file_path = tmp_path / 'test_file.py'
    file_path.write_text(file_content)
    return str(file_path)

def wrongly_indented_py_file(tmp_path):
    def foo():
            print("Hello, World!")
    """

def simple_correct_py_file(tmp_path):
    file_content = 'print("Hello, World!")\n'

def simple_correct_py_func_def(tmp_path):
    file_content = """def foo():
foo()
"""

def simple_correct_ruby_file(tmp_path):
    file_content = """def foo
  puts "Hello, World!"
end
foo
"""
    file_path = tmp_path / 'test_file.rb'

def simple_incorrect_ruby_file(tmp_path):
foo()
"""

def parenthesis_incorrect_ruby_file(tmp_path):
    file_content = """def print_hello_world()\n    puts 'Hello World'\n"""

from openhands_aci.linter.impl.treesitter import TreesitterBasicLinter

def test_syntax_error_py_file(syntax_error_py_file):
    linter = TreesitterBasicLinter()
    result = linter.lint(syntax_error_py_file)
        file=syntax_error_py_file,
        line=5,
        column=5,
        message='Syntax error',
    )

    assert (
        result[0].visualize()
        == (
            '2|    def foo():\n'
            '3|        print("Hello, World!")\n'
            '4|    print("Wrong indent")\n'
            '\033[91m5|    foo(\033[0m\n'  # color red
            '      ^ ERROR HERE: Syntax error\n'
            '6|'
        )
    )

    general_result = general_linter.lint(syntax_error_py_file)
    # NOTE: general linter returns different result
    # because it uses flake8 first, which is different from treesitter
    assert general_result != result

def test_simple_correct_ruby_file(simple_correct_ruby_file):
    result = linter.lint(simple_correct_ruby_file)
    assert isinstance(result, list) and len(result) == 0

    # Test that the general linter also returns the same result
    general_result = general_linter.lint(simple_correct_ruby_file)
    assert general_result == result

def test_simple_incorrect_ruby_file(simple_incorrect_ruby_file):
    result = linter.lint(simple_incorrect_ruby_file)
    assert isinstance(result, list) and len(result) == 2
        file=simple_incorrect_ruby_file,
        line=1,
        column=1,
    )
    assert (
        == (
            '\033[91m1|def foo():\033[0m\n'  # color red
            '  ^ ERROR HERE: Syntax error\n'
            '2|    print("Hello, World!")\n'
            '3|foo()'
        )
    )
    assert result[1] == LintResult(
        line=1,
        column=10,
    )
    print(result[1].visualize())
    assert (
        result[1].visualize()
        == (
            '           ^ ERROR HERE: Syntax error\n'
            '3|foo()'
        )
    )

    general_result = general_linter.lint(simple_incorrect_ruby_file)

def test_parenthesis_incorrect_ruby_file(parenthesis_incorrect_ruby_file):
    result = linter.lint(parenthesis_incorrect_ruby_file)
        file=parenthesis_incorrect_ruby_file,
        line=1,
        column=1,
    )
        '\033[91m1|def print_hello_world()\033[0m\n'
        '  ^ ERROR HERE: Syntax error\n'
        "2|    puts 'Hello World'"
    )

    general_result = general_linter.lint(parenthesis_incorrect_ruby_file)

from openhands_aci.code_search import initialize_code_search, search_code

def main():
    # Initialize search for the openhands-aci repo
    print("Initializing code search...")
        repo_path='/workspace/openhands-aci',
        save_dir='/workspace/openhands-aci/code_search_index',
        extensions=['.py'],  # only index Python files
        embedding_model='BAAI/bge-base-en-v1.5'
    )
    print("Initialization result:", result)

    if result['status'] == 'success':
        # Try some searches
        queries = [
            "code that handles file editing",
            "function for running bash commands",
            "code related to linting or code analysis"
        ]

        print("\nTesting searches:")
        for query in queries:
            print(f"\nQuery: {query}")
            result = search_code(
                save_dir='/workspace/openhands-aci/code_search_index',
                query=query,
                k=3
            )
            if result['status'] == 'success':
                for doc in result['results']:
                    print(f"\nFile: {doc['path']}")
                    print(f"Score: {doc['score']:.3f}")
                    print("Content preview:", doc['content'][:200] + "...")
            else:
                print("Search failed:", result['message'])

if __name__ == '__main__':
    main()

from collections import defaultdict
from typing import Literal, get_args

from openhands_aci.linter import DefaultLinter
from openhands_aci.utils.shell import run_shell_cmd

from .config import SNIPPET_CONTEXT_WINDOW
from .exceptions import (
)
from .results import CLIResult, maybe_truncate

Command = Literal[
    'view',
    'create',
    'str_replace',
    'insert',
    'undo_edit',
    # 'jump_to_definition', TODO:
    # 'find_references' TODO:
]

class OHEditor:
    """
    An filesystem editor tool that allows the agent to
    - view
    - create
    - navigate
    - edit files
    The tool parameters are defined by Anthropic and are not editable.

    Original implementation: https://github.com/anthropics/anthropic-quickstarts/blob/main/computer-use-demo/computer_use_demo/tools/edit.py
    """

    TOOL_NAME = 'oh_editor'

    def __init__(self):
        self._file_history: dict[Path, list[str]] = defaultdict(list)
        self._linter = DefaultLinter()

    def __call__(
        self,
        *,
        command: Command,
        path: str,
        file_text: str | None = None,
        view_range: list[int] | None = None,
        old_str: str | None = None,
        new_str: str | None = None,
        insert_line: int | None = None,
        enable_linting: bool = False,
        **kwargs,
    ) -> CLIResult:
        _path = Path(path)
        self.validate_path(command, _path)
        if command == 'view':
            return self.view(_path, view_range)
        elif command == 'create':
            if not file_text:
                raise EditorToolParameterMissingError(command, 'file_text')
            self.write_file(_path, file_text)
            self._file_history[_path].append(file_text)
            return CLIResult(
                path=str(_path),
                new_content=file_text,
                prev_exist=False,
                output=f'File created successfully at: {_path}',
            )
        elif command == 'str_replace':
            if not old_str:
                raise EditorToolParameterMissingError(command, 'old_str')
            if new_str == old_str:
                raise EditorToolParameterInvalidError(
                    'new_str',
                    new_str,
                    'No replacement was performed. `new_str` and `old_str` must be different.',
                )
            return self.str_replace(_path, old_str, new_str, enable_linting)
        elif command == 'insert':
            if insert_line is None:
                raise EditorToolParameterMissingError(command, 'insert_line')
            if not new_str:
                raise EditorToolParameterMissingError(command, 'new_str')
            return self.insert(_path, insert_line, new_str, enable_linting)
        elif command == 'undo_edit':
            return self.undo_edit(_path)

        raise ToolError(
            f'Unrecognized command {command}. The allowed commands for the {self.TOOL_NAME} tool are: {", ".join(get_args(Command))}'
        )

    def str_replace(
        self, path: Path, old_str: str, new_str: str | None, enable_linting: bool
        """
        Implement the str_replace command, which replaces old_str with new_str in the file content.
        """
        file_content = self.read_file(path).expandtabs()
        old_str = old_str.expandtabs()
        new_str = new_str.expandtabs() if new_str is not None else ''

        # Check if old_str is unique in the file
        occurrences = file_content.count(old_str)
        if occurrences == 0:
            raise ToolError(
                f'No replacement was performed, old_str `{old_str}` did not appear verbatim in {path}.'
            )
        if occurrences > 1:
            # Find starting line numbers for each occurrence
            line_numbers = []
            start_idx = 0
            while True:
                idx = file_content.find(old_str, start_idx)
                if idx == -1:
                    break
                # Count newlines before this occurrence to get the line number
                line_num = file_content.count('\n', 0, idx) + 1
                line_numbers.append(line_num)
                start_idx = idx + 1
                f'No replacement was performed. Multiple occurrences of old_str `{old_str}` in lines {line_numbers}. Please ensure it is unique.'
            )

        # Replace old_str with new_str
        new_file_content = file_content.replace(old_str, new_str)

        # Write the new content to the file
        self.write_file(path, new_file_content)

        # Save the content to history
        self._file_history[path].append(file_content)

        # Create a snippet of the edited section
        replacement_line = file_content.split(old_str)[0].count('\n')
        start_line = max(0, replacement_line - SNIPPET_CONTEXT_WINDOW)
        end_line = replacement_line + SNIPPET_CONTEXT_WINDOW + new_str.count('\n')
        snippet = '\n'.join(new_file_content.split('\n')[start_line : end_line + 1])

        # Prepare the success message
        success_message = f'The file {path} has been edited. '
        success_message += self._make_output(
            snippet, f'a snippet of {path}', start_line + 1
        )

        if enable_linting:
            # Run linting on the changes
            lint_results = self._run_linting(file_content, new_file_content, path)
            success_message += '\n' + lint_results + '\n'

        success_message += 'Review the changes and make sure they are as expected. Edit the file again if necessary.'
        return CLIResult(
            output=success_message,
            prev_exist=True,
            path=str(path),
            old_content=file_content,
            new_content=new_file_content,
        )

    def view(self, path: Path, view_range: list[int] | None = None) -> CLIResult:
        """
        View the contents of a file or a directory.
        """
        if path.is_dir():
            if view_range:
                    'view_range',
                    view_range,
                    'The `view_range` parameter is not allowed when `path` points to a directory.',
                )

            _, stdout, stderr = run_shell_cmd(
                rf"find -L {path} -maxdepth 2 -not -path '*/\.*'"
            )
            if not stderr:
                stdout = f"Here's the files and directories up to 2 levels deep in {path}, excluding hidden items:\n{stdout}\n"
                output=stdout,
                error=stderr,
                path=str(path),
                prev_exist=True,
            )

        file_content = self.read_file(path)
        start_line = 1
        if not view_range:
                output=self._make_output(file_content, str(path), start_line),
            )

        if len(view_range) != 2 or not all(isinstance(i, int) for i in view_range):
            raise EditorToolParameterInvalidError(
                'view_range',
                view_range,
                'It should be a list of two integers.',
            )

        file_content_lines = file_content.split('\n')
        num_lines = len(file_content_lines)
        start_line, end_line = view_range
        if start_line < 1 or start_line > num_lines:
                f'Its first element `{start_line}` should be within the range of lines of the file: {[1, num_lines]}.',
            )

        if end_line > num_lines:
                f'Its second element `{end_line}` should be smaller than the number of lines in the file: `{num_lines}`.',
            )

        if end_line != -1 and end_line < start_line:
                f'Its second element `{end_line}` should be greater than or equal to the first element `{start_line}`.',
            )

        if end_line == -1:
            file_content = '\n'.join(file_content_lines[start_line - 1 :])
        else:
            file_content = '\n'.join(file_content_lines[start_line - 1 : end_line])
            output=self._make_output(file_content, str(path), start_line),
        )

    def write_file(self, path: Path, file_text: str) -> None:
        """
        Write the content of a file to a given path; raise a ToolError if an error occurs.
        """
        try:
            path.write_text(file_text)
        except Exception as e:
            raise ToolError(f'Ran into {e} while trying to write to {path}') from None

    def insert(
        self, path: Path, insert_line: int, new_str: str, enable_linting: bool
        """
        Implement the insert command, which inserts new_str at the specified line in the file content.
        """
        try:
            file_text = self.read_file(path)
            raise ToolError(f'Ran into {e} while trying to read {path}') from None

        file_text = file_text.expandtabs()
        new_str = new_str.expandtabs()

        file_text_lines = file_text.split('\n')
        num_lines = len(file_text_lines)

        if insert_line < 0 or insert_line > num_lines:
                'insert_line',
                insert_line,
                f'It should be within the range of lines of the file: {[0, num_lines]}',
            )

        new_str_lines = new_str.split('\n')
        new_file_text_lines = (
            file_text_lines[:insert_line]
            + new_str_lines
            + file_text_lines[insert_line:]
        )
        snippet_lines = (
            file_text_lines[max(0, insert_line - SNIPPET_CONTEXT_WINDOW) : insert_line]
            + file_text_lines[
                insert_line : min(num_lines, insert_line + SNIPPET_CONTEXT_WINDOW)
            ]
        )
        new_file_text = '\n'.join(new_file_text_lines)
        snippet = '\n'.join(snippet_lines)

        self.write_file(path, new_file_text)
        self._file_history[path].append(file_text)

            snippet,
            'a snippet of the edited file',
            max(1, insert_line - SNIPPET_CONTEXT_WINDOW + 1),
        )

            lint_results = self._run_linting(file_text, new_file_text, path)

        success_message += 'Review the changes and make sure they are as expected (correct indentation, no duplicate lines, etc). Edit the file again if necessary.'
            old_content=file_text,
            new_content=new_file_text,
        )

    def validate_path(self, command: Command, path: Path) -> None:
        """
        Check that the path/command combination is valid.
        """
        # Check if its an absolute path
        if not path.is_absolute():
            suggested_path = Path('') / path
                'path',
                path,
                f'The path should be an absolute path, starting with `/`. Maybe you meant {suggested_path}?',
            )
        # Check if path and command are compatible
        if command == 'create' and path.exists():
                'path',
                path,
                f'File already exists at: {path}. Cannot overwrite files using command `create`.',
            )
        if command != 'create' and not path.exists():
                'path',
                path,
                f'The path {path} does not exist. Please provide a valid path.',
            )
        if command != 'view' and path.is_dir():
                'path',
                path,
                f'The path {path} is a directory and only the `view` command can be used on directories.',
            )

    def undo_edit(self, path: Path) -> CLIResult:
        """
        Implement the undo_edit command.
        """
        if not self._file_history[path]:
            raise ToolError(f'No edit history found for {path}.')

        current_text = self.read_file(path).expandtabs()
        old_text = self._file_history[path].pop()
        self.write_file(path, old_text)

            output=f'Last edit to {path} undone successfully. {self._make_output(old_text, str(path))}',
            old_content=current_text,
            new_content=old_text,
        )

    def read_file(self, path: Path) -> str:
        """
        Read the content of a file from a given path; raise a ToolError if an error occurs.
        """
        try:
            return path.read_text()

    def _make_output(
        self,
        snippet_content: str,
        snippet_description: str,
        start_line: int = 1,
        expand_tabs: bool = True,
    ) -> str:
        """
        Generate output for the CLI based on the content of a code snippet.
        """
        snippet_content = maybe_truncate(snippet_content)
        if expand_tabs:
            snippet_content = snippet_content.expandtabs()

        snippet_content = '\n'.join(
            [
                f'{i + start_line:6}\t{line}'
                for i, line in enumerate(snippet_content.split('\n'))
            ]
        )
        return (
            f"Here's the result of running `cat -n` on {snippet_description}:\n"
            + snippet_content
            + '\n'
        )

    def _run_linting(self, old_content: str, new_content: str, path: Path) -> str:
        """
        Run linting on file changes and return formatted results.
        """
        # Create a temporary directory
        with tempfile.TemporaryDirectory() as temp_dir:
            # Create paths with exact filenames in temp directory
            temp_old = Path(temp_dir) / f'old.{path.name}'
            temp_new = Path(temp_dir) / f'new.{path.name}'

            # Write content to temporary files
            temp_old.write_text(old_content)
            temp_new.write_text(new_content)

            results = self._linter.lint_file_diff(str(temp_old), str(temp_new))

            if not results:
                return 'No linting issues found in the changes.'

            # Format results
            output = ['Linting issues found in the changes:']
            for result in results:
                output.append(
                    f'- Line {result.line}, Column {result.column}: {result.message}'
                )
            return '\n'.join(output) + '\n'

class ToolError(Exception):
    """Raised when a tool encounters an error."""

    def __init__(self, message):
        self.message = message

class EditorToolParameterMissingError(ToolError):
    """Raised when a required parameter is missing for a tool command."""

    def __init__(self, command, parameter):
        self.command = command
        self.parameter = parameter
        self.message = f'Parameter `{parameter}` is required for command: {command}.'

class EditorToolParameterInvalidError(ToolError):
    """Raised when a parameter is invalid for a tool command."""

    def __init__(self, parameter, value, hint=None):
        self.value = value
        self.message = (
            f'Invalid `{parameter}` parameter: {value}. {hint}'
            if hint
            else f'Invalid `{parameter}` parameter: {value}.'
        )

import uuid

from .editor import Command, OHEditor
from .exceptions import ToolError
from .results import ToolResult

_GLOBAL_EDITOR = OHEditor()

def _make_api_tool_result(tool_result: ToolResult) -> str:
    """Convert an agent ToolResult to an API ToolResultBlockParam."""
    if tool_result.error:
        return f'ERROR:\n{tool_result.error}'

    assert tool_result.output, 'Expected output in file_editor.'
    return tool_result.output

def file_editor(
    command: Command,
    path: str,
    file_text: str | None = None,
    view_range: list[int] | None = None,
    old_str: str | None = None,
    new_str: str | None = None,
    insert_line: int | None = None,
    enable_linting: bool = False,
) -> str:
    result: ToolResult | None = None
    try:
        result = _GLOBAL_EDITOR(
            command=command,
            path=path,
            file_text=file_text,
            view_range=view_range,
            old_str=old_str,
            new_str=new_str,
            insert_line=insert_line,
            enable_linting=enable_linting,
        )
    except ToolError as e:
        result = ToolResult(error=e.message)

    formatted_output_and_error = _make_api_tool_result(result)
    marker_id = uuid.uuid4().hex
    return f"""<oh_aci_output_{marker_id}>
{json.dumps(result.to_dict(extra_field={'formatted_output_and_error': formatted_output_and_error}), indent=2)}
</oh_aci_output_{marker_id}>"""

CONTENT_TRUNCATED_NOTICE: str = '<response clipped><NOTE>To save on context only part of this file has been shown to you. You should retry this tool after you have searched inside the file with `grep -n` in order to find the line numbers of what you are looking for.</NOTE>'

MAX_RESPONSE_LEN_CHAR: int = 16000
SNIPPET_CONTEXT_WINDOW: int = 4

from dataclasses import asdict, dataclass, fields

from .config import MAX_RESPONSE_LEN_CHAR
from .prompts import CONTENT_TRUNCATED_NOTICE

@dataclass
class ToolResult:
    """Represents the result of a tool execution."""

    output: str | None = None
    error: str | None = None

    def __bool__(self):
        return any(getattr(self, field.name) for field in fields(self))

    def to_dict(self, extra_field: dict | None = None) -> dict:
        result = asdict(self)

        # Add extra fields if provided
        if extra_field:
            result.update(extra_field)
        return result

class CLIResult(ToolResult):
    """A ToolResult that can be rendered as a CLI output."""

    # Optional fields for file editing commands
    path: str | None = None
    prev_exist: bool = True
    old_content: str | None = None
    new_content: str | None = None

def maybe_truncate(
    content: str, truncate_after: int | None = MAX_RESPONSE_LEN_CHAR
) -> str:
    """
    Truncate content and append a notice if content exceeds the specified length.
    """
    return (
        content
        if not truncate_after or len(content) <= truncate_after
        else content[:truncate_after] + CONTENT_TRUNCATED_NOTICE
    )

"""Linter module for OpenHands ACI.

Part of this Linter module is adapted from Aider (Apache 2.0 License, [original code](https://github.com/paul-gauthier/aider/blob/main/aider/linter.py)). Please see the [original repository](https://github.com/paul-gauthier/aider) for more information.
"""

from .base import LintResult
from .linter import DefaultLinter

__all__ = ['DefaultLinter', 'LintResult']

import os
from difflib import SequenceMatcher

from ..linter.base import BaseLinter, LinterException, LintResult
from ..linter.impl.python import PythonLinter
from ..linter.impl.treesitter import TreesitterBasicLinter

class DefaultLinter(BaseLinter):
        self.linters: dict[str, list[BaseLinter]] = defaultdict(list)
        self.linters['.py'] = [PythonLinter()]

        # Add treesitter linter as a fallback for all linters
        self.basic_linter = TreesitterBasicLinter()
        for extension in self.basic_linter.supported_extensions:
            self.linters[extension].append(self.basic_linter)
        self._supported_extensions = list(self.linters.keys())

    @property
    def supported_extensions(self) -> list[str]:
        return self._supported_extensions

    def lint(self, file_path: str) -> list[LintResult]:
        if not os.path.isabs(file_path):
            raise LinterException(f'File path {file_path} is not an absolute path')
        file_extension = os.path.splitext(file_path)[1]

        linters: list[BaseLinter] = self.linters.get(file_extension, [])
        for linter in linters:
            res = linter.lint(file_path)
            # We always return the first linter's result (higher priority)
            if res:
                return res
        return []

    def lint_file_diff(
        self, original_file_path: str, updated_file_path: str
    ) -> list[LintResult]:
        """Only return lint errors that are introduced by the diff.

        Args:
            original_file_path: The original file path.
            updated_file_path: The updated file path.

        Returns:
            A list of lint errors that are introduced by the diff.
        """
        # 1. Lint the original and updated file
        original_lint_errors: list[LintResult] = self.lint(original_file_path)
        updated_lint_errors: list[LintResult] = self.lint(updated_file_path)

        # 2. Load the original and updated file content
        with open(original_file_path, 'r') as f:
            old_lines = f.readlines()
        with open(updated_file_path, 'r') as f:
            new_lines = f.readlines()

        # 3. Get line numbers that are changed & unchanged
        # Map the line number of the original file to the updated file
        # NOTE: this only works for lines that are not changed (i.e., equal)
        old_to_new_line_no_mapping: dict[int, int] = {}
        replace_or_inserted_lines: list[int] = []
        for (
            tag,
            old_idx_start,
            old_idx_end,
            new_idx_start,
            new_idx_end,
        ) in SequenceMatcher(
            isjunk=None,
            a=old_lines,
            b=new_lines,
        ).get_opcodes():
            if tag == 'equal':
                for idx, _ in enumerate(old_lines[old_idx_start:old_idx_end]):
                    old_to_new_line_no_mapping[old_idx_start + idx + 1] = (
                        new_idx_start + idx + 1
                    )
            elif tag == 'replace' or tag == 'insert':
                    replace_or_inserted_lines.append(new_idx_start + idx + 1)
            else:
                # omit the case of delete
                pass

        # 4. Get pre-existing errors in unchanged lines
        # increased error elsewhere introduced by the newlines
        # i.e., we omit errors that are already in original files and report new one
        new_line_no_to_original_errors: dict[int, list[LintResult]] = defaultdict(list)
        for error in original_lint_errors:
            if error.line in old_to_new_line_no_mapping:
                new_line_no_to_original_errors[
                    old_to_new_line_no_mapping[error.line]
                ].append(error)

        # 5. Select errors from lint results in new file to report
        selected_errors = []
        for error in updated_lint_errors:
            # 5.1. Error introduced by replace/insert
            if error.line in replace_or_inserted_lines:
                selected_errors.append(error)
            # 5.2. Error introduced by modified lines that impacted
            #      the unchanged lines that HAVE pre-existing errors
            elif error.line in new_line_no_to_original_errors:
                # skip if the error is already reported
                # or add if the error is new
                if not any(
                    original_error.message == error.message
                    and original_error.column == error.column
                    for original_error in new_line_no_to_original_errors[error.line]
                ):
                    selected_errors.append(error)
            # 5.3. Error introduced by modified lines that impacted
            #      the unchanged lines that have NO pre-existing errors
            else:

        # 6. Sort errors by line and column
        selected_errors.sort(key=lambda x: (x.line, x.column))
        return selected_errors

from abc import ABC, abstractmethod

from pydantic import BaseModel

class LintResult(BaseModel):
    file: str
    line: int  # 1-indexed
    column: int  # 1-indexed
    message: str

    def visualize(self, half_window: int = 3) -> str:
        """Visualize the lint result by print out all the lines where the lint result is found.

        Args:
            half_window: The number of context lines to display around the error on each side.
        """
        with open(self.file, 'r') as f:
            file_lines = f.readlines()

        # Add line numbers
        _span_size = len(str(len(file_lines)))
        file_lines = [
            f'{i + 1:>{_span_size}}|{line.rstrip()}'
            for i, line in enumerate(file_lines)
        ]

        # Get the window of lines to display
        assert self.line <= len(file_lines) and self.line > 0
        line_idx = self.line - 1
        begin_window = max(0, line_idx - half_window)
        end_window = min(len(file_lines), line_idx + half_window + 1)

        selected_lines = file_lines[begin_window:end_window]
        line_idx_in_window = line_idx - begin_window

        # Add character hint
        _character_hint = (
            _span_size * ' '
            + ' ' * (self.column)
            + '^'
            + ' ERROR HERE: '
            + self.message
        )
        selected_lines[line_idx_in_window] = (
            f'\033[91m{selected_lines[line_idx_in_window]}\033[0m'
            + '\n'
            + _character_hint
        )
        return '\n'.join(selected_lines)

class LinterException(Exception):
    """Base class for all linter exceptions."""

    pass

class BaseLinter(ABC):
    """Base class for all linters.

    Each linter should be able to lint files of a specific type and return a list of (parsed) lint results.
    """

    encoding: str = 'utf-8'

    @property
    @abstractmethod
        """The file extensions that this linter supports, such as .py or .tsx."""
        return []

        """Lint the given file.

        file_path: The path to the file to lint. Required to be absolute.
        """
        pass

from typing import List

from openhands_aci.utils.logger import oh_aci_logger as logger

from ..base import BaseLinter, LintResult

def python_compile_lint(fname: str) -> list[LintResult]:
    try:
        with open(fname, 'r') as f:
            code = f.read()
        compile(code, fname, 'exec')  # USE TRACEBACK BELOW HERE
        return []
    except SyntaxError as err:
        err_lineno = getattr(err, 'end_lineno', err.lineno)
        err_offset = getattr(err, 'end_offset', err.offset)
        if err_offset and err_offset < 0:
            err_offset = err.offset
        return [
            LintResult(
                file=fname, line=err_lineno, column=err_offset or 1, message=err.msg
            )
        ]

def flake_lint(filepath: str) -> list[LintResult]:
    fatal = 'F821,F822,F831,E112,E113,E999,E902'
    flake8_cmd = f'flake8 --select={fatal} --isolated {filepath}'

    try:
        cmd_outputs = run_shell_cmd(flake8_cmd, truncate_after=None)[1]
    except FileNotFoundError:
        return []
    results: list[LintResult] = []
    if not cmd_outputs:
        return results
    for line in cmd_outputs.splitlines():
        parts = line.split(':')
        if len(parts) >= 4:
            _msg = parts[3].strip()
            if len(parts) > 4:
                _msg += ': ' + parts[4].strip()

            try:
                line_num = int(parts[1])
            except ValueError as e:
                logger.warning(
                    f'Error parsing flake8 output for line: {e}. Parsed parts: {parts}. Skipping...'
                )
                continue

            try:
                column_num = int(parts[2])
                column_num = 1
                _msg = (
                    parts[2].strip() + ' ' + _msg
                )  # add the unparsed message to the original message
                    f'Error parsing flake8 output for column: {e}. Parsed parts: {parts}. Using default column 1.'
                )

            results.append(
                LintResult(
                    file=filepath,
                    line=line_num,
                    column=column_num,
                    message=_msg,
                )
            )
    return results

class PythonLinter(BaseLinter):
    @property
    def supported_extensions(self) -> List[str]:
        return ['.py']

        error = flake_lint(file_path)
        if not error:
            error = python_compile_lint(file_path)
        return error

    def compile_lint(self, file_path: str, code: str) -> List[LintResult]:
        try:
            compile(code, file_path, 'exec')
            return []
        except SyntaxError as e:
            return [
                    file=file_path,
                    line=e.lineno,
                    column=e.offset,
                    message=str(e),
                    rule='SyntaxError',
                )
            ]

import warnings

from grep_ast import TreeContext, filename_to_lang
from grep_ast.parsers import PARSERS
from tree_sitter_languages import get_parser

# tree_sitter is throwing a FutureWarning
warnings.simplefilter('ignore', category=FutureWarning)

def tree_context(fname, code, line_nums):
    context = TreeContext(
        fname,
        code,
        color=False,
        line_number=True,
        child_context=False,
        last_line=False,
        margin=0,
        mark_lois=True,
        loi_pad=3,
        # header_max=30,
        show_top_of_file_parent_scope=False,
    )
    line_nums = set(line_nums)
    context.add_lines_of_interest(line_nums)
    context.add_context()
    output = context.format()
    return output

def traverse_tree(node):
    """Traverses the tree to find errors."""
    errors = []
    if node.type == 'ERROR' or node.is_missing:
        line_no = node.start_point[0] + 1
        col_no = node.start_point[1] + 1
        error_type = 'Missing node' if node.is_missing else 'Syntax error'
        errors.append((line_no, col_no, error_type))

    for child in node.children:
        errors += traverse_tree(child)

    return errors

class TreesitterBasicLinter(BaseLinter):
    @property
        return list(PARSERS.keys())

        """Use tree-sitter to look for syntax errors, display them with tree context."""
        lang = filename_to_lang(file_path)
        if not lang:
            return []
        parser = get_parser(lang)
        with open(file_path, 'r') as f:
        tree = parser.parse(bytes(code, 'utf-8'))
        errors = traverse_tree(tree.root_node)
        if not errors:
            return []
        return [
                file=file_path,
                line=int(line),
                column=int(col),
                message=error_details,
            )
            for line, col, error_details in errors
        ]

from .core import CodeSearchIndex
from .tools import initialize_code_search, search_code

__all__ = ['CodeSearchIndex', 'initialize_code_search', 'search_code']

import os
from typing import List, Dict, Any, Optional

from git.exc import InvalidGitRepositoryError

def _get_files_from_repo(repo_path: str, extensions: Optional[List[str]] = None) -> List[Dict[str, Any]]:
    """Get all files from a git repository.
    
    Args:
        repo_path: Path to the git repository
        extensions: List of file extensions to include (e.g. ['.py', '.js'])
                  If None, includes all files
                  
    Returns:
        List of document dictionaries with 'id' and 'content' keys
    """
    try:
        repo = Repo(repo_path)
    except InvalidGitRepositoryError:
        raise ValueError(f"{repo_path} is not a valid git repository")

    documents = []
    repo_path = Path(repo_path)

    for root, _, files in os.walk(repo_path):
        if '.git' in root:
            continue

        for file in files:
            if extensions and not any(file.endswith(ext) for ext in extensions):
                continue

            file_path = Path(root) / file
            try:
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = f.read()
            except (UnicodeDecodeError, IOError):
                continue

            rel_path = file_path.relative_to(repo_path)
            documents.append({
                'id': str(rel_path),
                'content': content,
                'path': str(rel_path)
            })

    return documents

def initialize_code_search(repo_path: str, 
                         save_dir: str,
                         extensions: Optional[List[str]] = None,
                         embedding_model: Optional[str] = None,
                         batch_size: int = 32) -> Dict[str, Any]:
    """Initialize code search for a repository.
    
    Args:
        save_dir: Directory to save the search index
        extensions: List of file extensions to include
        embedding_model: Name or path of the embedding model to use
        batch_size: Batch size for embedding generation
        
    Returns:
        Dictionary with status and message
    """
    try:
        # Get all files from repo
        documents = _get_files_from_repo(repo_path, extensions)
        if not documents:
            return {
                'status': 'error',
                'message': f'No files found in repository {repo_path}'
            }

        # Create and save index
        index = CodeSearchIndex(embedding_model=embedding_model)
        index.add_documents(documents, batch_size=batch_size)
        index.save(save_dir)

        return {
            'status': 'success',
            'message': f'Successfully indexed {len(documents)} files from {repo_path}',
            'num_documents': len(documents)
        }

    except Exception as e:
        return {
            'status': 'error',
            'message': f'Error initializing code search: {str(e)}'
        }

def search_code(save_dir: str,
                query: str,
                k: int = 5) -> Dict[str, Any]:
    """Search code in an indexed repository.
    
    Args:
        save_dir: Directory containing the search index
        query: Search query
        k: Number of results to return
        
    Returns:
        Dictionary with status and search results
    """
    try:
        # Load index
        index = CodeSearchIndex.load(save_dir)
        
        # Search
        results = index.search(query, k=k)
        
        return {
            'results': results
        }

        return {
            'message': f'Error searching code: {str(e)}'
        }

import os
import pickle
from typing import List, Dict, Any, Optional, Tuple

import faiss
import numpy as np
import torch
from sentence_transformers import SentenceTransformer

class CodeSearchIndex:
    def __init__(self, embedding_model: Optional[str] = None):
        """Initialize the code search index.
        
        Args:
            embedding_model: Name or path of the sentence transformer model to use.
                           If None, will use the model specified in EMBEDDING_MODEL env var.
        """
        self.embedding_model = embedding_model or os.getenv('EMBEDDING_MODEL', 'BAAI/bge-base-en-v1.5')
        self.model = SentenceTransformer(self.embedding_model)
        self.index = None
        self.documents = []
        self.doc_ids = []

    def _embed_text(self, text: str) -> np.ndarray:
        """Embed a single text string."""
        with torch.no_grad():
            embedding = self.model.encode(text, convert_to_tensor=True)
            return embedding.cpu().numpy()

    def _embed_batch(self, texts: List[str], batch_size: int = 32) -> np.ndarray:
        """Embed a batch of text strings."""
        embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            with torch.no_grad():
                batch_embeddings = self.model.encode(batch, convert_to_tensor=True)
                embeddings.append(batch_embeddings.cpu().numpy())
        return np.vstack(embeddings)

    def add_documents(self, documents: List[Dict[str, Any]], batch_size: int = 32):
        """Add documents to the index.
        
        Args:
            documents: List of document dictionaries with 'id' and 'content' keys
            batch_size: Batch size for embedding generation
        """
        texts = [doc['content'] for doc in documents]
        embeddings = self._embed_batch(texts, batch_size)
        
        if self.index is None:
            self.index = faiss.IndexFlatIP(embeddings.shape[1])
        
        self.index.add(embeddings)
        self.documents.extend(documents)
        self.doc_ids.extend([doc['id'] for doc in documents])

    def search(self, query: str, k: int = 5) -> List[Dict[str, Any]]:
        """Search the index with a query string.
        
        Args:
            query: The search query
            k: Number of results to return
            
        Returns:
            List of document dictionaries with scores
        """
        query_embedding = self._embed_text(query)
        scores, indices = self.index.search(query_embedding.reshape(1, -1), k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < 0 or idx >= len(self.documents):
                continue
            doc = self.documents[idx].copy()
            doc['score'] = float(score)
            results.append(doc)
        
    def save(self, directory: str):
        """Save the index and documents to disk.
        
        Args:
            directory: Directory to save the index in
        """
        directory = Path(directory)
        directory.mkdir(parents=True, exist_ok=True)
        
        # Save the Faiss index
        faiss.write_index(self.index, str(directory / 'index.faiss'))
        
        # Save documents and metadata
        with open(directory / 'documents.pkl', 'wb') as f:
            pickle.dump({
                'documents': self.documents,
                'doc_ids': self.doc_ids,
                'embedding_model': self.embedding_model
            }, f)

    @classmethod
    def load(cls, directory: str) -> 'CodeSearchIndex':
        """Load an index from disk.
        
        Args:
            directory: Directory containing the saved index
            
        Returns:
            Loaded CodeSearchIndex instance
        """
        
        # Load metadata
        with open(directory / 'documents.pkl', 'rb') as f:
            data = pickle.load(f)
        
        # Create instance with same model
        instance = cls(embedding_model=data['embedding_model'])
        instance.documents = data['documents']
        instance.doc_ids = data['doc_ids']
        
        # Load Faiss index
        instance.index = faiss.read_index(str(directory / 'index.faiss'))
        
        return instance

import os
import time

from openhands_aci.editor.results import maybe_truncate

def run_shell_cmd(
    cmd: str,
    timeout: float | None = 120.0,  # seconds
    truncate_after: int | None = MAX_RESPONSE_LEN_CHAR,
) -> tuple[int, str, str]:
    """Run a shell command synchronously with a timeout.

    Args:
        cmd: The shell command to run.
        timeout: The maximum time to wait for the command to complete.
        truncate_after: The maximum number of characters to return for stdout and stderr.

    Returns:
        A tuple containing the return code, stdout, and stderr.
    """

    start_time = time.time()

    try:
        process = subprocess.Popen(
            cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )

        stdout, stderr = process.communicate(timeout=timeout)

        return (
            process.returncode or 0,
            maybe_truncate(stdout, truncate_after=truncate_after),
            maybe_truncate(stderr, truncate_after=truncate_after),
        )
    except subprocess.TimeoutExpired:
        process.kill()
        elapsed_time = time.time() - start_time
        raise TimeoutError(
            f"Command '{cmd}' timed out after {elapsed_time:.2f} seconds"
        )

def check_tool_installed(tool_name: str) -> bool:
    """Check if a tool is installed."""
    try:
        subprocess.run(
            [tool_name, '--version'],
            check=True,
            cwd=os.getcwd(),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        return True
    except (subprocess.CalledProcessError, FileNotFoundError):
        return False

import logging
import os

LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO').upper()

DEBUG = os.getenv('DEBUG', 'False').lower() in ['true', '1', 'yes']
if DEBUG:
    LOG_LEVEL = 'DEBUG'

oh_aci_logger = logging.getLogger('openhands_aci')

current_log_level = logging.INFO
if LOG_LEVEL in logging.getLevelNamesMapping():
    current_log_level = logging.getLevelNamesMapping()[LOG_LEVEL]

console_handler = logging.StreamHandler()
console_handler.setLevel(current_log_level)
formatter = logging.Formatter(
    '{asctime} - {name}:{levelname} - {message}',
    style='{',
    datefmt='%Y-%m-%d %H:%M',
)
console_handler.setFormatter(formatter)

oh_aci_logger.setLevel(current_log_level)
oh_aci_logger.addHandler(console_handler)
oh_aci_logger.propagate = False
oh_aci_logger.debug('Logger initialized')

import difflib

import whatthepatch

def get_diff(old_contents: str, new_contents: str, filepath: str = 'file') -> str:
    diff = list(
        difflib.unified_diff(
            old_contents.split('\n'),
            new_contents.split('\n'),
            fromfile=filepath,
            tofile=filepath,
            # do not output unchange lines
            # because they can cause `parse_diff` to fail
            n=0,
        )
    )
    return '\n'.join(map(lambda x: x.rstrip(), diff))

def parse_diff(diff_patch: str) -> list[whatthepatch.patch.Change]:
    # handle empty patch
    if diff_patch.strip() == '':
        return []

    patch = whatthepatch.parse_patch(diff_patch)
    patch_list = list(patch)
    assert len(patch_list) == 1, (
        'parse_diff only supports single file diff. But got:\nPATCH:\n'
        + diff_patch
        + '\nPATCH LIST:\n'
        + str(patch_list)
    )
    changes = patch_list[0].changes

    # ignore changes that are the same (i.e., old_lineno == new_lineno)
    output_changes = []
    for change in changes:
        if change.old != change.new:
            output_changes.append(change)
    return output_changes