Lars
95feba6204
All checks were successful
Deploy Development / deploy (push) Successful in 44s
Test Suite / pytest-backend (push) Successful in 1m19s
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 18s
Test Suite / playwright-smoke (push) Successful in 11s
Persistierter Standard Lifecycle pro Initiative, backend/steering/ Skeleton, Attention-Refactor und Hook-Dispatch als Basis für Method Registry. Co-authored-by: Cursor <cursoragent@cursor.com>
79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
"""Lifecycle orchestrator — persistierter Standard Lifecycle (AP1.0)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from services import steering_context as sc_service
|
|
from steering.lifecycle.states import validate_lifecycle_state
|
|
from tenant_context import TenantContext
|
|
|
|
|
|
def get_lifecycle_state(ctx: TenantContext, initiative_id: str) -> str:
|
|
row = sc_service.get_steering_context(
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
if row:
|
|
return row["lifecycle_state"]
|
|
created = sc_service.ensure_steering_context(
|
|
tenant_id=ctx.tenant_id,
|
|
initiative_id=initiative_id,
|
|
initial_state="intake",
|
|
)
|
|
return created["lifecycle_state"]
|
|
|
|
|
|
def transition(
|
|
ctx: TenantContext | None,
|
|
initiative_id: str,
|
|
to_state: str,
|
|
*,
|
|
reason: str = "",
|
|
tenant_id: Optional[str] = None,
|
|
user_id: Optional[str] = None,
|
|
) -> dict[str, Any]:
|
|
"""Lifecycle-Transition — AP1.0: freier Übergang (Guards in AP1.1)."""
|
|
validate_lifecycle_state(to_state)
|
|
tid = tenant_id or (ctx.tenant_id if ctx else None)
|
|
if not tid:
|
|
raise ValueError("tenant_id erforderlich")
|
|
|
|
existing = sc_service.get_steering_context(tenant_id=tid, initiative_id=initiative_id)
|
|
if not existing:
|
|
existing = sc_service.ensure_steering_context(
|
|
tenant_id=tid,
|
|
initiative_id=initiative_id,
|
|
initial_state=to_state,
|
|
)
|
|
return {
|
|
"initiative_id": initiative_id,
|
|
"from_state": None,
|
|
"to_state": to_state,
|
|
"reason": reason,
|
|
}
|
|
|
|
from_state = existing["lifecycle_state"]
|
|
if from_state == to_state:
|
|
return {
|
|
"initiative_id": initiative_id,
|
|
"from_state": from_state,
|
|
"to_state": to_state,
|
|
"reason": reason,
|
|
"unchanged": True,
|
|
}
|
|
|
|
updated = sc_service.update_lifecycle_state(
|
|
tenant_id=tid,
|
|
initiative_id=initiative_id,
|
|
lifecycle_state=to_state,
|
|
user_id=user_id,
|
|
reason=reason,
|
|
from_state=from_state,
|
|
)
|
|
return {
|
|
"initiative_id": initiative_id,
|
|
"from_state": from_state,
|
|
"to_state": updated["lifecycle_state"],
|
|
"reason": reason,
|
|
}
|