mtfyun/dify
1
0 Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-11-11 22:02:23 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
feat/dify-builder
dify/api/services/dify_builder/service.py
T
twwu b0045ab1f1 fix(builder): keep native tests in sync through approval 
Resume the original run after all approvals and restore terminal progress
and traversed edges on reload.

Show structured approval details and receipts, hide Stop while forms are
open, and recover when a running conversation is cleared.
2026-11-10 14:46:48 +08:00

2644 lines
116 KiB
Python
Raw Permalink Blame History

"""The Dify Builder usecase: session lifecycle + dispatch.
Port of dify-enterprise/server/pkg/enterprise/biz/dify_builder/usecase.go.
Dependency-injected (``repo``, ``session_lock``, ``enqueue_fn`` are
constructor args) so it stays unit-testable without importing the real
Redis-backed ``session_lock`` module or Celery -- those are wired in by the
caller (P3b Task 4: the Flask controller + the Celery task's ``.delay``).
"""
import json
import logging
from collections.abc import Callable, Iterator
from contextlib import AbstractContextManager, nullcontext, suppress
from dataclasses import asdict, dataclass, replace
from enum import StrEnum
from typing import TYPE_CHECKING, Protocol
from uuid import NAMESPACE_URL, uuid4, uuid5
from core.dify_builder.contract import (
CONFIRM_ACTION_ID,
ActionKind,
ActiveInteraction,
AppRevision,
CardOption,
CheckpointRef,
ConversationPage,
Decision,
InteractionResponseField,
InteractionResponseItem,
Phase,
PreflightContextCard,
PreflightIssue,
RecoveryClass,
RunContextCard,
RunStatus,
SessionModel,
SessionView,
UserItem,
)
from core.dify_builder.contract import Action as UiAction
from core.dify_builder.errors import BadRequestError, BusyError, ConflictError, NotFoundError
from core.dify_builder.models import (
CONFIRMED_TOOL_ACTION,
Action,
Actor,
ChecklistError,
ConversationItem,
DifyBuilderContext,
EntryMode,
PublicationDecision,
Run,
Session,
)
from core.dify_builder.state import PcState, canvas_read_only, is_terminal, is_waiting, is_working
from runtime_contracts.builder.copy_ids import BuilderCopyId
from runtime_contracts.builder.execution import canonical_digest
from runtime_contracts.builder.presentation import BuilderText, copy_text, stored_text, text_parts
from services.dify_builder import trace_diagnostics
from services.dify_builder.action_policy import requires_app_revision
from services.dify_builder.agent.model_resolver import validate_model_config
from services.dify_builder.application import recovery
from services.dify_builder.application.execution_policy import (
BuilderExecutionPolicyError,
HttpFixtureSetV1,
decode_http_fixture_submission,
)
from services.dify_builder.application.ports import Repository
from services.dify_builder.business_acceptance import context_publication_decision
from services.dify_builder.receipt_snapshot import apply_receipt_snapshot
from services.dify_builder.repair_gate import NO_REPAIR_ACTIONS, NO_REPAIR_DECISION_COPY
from services.dify_builder.test_actions import (
TEST_ACTIONS,
TestActionPort,
control_actions,
control_label,
control_title,
parse_test_action,
)
if TYPE_CHECKING:
from core.dify_builder.real_call_consent import ConsentReceipt
from services.dify_builder.attachment_intake import BuilderAttachmentIntake
from services.dify_builder.fix_intents import FixIntentRepository
from services.dify_builder.real_call_consent_service import BuilderRealCallConsentService
logger = logging.getLogger(__name__)
__all__ = [
"AppAccess",
"DifyBuilderService",
"SessionLock",
"SessionView",
"resolve_action_kind",
"resolve_submitted_action",
]
class AppAccess(StrEnum):
"""App-scoped permission tier required by one Builder operation.
Every tier includes the base app-edit check in the production authorizer;
test/run and release additionally enforce their dedicated RBAC scenes.
"""
EDIT = "edit"
TEST_AND_RUN = "test_and_run"
RELEASE = "release"
@dataclass(frozen=True)
class CommandSubmission:
"""Admission receipt. Pending work settles on the session stream."""
command_id: str
pending: bool
session: SessionView | None = None
class SessionLock(Protocol):
"""The per-session advance gate. Both the ``session_lock`` module and
the in-memory fake used in tests satisfy this structurally."""
def acquire(self, session_id: str, *, generation: int | None = None) -> str | None: ...
def release(self, session_id: str, token: str) -> None: ...
def exists(self, session_id: str) -> bool: ...
def current_owner(self, session_id: str) -> str | None: ...
class _SessionSubscription(Protocol):
"""The subscription surface used by command streams."""
def receive(self, timeout: float | None = 0.1) -> bytes | None: ...
def close(self) -> None: ...
_PHASE_FOR: dict[PcState, Phase] = {
# Fix.
PcState.FIX_DIAGNOSE: Phase.UNDERSTAND,
PcState.FIX_PROPOSE: Phase.PLAN,
PcState.FIX_AWAIT_APPROVAL: Phase.PLAN,
PcState.FIX_APPLY: Phase.MODIFY,
PcState.FIX_AWAIT_VERIFY: Phase.TEST,
PcState.FIX_AWAIT_TESTDATA: Phase.TEST,
PcState.FIX_VERIFY: Phase.TEST,
PcState.FIX_AWAIT_DECISION: Phase.REVIEW,
PcState.FIX_PUBLISH: Phase.PUBLISH,
# Checklist.
PcState.CHECKLIST_DIAGNOSE: Phase.UNDERSTAND,
PcState.CHECKLIST_PROPOSE: Phase.PLAN,
PcState.CHECKLIST_AWAIT_RECHECK: Phase.TEST,
# Terminal.
PcState.SUCCESS: Phase.COMPLETE,
PcState.FAILED: Phase.COMPLETE,
# Build.
PcState.BUILD_CAPABILITY_CHECK: Phase.UNDERSTAND,
PcState.BUILD_GOAL_ANALYSIS: Phase.CLARIFY,
PcState.BUILD_INITIAL_PLAN: Phase.PLAN,
PcState.BUILD_RESOURCE_RECOMMENDATION: Phase.RESOURCES,
PcState.BUILD_PLAN_APPROVAL: Phase.PLAN,
PcState.BUILD_EXECUTION: Phase.MODIFY,
PcState.BUILD_AWAIT_TESTDATA: Phase.TEST,
PcState.BUILD_TEST_AND_REPAIR: Phase.TEST,
PcState.BUILD_AWAIT_REPAIR: Phase.TEST,
PcState.BUILD_REVIEW: Phase.REVIEW,
PcState.BUILD_PUBLISH: Phase.PUBLISH,
PcState.BUILD_COMPLETE: Phase.COMPLETE,
PcState.BUILD_REVERTED: Phase.PLAN,
# Edit.
PcState.EDIT_CAPABILITY_CHECK: Phase.UNDERSTAND,
PcState.EDIT_IMPACT_ANALYSIS: Phase.CLARIFY,
PcState.EDIT_PLAN_APPROVAL: Phase.PLAN,
PcState.EDIT_APPLY_CHANGES: Phase.MODIFY,
PcState.EDIT_AWAIT_TESTDATA: Phase.TEST,
PcState.EDIT_TEST_AFFECTED_PATHS: Phase.TEST,
PcState.EDIT_AWAIT_REPAIR: Phase.TEST,
PcState.EDIT_REVIEW: Phase.REVIEW,
PcState.EDIT_PUBLISH: Phase.PUBLISH,
PcState.EDIT_COMPLETE: Phase.COMPLETE,
PcState.EDIT_REVERTED: Phase.PLAN,
}
def _phase_for(state: PcState) -> Phase:
"""Map a ``PcState`` to the coarse UX ``Phase`` shown in the panel
header (spec §2, §7). Defensive fallback for any unmapped state."""
return _PHASE_FOR.get(state, Phase.UNDERSTAND)
_ACTIONS_FOR: dict[PcState, list[UiAction]] = {
PcState.FIX_AWAIT_APPROVAL: [
UiAction(id="approve_plan", label=copy_text(BuilderCopyId.ACTIONS_APPROVE_FIX), kind=ActionKind.PRIMARY),
UiAction(id="reject_repair", label=copy_text(BuilderCopyId.ACTIONS_REJECT), kind=ActionKind.DESTRUCTIVE),
],
PcState.FIX_AWAIT_VERIFY: [
UiAction(id="run_validation", label=copy_text(BuilderCopyId.ACTIONS_RUN_VALIDATION), kind=ActionKind.PRIMARY),
UiAction(id="revert", label=copy_text(BuilderCopyId.ACTIONS_REVERT), kind=ActionKind.DESTRUCTIVE),
],
PcState.FIX_AWAIT_TESTDATA: [
UiAction(id="provide_testdata", label=copy_text(BuilderCopyId.FORMS_TEST_DATA_TITLE), kind=ActionKind.PRIMARY),
],
PcState.FIX_AWAIT_DECISION: [
UiAction(id="publish_fix", label=copy_text(BuilderCopyId.ACTIONS_PUBLISH_FIX), kind=ActionKind.PRIMARY),
UiAction(id="keep_draft", label=copy_text(BuilderCopyId.ACTIONS_KEEP_DRAFT), kind=ActionKind.SECONDARY),
UiAction(id="continue_adjusting", label=copy_text(BuilderCopyId.ACTIONS_FIX_AGAIN), kind=ActionKind.SECONDARY),
UiAction(id="revert", label=copy_text(BuilderCopyId.ACTIONS_REVERT), kind=ActionKind.DESTRUCTIVE),
],
PcState.CHECKLIST_AWAIT_RECHECK: [
UiAction(id="recheck", label=copy_text(BuilderCopyId.ACTIONS_RE_CHECK), kind=ActionKind.PRIMARY),
UiAction(id="revert", label=copy_text(BuilderCopyId.ACTIONS_REVERT), kind=ActionKind.DESTRUCTIVE),
],
PcState.FAILED: [
UiAction(
id="restart", label=copy_text(BuilderCopyId.ACTIONS_RESTART_FROM_CURRENT_DRAFT), kind=ActionKind.PRIMARY
),
],
# Build (Slice 2). next_state/canvas_event carry the frozen state-map hints.
PcState.BUILD_GOAL_ANALYSIS: [
UiAction(
id="submit_requirements",
label=copy_text(BuilderCopyId.ACTIONS_SUBMIT_REQUIREMENTS),
kind=ActionKind.PRIMARY,
next_state="build.initial_plan",
),
],
# build.initial_plan offers no UI action: it is a working/pass-through
# state now (state.py), not a waiting one -- the straight-through path no
# longer stops there (submit_requirements commits the abstract plan here
# before resource discovery), and the
# continue_adjusting/retry_after_revert loop-back that still lands there
# runs its handler unconditionally on entry, with no action to gate on.
PcState.BUILD_RESOURCE_RECOMMENDATION: [
UiAction(
id="confirm_resources",
label=copy_text(BuilderCopyId.ACTIONS_CONFIRM_RESOURCES),
kind=ActionKind.PRIMARY,
next_state="build.plan_approval",
),
],
PcState.BUILD_PLAN_APPROVAL: [
UiAction(
id="approve_plan",
label=copy_text(BuilderCopyId.ACTIONS_APPROVE_PLAN),
kind=ActionKind.PRIMARY,
next_state="build.execution",
),
UiAction(
id="continue_adjusting",
label=copy_text(BuilderCopyId.ACTIONS_CONTINUE_ADJUSTING),
kind=ActionKind.SECONDARY,
next_state="build.goal_analysis",
),
],
PcState.BUILD_EXECUTION: [
UiAction(
id="run_test",
label=copy_text(BuilderCopyId.ACTIONS_RUN_TEST),
kind=ActionKind.PRIMARY,
next_state="build.test_and_repair",
),
UiAction(
id="revert",
label=copy_text(BuilderCopyId.ACTIONS_REVERT),
kind=ActionKind.DESTRUCTIVE,
next_state="build.reverted",
canvas_event="graph_changed",
),
],
PcState.BUILD_AWAIT_TESTDATA: [
UiAction(id="provide_testdata", label=copy_text(BuilderCopyId.FORMS_TEST_DATA_TITLE), kind=ActionKind.PRIMARY),
],
PcState.BUILD_AWAIT_REPAIR: [
UiAction(
id="approve_plan",
label=copy_text(BuilderCopyId.ACTIONS_APPLY_THE_FIX),
kind=ActionKind.PRIMARY,
next_state="build.execution",
),
UiAction(
id="keep_draft",
label=copy_text(BuilderCopyId.ACTIONS_KEEP_DRAFT),
kind=ActionKind.SECONDARY,
next_state="build.complete",
),
UiAction(
id="revert",
label=copy_text(BuilderCopyId.ACTIONS_REVERT),
kind=ActionKind.DESTRUCTIVE,
next_state="build.reverted",
canvas_event="graph_changed",
),
],
PcState.BUILD_REVIEW: [
UiAction(
id="publish_workflow",
label=copy_text(BuilderCopyId.ACTIONS_PUBLISH),
kind=ActionKind.PRIMARY,
next_state="build.publish",
canvas_event="publish_workflow",
),
UiAction(
id="keep_draft",
label=copy_text(BuilderCopyId.ACTIONS_KEEP_DRAFT),
kind=ActionKind.SECONDARY,
next_state="build.governance_feedback",
),
UiAction(
id="continue_adjusting",
label=copy_text(BuilderCopyId.ACTIONS_CONTINUE_ADJUSTING),
kind=ActionKind.SECONDARY,
next_state="build.initial_plan",
),
UiAction(
id="revert",
label=copy_text(BuilderCopyId.ACTIONS_REVERT),
kind=ActionKind.DESTRUCTIVE,
next_state="build.reverted",
canvas_event="graph_changed",
),
],
PcState.BUILD_REVERTED: [
UiAction(
id="retry_after_revert",
label=copy_text(BuilderCopyId.ACTIONS_RETRY),
kind=ActionKind.PRIMARY,
next_state="build.initial_plan",
),
],
# Edit (Slice 3). next_state/canvas_event carry the frozen state-map hints.
PcState.EDIT_CAPABILITY_CHECK: [
UiAction(
id="send_edit_goal",
label=copy_text(BuilderCopyId.ACTIONS_SEND),
kind=ActionKind.PRIMARY,
next_state="edit.impact_analysis",
),
],
PcState.EDIT_IMPACT_ANALYSIS: [
UiAction(
id="submit_edit_rules",
label=copy_text(BuilderCopyId.ACTIONS_SUBMIT_RULES),
kind=ActionKind.PRIMARY,
next_state="edit.plan_approval",
),
],
# Approve is NOT the only exit: an approval the engine refuses leaves the
# user parked here with the plan unchanged, so the gate also offers a way
# back to the rules and a way out entirely (triage
# edit-branch-failure-2026-09-22, "Hard dead end").
PcState.EDIT_PLAN_APPROVAL: [
UiAction(
id="approve_plan",
label=copy_text(BuilderCopyId.ACTIONS_APPROVE_CHANGES),
kind=ActionKind.PRIMARY,
next_state="edit.apply_changes",
),
UiAction(
id="continue_adjusting",
label=copy_text(BuilderCopyId.ACTIONS_CONTINUE_ADJUSTING),
kind=ActionKind.SECONDARY,
next_state="edit.impact_analysis",
),
# Same ``revert`` id and the same perform_revert path as every other
# gate, but labelled for what it does HERE: nothing has been applied
# yet, so this discards the proposal rather than undoing an edit the
# user can see on the canvas.
UiAction(
id="revert",
label=copy_text(BuilderCopyId.ACTIONS_DISCARD_THIS_CHANGE_PLAN),
kind=ActionKind.DESTRUCTIVE,
next_state="edit.reverted",
canvas_event="graph_changed",
),
],
PcState.EDIT_APPLY_CHANGES: [
UiAction(
id="run_affected_tests",
label=copy_text(BuilderCopyId.ACTIONS_RUN_AFFECTED_TESTS),
kind=ActionKind.PRIMARY,
next_state="edit.test_affected_paths",
),
UiAction(
id="revert",
label=copy_text(BuilderCopyId.ACTIONS_REVERT),
kind=ActionKind.DESTRUCTIVE,
next_state="edit.reverted",
canvas_event="graph_changed",
),
],
PcState.EDIT_AWAIT_TESTDATA: [
UiAction(id="provide_testdata", label=copy_text(BuilderCopyId.FORMS_TEST_DATA_TITLE), kind=ActionKind.PRIMARY),
],
PcState.EDIT_AWAIT_REPAIR: [
UiAction(
id="approve_plan",
label=copy_text(BuilderCopyId.ACTIONS_APPLY_THE_FIX),
kind=ActionKind.PRIMARY,
next_state="edit.apply_changes",
),
UiAction(
id="keep_draft",
label=copy_text(BuilderCopyId.ACTIONS_KEEP_DRAFT),
kind=ActionKind.SECONDARY,
next_state="edit.complete",
),
UiAction(
id="revert",
label=copy_text(BuilderCopyId.ACTIONS_REVERT),
kind=ActionKind.DESTRUCTIVE,
next_state="edit.reverted",
canvas_event="graph_changed",
),
],
PcState.EDIT_REVIEW: [
UiAction(
id="publish_workflow",
label=copy_text(BuilderCopyId.ACTIONS_PUBLISH),
kind=ActionKind.PRIMARY,
next_state="edit.publish",
canvas_event="publish_workflow",
),
UiAction(
id="keep_draft",
label=copy_text(BuilderCopyId.ACTIONS_KEEP_DRAFT),
kind=ActionKind.SECONDARY,
next_state="edit.complete",
),
UiAction(
id="continue_adjusting",
label=copy_text(BuilderCopyId.ACTIONS_CONTINUE_ADJUSTING),
kind=ActionKind.SECONDARY,
next_state="edit.impact_analysis",
),
UiAction(
id="revert",
label=copy_text(BuilderCopyId.ACTIONS_REVERT),
kind=ActionKind.DESTRUCTIVE,
next_state="edit.reverted",
canvas_event="graph_changed",
),
],
PcState.EDIT_REVERTED: [
UiAction(
id="retry_after_revert",
label=copy_text(BuilderCopyId.ACTIONS_RETRY),
kind=ActionKind.PRIMARY,
next_state="edit.plan_approval",
),
],
}
def _actions_for(
state: PcState,
fc: DifyBuilderContext | None = None,
*,
interrupted: bool = False,
app_revision_conflicted: bool = False,
publication: PublicationDecision | None = None,
) -> list[UiAction]:
"""Return only actions legal for the projected lifecycle condition."""
if state == PcState.FAILED:
# A durable, deterministic failure -- re-running the same handler would
# just fail again, so only offer a restart from the current draft.
return [
UiAction(
id="restart", label=copy_text(BuilderCopyId.ACTIONS_RESTART_FROM_CURRENT_DRAFT), kind=ActionKind.PRIMARY
)
]
if fc is not None and fc.paused:
return []
recovery_ref = recovery.recovery_ref_for(fc.recovery_class) if fc is not None else None
if recovery_ref is not None:
actions: list[UiAction] = []
if recovery_ref.can_continue:
actions.append(
UiAction(
id="recovery_continue", label=copy_text(BuilderCopyId.ACTIONS_CONTINUE), kind=ActionKind.PRIMARY
)
)
if recovery_ref.can_restart:
actions.append(
UiAction(
id="restart",
label=copy_text(BuilderCopyId.ACTIONS_RESTART_FROM_CURRENT_DRAFT),
kind=ActionKind.SECONDARY,
)
)
return actions
if app_revision_conflicted:
return [
UiAction(
id="check_recovery",
label=copy_text(BuilderCopyId.ACTIONS_REVIEW_DRAFT_CHANGES),
kind=ActionKind.PRIMARY,
)
]
if interrupted:
# Retry re-runs a working handler. Draft drift must first use the
# classification and explicit recovery choices above, just as at a
# waiting gate; only an unchanged draft can retry immediately.
return [
UiAction(id="recovery_continue", label=copy_text(BuilderCopyId.ACTIONS_RETRY), kind=ActionKind.PRIMARY),
UiAction(
id="restart",
label=copy_text(BuilderCopyId.ACTIONS_RESTART_FROM_CURRENT_DRAFT),
kind=ActionKind.SECONDARY,
),
]
if (
state == PcState.BUILD_RESOURCE_RECOMMENDATION
and fc is not None
and fc.resource_selection.get("refresh_required")
):
# The existing client renders generic decisions only when no active
# resource form is projected. A second button beside the resource form
# would be invisible in that client, so offer rediscovery alone here.
return [
UiAction(
id="rediscover_resources",
label=copy_text(BuilderCopyId.PROGRESS_FIND_COMPATIBLE_RESOURCES),
kind=ActionKind.PRIMARY,
)
]
actions = list(_ACTIONS_FOR.get(state, []))
retest = {
PcState.BUILD_REVIEW: ("run_test", copy_text(BuilderCopyId.ACTIONS_RUN_TEST)),
PcState.EDIT_REVIEW: ("run_affected_tests", copy_text(BuilderCopyId.ACTIONS_RUN_AFFECTED_TESTS)),
PcState.FIX_AWAIT_DECISION: ("run_validation", copy_text(BuilderCopyId.ACTIONS_RUN_VALIDATION)),
}
if state in retest and (publication is None or not publication.allowed):
actions = [a for a in actions if a.id not in {"publish_workflow", "publish_fix"}]
action_id, label = retest[state]
actions.insert(0, UiAction(id=action_id, label=label, kind=ActionKind.PRIMARY))
if state in {PcState.FIX_AWAIT_APPROVAL, PcState.BUILD_AWAIT_REPAIR, PcState.EDIT_AWAIT_REPAIR} and (
fc is None or not fc.staged_repair
):
if state == PcState.FIX_AWAIT_APPROVAL:
return list(NO_REPAIR_ACTIONS)
return [action for action in actions if action.id != "approve_plan"]
return actions
_ACTION_ID_TO_KIND: dict[str, str] = {
"approve_plan": "approve_repair",
"run_validation": "run_verify",
"publish_fix": "publish",
"continue_adjusting": "re_fix",
"pause": "stop",
"revert": "undo",
"restart": "recovery_restart",
"retry_after_revert": "re_fix",
# provide_testdata / recheck / keep_draft already match handler kinds → passthrough
}
_DECISION_COPY_FOR: dict[PcState, tuple[BuilderText, BuilderText]] = {
PcState.FIX_AWAIT_APPROVAL: (
copy_text(BuilderCopyId.DECISION_HOW_SHOULD_BUILDER_PROCEED_WITH_THIS_FIX),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.FIX_AWAIT_VERIFY: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_BUILDER_DO_WITH_THE_APPLIED_FIX),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.FIX_AWAIT_DECISION: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_HAPPEN_TO_THIS_FIX),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.CHECKLIST_AWAIT_RECHECK: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_BUILDER_DO_NEXT),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.BUILD_PLAN_APPROVAL: (
copy_text(BuilderCopyId.DECISION_IS_THIS_WORKFLOW_PLAN_READY_TO_APPLY),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.BUILD_EXECUTION: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_BUILDER_DO_WITH_THE_BUILT_WORKFLOW),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.BUILD_AWAIT_REPAIR: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_BUILDER_DO_WITH_THE_PROPOSED_REPAIR),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.BUILD_REVIEW: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_HAPPEN_TO_THIS_WORKFLOW),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.BUILD_REVERTED: (
copy_text(BuilderCopyId.DECISION_HOW_SHOULD_BUILDER_CONTINUE_AFTER_THE_REVERT),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.EDIT_PLAN_APPROVAL: (
copy_text(BuilderCopyId.DECISION_IS_THIS_CHANGE_PLAN_READY_TO_APPLY),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.EDIT_APPLY_CHANGES: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_BUILDER_DO_WITH_THE_UPDATED_WORKFLOW),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.EDIT_AWAIT_REPAIR: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_BUILDER_DO_WITH_THE_PROPOSED_REPAIR),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.EDIT_REVIEW: (
copy_text(BuilderCopyId.DECISION_WHAT_SHOULD_HAPPEN_TO_THESE_CHANGES),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.EDIT_REVERTED: (
copy_text(BuilderCopyId.DECISION_HOW_SHOULD_BUILDER_CONTINUE_AFTER_THE_REVERT),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
PcState.FAILED: (
copy_text(BuilderCopyId.DECISION_HOW_SHOULD_BUILDER_RECOVER),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
}
_RECOVERY_DESCRIPTION: dict[str, BuilderText] = {
RecoveryClass.UNCHANGED: copy_text(BuilderCopyId.REPLIES_NO_CHANGES_SINCE_YOU_LAST_LEFT_RESUMING_WHERE_YOU_WERE),
RecoveryClass.CONFIG_ONLY: copy_text(
BuilderCopyId.REPLIES_ONLY_NODE_SETTINGS_CHANGED_SINCE_YOU_LEFT_CONTINUE_OR_RESTART
),
RecoveryClass.STRUCTURAL_COMPATIBLE: copy_text(BuilderCopyId.REPLIES_THE_WORKFLOW_STRUCTURE_CHANGED_BUT_THE),
RecoveryClass.STRUCTURAL_INVALIDATING: copy_text(BuilderCopyId.REPLIES_THE_NODES_THIS_TASK_WAS_WORKING),
}
def _decision_copy_for(state: PcState, actions: list[UiAction]) -> tuple[BuilderText, BuilderText]:
if state == PcState.FIX_AWAIT_APPROVAL and any(action.id == "retry_repair" for action in actions):
return NO_REPAIR_DECISION_COPY
if state in {PcState.BUILD_AWAIT_REPAIR, PcState.EDIT_AWAIT_REPAIR} and not any(
action.id == "approve_plan" for action in actions
):
return (
copy_text(BuilderCopyId.DECISION_HOW_SHOULD_BUILDER_CONTINUE),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
)
return _DECISION_COPY_FOR.get(
state,
(
copy_text(BuilderCopyId.DECISION_HOW_SHOULD_BUILDER_CONTINUE),
copy_text(BuilderCopyId.LABELS_CHOOSE_ONE_OPTION_TO_CONTINUE),
),
)
def _decision_for(state: PcState, actions: list[UiAction]) -> Decision | None:
"""Project a non-form gate as one blocking choice interaction."""
renderable = [a for a in actions if a.kind != ActionKind.AUTOMATIC]
if not renderable or any(action.id in _ACTIVE_INTERACTION_CARD_FOR_ACTION for action in renderable):
return None
# Only the first primary is badged: two recommendations is no recommendation.
default_index = next((i for i, a in enumerate(renderable) if a.kind == ActionKind.PRIMARY), -1)
options = [
CardOption(
id=a.id,
label=a.label,
is_default=(i == default_index),
tone="destructive" if a.kind == ActionKind.DESTRUCTIVE else "neutral",
next_state=a.next_state,
canvas_event=a.canvas_event,
)
for i, a in enumerate(renderable)
]
title, description = _decision_copy_for(state, actions)
return Decision(
title=title,
description=description,
options=options,
submit=UiAction(id=CONFIRM_ACTION_ID, label=copy_text(BuilderCopyId.ACTIONS_SUBMIT), kind=ActionKind.PRIMARY),
default_option_id=options[default_index].id if default_index >= 0 else "",
)
_ACTIVE_INTERACTION_CARD_FOR_ACTION: dict[str, tuple[str, str | None]] = {
"submit_requirements": ("form", "build_requirements"),
"submit_edit_rules": ("form", "edit_rules"),
"provide_testdata": ("form", "testdata"),
"confirm_resources": ("resource_select", None),
}
_FORM_COPY_FOR_VARIANT: dict[str, tuple[BuilderText, BuilderText]] = {
"build_requirements": (
copy_text(BuilderCopyId.FORMS_REQUIREMENTS_TITLE),
copy_text(BuilderCopyId.FORMS_REQUIREMENTS_DESCRIPTION),
),
"edit_rules": (
copy_text(BuilderCopyId.FORMS_EDIT_RULES_TITLE),
copy_text(BuilderCopyId.FORMS_EDIT_RULES_DESCRIPTION),
),
"testdata": (
copy_text(BuilderCopyId.FORMS_TEST_DATA_TITLE),
copy_text(BuilderCopyId.FORMS_TEST_DATA_DESCRIPTION),
),
}
def _display_interaction_value(value: object) -> BuilderText:
if value is None or value == "":
return copy_text(BuilderCopyId.VALUES_EMPTY)
if isinstance(value, bool):
return copy_text(BuilderCopyId.VALUES_YES if value else BuilderCopyId.VALUES_NO)
if isinstance(value, list):
return (
text_parts(
*[
part
for i, item in enumerate(value)
for part in ((", ", _display_interaction_value(item)) if i else (_display_interaction_value(item),))
]
)
if value
else copy_text(BuilderCopyId.VALUES_EMPTY)
)
if isinstance(value, dict):
name = value.get("name") or value.get("filename")
if isinstance(name, str) and name:
return name
return json.dumps(value, ensure_ascii=False, default=str, separators=(",", ":"))
return str(value)
def _selected_ui_action(actions: list[UiAction], action: Action) -> UiAction | None:
option_id = action.payload.get("option_id")
if isinstance(option_id, str) and option_id:
selected = next((candidate for candidate in actions if candidate.id == option_id), None)
if selected is not None:
return selected
matching = [candidate for candidate in actions if resolve_action_kind(candidate.id) == action.kind]
return matching[0] if len(matching) == 1 else None
def _interaction_response_for(
repo: Repository,
session: Session,
action: Action,
visible_actions: list[UiAction],
reply_language: str,
) -> InteractionResponseItem | None:
selected_action = _selected_ui_action(visible_actions, action)
if selected_action is None:
return None
card_contract = _ACTIVE_INTERACTION_CARD_FOR_ACTION.get(selected_action.id)
if card_contract is None:
title, _description = _decision_copy_for(session.current_state, visible_actions)
return InteractionResponseItem(
reply_language=reply_language,
interaction_kind="choice",
question=title,
answer=selected_action.label,
submitted_data={"option_id": selected_action.id},
)
card_kind, expected_variant = card_contract
card = repo.get_latest_conversation_item(session.id, frozenset({card_kind}))
if card is None:
return None
if card_kind == "resource_select":
raw_ids = action.payload.get("resource_ids")
selected_ids = [value for value in raw_ids if isinstance(value, str)] if isinstance(raw_ids, list) else []
resources = card.payload.get("recommended")
resource_options = resources if isinstance(resources, list) else []
selected_labels = [
str(resource.get("label") or resource.get("id") or "")
for resource in resource_options
if isinstance(resource, dict) and resource.get("id") in selected_ids
]
return InteractionResponseItem(
reply_language=reply_language,
interaction_kind="resource",
question=stored_text(card.payload.get("title") or copy_text(BuilderCopyId.RESOURCES_TITLE)),
answer=", ".join(label for label in selected_labels if label)
or copy_text(BuilderCopyId.RECEIPTS_NO_RESOURCES_SELECTED),
submitted_data={"resource_ids": selected_ids},
)
variant = str(card.payload.get("variant") or "")
if expected_variant is not None and variant != expected_variant:
return None
raw_values = action.payload.get("inputs") if selected_action.id == "provide_testdata" else action.payload
submitted_values = raw_values if isinstance(raw_values, dict) else {}
raw_fields = card.payload.get("fields")
fields = raw_fields if isinstance(raw_fields, list) else []
response_fields: list[InteractionResponseField] = []
normalized_values: dict[str, object] = {}
for raw_field in fields:
if not isinstance(raw_field, dict):
continue
key = raw_field.get("key")
if not isinstance(key, str) or not key:
continue
value = submitted_values.get(key)
normalized_values[key] = value
response_fields.append(
InteractionResponseField(
key=key,
label=stored_text(raw_field.get("label") or key),
value=value,
display_value=_display_interaction_value(value),
)
)
fallback_title, _fallback_description = _FORM_COPY_FOR_VARIANT.get(
variant,
(copy_text(BuilderCopyId.RECEIPTS_SUBMITTED_FORM), ""),
)
return InteractionResponseItem(
reply_language=reply_language,
interaction_kind="form",
question=stored_text(card.payload.get("title") or fallback_title),
fields=response_fields,
submitted_data=normalized_values,
)
# Handler-facing kinds accepted at each state. This is deliberately explicit:
# several handlers historically treated every unknown kind as a default branch
# (notably fix.await_decision -> keep_draft), so dispatch must never be the
# component that decides whether an action is valid.
_BACKEND_ACTIONS_FOR: dict[PcState, frozenset[str]] = {
PcState.FIX_AWAIT_APPROVAL: frozenset({"approve_repair", "reject_repair", "retry_repair", "finish_repair"}),
PcState.FIX_AWAIT_VERIFY: frozenset({"run_verify", "undo"}),
PcState.FIX_AWAIT_TESTDATA: frozenset({"provide_testdata"}),
PcState.FIX_AWAIT_DECISION: frozenset({"run_verify", "publish", "keep_draft", "re_fix", "undo"}),
PcState.CHECKLIST_AWAIT_RECHECK: frozenset({"recheck", "undo"}),
PcState.BUILD_GOAL_ANALYSIS: frozenset({"submit_requirements"}),
# BUILD_INITIAL_PLAN is a working/pass-through state now (state.py) -- it
# no longer gates on any action, so it has no entry here, matching every
# other working Build/Edit state (BUILD_TEST_AND_REPAIR, BUILD_PUBLISH, ...).
PcState.BUILD_RESOURCE_RECOMMENDATION: frozenset({"confirm_resources", "rediscover_resources"}),
PcState.BUILD_PLAN_APPROVAL: frozenset({"approve_repair", "re_fix"}),
PcState.BUILD_EXECUTION: frozenset({"run_test", "undo"}),
PcState.BUILD_AWAIT_TESTDATA: frozenset({"provide_testdata"}),
PcState.BUILD_AWAIT_REPAIR: frozenset({"approve_repair", "keep_draft", "undo"}),
PcState.BUILD_REVIEW: frozenset({"run_test", "publish_workflow", "keep_draft", "re_fix", "undo"}),
PcState.BUILD_REVERTED: frozenset({"re_fix"}),
PcState.EDIT_CAPABILITY_CHECK: frozenset({"send_edit_goal"}),
PcState.EDIT_IMPACT_ANALYSIS: frozenset({"submit_edit_rules"}),
PcState.EDIT_PLAN_APPROVAL: frozenset({"approve_repair", "re_fix", "undo"}),
PcState.EDIT_APPLY_CHANGES: frozenset({"run_affected_tests", "undo"}),
PcState.EDIT_AWAIT_TESTDATA: frozenset({"provide_testdata"}),
PcState.EDIT_AWAIT_REPAIR: frozenset({"approve_repair", "keep_draft", "undo"}),
PcState.EDIT_REVIEW: frozenset({"run_affected_tests", "publish_workflow", "keep_draft", "re_fix", "undo"}),
PcState.EDIT_REVERTED: frozenset({"re_fix"}),
}
_TEST_AND_RUN_ACTIONS = frozenset(
{
"provide_testdata",
"recheck",
"run_affected_tests",
"run_test",
"run_verify",
"stop",
"resume",
}
)
_RELEASE_ACTIONS = frozenset({"publish", "publish_workflow"})
_REPAIR_THEN_RETEST_STATES = frozenset({PcState.BUILD_AWAIT_REPAIR, PcState.EDIT_AWAIT_REPAIR})
# Working states whose interrupted-step Retry (recovery_continue) re-runs a
# privileged op: publish (RELEASE) or a draft test run (TEST_AND_RUN). A Retry
# re-invokes the state's handler, so it must be gated at that op's tier, not the
# base EDIT the recovery actions otherwise carry.
_PUBLISH_WORKING_STATES = frozenset({PcState.BUILD_PUBLISH, PcState.FIX_PUBLISH, PcState.EDIT_PUBLISH})
_TEST_WORKING_STATES = frozenset({PcState.BUILD_TEST_AND_REPAIR, PcState.FIX_VERIFY, PcState.EDIT_TEST_AFFECTED_PATHS})
def _app_access_for_action(state: PcState, kind: str) -> AppAccess:
if kind in _RELEASE_ACTIONS:
return AppAccess.RELEASE
if kind in _TEST_AND_RUN_ACTIONS or (kind == "approve_repair" and state in _REPAIR_THEN_RETEST_STATES):
return AppAccess.TEST_AND_RUN
# A recovery_continue (Retry) re-runs the interrupted working handler, which
# re-invokes the state's privileged op -- enforce that op's tier, not base EDIT.
# (recovery_restart only resets + re-enters the entry state, so it stays EDIT.)
if kind == "recovery_continue":
if state in _PUBLISH_WORKING_STATES:
return AppAccess.RELEASE
if state in _TEST_WORKING_STATES:
return AppAccess.TEST_AND_RUN
return AppAccess.EDIT
def _internal_action_allowed(state: PcState, fc: DifyBuilderContext, kind: str) -> bool:
"""Validate runner/service short-circuits against their product state.
These commands do not appear in ``_ACTIONS_FOR``, but they are not global:
transcript/model commands operate only at a waiting gate, pause transitions
must change the current pause flag, and recovery choices must match the
current terminal/interrupted/drift-recovery condition.
"""
if state == PcState.FAILED:
# Durable, deterministic failure -- only a restart is meaningful.
return kind == "recovery_restart"
if is_terminal(state):
return kind == "message" and not fc.paused and not fc.recovery_class
if not (is_waiting(state) or is_working(state)):
return False
if fc.paused:
return kind == "resume"
recovery_ref = recovery.recovery_ref_for(fc.recovery_class)
if recovery_ref is not None:
if kind == "recovery_continue":
return recovery_ref.can_continue
if kind == "recovery_restart":
return recovery_ref.can_restart
return False
# Classifying a working-state interrupt must settle without re-running its
# handler. A live worker is protected by dispatch's advance lock.
if is_working(state):
return kind in {"check_recovery", "recovery_continue", "recovery_restart"}
if kind in {"check_recovery", "message", "update_model"}:
return True
if kind == "stop":
return not fc.paused
if kind == "resume":
return fc.paused
return False
def resolve_action_kind(raw: str) -> str:
"""Map a frontend action_id to the engine handler kind.
Action IDs that already match their handler kind pass through unchanged.
"""
return _ACTION_ID_TO_KIND.get(raw, raw)
def resolve_submitted_action(action_id: str, payload: dict | None) -> str:
"""Map what the client posted to the engine handler kind.
The interaction dock posts ``confirm`` and names its choice in
``payload["option_id"]``. Handler-facing callers may already use the
resolved action id. A ``confirm`` naming no option resolves to "", which
the per-state legality check then rejects.
"""
if action_id == CONFIRM_ACTION_ID:
action_id = str(payload.get("option_id") or "") if isinstance(payload, dict) else ""
return resolve_action_kind(action_id)
_WAITING_INPUT_STATES = frozenset(
{
PcState.FIX_AWAIT_TESTDATA,
PcState.BUILD_GOAL_ANALYSIS,
PcState.BUILD_AWAIT_TESTDATA,
PcState.EDIT_CAPABILITY_CHECK,
PcState.EDIT_IMPACT_ANALYSIS,
PcState.EDIT_AWAIT_TESTDATA,
}
)
def _run_status(state: PcState, paused: bool = False) -> RunStatus:
"""Port of Go ``runStatusFor``, widened to the ``RunStatus`` enum (spec
§2). Deliberate wire-value change from the old string: ``waiting-input``
(hyphen) -> ``RunStatus.WAITING_INPUT`` = ``"waiting_input"``
(underscore).
Terminal check comes before waiting/working: ``PcState.BUILD_COMPLETE``
and ``PcState.EDIT_COMPLETE`` are terminal (spec §7.1/§7.2, ``run_status:
complete``) but are not in ``_WORKING``/``_WAITING`` and are not
``SUCCESS``/``FAILED`` -- without this ordering they'd wrongly fall
through to PROCESSING.
A safely settled task Stop pauses working and waiting states while retaining
their program counter. Terminal outcomes always win over the paused flag.
"""
if state == PcState.FAILED:
return RunStatus.FAILED
if is_terminal(state): # SUCCESS, BUILD_COMPLETE, EDIT_COMPLETE
return RunStatus.COMPLETE
if paused:
return RunStatus.PAUSED
if state in _WAITING_INPUT_STATES:
return RunStatus.WAITING_INPUT
if is_waiting(state):
return RunStatus.WAITING_CONFIRMATION
if is_working(state):
return RunStatus.PROCESSING
return RunStatus.PROCESSING # defensive; unreachable for classified states
class DifyBuilderService:
"""Session lifecycle (create/view/action/message) + dispatch.
Port of Go ``usecase.go``'s ``Usecase``.
"""
def __init__(
self,
repo: Repository,
session_lock: SessionLock,
enqueue_fn: Callable[[str, Action, Actor, str], None],
subscribe_fn: Callable[[str], _SessionSubscription | None] | None = None,
authorize_app_fn: Callable[[Actor, str, AppAccess], None] | None = None,
get_app_revision_fn: Callable[[str, Actor], str] | None = None,
get_app_name_fn: Callable[[str, Actor], str] | None = None,
get_verification_identity_fn: Callable[[str, Actor], tuple[str, str]] | None = None,
stamp_http_fixtures_fn: Callable[..., HttpFixtureSetV1] | None = None,
real_call_consent: "BuilderRealCallConsentService | None" = None,
test_actions: TestActionPort | None = None,
read_task_dsl_fn: Callable | None = None,
stream_read_scope_fn: Callable[[], AbstractContextManager] | None = None,
attachment_intake: "BuilderAttachmentIntake | None" = None,
cognition_policy_fn: Callable[[Actor, dict | None], dict] | None = None,
fix_intent_repo: "FixIntentRepository | None" = None,
) -> None:
# None: this instance cannot execute confirmed Tool tests (all existing callers).
self._fix_intent_repo = fix_intent_repo
self._cognition_policy_fn = cognition_policy_fn
self._real_call_consent = real_call_consent
self._attachment_intake = attachment_intake
self._test_actions = test_actions
if read_task_dsl_fn is None:
def read_task_dsl_fn(app_id, actor):
from services.dify_builder.dify_port import WorkflowServiceDifyPort
return WorkflowServiceDifyPort().read_task_dsl(app_id, actor)
self._read_task_dsl_fn = read_task_dsl_fn
self._stream_read_scope_fn = stream_read_scope_fn or nullcontext
self._stamp_http_fixtures_fn = stamp_http_fixtures_fn
self._repo = repo
self._session_lock = session_lock
self._enqueue_fn = enqueue_fn
self._subscribe_fn = subscribe_fn or (lambda _sid: None)
self._authorize_app_fn = authorize_app_fn or (lambda _actor, _app_id, _access: None)
self._get_app_revision_fn = get_app_revision_fn or (lambda _app_id, _actor: "")
self._get_verification_identity_fn = get_verification_identity_fn or (lambda _app_id, _actor: ("", ""))
self._get_app_name_fn = get_app_name_fn or (lambda _app_id, _actor: "")
def _validate_create_material(self, actor: Actor, attachments: list[str] | None) -> list[str]:
from services.dify_builder.attachment_intake import validated_upload_ids
ids = validated_upload_ids(attachments)
if ids:
if self._attachment_intake is None:
raise BadRequestError("Attachment intake is unavailable")
self._attachment_intake.authorize_uploads(actor, ids)
return ids
def _seed_material(
self,
session: Session,
context: DifyBuilderContext,
actor: Actor,
ids: list[str],
*,
client_turn_id: str | None = None,
) -> None:
if ids:
assert self._attachment_intake is not None
self._attachment_intake.submit(
session,
context,
actor,
text="",
upload_file_ids=ids,
client_turn_id=client_turn_id or "create:" + session.id,
)
def _get_app_revision(self, app_id: str, actor: Actor) -> str:
revision = self._get_app_revision_fn(app_id, actor)
return revision if isinstance(revision, str) else ""
def _get_app_name(self, app_id: str, actor: Actor) -> str:
name = self._get_app_name_fn(app_id, actor)
return name if isinstance(name, str) else ""
def _authorize_app(self, app_id: str, actor: Actor, access: AppAccess = AppAccess.EDIT) -> str:
if not isinstance(app_id, str) or not (app_id := app_id.strip()):
raise BadRequestError("app_id is required")
self._authorize_app_fn(actor, app_id, access)
return app_id
def _get_authorized_session(
self, session_id: str, actor: Actor, access: AppAccess = AppAccess.EDIT
) -> tuple[Session, DifyBuilderContext]:
s, fc = self._repo.get_session(session_id)
if s.owner_account_id != actor.account_id or s.tenant_id != actor.tenant_id:
raise NotFoundError("session not found")
self._authorize_app(s.app_id, actor, access)
from services.dify_builder.conversation_tasks import active_entry_mode
return replace(s, entry_mode=active_entry_mode(s, fc)), fc
def authorize_session(self, session_id: str, actor: Actor) -> None:
"""Authorize a session before opening an SSE subscription.
The stream route calls this once before subscribing, then reads the
bounded state after subscription through ``get_session_view`` to
preserve the no-lost-state-event ordering.
"""
self._get_authorized_session(session_id, actor)
def _validate_model_config(self, actor: Actor, model_config: dict | None) -> dict:
if model_config is not None and not isinstance(model_config, dict):
raise BadRequestError("model_config must be an object")
if self._cognition_policy_fn is not None:
return self._cognition_policy_fn(actor, model_config)
return validate_model_config(actor.tenant_id, model_config)
@staticmethod
def _parse_checklist_errors(
checklist_errors: list[ChecklistError] | list[dict] | None,
) -> list[ChecklistError] | None:
if checklist_errors is None:
return None
if not isinstance(checklist_errors, list):
raise BadRequestError("checklist_errors must be a list")
expected_fields = {"messages", "node_id", "node_type", "plugin_missing", "title", "unconnected"}
parsed: list[ChecklistError] = []
for entry in checklist_errors:
if isinstance(entry, ChecklistError):
values = {
"node_id": entry.node_id,
"node_type": entry.node_type,
"title": entry.title,
"messages": entry.messages,
"unconnected": entry.unconnected,
"plugin_missing": entry.plugin_missing,
}
elif isinstance(entry, dict) and set(entry) == expected_fields:
values = entry
else:
raise BadRequestError("invalid checklist_errors item")
messages = values["messages"]
if (
not isinstance(values["node_id"], str)
or not isinstance(values["node_type"], str)
or not isinstance(values["title"], str)
or not isinstance(messages, list)
or not all(isinstance(message, str) for message in messages)
or type(values["unconnected"]) is not bool
or type(values["plugin_missing"]) is not bool
):
raise BadRequestError("invalid checklist_errors item")
parsed.append(
ChecklistError(
node_id=values["node_id"],
node_type=values["node_type"],
title=values["title"],
messages=list(messages),
unconnected=values["unconnected"],
plugin_missing=values["plugin_missing"],
)
)
return parsed
def prepare_fix_session(
self,
app_id: str,
actor: Actor,
failed_run_id: str | None = None,
checklist_errors: list[ChecklistError] | list[dict] | None = None,
model_config: dict | None = None,
attachments: list[str] | None = None,
session_id: str | None = None,
client_action_id: str | None = None,
defer_existing: bool = False,
) -> tuple[str, Action | None]:
"""Shared setup for ``create_fix_session``/``create_fix_session_stream``:
everything through persisting the session (and the failed-run record)
up to -- but not including -- the ``request_fix`` dispatch. When
``defer_existing`` is set, return a selected conversation without
submitting a Fix intent before its observer is connected.
``failed_run_id`` is the id of the **Dify workflow run** that failed
(what the frontend has). In fix mode we record it as an immutable
``original-failed`` ``DifyBuilderRun`` and point ``fc.failed_run_id`` at
that row, so the async ``diagnose`` step can resolve it
(``repo.get_run(fc.failed_run_id)`` -> ``run.dify_run_id`` ->
``dify.node_outputs``). Checklist takes precedence when errors are
present (no failed run on that path).
"""
requested_model_config = model_config
if failed_run_id is not None and not isinstance(failed_run_id, str):
raise BadRequestError("failed_run_id must be a string")
failed_run_id = (failed_run_id or "").strip() or None
checklist_errors = self._parse_checklist_errors(checklist_errors)
if not failed_run_id and not checklist_errors:
raise BadRequestError("failed_run_id or checklist_errors is required")
app_id = self._authorize_app(app_id, actor)
if self._fix_intent_repo is not None:
selected = self._fix_intent_repo.select_conversation(actor, app_id, session_id)
if selected is not None and defer_existing:
return selected, None
if selected is not None:
from services.dify_builder.fix_intents import prepare_fix_intent
from services.dify_builder.task_lifecycle import task_identity
selected_head, selected_context = self._repo.get_session(selected)
receipt = prepare_fix_intent(
self,
selected,
actor,
failed_run_id=failed_run_id,
checklist_errors=checklist_errors,
model_config=model_config,
client_action_id=client_action_id
or "fix:"
+ canonical_digest(
{
"task": task_identity(selected_head, selected_context),
"run": failed_run_id,
"checklist": [asdict(item) for item in (checklist_errors or [])],
"model": requested_model_config,
"attachments": attachments,
}
),
attachments=attachments,
)
return receipt.session.session_id, None
if self._fix_intent_repo is not None:
self._fix_intent_repo.validate_source(
actor,
app_id,
{"kind": "checklist"} if checklist_errors else {"kind": "run", "failed_run_id": failed_run_id},
)
attachment_ids = self._validate_create_material(actor, attachments)
model_config = self._validate_model_config(actor, model_config)
app_revision = self._get_app_revision(app_id, actor)
failed_run: Run | None = None
if checklist_errors:
entry_mode, state = EntryMode.FIX_CHECKLIST, PcState.CHECKLIST_DIAGNOSE
fc = DifyBuilderContext(
source="checklist",
checklist_errors=checklist_errors,
model_config=model_config,
last_snapshot_hash=app_revision,
)
else:
entry_mode, state = EntryMode.FIX, PcState.FIX_DIAGNOSE
failed_run = Run(
id=str(uuid4()),
kind="original-failed",
dify_run_id=failed_run_id or "",
status="failed",
immutable=True,
)
fc = DifyBuilderContext(
failed_run_id=failed_run.id,
source="run",
model_config=model_config,
last_snapshot_hash=app_revision,
)
s = Session(
app_id=app_id,
tenant_id=actor.tenant_id,
owner_account_id=actor.account_id,
entry_mode=entry_mode,
current_state=state,
)
if checklist_errors:
issues = [
PreflightIssue(
node_id=issue.node_id,
node_type=issue.node_type,
title=issue.title,
messages=list(issue.messages),
unconnected=issue.unconnected,
plugin_missing=issue.plugin_missing,
)
for issue in checklist_errors
]
context_card = PreflightContextCard(
node_count=len({issue.node_id for issue in checklist_errors if issue.node_id}),
issue_count=len(checklist_errors),
issues=issues,
)
else:
context_card = RunContextCard(
run_id=failed_run_id or "",
title="",
error_code="",
message="",
trace_ref="",
)
items = [context_card.to_item(seq=0, at_version=0)]
if self._fix_intent_repo is not None:
selected, created = self._repo.create_fix_session_or_select(s, fc, items, source_run=failed_run)
if not created and defer_existing:
return selected.id, None
if not created:
from services.dify_builder.fix_intents import prepare_fix_intent
from services.dify_builder.task_lifecycle import task_identity
selected_head, selected_context = self._repo.get_session(selected.id)
receipt = prepare_fix_intent(
self,
selected.id,
actor,
failed_run_id=failed_run_id,
checklist_errors=checklist_errors,
model_config=requested_model_config,
client_action_id=client_action_id
or "fix:"
+ canonical_digest(
{
"task": task_identity(selected_head, selected_context),
"run": failed_run_id,
"checklist": [asdict(item) for item in (checklist_errors or [])],
"model": requested_model_config,
"attachments": attachments,
}
),
attachments=attachments,
)
return receipt.session.session_id, None
else:
self._repo.create_session(s, fc, items) # assigns s.id, s.version = 1
self._seed_material(s, fc, actor, attachment_ids)
if failed_run is not None and self._fix_intent_repo is None:
# Persist the failed-run record BEFORE dispatch so the enqueued
# advance's diagnose can resolve fc.failed_run_id.
self._repo.save_run(s.id, failed_run)
return s.id, Action(kind="request_fix", base_version=1, base_app_revision=app_revision)
def create_fix_session(
self,
app_id: str,
actor: Actor,
failed_run_id: str | None = None,
checklist_errors: list[ChecklistError] | list[dict] | None = None,
model_config: dict | None = None,
attachments: list[str] | None = None,
session_id: str | None = None,
client_action_id: str | None = None,
) -> SessionView:
"""Port of Go ``CreateFixSession``, extended to record the failed run."""
session_id, action = self.prepare_fix_session(
app_id, actor, failed_run_id, checklist_errors, model_config, attachments, session_id, client_action_id
)
if action is not None:
self.dispatch(session_id, action, actor)
return self.get_session_view(session_id, actor)
def create_fix_session_stream(
self,
app_id: str,
actor: Actor,
failed_run_id: str | None = None,
checklist_errors: list[ChecklistError] | list[dict] | None = None,
model_config: dict | None = None,
attachments: list[str] | None = None,
session_id: str | None = None,
client_action_id: str | None = None,
) -> Iterator[str]:
"""Streaming counterpart of ``create_fix_session``: subscribes to the
session's progress channel BEFORE dispatching the initial
``request_fix`` advance, so no progress frames are lost."""
from services.dify_builder.wiring import stream_advance_frames
session_id, action = self.prepare_fix_session(
app_id, actor, failed_run_id, checklist_errors, model_config, attachments, session_id, client_action_id
)
initial_items = self._repo.list_conversation(session_id) if action is not None else []
subscription = self._subscribe_fn(session_id) # BEFORE dispatch
try:
command_id = self.dispatch(session_id, action, actor) if action is not None else ""
view_dict = asdict(self.get_session_view(session_id, actor))
except Exception:
if subscription is not None:
subscription.close()
raise
return stream_advance_frames(
view_dict,
subscription,
expect_advance=action is not None,
emit_command_finished_when_settled=True,
command_id=command_id,
initial_items=initial_items,
read_session_view=lambda: self._read_stream_session_view(session_id, actor),
conversation_cleared=lambda: self.is_conversation_cleared(session_id, actor),
)
def prepare_build_session(
self,
app_id: str,
actor: Actor,
goal_text: str,
model_config: dict | None = None,
derive_app_name: bool = False,
attachments: list[str] | None = None,
) -> tuple[str, Action]:
"""Shared setup for ``create_build_session``/``create_build_session_stream``:
everything through persisting the session up to -- but not including --
the internal ``start_build`` dispatch."""
if not isinstance(goal_text, str) or (not goal_text.strip() and not attachments):
raise BadRequestError("goal_text is required")
goal_text = goal_text.strip() or "Analyze the attached material and clarify the workflow goal."
app_id = self._authorize_app(app_id, actor)
attachment_ids = self._validate_create_material(actor, attachments)
model_config = self._validate_model_config(actor, model_config)
app_revision = self._get_app_revision(app_id, actor)
fc = DifyBuilderContext(
goal_text=goal_text,
model_config=model_config,
last_snapshot_hash=app_revision,
app_name=self._get_app_name(app_id, actor),
app_name_auto=derive_app_name,
)
s = Session(
app_id=app_id,
tenant_id=actor.tenant_id,
owner_account_id=actor.account_id,
entry_mode=EntryMode.BUILD,
current_state=PcState.BUILD_CAPABILITY_CHECK,
)
items = [UserItem(text=goal_text, turn_id=str(uuid4())).to_item(seq=0, at_version=0)]
self._repo.create_session(s, fc, items) # assigns s.id, s.version = 1
self._seed_material(s, fc, actor, attachment_ids)
return s.id, Action(
kind="start_build",
base_version=1,
base_app_revision=app_revision,
)
def create_build_session(
self,
app_id: str,
actor: Actor,
goal_text: str,
model_config: dict | None = None,
derive_app_name: bool = False,
attachments: list[str] | None = None,
) -> SessionView:
"""Start a Build session at build.capability_check and dispatch the
initial ``start_build`` command (parallels ``create_fix_session``). The goal is
seeded as the user's opening bubble; the first advance's
``handle_capability_check`` analyzes it into requirements."""
session_id, action = self.prepare_build_session(
app_id, actor, goal_text, model_config, derive_app_name, attachments
)
if action is not None:
self.dispatch(session_id, action, actor)
return self.get_session_view(session_id, actor)
def create_build_session_stream(
self,
app_id: str,
actor: Actor,
goal_text: str,
model_config: dict | None = None,
derive_app_name: bool = False,
attachments: list[str] | None = None,
) -> Iterator[str]:
"""Streaming counterpart of ``create_build_session``: subscribes BEFORE
dispatching the initial ``start_build`` advance."""
from services.dify_builder.wiring import stream_advance_frames
session_id, action = self.prepare_build_session(
app_id, actor, goal_text, model_config, derive_app_name, attachments
)
initial_items = self._repo.list_conversation(session_id)
subscription = self._subscribe_fn(session_id) # BEFORE dispatch
try:
command_id = self.dispatch(session_id, action, actor)
view_dict = asdict(self.get_session_view(session_id, actor))
except Exception:
if subscription is not None:
subscription.close()
raise
return stream_advance_frames(
view_dict,
subscription,
expect_advance=True,
command_id=command_id,
initial_items=initial_items,
read_session_view=lambda: self._read_stream_session_view(session_id, actor),
conversation_cleared=lambda: self.is_conversation_cleared(session_id, actor),
)
def prepare_edit_session(
self,
app_id: str,
actor: Actor,
goal_text: str,
model_config: dict | None = None,
attachments: list[str] | None = None,
) -> tuple[str, Action]:
"""Shared setup for ``create_edit_session``/``create_edit_session_stream``:
validate and persist the opening goal, then return the initial
``send_edit_goal`` action for the caller to dispatch.
"""
if not isinstance(goal_text, str) or (not goal_text.strip() and not attachments):
raise BadRequestError("goal_text is required")
goal_text = goal_text.strip() or "Analyze the attached material and clarify the workflow goal."
app_id = self._authorize_app(app_id, actor)
attachment_ids = self._validate_create_material(actor, attachments)
model_config = self._validate_model_config(actor, model_config)
app_revision = self._get_app_revision(app_id, actor)
fc = DifyBuilderContext(
goal_text=goal_text,
model_config=model_config,
last_snapshot_hash=app_revision,
)
s = Session(
app_id=app_id,
tenant_id=actor.tenant_id,
owner_account_id=actor.account_id,
entry_mode=EntryMode.EDIT,
current_state=PcState.EDIT_CAPABILITY_CHECK,
)
items = [UserItem(text=goal_text, turn_id=str(uuid4())).to_item(seq=0, at_version=0)]
self._repo.create_session(s, fc, items) # assigns s.id, s.version = 1
self._seed_material(s, fc, actor, attachment_ids)
return s.id, Action(
kind="send_edit_goal",
payload={"text": goal_text},
base_version=1,
base_app_revision=app_revision,
)
def create_edit_session(
self,
app_id: str,
actor: Actor,
goal_text: str,
model_config: dict | None = None,
attachments: list[str] | None = None,
) -> SessionView:
"""Start an Edit session and dispatch its opening goal in one command.
The goal is seeded as the opening user bubble and the initial
``send_edit_goal`` advance reads the graph and analyzes impact.
"""
session_id, action = self.prepare_edit_session(app_id, actor, goal_text, model_config, attachments)
self.dispatch(session_id, action, actor)
return self.get_session_view(session_id, actor)
def create_edit_session_stream(
self,
app_id: str,
actor: Actor,
goal_text: str,
model_config: dict | None = None,
attachments: list[str] | None = None,
) -> Iterator[str]:
"""Streaming counterpart of ``create_edit_session``.
Subscribe before dispatching ``send_edit_goal`` and relay through its
terminal frame.
"""
from services.dify_builder.wiring import stream_advance_frames
session_id, action = self.prepare_edit_session(app_id, actor, goal_text, model_config, attachments)
initial_items = self._repo.list_conversation(session_id)
subscription = self._subscribe_fn(session_id) # BEFORE dispatch
try:
command_id = self.dispatch(session_id, action, actor)
view_dict = asdict(self.get_session_view(session_id, actor))
except Exception:
if subscription is not None:
subscription.close()
raise
return stream_advance_frames(
view_dict,
subscription,
expect_advance=True,
command_id=command_id,
initial_items=initial_items,
read_session_view=lambda: self._read_stream_session_view(session_id, actor),
conversation_cleared=lambda: self.is_conversation_cleared(session_id, actor),
)
def start_session(self, session_id: str, actor: Actor, request: Action) -> CommandSubmission:
"""Dispatch the opening command for an authorized, untouched session.
The caller opens the session stream first. Derive the command and its
payload from the saved opening context. The writer lease and the
initial version fence prevent a second start from enqueueing.
"""
session, context = self._get_authorized_session(session_id, actor)
kind = {
(EntryMode.BUILD, PcState.BUILD_CAPABILITY_CHECK): "start_build",
(EntryMode.EDIT, PcState.EDIT_CAPABILITY_CHECK): "send_edit_goal",
(EntryMode.FIX, PcState.FIX_DIAGNOSE): "request_fix",
(EntryMode.FIX_CHECKLIST, PcState.CHECKLIST_DIAGNOSE): "request_fix",
}.get((session.entry_mode, session.current_state))
if kind is None or session.version != 1 or request.base_version != session.version:
raise ConflictError("opening command is no longer current")
revision = self._get_app_revision(session.app_id, actor)
if request.base_app_revision != revision or context.last_snapshot_hash != revision:
raise ConflictError("draft changed before starting Builder")
action = Action(
kind=kind,
payload={"text": context.goal_text} if kind == "send_edit_goal" else {},
base_version=session.version,
base_app_revision=revision,
command_id=request.command_id,
)
return CommandSubmission(self.dispatch(session_id, action, actor, opening=True), pending=True)
def _read_stream_session_view(self, session_id: str, actor: Actor) -> dict[str, object] | None:
with self._stream_read_scope_fn():
_, context = self._get_authorized_session(session_id, actor)
from services.dify_builder.conversation_lifecycle import is_conversation_cleared
if (
not is_conversation_cleared(context)
and self._test_actions is not None
and self._test_actions.observation_initializing(
actor=actor, session_id=session_id, test_input_id=context.test_input_ref
)
):
return None
view = asdict(self.get_session_view(session_id, actor))
_, context = self._repo.get_session(session_id)
return {**view, "_task_paused": context.paused}
def stop_task(
self, session_id: str, actor: Actor, *, base_version: int, client_action_id: str, base_app_revision: str = ""
) -> SessionView:
from services.dify_builder.task_control_service import control
return control(
self,
session_id,
actor,
kind="stop",
base_version=base_version,
client_action_id=client_action_id,
base_app_revision=base_app_revision,
)
def resume_task(
self, session_id: str, actor: Actor, *, base_version: int, client_action_id: str, base_app_revision: str = ""
) -> SessionView:
from services.dify_builder.task_control_service import control
return control(
self,
session_id,
actor,
kind="resume",
base_version=base_version,
client_action_id=client_action_id,
base_app_revision=base_app_revision,
)
def clear_conversation(self, session_id: str, actor: Actor, **payload) -> dict:
from services.dify_builder.conversation_lifecycle import clear
return clear(self, session_id, actor, **payload)
def abandon_task(self, session_id: str, actor: Actor, **payload) -> dict:
from services.dify_builder.conversation_lifecycle import abandon
return abandon(self, session_id, actor, **payload)
def prepare_fix_intent(self, session_id: str, actor: Actor, *, base_version: int, **payload):
from services.dify_builder.fix_intents import prepare_fix_intent
from services.dify_builder.task_lifecycle import task_identity
current, context = self._get_authorized_session(session_id, actor)
if type(base_version) is not int or base_version < 1:
raise BadRequestError("base_version must be a positive integer")
observed, previous = self._repo.get_session_at_version(session_id, base_version)
if self._fix_intent_repo is None:
raise BadRequestError("Fix intake is unavailable")
client_action_id = payload.get("client_action_id")
saved = (
self._fix_intent_repo.find(actor, session_id, client_action_id)
if isinstance(client_action_id, str)
else None
)
target = task_identity(observed, previous)
if saved is not None:
if saved.target_task_id != target:
raise ConflictError("Fix replay base targets a different task")
# The use case still validates source/model/material and the saved
# body digest. A known key permits replay after its own activation,
# never arbitrary reuse of an old task's base for a new request.
elif task_identity(current, context) != target:
raise ConflictError("Fix intent targets a superseded task")
return prepare_fix_intent(self, session_id, actor, **payload)
def get_fix_control_receipt(
self,
session_id: str,
actor: Actor,
*,
operation: str,
client_action_id: str,
base_version: int,
fix_intent_id: str | None = None,
expected_task_id: str | None = None,
) -> dict:
"""Read an exact durable Fix admission/decision; never replay its effects."""
from services.dify_builder.fix_intents import get_fix_control_receipt
return get_fix_control_receipt(
self,
session_id,
actor,
operation=operation,
client_action_id=client_action_id,
base_version=base_version,
fix_intent_id=fix_intent_id,
expected_task_id=expected_task_id,
)
def is_conversation_cleared(self, session_id: str, actor: Actor) -> bool:
from services.dify_builder.conversation_lifecycle import is_conversation_cleared
with self._stream_read_scope_fn():
_, context = self._get_authorized_session(session_id, actor)
return is_conversation_cleared(context)
def get_session_view(self, session_id: str, actor: Actor) -> SessionView:
"""Return the bounded session projection; history has its own API."""
s, fc = self._get_authorized_session(session_id, actor)
from services.dify_builder.conversation_lifecycle import cleared_session_view, is_conversation_cleared
if is_conversation_cleared(fc):
view = cleared_session_view(s, fc)
if view.lifecycle_settlement == "pending" and self._repo.cleared_execution_settled(s):
view.lifecycle_settlement = "settled"
return view
if fc.paused and fc.task_control.get("settlement") in {"pending", "unknown"}:
from services.dify_builder.task_control_service import observe_stop
observe_stop(self._repo, self._test_actions, session_id, actor)
s, fc = self._get_authorized_session(session_id, actor)
return self._build_session_view(s, fc)
def get_conversation_page(
self,
session_id: str,
actor: Actor,
*,
limit: int,
before_seq: int | None = None,
after_seq: int | None = None,
) -> ConversationPage:
if type(limit) is not int or not 1 <= limit <= 100:
raise BadRequestError("limit must be between 1 and 100")
if before_seq is not None and (type(before_seq) is not int or before_seq < 0):
raise BadRequestError("before_seq must be a non-negative integer")
if after_seq is not None and (type(after_seq) is not int or after_seq < -1):
raise BadRequestError("after_seq must be an integer greater than or equal to -1")
if before_seq is not None and after_seq is not None:
raise BadRequestError("before_seq and after_seq are mutually exclusive")
_, context = self._get_authorized_session(session_id, actor)
from services.dify_builder.conversation_lifecycle import is_conversation_cleared
if is_conversation_cleared(context):
return ConversationPage(data=[], has_more=False, first_seq=None, last_seq=None)
return self._repo.list_conversation_page(
session_id,
limit=limit,
before_seq=before_seq,
after_seq=after_seq,
)
def _active_interaction_for(
self, session_id: str, version: int, actions: list[UiAction]
) -> ActiveInteraction | None:
for action in actions:
card_contract = _ACTIVE_INTERACTION_CARD_FOR_ACTION.get(action.id)
if card_contract is None:
continue
card_kind, expected_variant = card_contract
card = self._repo.get_latest_conversation_item(session_id, frozenset({card_kind}))
if card is None:
return None
if expected_variant is not None and card.payload.get("variant") != expected_variant:
return None
return ActiveInteraction(action_id=action.id, card=card, valid_at_version=version)
return None
def _publication_for_session(self, s: Session, fc: DifyBuilderContext, actor: Actor) -> PublicationDecision:
run = None
if fc.verify_run_id:
with suppress(NotFoundError):
run = self._repo.get_run(fc.verify_run_id)
revision, graph_revision = self._get_verification_identity_fn(s.app_id, actor)
return context_publication_decision(
run, s, fc, current_revision=revision, current_graph_revision=graph_revision, load_run=self._repo.get_run
)
def _build_session_view(self, s: Session, fc: DifyBuilderContext) -> SessionView:
from services.dify_builder.conversation_lifecycle import (
cleared_session_view,
is_conversation_cleared,
lifecycle_settlement,
)
from services.dify_builder.conversation_tasks import active_entry_mode
from services.dify_builder.task_lifecycle import task_identity
if is_conversation_cleared(fc):
return cleared_session_view(s, fc)
s = replace(s, entry_mode=active_entry_mode(s, fc))
st = s.current_state
continuation_pending = not fc.paused and fc.task_control.get("receipt", {}).get("enqueue") in {
"pending",
"dispatching",
"claimed",
}
lock_held = self._session_lock.exists(s.id)
current_app_revision = self._get_app_revision(
s.app_id,
Actor(account_id=s.owner_account_id, tenant_id=s.tenant_id),
)
app_revision_conflicted = bool(
fc.last_snapshot_hash and current_app_revision and fc.last_snapshot_hash != current_app_revision
)
stopped = fc.paused and not is_terminal(st) and fc.task_control.get("settlement", "settled") == "settled"
interrupted = is_working(st) and not lock_held and not fc.paused
actions = _actions_for(
st,
fc,
interrupted=interrupted,
app_revision_conflicted=app_revision_conflicted,
publication=self._publication_for_session(
s, fc, Actor(account_id=s.owner_account_id, tenant_id=s.tenant_id)
),
)
test_execution = None
receipt = None
native_controls = False
if self._test_actions is not None:
receipt = self._test_actions.current(
actor=Actor(account_id=s.owner_account_id, tenant_id=s.tenant_id),
session_id=s.id,
test_input_id=fc.test_input_ref,
)
if receipt is not None:
test_execution = receipt.model_dump(mode="json")
native_controls = receipt.status != "terminal" or receipt.outcome == "unknown"
if native_controls:
actions = control_actions(receipt)
decision = _decision_for(st, actions)
if native_controls and decision is not None:
decision.title = control_title(decision.options[0].id)
decision.description = (
copy_text(BuilderCopyId.DECISION_TEST_OPERATION_APPROVAL_DESCRIPTION)
if decision.options[0].id == "confirm_test_operation"
else ""
)
decision.default_option_id = ""
for option in decision.options:
option.is_default = False
if decision is not None and fc.recovery_class:
decision.description = _RECOVERY_DESCRIPTION[fc.recovery_class]
checkpoint = (
CheckpointRef(
checkpoint_id=fc.checkpoint_id, label=copy_text(BuilderCopyId.LABELS_RESTORE_POINT), created_at=""
)
if fc.checkpoint_id
else None
)
recovery_ref = recovery.recovery_ref_for(fc.recovery_class)
diagnostics = trace_diagnostics.from_history(
self._repo.list_recent_conversation(s.id, limit=24), session_id=s.id, version=s.version
)
if self._test_actions is not None and receipt is not None:
with suppress(Exception):
diagnostics = trace_diagnostics.merge_current(
diagnostics,
self._test_actions.current_diagnostics(
actor=Actor(account_id=s.owner_account_id, tenant_id=s.tenant_id),
session_id=s.id,
test_input_id=fc.test_input_ref,
receipt=receipt,
),
)
if fc.task_status == "abandoned":
actions = []
decision = None
pending_context = None
if self._fix_intent_repo is not None:
from services.dify_builder.fix_intents import pending_fix_projection
projection = pending_fix_projection(
self, s, fc, Actor(account_id=s.owner_account_id, tenant_id=s.tenant_id)
)
if projection is not None:
actions, decision, pending_context = projection
return SessionView(
lifecycle_settlement=lifecycle_settlement(fc),
task_id=task_identity(s, fc),
task_status="completed"
if is_terminal(st)
else "paused"
if fc.paused and fc.task_status != "abandoned"
else fc.task_status or "active",
task_summaries=fc.task_summaries,
pending_fix_context=pending_context,
test_execution=test_execution,
diagnostics=diagnostics,
attachments=self._attachment_intake.receipts(s, Actor(account_id=s.owner_account_id, tenant_id=s.tenant_id))
if self._attachment_intake is not None
else [],
session_id=s.id,
app_id=s.app_id,
version=s.version,
state=str(st),
canvas_read_only=False
if stopped
else lock_held or canvas_read_only(st) or fc.paused or continuation_pending,
run_status=(
RunStatus.PAUSED
if stopped
else RunStatus.PROCESSING
if fc.paused and fc.task_control.get("settlement") == "pending"
else RunStatus.FAILED
if fc.paused and fc.task_control.get("settlement") == "unknown"
else RunStatus.WAITING_CONFIRMATION
if receipt is not None and receipt.status == "waiting"
else RunStatus.PROCESSING
if receipt is not None and receipt.status in {"prepared", "running"}
else RunStatus.FAILED
if receipt is not None and receipt.outcome == "unknown"
else RunStatus.PROCESSING
if lock_held or continuation_pending
else _run_status(st, paused=fc.paused)
),
interrupted=interrupted,
conversation_last_seq=fc.next_seq - 1,
entry_mode=s.entry_mode,
phase=_phase_for(st),
actions=actions,
decision=decision,
active_interaction=self._active_interaction_for(s.id, s.version, actions),
checkpoint=checkpoint,
recovery=recovery_ref,
model=(
SessionModel(
provider=fc.model_config.get("provider", ""),
name=fc.model_config.get("name", ""),
mode=fc.model_config.get("mode", ""),
completion_params=fc.model_config.get("completion_params", {}),
)
if fc.model_config
else None
),
app_revision=AppRevision(
observed=fc.last_snapshot_hash,
current=current_app_revision,
conflicted=app_revision_conflicted,
),
last_command_id=fc.last_command_id,
command_pending=lock_held or continuation_pending,
reply_language=fc.reply_language,
)
def _prepare_action(
self, session_id: str, actor: Actor, action: Action
) -> tuple[SessionView, bool, list[ConversationItem]]:
"""Shared synchronous validation + settle for ``submit_action`` and the
streaming submit methods. Returns ``(view, expect_advance, items)``:
``expect_advance`` is ``True`` only when the caller must still call
``dispatch``; ``items`` contains rows committed synchronously before a
worker is needed. This method never dispatches itself, so a streaming
caller can subscribe before the advance is enqueued."""
if not action.command_id and action.kind in TEST_ACTIONS | {"stop", "resume"}:
action.command_id = str(
uuid5(
NAMESPACE_URL,
canonical_digest(
{
"session": session_id,
"actor": actor.account_id,
"tenant": actor.tenant_id,
"kind": action.kind,
"payload": action.payload,
"base_version": action.base_version,
"base_app_revision": action.base_app_revision,
}
),
)
)
action.command_id = action.command_id or str(uuid4())
kind = action.kind.strip() if isinstance(action.kind, str) else ""
if not kind:
raise BadRequestError("action kind is required")
if kind in {CONFIRMED_TOOL_ACTION, "task_continue"}:
# Only the confirmed dispatch owner's reservation winner may enqueue it.
raise BadRequestError(f"action {kind} is not allowed")
action.kind = kind
if not isinstance(action.payload, dict):
raise BadRequestError("action payload must be an object")
if type(action.base_version) is not int:
raise BadRequestError("base_version must be an integer")
if not isinstance(action.base_app_revision, str):
raise BadRequestError("base_app_revision must be a string")
if action.kind == "message":
text = action.payload.get("text")
if not isinstance(text, str) or not (text := text.strip()):
raise BadRequestError("message text is required")
client_turn_id = action.payload.get("client_turn_id")
if not isinstance(client_turn_id, str) or not (client_turn_id := client_turn_id.strip()):
raise BadRequestError("client_turn_id is required")
if len(client_turn_id) > 128:
raise BadRequestError("client_turn_id is too long")
action.payload = {**action.payload, "text": text, "client_turn_id": client_turn_id}
s, fc = self._get_authorized_session(session_id, actor)
from services.dify_builder.conversation_lifecycle import is_conversation_cleared
if is_conversation_cleared(fc):
raise ConflictError("conversation was cleared")
if action.kind in {"dismiss_fix", "stop_and_switch_fix"}:
from services.dify_builder.fix_intents import submit_fix_intent
return submit_fix_intent(self, session_id, actor, action), False, []
if action.kind in {"stop", "resume"}:
control = self.stop_task if action.kind == "stop" else self.resume_task
return (
control(
session_id,
actor,
base_version=action.base_version,
client_action_id=action.command_id,
base_app_revision=action.base_app_revision,
),
False,
[],
)
if action.kind == "message" and (
fc.task_status == "abandoned"
or fc.task_control.get("fresh_goal_receipt", {}).get("id") == action.payload["client_turn_id"]
):
from services.dify_builder.task_controls import generation
from services.dify_builder.task_lifecycle import task_identity
self._validate_model_config(actor, fc.model_config)
dsl, revision = self._read_task_dsl_fn(s.app_id, actor)
observed, previous = self._repo.get_session_at_version(session_id, action.base_version)
fresh, context, activated = self._repo.activate_abandoned_continuation(
session_id,
actor,
base_version=action.base_version,
client_turn_id=action.payload["client_turn_id"],
text=action.payload["text"],
app_revision=revision,
dsl=dsl,
expected_task_id=task_identity(observed, previous),
)
action.kind = "send_edit_goal"
action.base_version = fresh.version
action.base_app_revision = revision
action.execution_generation = generation(context)
action.command_id = action.payload["client_turn_id"]
return self._build_session_view(fresh, context), activated, []
if fc.task_control.get("receipt", {}).get("enqueue") in {"pending", "dispatching", "claimed"}:
raise BusyError("task continuation is pending")
if action.kind in TEST_ACTIONS:
return self._prepare_test_action(s, fc, actor, action)
if self._test_actions is not None:
current_test = self._test_actions.current(actor=actor, session_id=s.id, test_input_id=fc.test_input_ref)
if current_test is not None and (current_test.status != "terminal" or current_test.outcome == "unknown"):
raise BusyError("observe or finish the current test before starting another action")
if action.kind == "message":
client_turn_id = action.payload["client_turn_id"]
turn_kinds = self._repo.get_conversation_turn_kinds(session_id, client_turn_id)
if turn_kinds and self._attachment_intake is not None:
receipt = self._attachment_intake.submit(
s,
fc,
actor,
text=action.payload["text"],
upload_file_ids=[],
client_turn_id=client_turn_id,
as_message=True,
)
if receipt.status not in {"chat_pending", "chat_consumed"}:
return self._build_session_view(s, fc), False, []
if "assistant_turn" in turn_kinds:
return self._build_session_view(s, fc), False, []
if "user" in turn_kinds:
action.base_version = s.version
lifecycle_limited = bool(
fc.paused or fc.recovery_class or is_working(s.current_state) or is_terminal(s.current_state)
)
internal_allowed = _internal_action_allowed(s.current_state, fc, action.kind)
if s.current_state == PcState.BUILD_RESOURCE_RECOMMENDATION and action.kind in {
"confirm_resources",
"rediscover_resources",
}:
expected = "rediscover_resources" if fc.resource_selection.get("refresh_required") else "confirm_resources"
if action.kind != expected:
raise BadRequestError(f"action {action.kind} is not allowed for the current resource selection")
if not internal_allowed and (
lifecycle_limited or action.kind not in _BACKEND_ACTIONS_FOR.get(s.current_state, frozenset())
):
raise BadRequestError(f"action {action.kind} is not allowed in state {s.current_state}")
if action.kind in _RELEASE_ACTIONS or (
action.kind == "recovery_continue" and s.current_state in _PUBLISH_WORKING_STATES
):
publication = self._publication_for_session(s, fc, actor)
if not publication.allowed:
raise BadRequestError(f"publication verification denied: {publication.reason}; run validation again")
access = _app_access_for_action(s.current_state, action.kind)
if access != AppAccess.EDIT:
self._authorize_app(s.app_id, actor, access)
fixture_revision: str | None = None
if action.kind == "provide_testdata" and "http_fixtures" in action.payload:
try:
fixtures = decode_http_fixture_submission(action.payload["http_fixtures"])
if not action.base_app_revision:
raise BuilderExecutionPolicyError("fixture_base_revision_required")
if self._stamp_http_fixtures_fn is None:
raise BuilderExecutionPolicyError("fixture_admission_unavailable")
envelope = self._stamp_http_fixtures_fn(
s.app_id, actor, base_app_revision=action.base_app_revision, fixtures=fixtures
)
fixture_revision = envelope.execution_revision
# Keep the public action JSON-safe through the native Celery serializer.
action.payload = {**action.payload, "http_fixtures": [f.model_dump(mode="json") for f in fixtures]}
except BuilderExecutionPolicyError as exc:
raise BadRequestError(str(exc)) from None
if action.base_version != s.version:
raise ConflictError(f"stale base_version {action.base_version} for session {session_id}")
current_app_revision = fixture_revision or self._get_app_revision(s.app_id, actor)
app_revision_conflicted = bool(
fc.last_snapshot_hash and current_app_revision and fc.last_snapshot_hash != current_app_revision
)
# Inputs survive draft drift while the user fills the form; fixtures
# remain bound to the exact revision stamped above. Saving either form
# variant never rebases a staged repair or accepts its baseline drift.
baseline_exempt = action.kind == "provide_testdata"
if requires_app_revision(action):
if current_app_revision and not action.base_app_revision:
raise BadRequestError("base_app_revision is required")
if action.base_app_revision and action.base_app_revision != current_app_revision:
raise ConflictError(f"stale app revision for app {s.app_id}")
if (
app_revision_conflicted
and not fc.recovery_class
and not baseline_exempt
and action.kind not in {"check_recovery", "recovery_restart", "resume"}
# A completed-task message can only prepare a fresh plan/form.
# Its owner reads the current graph and invalidates old approvals.
and not (action.kind == "message" and is_terminal(s.current_state))
):
raise ConflictError(f"draft changed outside Builder for app {s.app_id}")
if (
action.kind == "approve_repair"
and s.current_state in {PcState.BUILD_AWAIT_REPAIR, PcState.EDIT_AWAIT_REPAIR, PcState.FIX_AWAIT_APPROVAL}
and not fc.staged_repair
):
raise BadRequestError("no repair is staged for approval")
if s.current_state == PcState.FIX_AWAIT_APPROVAL and (
(fc.staged_repair and action.kind in {"retry_repair", "finish_repair"})
or (not fc.staged_repair and action.kind == "reject_repair")
):
raise BadRequestError(f"action {action.kind} is not allowed for the current repair proposal")
visible_actions = _actions_for(
s.current_state,
fc,
interrupted=is_working(s.current_state) and not self._session_lock.exists(session_id),
app_revision_conflicted=app_revision_conflicted,
publication=self._publication_for_session(
s, fc, Actor(account_id=s.owner_account_id, tenant_id=s.tenant_id)
),
)
interaction_response = _interaction_response_for(self._repo, s, action, visible_actions, fc.reply_language)
if interaction_response is not None and action.receipt_snapshot is not None:
apply_receipt_snapshot(interaction_response, action.receipt_snapshot)
action.interaction_response = asdict(interaction_response) if interaction_response is not None else None
if action.kind == "update_model":
if is_working(s.current_state) or self._session_lock.exists(session_id):
raise BusyError(f"session {session_id} is busy")
model_config = action.payload.get("model_config")
if not isinstance(model_config, dict) or not model_config:
raise BadRequestError("model_config is required")
fc.model_config = self._validate_model_config(actor, model_config)
fc.last_command_id = action.command_id
self._repo.compare_and_advance(session_id, s.version, s.current_state, fc, [])
return self.get_session_view(session_id, actor), False, []
if action.kind == "message" and self._attachment_intake is not None:
self._attachment_intake.submit(
s,
fc,
actor,
text=action.payload["text"],
upload_file_ids=[],
client_turn_id=action.payload["client_turn_id"],
as_message=True,
)
return self._build_session_view(s, fc), True, []
def _prepare_test_action(self, s: Session, fc: DifyBuilderContext, actor: Actor, action: Action):
"""Apply an authorized native control once and commit its choice receipt.
The command digest and native owner retain retry identity. Public
presentation records the chosen action, independently of its scope.
"""
if self._test_actions is None:
raise BadRequestError("native test controls unavailable")
self._authorize_app(s.app_id, actor, AppAccess.TEST_AND_RUN)
command = parse_test_action(action.kind, action.payload)
fingerprint = canonical_digest(
{
"kind": action.kind,
"payload": action.payload,
"base_version": action.base_version,
"base_app_revision": action.base_app_revision,
}
)
prior = self._test_actions.find_command(actor=actor, session_id=s.id, command_id=action.command_id)
if prior is not None:
if prior.get("test_action_digest") != fingerprint:
raise ConflictError("action identity was reused with different content")
return self.get_session_view(s.id, actor), False, []
token = self._acquire_workflow_writer(s.id)
if token is None:
raise BusyError("Builder is still preparing the current test")
try:
s, fc = self._get_authorized_session(s.id, actor)
if (fc.paused or fc.recovery_class) and command.method != "cancel":
raise ConflictError("resolve Builder recovery before continuing the test")
if action.base_version != s.version:
raise ConflictError("stale test action session version")
if self._get_app_revision(s.app_id, actor) != action.base_app_revision:
raise ConflictError("stale test action draft revision")
current_request_id = self._test_actions.current_request_id(
actor=actor, session_id=s.id, test_input_id=fc.test_input_ref
)
if current_request_id != command.request_id:
raise ConflictError("test request is no longer current")
response = InteractionResponseItem(
reply_language=fc.reply_language,
interaction_kind="choice",
question=control_title(action.kind),
answer=control_label(action.kind),
submitted_data={"option_id": action.kind},
)
if action.receipt_snapshot is not None:
apply_receipt_snapshot(response, action.receipt_snapshot)
self._test_actions.execute(
actor=actor,
session_id=s.id,
test_input_id=fc.test_input_ref,
command=command,
idempotency_key=action.command_id,
)
# Native ownership has already committed; a crash here cannot authorize
# another dispatch. Repeating the same command reuses its durable key.
s, fc = self._get_authorized_session(s.id, actor)
fc.last_command_id = action.command_id
item = ConversationItem(
seq=fc.next_seq,
kind="interaction_response",
at_version=s.version + 1,
payload={
**asdict(response),
"test_action_kind": action.kind,
"test_command_id": action.command_id,
"test_action_digest": fingerprint,
"request_id": command.request_id,
},
)
fc.next_seq += 1
self._repo.compare_and_advance(s.id, s.version, s.current_state, fc, [item])
self._release_workflow_writer(s.id, token)
self._test_actions.settle(actor=actor, session_id=s.id, request_id=command.request_id)
return self.get_session_view(s.id, actor), False, [item]
finally:
self._release_workflow_writer(s.id, token)
def submit_action(self, session_id: str, actor: Actor, action: Action) -> SessionView:
"""Submit an action and return the latest bounded session view."""
receipt = self.submit_action_command(session_id, actor, action)
return receipt.session or self.get_session_view(session_id, actor)
def submit_action_command(self, session_id: str, actor: Actor, action: Action) -> CommandSubmission:
"""Validate and admit once. Return the projection only for synchronous settlement."""
view, expect_advance, _ = self._prepare_action(session_id, actor, action)
if expect_advance:
return CommandSubmission(self.dispatch(session_id, action, actor), pending=True)
return CommandSubmission(action.command_id, pending=False, session=view)
def submit_message(
self,
session_id: str,
actor: Actor,
text: str,
base_version: int,
client_turn_id: str,
attachments: list[str] | None = None,
) -> SessionView:
"""Submit a message and return the latest bounded session view."""
receipt = self.submit_message_command(session_id, actor, text, base_version, client_turn_id, attachments)
return receipt.session or self.get_session_view(session_id, actor)
def submit_message_command(
self,
session_id: str,
actor: Actor,
text: str,
base_version: int,
client_turn_id: str,
attachments: list[str] | None = None,
) -> CommandSubmission:
"""Admit a chat turn or material. Material intake returns its durable receipt immediately."""
queued = self._prepare_material_message(session_id, actor, text, base_version, client_turn_id, attachments)
if queued is not None:
session, action = queued
command_id = self._dispatch_material_action(session.id, action, actor) if action is not None else ""
return CommandSubmission(
command_id, pending=bool(command_id), session=self.get_session_view(session.id, actor)
)
if not isinstance(text, str) or not (text := text.strip()):
raise BadRequestError("message text is required")
return self.submit_action_command(
session_id,
actor,
Action(
kind="message",
payload={"text": text, "client_turn_id": client_turn_id},
base_version=base_version,
),
)
def submit_action_stream(self, session_id: str, actor: Actor, action: Action) -> Iterator[str]:
"""Streaming counterpart of ``submit_action``: performs the same eager
validation via ``_prepare_action`` (raising before any streaming begins),
then -- for the dispatch case -- subscribes BEFORE enqueuing the advance
so no progress frames are lost. Synchronously committed items are sent
individually before the terminal projection."""
from services.dify_builder.wiring import stream_advance_frames
view, expect_advance, settled_items = self._prepare_action(session_id, actor, action)
if not expect_advance:
# ``message`` (replay of a settled turn) and ``update_model`` both
# settle synchronously while changing the version, so they must emit
# a terminal ``command_finished`` frame carrying the new version --
# otherwise a client keeps a stale held version and its next action
# conflicts.
return stream_advance_frames(
asdict(view),
None,
expect_advance=False,
emit_command_finished_when_settled=action.kind
in {"message", "send_edit_goal", "update_model", "stop", "resume", "dismiss_fix", "stop_and_switch_fix"}
| TEST_ACTIONS,
command_id=action.command_id,
initial_items=settled_items,
conversation_cleared=lambda: self.is_conversation_cleared(session_id, actor),
)
subscription = self._subscribe_fn(session_id) # BEFORE dispatch
try:
command_id = self.dispatch(session_id, action, actor)
view_dict = asdict(view)
except Exception:
if subscription is not None:
subscription.close()
raise
return stream_advance_frames(
view_dict,
subscription,
expect_advance=True,
command_id=command_id,
read_session_view=lambda: self._read_stream_session_view(session_id, actor),
conversation_cleared=lambda: self.is_conversation_cleared(session_id, actor),
)
def submit_message_stream(
self,
session_id: str,
actor: Actor,
text: str,
base_version: int,
client_turn_id: str,
attachments: list[str] | None = None,
) -> Iterator[str]:
queued = self._prepare_material_message(session_id, actor, text, base_version, client_turn_id, attachments)
if queued is not None:
from services.dify_builder.wiring import stream_advance_frames
session, action = queued
subscription = self._subscribe_fn(session_id) if action is not None else None
try:
command_id = self._dispatch_material_action(session.id, action, actor) if action is not None else ""
if not command_id and subscription is not None:
subscription.close()
subscription = None
view = asdict(self.get_session_view(session.id, actor))
except Exception:
if subscription is not None:
subscription.close()
raise
return stream_advance_frames(
view,
subscription,
expect_advance=bool(command_id),
emit_command_finished_when_settled=not bool(command_id),
command_id=command_id,
read_session_view=lambda: self._read_stream_session_view(session_id, actor),
conversation_cleared=lambda: self.is_conversation_cleared(session_id, actor),
)
if not isinstance(text, str) or not (text := text.strip()):
raise BadRequestError("message text is required")
return self.submit_action_stream(
session_id,
actor,
Action(
kind="message",
payload={"text": text, "client_turn_id": client_turn_id},
base_version=base_version,
),
)
def _prepare_material_message(self, session_id, actor, text, base_version, client_turn_id, attachments):
"""Queue new material without interrupting an in-flight plan or taking its lease."""
if not attachments and self._attachment_intake is None:
return None
session, context = self._get_authorized_session(session_id, actor)
from services.dify_builder.conversation_lifecycle import is_conversation_cleared
if is_conversation_cleared(context):
raise ConflictError("conversation was cleared")
working = self._session_lock.exists(session.id) or is_working(session.current_state)
if self._attachment_intake is None:
raise BadRequestError("Attachment intake is unavailable")
if type(base_version) is not int or base_version > session.version:
raise ConflictError("stale base_version")
existing = self._attachment_intake.find_turn(session, actor, client_turn_id)
if not attachments and not working and existing is None:
return None
receipt = self._attachment_intake.submit(
session,
context,
actor,
text=text,
upload_file_ids=attachments,
client_turn_id=client_turn_id,
as_message=not attachments and not working,
)
if receipt.status == "chat_pending":
return None
session, context = self._get_authorized_session(session_id, actor)
working = self._session_lock.exists(session.id) or is_working(session.current_state)
action = None
if receipt.status == "pending" and not working and not context.paused:
action = Action(
kind="consume_material",
base_version=session.version,
command_id=str(uuid5(NAMESPACE_URL, f"builder-material:{session.id}:{receipt.id}")),
)
return session, action
def _dispatch_material_action(self, session_id: str, action: Action, actor: Actor) -> str:
try:
return self.dispatch(session_id, action, actor)
except BusyError:
return "" # The existing owner consumes this already accepted queue.
def dispatch_pending_material(self, session_id: str, actor: Actor) -> str:
"""Handoff after lease release; a concurrent winner keeps the same lease owner."""
if self._attachment_intake is None:
return ""
session, context = self._get_authorized_session(session_id, actor)
if context.paused or is_working(session.current_state) or self._session_lock.exists(session_id):
return ""
pending = self._attachment_intake.pending(session, context, actor)
if not pending:
return ""
return self._dispatch_material_action(
session_id,
Action(
kind="consume_material",
base_version=session.version,
command_id=str(uuid5(NAMESPACE_URL, f"builder-material:{session.id}:{pending[-1].id}")),
),
actor,
)
def _acquire_workflow_writer(self, session_id: str, *, generation: int | None = None) -> str | None:
from configs import dify_config
from services.dify_builder.task_controls import generation as task_generation
if generation is None:
_, context = self._repo.get_session(session_id)
generation = task_generation(context)
token = self._session_lock.acquire(session_id, generation=generation)
if token is None:
return None
try:
reserved = self._repo.reserve_workflow_writer(
session_id, token, generation, dify_config.DIFY_BUILDER_MAX_ADVANCE_SECONDS
)
except BaseException:
self._session_lock.release(session_id, token)
raise
if not reserved:
self._session_lock.release(session_id, token)
return None
return token
def _release_workflow_writer(self, session_id: str, token: str) -> None:
self._repo.release_workflow_writer(session_id, token=token)
self._session_lock.release(session_id, token)
def execute_confirmed_tool(
self,
session_id: str,
actor: Actor,
*,
consent_id: str,
scope_digest: str,
at_version: int,
idempotency_key: str,
) -> "ConsentReceipt":
"""Confirmed dispatch coordinator: one lease, one reservation, at most one enqueue.
Acquires the session's advance lease BEFORE the consent owner's atomic
SQL reservation and hands that same token only to the winning enqueue
of the already reserved command (no ``_prepare_action``, no second
command or state commit). A held lease answers a matching duplicate
from the consent owner's replay (current receipt, no reservation or
enqueue) and is busy otherwise. Refusal or a non-winning replay
releases only this call's own token. A failed enqueue keeps the lease:
its acknowledgement may be lost while the worker already holds the same
token, and the reservation can never be enqueued again; the reconcile
sweeper settles a launch that never starts.
"""
if self._real_call_consent is None:
raise BadRequestError("confirmed tool execution is unavailable")
consent = self._real_call_consent
s, _fc = self._get_authorized_session(session_id, actor, AppAccess.TEST_AND_RUN)
if _fc.paused:
raise ConflictError("task is paused")
token = self._acquire_workflow_writer(session_id)
if token is None:
replay = consent.replay_execution(
actor=actor,
session_id=session_id,
consent_id=consent_id,
scope_digest=scope_digest,
at_version=at_version,
idempotency_key=idempotency_key,
)
if replay is None:
raise BusyError(f"session {session_id} is busy")
return replay
try:
receipt, won = consent.reserve_execution(
actor=actor,
session_id=session_id,
consent_id=consent_id,
scope_digest=scope_digest,
at_version=at_version,
idempotency_key=idempotency_key,
)
except BaseException:
self._release_workflow_writer(session_id, token)
raise
if not won or receipt.command_id is None:
self._release_workflow_writer(session_id, token)
return receipt
action = Action(
kind=CONFIRMED_TOOL_ACTION,
payload={"consent_id": receipt.consent_id},
base_version=receipt.at_version,
command_id=receipt.command_id,
)
try:
self._enqueue_fn(session_id, action, actor, token)
except Exception:
logger.warning(
"dify_builder confirmed enqueue acknowledgement unavailable (session=%s, command=%s)",
s.id,
receipt.command_id,
)
return receipt
def dispatch(self, session_id: str, action: Action, actor: Actor, *, opening: bool = False) -> str:
"""Reserve one workflow writer and enqueue once; uncertain ACKs retain ownership."""
action.command_id = action.command_id or str(uuid4())
from services.dify_builder.task_controls import generation
_, context = self._repo.get_session(session_id)
from services.dify_builder.application.cognition_policy import cognition_admission_required
if self._cognition_policy_fn is not None and cognition_admission_required(action.kind):
self._validate_model_config(actor, context.model_config)
if context.paused:
raise ConflictError("task is paused")
if action.execution_generation is None:
if action.base_version > 0:
_, observed = self._repo.get_session_at_version(session_id, action.base_version)
action.execution_generation = generation(observed)
else:
action.execution_generation = generation(context)
if generation(context) != action.execution_generation:
raise ConflictError("task dispatch stopped or superseded")
token = self._acquire_workflow_writer(session_id, generation=action.execution_generation)
if token is None:
raise BusyError(f"session {session_id} is busy")
try:
session, current = self._repo.get_session(session_id)
if current.paused or generation(current) != action.execution_generation:
raise ConflictError("task dispatch stopped or superseded")
if opening and (session.version != action.base_version or current.last_command_id):
raise ConflictError("opening command is no longer current")
if action.kind == "consume_material":
action.base_version = session.version
if action.kind == "send_edit_goal" and session.current_state == PcState.EDIT_CAPABILITY_CHECK:
text = action.payload.get("text")
if isinstance(text, str) and text and text != current.goal_text:
from services.dify_builder.business_acceptance import clear_task_acceptance
# After acquiring the lease, preserve ordinary input before
# enqueue. Stop may fence delivery without losing this goal.
clear_task_acceptance(current)
current.goal_text = text
action.base_version = self._repo.compare_and_advance(
session_id, action.base_version, session.current_state, current, []
)
except Exception:
self._release_workflow_writer(session_id, token)
raise
# A broker acknowledgement can be lost before worker activation. Retain
# both leases on any enqueue error; Stop/expiry supplies recovery, never replay.
from services.dify_builder.errors import EnqueueAcknowledgementError
try:
self._enqueue_fn(session_id, action, actor, token)
except Exception as error:
raise EnqueueAcknowledgementError(str(error)) from error
return action.command_id
Reference in new issue View Git Blame Copy Permalink