17759ef564
- apps/api: psycopg v3 repositories, schema.sql + migration runner, state machine implementing the data-model transition table, approval gate with sha256 hash match + 24h expiry enforced at send time, EchoTransport pluggable sender, LLM calls routed through packages.llm-gateway with mock mode, 47 pytest tests green - apps/api/Dockerfile.test: python 3.13-slim test image (DinD-safe: migrations copied as directory) - docker-compose.yml: add api-test service on compose network (host port publishing is broken in this sandbox; container-to-container networking used) - Fix: replace masked placeholder password in config.py/conftest.py defaults
105 lines
No EOL
3 KiB
Python
105 lines
No EOL
3 KiB
Python
"""State machine for application transitions.
|
|
|
|
Implements the transition table from docs/data-model.md.
|
|
Any transition not listed raises InvalidTransition (409 in API).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
|
|
class InvalidTransition(Exception):
|
|
"""Raised when a state transition is not allowed."""
|
|
|
|
def __init__(self, from_state: str, to_state: str, reason: str = ""):
|
|
self.from_state = from_state
|
|
self.to_state = to_state
|
|
self.reason = reason
|
|
super().__init__(
|
|
f"Invalid transition: {from_state} -> {to_state}"
|
|
+ (f": {reason}" if reason else "")
|
|
)
|
|
|
|
|
|
# All valid states
|
|
STATES = frozenset({
|
|
"discovered",
|
|
"scored",
|
|
"approved",
|
|
"rejected",
|
|
"drafting",
|
|
"sent",
|
|
"interviewing",
|
|
"offer",
|
|
"closed",
|
|
"expired",
|
|
})
|
|
|
|
|
|
# Transition table: (from, to) -> guard name (or None for no guard)
|
|
# Guard names map to guard functions below.
|
|
TRANSITIONS: dict[tuple[str, str], str | None] = {
|
|
("discovered", "scored"): "scoring_completed",
|
|
("discovered", "rejected"): None,
|
|
("scored", "approved"): None,
|
|
("scored", "rejected"): None,
|
|
("approved", "drafting"): None,
|
|
("drafting", "sent"): "confirmed_approval",
|
|
("sent", "interviewing"): None,
|
|
("interviewing", "offer"): None,
|
|
("interviewing", "closed"): None,
|
|
("offer", "closed"): None,
|
|
("scored", "expired"): None,
|
|
("approved", "expired"): None,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class TransitionContext:
|
|
"""Context passed to guard functions for validation."""
|
|
|
|
application_id: str
|
|
from_state: str
|
|
to_state: str
|
|
has_score: bool = False
|
|
has_confirmed_approval: bool = False
|
|
artifact_hash_match: bool = False
|
|
|
|
|
|
# Guard implementations
|
|
GUARDS: dict[str, callable] = {
|
|
"scoring_completed": lambda ctx: ctx.has_score,
|
|
"confirmed_approval": lambda ctx: ctx.has_confirmed_approval and ctx.artifact_hash_match,
|
|
}
|
|
|
|
|
|
def can_transition(from_state: str, to_state: str) -> bool:
|
|
"""Check if a transition is in the table (does not evaluate guards)."""
|
|
return (from_state, to_state) in TRANSITIONS
|
|
|
|
|
|
def check_transition(ctx: TransitionContext) -> None:
|
|
"""Validate a transition. Raises InvalidTransition if not allowed."""
|
|
if ctx.from_state not in STATES:
|
|
raise InvalidTransition(ctx.from_state, ctx.to_state, f"unknown state: {ctx.from_state}")
|
|
if ctx.to_state not in STATES:
|
|
raise InvalidTransition(ctx.from_state, ctx.to_state, f"unknown state: {ctx.to_state}")
|
|
|
|
key = (ctx.from_state, ctx.to_state)
|
|
if key not in TRANSITIONS:
|
|
raise InvalidTransition(
|
|
ctx.from_state,
|
|
ctx.to_state,
|
|
f"transition {ctx.from_state} -> {ctx.to_state} is not in the transition table",
|
|
)
|
|
|
|
guard_name = TRANSITIONS[key]
|
|
if guard_name is not None:
|
|
guard_fn = GUARDS[guard_name]
|
|
if not guard_fn(ctx):
|
|
raise InvalidTransition(
|
|
ctx.from_state,
|
|
ctx.to_state,
|
|
f"guard failed: {guard_name}",
|
|
) |