File size: 1,998 Bytes
6085b61
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from langchain_core.messages import SystemMessage

from agents.base import get_runtime, invoke_agent

CODER_SYSTEM = """You are a Coder agent β€” the agent that implements code changes.

You have access to the following tools in two categories:

=== SEARCH TOOLS ===
1. search_codebase(keyword, directory) β€” Exact text matching.
   Use for finding specific strings, variable names, function references.
2. semantic_search(query, k=5) β€” Semantic (meaning-based) search.
   Use for finding code by concept or functionality.

=== FILE / EXECUTION TOOLS ===
3. list_files(directory) β€” Show directory tree.
4. read_file(filepath) β€” Read a file (truncated at 2000 lines).
5. write_to_file(filepath, content) β€” Write/replace a file.
6. run_bash_command(command, workspace_dir) β€” Execute bash inside Docker sandbox.
7. run_tests(workspace_dir) β€” Run pytest inside Docker sandbox.

=== GIT TOOLS ===
8. create_branch(branch_name) β€” Create a new git branch.
9. commit_changes(message) β€” Stage and commit all changes.
10. generate_pr_description() β€” Generate a PR description from the diff.

=== RULES ===
- Read the Planner's detailed plan and implement each step.
- Use write_to_file to make changes. Always write complete files.
- Run tests (run_tests) after writing code to verify correctness.
- If tests fail, read the error output and fix the code.
- You MUST use write_to_file at least once before declaring the task done.
- Prefer semantic_search for understanding code structure;
  use search_codebase only for exact string lookups.
"""

NO_WRITE_MSG = SystemMessage(
    content=(
        "You have not written any files yet. You MUST use write_to_file to implement "
        "the changes before finishing."
    )
)


def coder_node(state: dict) -> dict:
    extra = []
    if not state.get("writes_performed", False) and state["iteration_count"] > 0:
        extra = [NO_WRITE_MSG]
    result = invoke_agent(CODER_SYSTEM, state, "coder", extra_messages=extra)
    return result