feat(G): G-APPROVAL/G-MAN/G-WEB/G-LOG/G-RUN-resume — workflow routes (resume, manual trigger, webhook trigger, step history, approve/reject), 30 tests passing

2026-08-18 08:31:18 +02:00
parent db41e60042
commit 9866fb2d14
2 changed files with 660 additions and 1 deletions
@@ -4,7 +4,7 @@ from __future__ import annotations

 import uuid

-from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
 from sqlalchemy.ext.asyncio import AsyncSession

 from app.core.db import get_db
@@ -291,3 +291,290 @@ async def cancel_instance(
            detail={"detail": result["error"], "code": "invalid_state"},
        )
    return result
+
+
+# ─── G-RUN: Resume waiting workflow instance ──────────────────────────────────
+
+
+@router.post("/instances/{instance_id}/resume")
+async def resume_instance(
+    instance_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Resume a waiting workflow instance (e.g. after wait timer expired)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    from app.workflows.engine import WorkflowEngine
+    from app.models.workflow import WorkflowInstance
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowInstance).where(
+            WorkflowInstance.id == uuid.UUID(instance_id),
+            WorkflowInstance.tenant_id == tenant_id,
+            WorkflowInstance.deleted_at.is_(None),
+        )
+    )
+    instance = result.scalar_one_or_none()
+    if instance is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Instance not found", "code": "not_found"},
+        )
+    if instance.status != "waiting":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={"detail": "Instance is not waiting", "code": "invalid_state"},
+        )
+
+    engine = WorkflowEngine(db, tenant_id)
+    return await engine.resume(instance)
+
+
+# ─── G-MAN: Manual trigger — start workflow from UI button ────────────────────
+
+
+@router.post("/{workflow_id}/trigger", status_code=status.HTTP_201_CREATED)
+async def manual_trigger(
+    workflow_id: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Manually trigger a workflow — starts a new instance with optional context."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+
+    result = await workflow_service.create_instance(
+        db,
+        tenant_id,
+        user_id,
+        workflow_id=workflow_id,
+        context=body,
+    )
+    if result is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Workflow not found", "code": "not_found"},
+        )
+    return result
+
+
+# ─── G-WEB: Incoming webhook trigger ──────────────────────────────────────────
+
+
+@router.post("/webhook/{token}", status_code=status.HTTP_200_OK)
+async def webhook_trigger(
+    token: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+):
+    """Incoming webhook trigger — starts a workflow via secure token."""
+    from app.models.webhook import Webhook
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(Webhook).where(
+            Webhook.token == token,
+            Webhook.is_active.is_(True),
+        )
+    )
+    webhook = result.scalar_one_or_none()
+    if webhook is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Webhook not found", "code": "not_found"},
+        )
+
+    try:
+        payload = await request.json()
+    except Exception:
+        payload = {}
+
+    result = await workflow_service.create_instance(
+        db,
+        webhook.tenant_id,
+        None,
+        workflow_id=str(webhook.workflow_id) if hasattr(webhook, "workflow_id") else str(webhook.entity_id),
+        context={"webhook_payload": payload, "webhook_token": token},
+    )
+    if result is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Workflow not found", "code": "not_found"},
+        )
+    return {"status": "triggered", "instance": result}
+
+
+# ─── G-LOG: Step history for a workflow instance ─────────────────────────────
+
+
+@router.get("/instances/{instance_id}/history")
+async def get_instance_history(
+    instance_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:read")),
+):
+    """Get step history for a workflow instance (execution log)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    from app.models.workflow import WorkflowStepHistory
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowStepHistory)
+        .where(
+            WorkflowStepHistory.tenant_id == tenant_id,
+            WorkflowStepHistory.instance_id == uuid.UUID(instance_id),
+        )
+        .order_by(WorkflowStepHistory.created_at.asc())
+    )
+    history = result.scalars().all()
+    return {
+        "items": [
+            {
+                "id": str(h.id),
+                "step_index": h.step_index,
+                "step_type": h.step_type,
+                "action": h.action,
+                "actor_id": str(h.actor_id) if h.actor_id else None,
+                "details": h.details,
+                "created_at": h.created_at.isoformat() if h.created_at else None,
+            }
+            for h in history
+        ],
+        "total": len(history),
+    }
+
+
+# ─── G-APPROVAL: Approval step using central ApprovalRequest ─────────────────
+
+
+@router.post("/instances/{instance_id}/approve")
+async def approve_workflow_step(
+    instance_id: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Approve the current approval step of a workflow instance."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+    comment = body.get("comment", "")
+
+    from app.core.approval import create_approval_request, decide_approval
+    from app.models.workflow import WorkflowInstance
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowInstance).where(
+            WorkflowInstance.id == uuid.UUID(instance_id),
+            WorkflowInstance.tenant_id == tenant_id,
+            WorkflowInstance.deleted_at.is_(None),
+        )
+    )
+    instance = result.scalar_one_or_none()
+    if instance is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Instance not found", "code": "not_found"},
+        )
+
+    approval = await create_approval_request(
+        db=db,
+        tenant_id=tenant_id,
+        entity_type="workflow_instance",
+        entity_id=uuid.UUID(instance_id),
+        action="workflow_step_approval",
+        requested_by=user_id,
+        requested_by_type="user",
+    )
+    await decide_approval(
+        db=db,
+        approval_id=approval["id"],
+        decision="approved",
+        decided_by=user_id,
+        comment=comment,
+    )
+
+    return await workflow_service.advance_instance(
+        db,
+        tenant_id,
+        user_id,
+        instance_id=instance_id,
+        decision="approved",
+        comment=comment,
+        is_system_admin=current_user.get("is_system_admin", False),
+    )
+
+
+@router.post("/instances/{instance_id}/reject")
+async def reject_workflow_step(
+    instance_id: str,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("workflows:write")),
+):
+    """Reject the current approval step of a workflow instance."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+    comment = body.get("comment", "")
+
+    from app.core.approval import create_approval_request, decide_approval
+    from app.models.workflow import WorkflowInstance
+    from sqlalchemy import select
+
+    result = await db.execute(
+        select(WorkflowInstance).where(
+            WorkflowInstance.id == uuid.UUID(instance_id),
+            WorkflowInstance.tenant_id == tenant_id,
+            WorkflowInstance.deleted_at.is_(None),
+        )
+    )
+    instance = result.scalar_one_or_none()
+    if instance is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"detail": "Instance not found", "code": "not_found"},
+        )
+
+    approval = await create_approval_request(
+        db=db,
+        tenant_id=tenant_id,
+        entity_type="workflow_instance",
+        entity_id=uuid.UUID(instance_id),
+        action="workflow_step_approval",
+        requested_by=user_id,
+        requested_by_type="user",
+    )
+    await decide_approval(
+        db=db,
+        approval_id=approval["id"],
+        decision="rejected",
+        decided_by=user_id,
+        comment=comment,
+    )
+
+    return await workflow_service.cancel_instance(
+        db,
+        tenant_id,
+        user_id,
+        instance_id=instance_id,
+        is_system_admin=current_user.get("is_system_admin", False),
+    )