91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
"""Run orchestration: our DB rows + LFX execution + terminal-state mapping."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from uuid import UUID
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from app.models import ApprovalTask, FlowRun, FlowVersion
|
|
|
|
from . import store
|
|
from .lfx_client import LfxClient, LfxRunFailed
|
|
from .store import InvalidError
|
|
|
|
|
|
def build_envelope(
|
|
*,
|
|
payload: dict,
|
|
run_id: UUID,
|
|
callback_base_url: str,
|
|
continuation_flow_id: UUID | None,
|
|
callback_token: str,
|
|
) -> str:
|
|
return json.dumps(
|
|
{
|
|
"payload": payload,
|
|
"run_id": str(run_id),
|
|
"callback_url": callback_base_url.rstrip("/"),
|
|
"continuation_flow_id": (str(continuation_flow_id) if continuation_flow_id else None),
|
|
"callback_token": callback_token,
|
|
}
|
|
)
|
|
|
|
|
|
def _task_exists(session: Session, run_id: UUID) -> bool:
|
|
return (
|
|
session.exec(
|
|
select(ApprovalTask).where(ApprovalTask.run_id == run_id, ApprovalTask.status == "open")
|
|
).first()
|
|
is not None
|
|
)
|
|
|
|
|
|
async def execute_run(
|
|
session: Session,
|
|
lfx: LfxClient,
|
|
version: FlowVersion,
|
|
*,
|
|
payload: dict,
|
|
tenant_id: str = "default",
|
|
session_id: str | None = None,
|
|
callback_base_url: str,
|
|
continuation_flow_id: UUID | None = None,
|
|
callback_token: str = "",
|
|
) -> FlowRun:
|
|
"""Run a deployed version to a terminal state (no interrupts).
|
|
|
|
The flow always terminates: either it completes, or its approval branch
|
|
POSTs back to /api/approvals (creating a task) and terminates with a
|
|
pending marker. Status is derived from that side effect, not from
|
|
result-text parsing.
|
|
"""
|
|
if not version.lfx_flow_id:
|
|
raise InvalidError(f"version {version.id} is not deployed to LFX")
|
|
run = store.create_run(
|
|
session, version, input=payload, tenant_id=tenant_id, session_id=session_id
|
|
)
|
|
envelope = build_envelope(
|
|
payload=payload,
|
|
run_id=run.id,
|
|
callback_base_url=callback_base_url,
|
|
continuation_flow_id=continuation_flow_id,
|
|
callback_token=callback_token,
|
|
)
|
|
try:
|
|
result = await lfx.run_flow(
|
|
version.lfx_flow_id,
|
|
input_value=envelope,
|
|
session_id=session_id or str(run.id),
|
|
)
|
|
except LfxRunFailed as exc:
|
|
return store.set_run_result(session, run, "failed", {"error": str(exc)})
|
|
output = {
|
|
"result": result.result,
|
|
"component": result.component,
|
|
"type": result.type,
|
|
}
|
|
if _task_exists(session, run.id):
|
|
return store.set_run_result(session, run, "pending_approval", output)
|
|
return store.set_run_result(session, run, "completed", output)
|