RepoForge/execution_plan.py at main · Gitcoder12/RepoForge · GitHub

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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
"""
RepoForge Execution Plan

Creates a safe, structured execution blueprint
before applying repository changes.
"""

from dataclasses import dataclass, asdict
from typing import Dict, Any, List


@dataclass
class ExecutionTask:
    name: str
    description: str
    priority: int
    risk: str
    files: List[str]
    reversible: bool = True


class ExecutionPlan:

    def __init__(self):
        self.tasks = []


    def create(
        self,
        action: Dict[str, Any],
        risk: Dict[str, Any],
        repository: Dict[str, Any],
    ) -> Dict[str, Any]:

        task = ExecutionTask(

            name=action.get(
                "title",
                "Repository improvement"
            ),

            description=action.get(
                "description",
                ""
            ),

            priority=action.get(
                "priority",
                1
            ),

            risk=risk.get(
                "level",
                "UNKNOWN"
            ),

            files=action.get(
                "files",
                []
            ),

            reversible=risk.get(
                "rollback_required",
                True
            )

        )


        self.tasks.append(task)


        return {

            "status": "READY",

            "repository":

                self._repository_name(
                    repository
                ),

            "risk":

                risk,

            "tasks":

                [
                    asdict(t)
                    for t in self.tasks
                ],

            "approval_required":

                self._approval_required(
                    risk
                ),

            "execution_order":

                self._order()

        }


    def _repository_name(
        self,
        repository
    ):

        return repository.get(
            "name",
            "unknown"
        )


    def _approval_required(
        self,
        risk
    ):

        level = risk.get(
            "level",
            "LOW"
        )


        return level in {
            "HIGH",
            "CRITICAL"
        }


    def _order(self):

        ordered = sorted(
            self.tasks,
            key=lambda x: x.priority
        )


        return [
            task.name
            for task in ordered
        ]


    def clear(self):

        self.tasks = []