File size: 3,230 Bytes
2bf281f
 
 
 
 
e9e55ec
2bf281f
 
 
e9e55ec
 
2bf281f
 
 
 
 
 
37f9abc
 
 
 
 
 
 
 
c706455
 
 
c6231ab
c706455
37f9abc
 
 
 
 
 
 
 
c706455
 
37f9abc
 
 
c6231ab
37f9abc
 
 
 
 
 
 
 
c706455
37f9abc
c706455
c6231ab
c706455
37f9abc
 
 
 
 
 
 
 
 
 
c706455
 
37f9abc
 
 
ea15e09
37f9abc
 
 
 
 
 
 
 
 
c706455
37f9abc
 
 
 
 
 
 
 
e9e55ec
 
 
2bf281f
e9e55ec
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
from langchain.tools import ToolRuntime, tool

from multi_agent_sdlc.runtime.process import execute_process
from multi_agent_sdlc.runtime.workspace import get_project_directory
from multi_agent_sdlc.state import DevState
from multi_agent_sdlc.tools.coder.descriptions import (
    RUN_APPLICATION_DESCRIPTION,
    RUN_PYTHON_MODULE_DESCRIPTION,
    RUN_SYNC_PROJECT,
    RUN_VERIFICATION_COMMAND_DESCRIPTION,
)
from multi_agent_sdlc.tools.coder.validation import (
    ApplicationArguments,
    EntryPoint,
    ExecutionTimeout,
    PythonModuleName,
    StandardInput,
)


@tool(
    "run_application",
    description=RUN_APPLICATION_DESCRIPTION,
)
def coder_run_application(
    entry_point: EntryPoint,
    runtime: ToolRuntime[DevState],
    arguments: ApplicationArguments | None = None,
    stdin_text: StandardInput | None = None,
    timeout_seconds: ExecutionTimeout = 15,
) -> dict[str, object]:

    project_directory = get_project_directory(runtime)

    return execute_process(
        [
            "uv",
            "run",
            entry_point,
            *(arguments or []),
        ],
        project_directory=project_directory,
        timeout_seconds=timeout_seconds,
        stdin_text=stdin_text,
    )


@tool(
    "run_python_module",
    description=RUN_PYTHON_MODULE_DESCRIPTION,
)
def coder_run_python_module(
    module: PythonModuleName,
    runtime: ToolRuntime[DevState],
    arguments: ApplicationArguments | None = None,
    stdin_text: StandardInput | None = None,
    timeout_seconds: ExecutionTimeout = 15,
) -> dict[str, object]:

    project_directory = get_project_directory(runtime)

    return execute_process(
        [
            "uv",
            "run",
            "python",
            "-m",
            module,
            *(arguments or []),
        ],
        project_directory=project_directory,
        timeout_seconds=timeout_seconds,
        stdin_text=stdin_text,
    )


@tool(
    "sync_project",
    description=RUN_SYNC_PROJECT,
)
def coder_sync_project(
    runtime: ToolRuntime[DevState],
    timeout_seconds: ExecutionTimeout = 120,
) -> dict[str, object]:
    project_directory = get_project_directory(runtime)

    return execute_process(
        ["uv", "sync"],
        project_directory=project_directory,
        timeout_seconds=timeout_seconds,
    )


from typing import Annotated, Literal

from pydantic import Field

VerificationCommand = Annotated[
    Literal[
        "ruff",
        "mypy",
        "coverage",
    ],
    Field(
        description=(
            "Approved development or verification executable to run "
            "inside the project's uv-managed environment."
        )
    ),
]


@tool(
    "run_verification_command",
    description=RUN_VERIFICATION_COMMAND_DESCRIPTION,
)
def coder_run_verification_command(
    command: VerificationCommand,
    runtime: ToolRuntime[DevState],
    arguments: ApplicationArguments | None = None,
    timeout_seconds: ExecutionTimeout = 120,
) -> dict[str, object]:
    project_directory = get_project_directory(runtime)

    return execute_process(
        ["uv", "run", command, *(arguments or [])],
        project_directory=project_directory,
        timeout_seconds=timeout_seconds,
    )