diff --git a/backend/app/flows/api.py b/backend/app/flows/api.py index a18d4a9..aa155b9 100644 --- a/backend/app/flows/api.py +++ b/backend/app/flows/api.py @@ -34,6 +34,8 @@ from .lfx_client import ( from .schemas import ( ApprovalCreate, ApprovalRead, + DecisionRead, + DecisionRequest, FlowDefinitionCreate, FlowDefinitionRead, FlowImportRead, @@ -290,9 +292,13 @@ async def _run_stream(session: Session, lfx: LfxClient, version, data: RunReques @flows_router.get("/{flow_id}/runs", response_model=list[RunRead]) -def list_runs(flow_id: UUID, session: Session = Depends(get_db)): +def list_runs( + flow_id: UUID, + tenant_id: str | None = Query(default=None), + session: Session = Depends(get_db), +): try: - return [_read_run(r) for r in store.list_runs(session, flow_id)] + return [_read_run(r) for r in store.list_runs(session, flow_id, tenant_id)] except store.NotFoundError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @@ -350,11 +356,19 @@ async def import_flow( @flows_router.get("/runs/{run_id}", response_model=RunRead) -def get_run(run_id: UUID, session: Session = Depends(get_db)): +def get_run( + run_id: UUID, + tenant_id: str | None = Query(default=None), + session: Session = Depends(get_db), +): + """Read a run. With tenant_id, cross-tenant reads 404 (v1 scoping).""" try: - return _read_run(store.get_run(session, run_id)) + run = store.get_run(session, run_id) except store.NotFoundError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc + if tenant_id is not None and run.tenant_id != tenant_id: + raise HTTPException(status_code=404, detail=f"run {run_id} not found") + return _read_run(run) @approvals_router.post("", response_model=ApprovalRead, status_code=201) @@ -376,3 +390,66 @@ def list_approvals( session: Session = Depends(get_db), ): return [_read_task(t) for t in store.list_approval_tasks(session, tenant_id)] + + +@approvals_router.post("/{task_id}/decision", response_model=DecisionRead) +async def decide_approval( + task_id: UUID, + data: DecisionRequest, + session: Session = Depends(get_db), + lfx: LfxClient = Depends(get_lfx), +): + """Decide an open task. Approve starts the continuation flow (if any).""" + try: + task = store.get_task(session, task_id) + except store.NotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + + # Resolve the continuation first: bad references fail before any state change. + continuation_version = None + if data.decision == "approve" and task.continuation_flow_id is not None: + try: + continuation_version = store.get_latest_deployed_version( + session, task.continuation_flow_id + ) + except (store.NotFoundError, store.InvalidError) as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + try: + task = store.decide_task( + session, + task, + decision="approved" if data.decision == "approve" else "rejected", + approver=data.approver, + note=data.note, + ) + except store.ConflictError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + # Note: the decision is committed before the continuation runs. If LFX is + # down, the task stays decided and the continuation can be started + # manually from the preserved task payload. + continuation_run = None + if continuation_version is not None: + payload = task.payload + if isinstance(payload.get("facts"), dict): + payload = payload["facts"] + try: + continuation_run = await runner.execute_run( + session, + lfx, + continuation_version, + payload=payload, + tenant_id=task.tenant_id, + session_id=f"continuation-of-{task.run_id}", + callback_base_url=data.callback_base_url or settings.callback_base_url, + ) + except LfxNotFound as exc: + raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc + except LfxUnavailable as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + except LfxError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + return DecisionRead( + task=_read_task(task), + continuation_run=_read_run(continuation_run) if continuation_run else None, + ) diff --git a/backend/app/flows/schemas.py b/backend/app/flows/schemas.py index 84270b5..c024c34 100644 --- a/backend/app/flows/schemas.py +++ b/backend/app/flows/schemas.py @@ -98,3 +98,15 @@ class FlowImportRequest(BaseModel): class FlowImportRead(BaseModel): definition: FlowDefinitionRead version: FlowVersionRead + + +class DecisionRequest(BaseModel): + decision: str = Field(pattern="^(approve|reject)$") + approver: str = Field(default="", max_length=200) + note: str = Field(default="", max_length=2000) + callback_base_url: str | None = None + + +class DecisionRead(BaseModel): + task: ApprovalRead + continuation_run: RunRead | None diff --git a/backend/app/flows/store.py b/backend/app/flows/store.py index ac9527c..c5039f0 100644 --- a/backend/app/flows/store.py +++ b/backend/app/flows/store.py @@ -174,16 +174,19 @@ def get_run(session: Session, run_id: UUID) -> FlowRun: return run -def list_runs(session: Session, flow_id: UUID) -> list[FlowRun]: +def list_runs( + session: Session, flow_id: UUID, tenant_id: str | None = None +) -> list[FlowRun]: get_definition(session, flow_id) - return list( - session.exec( - select(FlowRun) - .join(FlowVersion, FlowVersion.id == FlowRun.version_id) - .where(FlowVersion.flow_id == flow_id) - .order_by(FlowRun.created_at.desc()) - ).all() + query = ( + select(FlowRun) + .join(FlowVersion, FlowVersion.id == FlowRun.version_id) + .where(FlowVersion.flow_id == flow_id) + .order_by(FlowRun.created_at.desc()) ) + if tenant_id is not None: + query = query.where(FlowRun.tenant_id == tenant_id) + return list(session.exec(query).all()) def create_approval_task(session: Session, data: ApprovalCreate) -> ApprovalTask: @@ -223,3 +226,33 @@ def list_approval_tasks(session: Session, tenant_id: str | None = None) -> list[ if tenant_id is not None: query = query.where(ApprovalTask.tenant_id == tenant_id) return list(session.exec(query).all()) + + +def get_task(session: Session, task_id: UUID) -> ApprovalTask: + task = session.get(ApprovalTask, task_id) + if task is None: + raise NotFoundError(f"approval task {task_id} not found") + return task + + +def decide_task( + session: Session, + task: ApprovalTask, + *, + decision: str, + approver: str = "", + note: str = "", +) -> ApprovalTask: + """Approve or reject an open task. Decided tasks are immutable (409).""" + if task.status != "open": + raise ConflictError(f"task {task.id} is already {task.status}") + if decision not in ("approved", "rejected"): + raise InvalidError(f"unknown decision {decision!r}") + task.status = decision + task.approver = approver + task.decision_note = note + task.decided_at = datetime.now(timezone.utc) + session.add(task) + session.commit() + session.refresh(task) + return task diff --git a/backend/tests/test_approvals.py b/backend/tests/test_approvals.py new file mode 100644 index 0000000..bc8729d --- /dev/null +++ b/backend/tests/test_approvals.py @@ -0,0 +1,255 @@ +"""Step 5 tests: decisions + continuations, tenant isolation, error paths.""" + +import json +from pathlib import Path +from uuid import UUID, uuid4 + +from app.flows.schemas import ApprovalCreate + +FIXTURES = Path(__file__).parent / "fixtures" / "flows" + + +def load_fixture(name: str) -> dict: + return json.loads((FIXTURES / name).read_text()) + + +async def _create_flow(client, name): + response = await client.post("/api/flows", json={"name": name}) + assert response.status_code == 201, response.text + return response.json() + + +async def _version_and_deploy(client, flow_id, fixture="simple_rule.json"): + response = await client.post( + f"/api/flows/{flow_id}/versions", + json={"langflow_json": load_fixture(fixture)}, + ) + assert response.status_code == 201, response.text + response = await client.post(f"/api/flows/{flow_id}/versions/1/deploy") + assert response.status_code == 200, response.text + return response.json() + + +async def _handoff_task(client, live_server, continuation_flow_id=None): + """Run approval_handoff over threshold; returns (run, task).""" + flow = await _create_flow(client, f"handoff-{uuid4().hex[:8]}") + await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json") + body = { + "input": {"betrag": 250000, "schwelle": 200000}, + "callback_base_url": live_server, + } + if continuation_flow_id is not None: + body["continuation_flow_id"] = continuation_flow_id + run = ( + await client.post(f"/api/flows/{flow['id']}/run", json=body) + ).json() + assert run["status"] == "pending_approval" + tasks = (await client.get("/api/approvals")).json() + task = next(t for t in tasks if t["run_id"] == run["id"]) + return run, task + + +async def test_approve_without_continuation(client, lfx_server, live_server): + _, task = await _handoff_task(client, live_server) + response = await client.post( + f"/api/approvals/{task['id']}/decision", + json={"decision": "approve", "approver": "sachbearbeiter", "note": "ok"}, + ) + assert response.status_code == 200, response.text + body = response.json() + assert body["task"]["status"] == "approved" + assert body["task"]["approver"] == "sachbearbeiter" + assert body["task"]["decided_at"] is not None + assert body["continuation_run"] is None + + +async def test_approve_with_continuation_runs_facts(client, lfx_server, live_server): + cont = await _create_flow(client, f"cont-{uuid4().hex[:8]}") + await _version_and_deploy(client, cont["id"], fixture="simple_rule.json") + + _, task = await _handoff_task( + client, live_server, continuation_flow_id=cont["id"] + ) + assert task["continuation_flow_id"] == cont["id"] + + response = await client.post( + f"/api/approvals/{task['id']}/decision", + json={"decision": "approve", "approver": "chef"}, + ) + assert response.status_code == 200, response.text + body = response.json() + assert body["task"]["status"] == "approved" + cont_run = body["continuation_run"] + assert cont_run is not None + assert cont_run["status"] == "completed" + assert cont_run["tenant_id"] == task["tenant_id"] + # Facts (betrag 250000 > Schwelle) flowed into the continuation. + assert "Sonderfall" in cont_run["output"]["result"] + + +async def test_reject_task(client, lfx_server, live_server): + _, task = await _handoff_task(client, live_server) + response = await client.post( + f"/api/approvals/{task['id']}/decision", + json={"decision": "reject", "approver": "chef", "note": "zu teuer"}, + ) + assert response.status_code == 200, response.text + body = response.json() + assert body["task"]["status"] == "rejected" + assert body["task"]["decision_note"] == "zu teuer" + assert body["continuation_run"] is None + + +async def test_decision_guards(client, lfx_server, live_server): + _, task = await _handoff_task(client, live_server) + url = f"/api/approvals/{task['id']}/decision" + + response = await client.post(url, json={"decision": "approve"}) + assert response.status_code == 200 + response = await client.post(url, json={"decision": "reject"}) + assert response.status_code == 409 + + response = await client.post( + f"/api/approvals/{uuid4()}/decision", json={"decision": "approve"} + ) + assert response.status_code == 404 + + _, task2 = await _handoff_task(client, live_server) + response = await client.post( + f"/api/approvals/{task2['id']}/decision", json={"decision": "maybe"} + ) + assert response.status_code == 422 + + response = await client.post( + f"/api/approvals/{task2['id']}/decision", + json={"decision": "approve", "note": "x"}, + ) + # continuation unset -> fine, resolves to no continuation run + assert response.status_code == 200 + + +async def test_decision_bad_continuation_reference(client, session): + from app.flows import store as flow_store + + flow = await _create_flow(client, f"badcont-{uuid4().hex[:8]}") + version = ( + await client.post( + f"/api/flows/{flow['id']}/versions", + json={"langflow_json": load_fixture("simple_rule.json")}, + ) + ).json() + run = flow_store.create_run( + session, + flow_store.get_version(session, UUID(version["id"])), + input={}, + ) + task = flow_store.create_approval_task( + session, ApprovalCreate(run_id=run.id, title="t") + ) + # plant unknown continuation directly (store allows it pre-validation) + task.continuation_flow_id = uuid4() + session.add(task) + session.commit() + + response = await client.post( + f"/api/approvals/{task.id}/decision", json={"decision": "approve"} + ) + assert response.status_code == 422 + + +async def test_tenant_isolation(client, lfx_server): + flow = await _create_flow(client, f"tenant-{uuid4().hex[:8]}") + await _version_and_deploy(client, flow["id"]) + + for tenant in ("pfarramt-a", "pfarramt-b"): + response = await client.post( + f"/api/flows/{flow['id']}/run", + json={"input": {"betrag": 1}, "tenant_id": tenant}, + ) + assert response.status_code == 200 + + response = await client.get( + f"/api/flows/{flow['id']}/runs", params={"tenant_id": "pfarramt-a"} + ) + runs_a = response.json() + assert len(runs_a) == 1 + assert runs_a[0]["tenant_id"] == "pfarramt-a" + + response = await client.get(f"/api/flows/{flow['id']}/runs") + assert len(response.json()) == 2 + + # Cross-tenant read with scoping 404s. + response = await client.get( + f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-b"} + ) + assert response.status_code == 404 + response = await client.get( + f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-a"} + ) + assert response.status_code == 200 + + +async def test_lfx_down_maps_to_503(client, lfx_server, monkeypatch): + from app.settings import settings + + flow = await _create_flow(client, f"down-{uuid4().hex[:8]}") + await client.post( + f"/api/flows/{flow['id']}/versions", + json={"langflow_json": load_fixture("simple_rule.json")}, + ) + monkeypatch.setattr(settings, "lfx_url", "http://127.0.0.1:9") + response = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy") + assert response.status_code == 503 + + # Deploy for real, then break the URL again for the run call. + monkeypatch.setattr(settings, "lfx_url", lfx_server) + deployed = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy") + assert deployed.status_code == 200 + monkeypatch.setattr(settings, "lfx_url", "http://127.0.0.1:9") + response = await client.post( + f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}} + ) + assert response.status_code == 503 + + +async def test_lfx_drift_maps_to_502(client, lfx_server): + from app.flows.lfx_client import LfxClient + from app.settings import settings + + flow = await _create_flow(client, f"drift-{uuid4().hex[:8]}") + deployed = await _version_and_deploy(client, flow["id"]) + await LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None).delete_flow( + deployed["lfx_flow_id"] + ) + response = await client.post( + f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}} + ) + assert response.status_code == 502 + + +async def test_corrupt_flow_json_deploy_maps_to_422(client, lfx_server): + flow = await _create_flow(client, f"corrupt-{uuid4().hex[:8]}") + bogus = { + "name": "bogus", + "data": { + "nodes": [ + { + "id": "Nope-x1", + "data": { + "id": "Nope-x1", + "type": "Nope", + "showNode": True, + "node": {"template": {}, "outputs": []}, + }, + "type": "genericNode", + } + ], + "edges": [], + }, + } + response = await client.post( + f"/api/flows/{flow['id']}/versions", json={"langflow_json": bogus} + ) + assert response.status_code == 201, response.text + response = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy") + assert response.status_code == 422