openorbit/backend/app/models.py at main · ghzhost/openorbit · 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
from __future__ import annotations

from datetime import datetime
from typing import Any, Literal

from pydantic import BaseModel, Field, field_validator

# Legacy keys remain accepted when reading stored workflows and integrations.
# New assets and run records always use these canonical, task-agnostic names.
PHASE_ALIASES = {
    "init": "before_all",
    "setup": "before_each",
    "run": "execute",
    "eval": "verify",
    "teardown": "after_each",
    "finalize": "after_all",
}
LIFECYCLE_PHASES = ("before_all", "before_each", "execute", "verify", "after_each", "after_all")


class Step(BaseModel):
    id: str
    phase: Literal["before_all", "before_each", "execute", "verify", "after_each", "after_all"]
    name: str
    command: list[str]
    working_directory: str
    timeout_seconds: int = Field(default=300, ge=1, le=86_400)
    approval: Literal["not_required", "required"] = "not_required"
    on_failure: Literal["stop", "continue"] = "stop"
    minimum_interval_seconds: int = Field(default=0, ge=0, le=86_400)

    @field_validator("phase", mode="before")
    @classmethod
    def normalize_legacy_phase(cls, value: object) -> object:
        return PHASE_ALIASES.get(value, value) if isinstance(value, str) else value


class Workflow(BaseModel):
    id: str
    behavior_bundle: str | None = None
    name: str
    description: str
    kind: Literal["simulation", "improvement"]
    enabled: bool = False
    risk: Literal["low", "medium", "high"]
    tags: list[str] = []
    runner_id: str | None = None
    steps: list[Step]
    test_steps: list[Step] | None = None

    def steps_for(self, execution_mode: Literal["run", "test"]) -> list[Step]:
        if execution_mode == "test" and self.test_steps is not None:
            return self.test_steps
        return self.steps

    def lifecycle_is_complete(self, execution_mode: Literal["run", "test"] = "run") -> bool:
        phases = [step.phase for step in self.steps_for(execution_mode)]
        return phases in (
            ["before_all", "before_each", "execute", "verify", "after_each"],
            ["before_all", "before_each", "execute", "verify", "after_each", "after_all"],
        )


class Run(BaseModel):
    id: str
    workflow_id: str
    workflow_name: str
    build_id: str | None = None
    build_name: str | None = None
    repository: str | None = None
    supervisor_profile_name: str | None = None
    prompt_source: str | None = None
    prompt_snapshot: str | None = None
    execution_mode: Literal["run", "test"] = "run"
    execution_type: Literal["pipeline", "invoke"] = "pipeline"
    loop_limit: int = 1
    timezone: str = "UTC"
    schedule_enabled: bool = False
    schedule_weekdays: list[int] = Field(default_factory=list)
    schedule_start_time: str = "00:00"
    schedule_end_time: str = "23:59"
    start_iteration: int = 1
    retry_of_run_id: str | None = None
    retry_mode: Literal["restart", "resume"] | None = None
    iteration_strategy: Literal["linear", "score_select"] = "linear"
    candidates_per_iteration: int = 1
    iteration_candidates: list[dict[str, Any]] = Field(default_factory=list)
    repeat_interval_minutes: int = 0
    cadence_mode: Literal["after_completion", "fixed"] = "after_completion"
    overrun_policy: Literal["wait", "interrupt_eval"] = "wait"
    iteration_deadline_at: datetime | None = None
    advance_requested: bool = False
    approval_score: int | None = None
    status: Literal["queued", "awaiting_approval", "running", "succeeded", "failed", "cancelled"]
    created_at: datetime
    updated_at: datetime
    finished_at: datetime | None = None
    current_step: str | None = None
    current_phase: str | None = None
    pid: int | None = None
    last_pid: int | None = None
    telemetry_trace_id: str | None = None
    supervisor_status: Literal["pending", "completed", "not_configured", "invalid_response", "failed"] = (
        "pending"
    )
    supervisor_response: dict[str, Any] | None = None
    supervisor_error: str | None = None
    supervisor_results: list[dict[str, Any]] = Field(default_factory=list)
    runner_output: str = ""
    step_results: list[dict] = []
    approval_reason: str | None = None