mirror of
https://github.com/langgenius/dify.git
synced 2026-10-04 13:47:55 +08:00
452 lines
21 KiB
Python
452 lines
21 KiB
Python
"""Celery task that runs the dify_builder engine's ``Runner.advance`` off the
|
|
request thread (P3b Task 4).
|
|
|
|
The usecase's ``dispatch`` (Task 3) acquires the cross-process advance lock
|
|
(``services.dify_builder.session_lock``) in the web process and enqueues
|
|
this task; this task runs the actual engine step in the Celery process and
|
|
releases the lock in ``finally`` -- regardless of whether the step succeeded,
|
|
lost a version-CAS race, or raised. Progress (curated phase deltas,
|
|
native workflow events during test runs, plus the settled command projection) is
|
|
forwarded to the session's progress bus
|
|
(``services.dify_builder.progress_bus``) for the P3c SSE endpoint to relay.
|
|
|
|
Each invocation opens its own ``sessionmaker(bind=db.engine, ...)`` --
|
|
mirrors ``services.dify_builder.dify_port._session_factory`` -- rather
|
|
than reusing the Flask-request-scoped ``db.session``, since this runs in a
|
|
Celery worker, not a Flask request. The ``FlaskTask`` base
|
|
(``extensions.ext_celery.init_app``) already wraps the task body in
|
|
``app.app_context()``, so this module must NOT push its own context.
|
|
|
|
Once a message's user half is durable, a failed assistant half remains at
|
|
the current waiting gate. The client can then retry with the same
|
|
``client_turn_id`` without duplicating the user bubble.
|
|
"""
|
|
|
|
import logging
|
|
from collections.abc import Mapping
|
|
from dataclasses import asdict, replace
|
|
|
|
from celery import shared_task
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from configs import dify_config
|
|
from core.dify_builder.contract import (
|
|
AgentMessageEventData,
|
|
AssistantTurnItem,
|
|
ConversationItemAppendedEventData,
|
|
ExecutionActivity,
|
|
ExecutionProgress,
|
|
ProgressEventData,
|
|
ReasoningEventData,
|
|
WorkflowEventData,
|
|
post_canvas_action_id,
|
|
)
|
|
from core.dify_builder.errors import ConflictError
|
|
from core.dify_builder.handlers_build import build_registry
|
|
from core.dify_builder.handlers_edit import edit_registry
|
|
from core.dify_builder.handlers_fix import fix_registry
|
|
from core.dify_builder.models import Action, Actor, Session, Turn
|
|
from core.dify_builder.runner import Env, Runner
|
|
from core.dify_builder.state import PcState, is_terminal
|
|
from extensions.ext_database import db
|
|
from libs.datetime_utils import naive_utc_now
|
|
from services.dify_builder import app_naming, progress_bus, session_lock
|
|
from services.dify_builder.agent.localize import Localizer
|
|
from services.dify_builder.agent_factory import build_dify_builder_agent
|
|
from services.dify_builder.dify_port import WorkflowServiceDifyPort
|
|
from services.dify_builder.errors import HashMismatchError
|
|
from services.dify_builder.repository import SqlDifyBuilderRepository
|
|
from services.dify_builder.service import DifyBuilderService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
__all__ = ["advance_session"]
|
|
|
|
_COMMAND_FINISHED_UNAVAILABLE_CODE = "command_finished_unavailable"
|
|
|
|
|
|
def _build_repo() -> SqlDifyBuilderRepository:
|
|
# Per-task sessionmaker (never the Flask-request-scoped db.session) --
|
|
# mirrors dify_port._session_factory().
|
|
return SqlDifyBuilderRepository(sessionmaker(bind=db.engine, expire_on_commit=False))
|
|
|
|
|
|
def _persist_failed_state(
|
|
repo: SqlDifyBuilderRepository,
|
|
session_id: str,
|
|
*,
|
|
expected_version: int,
|
|
expected_state: PcState,
|
|
command_id: str,
|
|
) -> int | None:
|
|
"""Persist a generic terminal failure even when agent construction failed."""
|
|
session, context = repo.get_session(session_id)
|
|
if is_terminal(session.current_state):
|
|
return None
|
|
if session.version != expected_version or session.current_state != expected_state:
|
|
raise ConflictError(
|
|
f"dify_builder: refusing stale setup failure for session {session_id} "
|
|
f"at version {session.version} state {session.current_state}"
|
|
)
|
|
at_version = session.version + 1
|
|
item = AssistantTurnItem(
|
|
turn_id=command_id,
|
|
stage_id=str(session.current_state),
|
|
execution=ExecutionProgress(status="error"),
|
|
reply_text="The operation could not be completed. Restart from the current draft to continue.",
|
|
).to_item(seq=context.next_seq, at_version=at_version)
|
|
context.next_seq += 1
|
|
context.last_command_id = command_id
|
|
return repo.compare_and_advance(session.id, session.version, PcState.FAILED, context, [item])
|
|
|
|
|
|
@shared_task(
|
|
queue="dify_builder",
|
|
soft_time_limit=dify_config.DIFY_BUILDER_MAX_ADVANCE_SECONDS,
|
|
time_limit=session_lock.active_advance_time_limit(),
|
|
)
|
|
def advance_session(session_id: str, action_dict: dict, actor_dict: dict, token: str) -> None:
|
|
"""Run one ``Runner.advance`` for ``session_id``, then release the
|
|
advance lock held under ``token``. Always releases the lock, even on a
|
|
lost CAS race or an unexpected engine failure -- a stuck lock would wedge
|
|
the session for its full TTL (``DIFY_BUILDER_MAX_ADVANCE_SECONDS``)."""
|
|
if not session_lock.activate(session_id, token):
|
|
logger.warning("dify_builder stale queued advance ignored for session %s", session_id)
|
|
return
|
|
|
|
terminal_error: dict | None = None
|
|
completed: tuple[SqlDifyBuilderRepository, WorkflowServiceDifyPort, Actor] | None = None
|
|
repo: SqlDifyBuilderRepository | None = None
|
|
dify: WorkflowServiceDifyPort | None = None
|
|
actor: Actor | None = None
|
|
runner: Runner | None = None
|
|
env: Env | None = None
|
|
loaded_session: Session | None = None
|
|
last_progress: ProgressEventData | None = None
|
|
progress_activities: dict[str, ExecutionActivity] = {}
|
|
progress_operation_id = ""
|
|
action: Action | None = None
|
|
try:
|
|
repo = _build_repo()
|
|
dify = WorkflowServiceDifyPort()
|
|
actor = Actor(**actor_dict)
|
|
action = Action(**action_dict)
|
|
# The per-session model choice lives on the context (stable for the session).
|
|
# Read the head so the real agent is constructed with the user's chosen model.
|
|
loaded_session, _fc = repo.get_session(session_id)
|
|
agent = build_dify_builder_agent(tenant_id=actor.tenant_id, model_config=_fc.model_config)
|
|
|
|
def emit_workflow(payload: Mapping[str, object]) -> None:
|
|
assert env is not None
|
|
try:
|
|
progress_bus.publish(
|
|
session_id,
|
|
asdict(
|
|
WorkflowEventData(
|
|
session_id=session_id,
|
|
operation_id=env.operation_id,
|
|
stage_id=env.stage_id,
|
|
at_version=env.at_version,
|
|
revision=env.next_event_revision(),
|
|
payload=dict(payload),
|
|
)
|
|
),
|
|
)
|
|
except Exception:
|
|
logger.exception("dify_builder workflow event publish failed for session %s", session_id)
|
|
|
|
def emit_canvas(event: dict) -> None:
|
|
assert env is not None
|
|
try:
|
|
progress_bus.publish(
|
|
session_id,
|
|
{
|
|
"kind": "canvas",
|
|
"session_id": session_id,
|
|
"operation_id": env.operation_id,
|
|
"stage_id": env.stage_id,
|
|
"at_version": env.at_version,
|
|
"revision": env.next_event_revision(),
|
|
**event,
|
|
},
|
|
)
|
|
except Exception:
|
|
logger.exception("dify_builder canvas event publish failed for session %s", session_id)
|
|
|
|
def emit_message(message: AgentMessageEventData) -> None:
|
|
try:
|
|
progress_bus.publish(session_id, asdict(message))
|
|
except Exception:
|
|
# Streaming delivery is best effort. The terminal sequence
|
|
# watermark exposes the missing assistant item so the client
|
|
# can recover it through the history endpoint.
|
|
logger.exception(
|
|
"dify_builder agent_message event publish failed for session %s turn %s",
|
|
session_id,
|
|
message.turn_id,
|
|
)
|
|
|
|
def emit_item(event: ConversationItemAppendedEventData) -> None:
|
|
try:
|
|
progress_bus.publish(session_id, asdict(event))
|
|
except Exception:
|
|
# The terminal sequence watermark lets the client detect this
|
|
# missing item and recover it through the history endpoint.
|
|
logger.exception(
|
|
"dify_builder conversation item event publish failed for session %s seq %s",
|
|
session_id,
|
|
event.item.seq,
|
|
)
|
|
|
|
def emit_reasoning(reasoning: ReasoningEventData) -> None:
|
|
try:
|
|
progress_bus.publish(session_id, asdict(reasoning))
|
|
except Exception:
|
|
logger.exception(
|
|
"dify_builder reasoning event publish failed for session %s operation %s",
|
|
session_id,
|
|
reasoning.operation_id,
|
|
)
|
|
|
|
def emit_progress(progress: ProgressEventData) -> None:
|
|
nonlocal last_progress, progress_operation_id
|
|
if progress.operation_id != progress_operation_id:
|
|
progress_activities.clear()
|
|
progress_operation_id = progress.operation_id
|
|
if progress.activity is not None:
|
|
progress_activities[progress.activity.id] = replace(progress.activity)
|
|
last_progress = progress
|
|
try:
|
|
progress_bus.publish(session_id, asdict(progress))
|
|
except Exception:
|
|
# Phase progress is an observer notification. The following
|
|
# command_finished still carries the authoritative projection.
|
|
logger.exception(
|
|
"dify_builder progress event publish failed for session %s operation %s",
|
|
session_id,
|
|
progress.operation_id,
|
|
)
|
|
|
|
# `model_or_none` is part of the DifyBuilderAgent protocol; the Localizer
|
|
# uses it as its model provider (None -> reply-language localization is a no-op).
|
|
localizer = Localizer(agent.model_or_none)
|
|
|
|
def rename_app(proposed: str) -> str:
|
|
assert actor is not None
|
|
assert loaded_session is not None
|
|
return app_naming.rename_app_from_proposal(
|
|
app_id=loaded_session.app_id,
|
|
tenant_id=actor.tenant_id,
|
|
account_id=actor.account_id,
|
|
proposed=proposed,
|
|
)
|
|
|
|
env = Env(
|
|
dify=dify,
|
|
agent=agent,
|
|
repo=repo,
|
|
now=naive_utc_now,
|
|
emit_workflow=emit_workflow,
|
|
emit_canvas=emit_canvas,
|
|
emit_message=emit_message,
|
|
emit_item=emit_item,
|
|
emit_progress=emit_progress,
|
|
emit_reasoning=emit_reasoning,
|
|
detect_language=localizer.detect_language,
|
|
localize_items=localizer.localize_items,
|
|
rename_app=rename_app,
|
|
command_id=action.command_id,
|
|
)
|
|
runner = Runner(env, fix_registry() | build_registry() | edit_registry())
|
|
if action.base_app_revision:
|
|
_graph, current_app_revision = dify.read_graph(loaded_session.app_id, actor)
|
|
if action.base_app_revision != current_app_revision:
|
|
raise ConflictError(f"stale app revision for app {loaded_session.app_id}")
|
|
runner.advance(session_id, Turn(action=action, actor=actor))
|
|
completed = (repo, dify, actor)
|
|
except (ConflictError, HashMismatchError):
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "conflict",
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
except Exception:
|
|
# Generic message only -- never leak exception detail into the
|
|
# progress event (it is relayed to the end user via SSE in P3c).
|
|
logger.exception("dify_builder advance failed for session %s", session_id)
|
|
if last_progress is not None and last_progress.status == "running":
|
|
active_activities = [activity for activity in progress_activities.values() if activity.state == "active"]
|
|
failure_deltas = [replace(activity, state="failed") for activity in active_activities] or [None]
|
|
try:
|
|
for revision_offset, activity in enumerate(failure_deltas, start=1):
|
|
progress_bus.publish(
|
|
session_id,
|
|
asdict(
|
|
replace(
|
|
last_progress,
|
|
revision=last_progress.revision + revision_offset,
|
|
status="error",
|
|
activity=activity,
|
|
)
|
|
),
|
|
)
|
|
except Exception:
|
|
logger.exception("dify_builder failed progress publish failed for session %s", session_id)
|
|
retryable_message_failure = False
|
|
failed_action_kind = action.kind if action is not None else action_dict.get("kind")
|
|
failed_action_payload = action.payload if action is not None else action_dict.get("payload")
|
|
if repo is not None and failed_action_kind == "message" and isinstance(failed_action_payload, dict):
|
|
client_turn_id = failed_action_payload.get("client_turn_id")
|
|
if isinstance(client_turn_id, str) and client_turn_id:
|
|
try:
|
|
turn_kinds = repo.get_conversation_turn_kinds(session_id, client_turn_id)
|
|
retryable_message_failure = "user" in turn_kinds and "assistant_turn" not in turn_kinds
|
|
except Exception:
|
|
logger.exception(
|
|
"dify_builder could not inspect failed message turn for session %s",
|
|
session_id,
|
|
)
|
|
|
|
if retryable_message_failure:
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "step failed",
|
|
"recoverable": True,
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
elif runner is not None:
|
|
try:
|
|
assert repo is not None
|
|
# ``Runner.fail`` normally fences with the operation metadata
|
|
# established by ``Runner.advance``. If a failure happens
|
|
# before that metadata is initialized, compare against the
|
|
# head loaded by this worker so an expired/stale worker cannot
|
|
# overwrite a transition committed by a newer one.
|
|
if env is not None and env.at_version <= 0 and loaded_session is not None:
|
|
current, _context = repo.get_session(session_id)
|
|
if (
|
|
current.version != loaded_session.version
|
|
or current.current_state != loaded_session.current_state
|
|
):
|
|
raise ConflictError(
|
|
f"dify_builder: refusing stale failure for session {session_id} "
|
|
f"at version {current.version} state {current.current_state}"
|
|
)
|
|
runner.fail(session_id)
|
|
assert dify is not None
|
|
assert actor is not None
|
|
completed = (repo, dify, actor)
|
|
except ConflictError:
|
|
logger.warning("dify_builder stale worker failure ignored for session %s", session_id)
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "conflict",
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
except Exception:
|
|
logger.exception("dify_builder could not persist failed state for session %s", session_id)
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "step failed",
|
|
"recoverable": True,
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
elif repo is not None and loaded_session is not None:
|
|
try:
|
|
_persist_failed_state(
|
|
repo,
|
|
session_id,
|
|
expected_version=loaded_session.version,
|
|
expected_state=loaded_session.current_state,
|
|
command_id=action.command_id if action is not None else str(action_dict.get("command_id") or ""),
|
|
)
|
|
if dify is not None and actor is not None:
|
|
completed = (repo, dify, actor)
|
|
else:
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "step failed",
|
|
"recoverable": True,
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
except ConflictError:
|
|
logger.warning("dify_builder stale setup failure ignored for session %s", session_id)
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "conflict",
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
except Exception:
|
|
logger.exception("dify_builder could not persist early failed state for session %s", session_id)
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "step failed",
|
|
"recoverable": True,
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
else:
|
|
terminal_error = {
|
|
"kind": "error",
|
|
"error": "step failed",
|
|
"recoverable": True,
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
}
|
|
finally:
|
|
session_lock.release(session_id, token)
|
|
|
|
# Publish terminal frames only after releasing the lock so the
|
|
# authoritative SSE state no longer projects a completed/waiting
|
|
# transition as executing.
|
|
if terminal_error is not None:
|
|
progress_bus.publish(session_id, terminal_error)
|
|
elif completed is not None:
|
|
# The advance already succeeded and the lock is released; the session is
|
|
# durable. But the view projection (read_graph) or the publish itself can
|
|
# still raise -- and if this block let that escape, NO terminal frame
|
|
# would reach the SSE client, hanging its stream to the deadline with a
|
|
# stale held version (-> 409 on its next action). Fall back to a terminal
|
|
# error frame so the client stream always ends.
|
|
try:
|
|
repo, dify, actor = completed
|
|
view = DifyBuilderService(
|
|
repo,
|
|
session_lock,
|
|
lambda *a, **k: None,
|
|
get_app_revision_fn=lambda app_id, owner: dify.read_graph(app_id, owner)[1],
|
|
).get_session_view(session_id, actor)
|
|
progress_bus.publish(
|
|
session_id,
|
|
{
|
|
"kind": "command_finished",
|
|
**asdict(view),
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id", ""),
|
|
"post_canvas_action_id": post_canvas_action_id(
|
|
str(loaded_session.current_state) if loaded_session is not None else None,
|
|
action.kind if action is not None else action_dict.get("kind"),
|
|
),
|
|
},
|
|
)
|
|
except Exception:
|
|
logger.exception("dify_builder command_finished projection/publish failed for session %s", session_id)
|
|
try:
|
|
progress_bus.publish(
|
|
session_id,
|
|
{
|
|
"kind": "error",
|
|
"code": _COMMAND_FINISHED_UNAVAILABLE_CODE,
|
|
"error": "step failed",
|
|
"session_id": session_id,
|
|
"command_id": action.command_id if action is not None else action_dict.get("command_id"),
|
|
},
|
|
)
|
|
except Exception:
|
|
logger.exception("dify_builder terminal error frame also failed to publish for session %s", session_id)
|