dify/api/services/dify_builder/task_native_cancel.py

"""Cancel ordinary native debug runs using the existing workflow stop owner."""

from sqlalchemy import select

from core.dify_builder.errors import NotFoundError
from models.dify_builder import DifyBuilderExecutionRequest, DifyBuilderRun, DifyBuilderSession
from models.enums import CreatorUserRole
from models.model import App, AppMode
from models.workflow import WorkflowRun


def task_execution_settlement(sql, session_id, context, *, include_controls=True):
    """Classify current execution evidence without reading Workflow or payloads.

    An immutable unknown Run is still unknown. A new task cannot discard that
    fact merely because there is no live request left to cancel.
    """
    result = "settled"
    if include_controls:
        controls = context.task_control
        if controls.get("settlement") == "unknown" or controls.get("receipt", {}).get("enqueue") == "unknown":
            return "unknown"
        if controls.get("settlement") == "pending" or controls.get("receipt", {}).get("enqueue") in {
            "pending",
            "dispatching",
            "claimed",
        }:
            result = "pending"
    if context.verify_run_id:
        run = sql.scalar(
            select(DifyBuilderRun).where(
                DifyBuilderRun.id == context.verify_run_id,
                DifyBuilderRun.session_id == session_id,
            )
        )
        if run is None or run.status not in {"succeeded", "failed", "stopped", "running"}:
            return "unknown"
        evidence = (run.verification or {}).get("execution_evidence")
        if evidence and evidence.get("safety_outcome") == "execution_evidence_unknown":
            return "unknown"
        if run.status == "running" or (evidence and not evidence.get("sealed")):
            result = "pending"
    if context.test_input_ref:
        for request in sql.scalars(
            select(DifyBuilderExecutionRequest).where(
                DifyBuilderExecutionRequest.session_id == session_id,
                DifyBuilderExecutionRequest.test_input_id == context.test_input_ref,
            )
        ):
            if (request.completion_summary or {}).get("outcome") == "unknown":
                return "unknown"
            if request.state != "sealed":
                result = "pending"
    return result


def settle_ordinary(repo, *, session_id, actor, run_ids, request_ids, deliver=True):
    from services.app_task_service import AppTaskService

    with repo._session_factory() as sql:
        session = sql.get(DifyBuilderSession, session_id)
        if session is None or (session.tenant_id, session.owner_account_id) != (actor.tenant_id, actor.account_id):
            raise NotFoundError("session not found")
        app = sql.scalar(select(App).where(App.id == session.app_id, App.tenant_id == actor.tenant_id))
        if app is None:
            raise NotFoundError("app not found")
        native_ids = set()
        pending = False
        for row in sql.scalars(
            select(DifyBuilderRun).where(DifyBuilderRun.session_id == session_id, DifyBuilderRun.id.in_(run_ids))
        ):
            if row.dify_run_id:
                native_ids.add(row.dify_run_id)
            elif not row.immutable:
                pending = True  # Launch without acknowledgement cannot be replayed.
        for row in sql.scalars(
            select(DifyBuilderExecutionRequest).where(
                DifyBuilderExecutionRequest.session_id == session_id,
                DifyBuilderExecutionRequest.tenant_id == actor.tenant_id,
                DifyBuilderExecutionRequest.actor_id == actor.account_id,
                DifyBuilderExecutionRequest.id.in_(request_ids),
            )
        ):
            if row.native_run_id:
                native_ids.add(row.native_run_id)
            elif row.state not in {"prepared", "cancelled", "sealed", "terminal"}:
                pending = True
        runs = list(
            sql.scalars(
                select(WorkflowRun).where(
                    WorkflowRun.id.in_(native_ids),
                    WorkflowRun.tenant_id == actor.tenant_id,
                    WorkflowRun.app_id == app.id,
                    WorkflowRun.created_by == actor.account_id,
                )
            )
        )
        if len(runs) != len(native_ids):
            pending = True
        commands = [(r.id, r.task_id, str(r.status)) for r in runs]
        app_id, mode = app.id, AppMode.value_of(app.mode)
    for _, task_id, status in commands:
        if status in {"running", "paused"} and deliver:
            if not task_id:
                pending = True
                continue
            AppTaskService.stop_workflow_task(
                tenant_id=actor.tenant_id,
                app_id=app_id,
                task_id=task_id,
                app_mode=mode,
                owner=(CreatorUserRole.ACCOUNT, actor.account_id),
            )
    with repo._session_factory() as sql, sql.begin():
        stopped_ids = set()
        for native_id, _, _ in commands:
            run = sql.get(WorkflowRun, native_id)
            if run is None or str(run.status) in {"running", "paused"}:
                pending = True
            elif str(run.status) in {"stopped", "succeeded", "failed", "partial-succeeded", "partial_succeeded"}:
                stopped_ids.add(native_id)
            else:
                return "unknown"
        # Seal only proved terminal history. Late worker settlement cannot replace it.
        for row in sql.scalars(
            select(DifyBuilderRun)
            .where(DifyBuilderRun.session_id == session_id, DifyBuilderRun.id.in_(run_ids))
            .with_for_update()
        ):
            if row.dify_run_id in stopped_ids and not row.immutable:
                native = sql.get(WorkflowRun, row.dify_run_id)
                row.status = str(native.status)
                row.immutable = True
    return "pending" if pending else "settled"