56 lines
2.4 KiB
Python
56 lines
2.4 KiB
Python
|
# orchestrator/step_result.py
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import Any, Literal
|
||
|
|
|
||
|
|
# Terminal reasons are strings, not an enum, so you can add domain-specific
|
||
|
|
# reasons (e.g. "escalated:ci_infra") without touching this module.
|
||
|
|
TerminalReason = str
|
||
|
|
AwaitState = str # must be one of the AWAITING_* names, or "ABANDONED"
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class StepResult:
|
||
|
|
"""
|
||
|
|
The outcome of a step handler.
|
||
|
|
|
||
|
|
This is control flow, not data. The step's contract *output* lives in
|
||
|
|
`output`; the fields here tell the runner what to do next. The runner
|
||
|
|
is the only thing that reads these fields; handlers only construct them.
|
||
|
|
"""
|
||
|
|
kind: Literal["continue", "await", "terminal"]
|
||
|
|
next_step: str | None = None # set iff kind == "continue"
|
||
|
|
await_state: AwaitState | None = None # set iff kind == "await"
|
||
|
|
terminal_reason: TerminalReason | None = None # set iff kind == "terminal"
|
||
|
|
output: Any = None # the step's contract output
|
||
|
|
|
||
|
|
# ── constructors ─────────────────────────────────────────────
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def continue_to(cls, step: str, output: Any = None) -> "StepResult":
|
||
|
|
"""Advance to another step in the pipeline."""
|
||
|
|
return cls(kind="continue", next_step=step, output=output)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def awaiting(cls, state: AwaitState, output: Any = None) -> "StepResult":
|
||
|
|
"""Stop this call; resume when the await trigger fires."""
|
||
|
|
if not (state.startswith("AWAITING_") or state == "ABANDONED"):
|
||
|
|
raise ValueError(f"not an await state: {state!r}")
|
||
|
|
return cls(kind="await", await_state=state, output=output)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def complete(cls, output: Any = None) -> "StepResult":
|
||
|
|
"""Terminal: MR merged and task closed."""
|
||
|
|
return cls(kind="terminal", terminal_reason="completed", output=output)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def abandoned(cls, output: Any = None) -> "StepResult":
|
||
|
|
"""Terminal: no response within the timeout window."""
|
||
|
|
return cls(kind="terminal", terminal_reason="abandoned", output=output)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def escalated(cls, reason: str, output: Any = None) -> "StepResult":
|
||
|
|
"""Terminal: stopping because a human must intervene."""
|
||
|
|
return cls(kind="terminal", terminal_reason=f"escalated:{reason}",
|
||
|
|
output=output)
|