cetdev/dify
0
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into feat/agent-v2

Browse Source 
# Conflicts:
#	api/controllers/console/app/agent_drive_inspector.py
#	api/migrations/versions/2026_06_18_2300-b2515f9d4c2a_agent_drive_skill_metadata_refactor.py
#	api/openapi/markdown/console-openapi.md
#	api/services/agent/skill_standardize_service.py
#	api/services/agent_drive_service.py
#	api/tests/unit_tests/controllers/console/app/test_agent_drive_inspector.py
#	api/tests/unit_tests/services/agent/test_skill_standardize_service.py
#	packages/contracts/generated/api/console/agent/orpc.gen.ts
#	packages/contracts/generated/api/console/agent/types.gen.ts
#	packages/contracts/generated/api/console/agent/zod.gen.ts
#	packages/contracts/generated/api/console/apps/orpc.gen.ts
#	packages/contracts/generated/api/console/apps/types.gen.ts
#	packages/contracts/generated/api/console/apps/zod.gen.ts
This commit is contained in:
Yanli 盐粒
2026-06-22 16:43:54 +08:00
parent 0262fff48d 084f122814
commit c3bd42ab93
215 changed files with 6267 additions and 3468 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/core/app/apps/common/workflow_response_converter.py
+12 -6
View File
@@ -51,8 +51,11 @@ from core.tools.entities.tool_entities import ToolProviderType
from core.tools.tool_manager import ToolManager
from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
from core.trigger.trigger_manager import TriggerManager
from core.workflow.human_input_forms import load_form_tokens_by_form_id
from core.workflow.human_input_forms import (
load_form_dispositions_by_form_id,
)
from core.workflow.human_input_policy import (
FormDisposition,
HumanInputSurface,
enrich_human_input_pause_reasons,
resolve_human_input_pause_reason_inputs,
@@ -340,13 +343,14 @@ class WorkflowResponseConverter:
human_input_form_ids = [reason.form_id for reason in resolved_reasons if isinstance(reason, HumanInputRequired)]
expiration_times_by_form_id: dict[str, datetime] = {}
display_in_ui_by_form_id: dict[str, bool] = {}
form_token_by_form_id: dict[str, str] = {}
dispositions_by_form_id: dict[str, FormDisposition] = {}
if human_input_form_ids:
stmt = select(
HumanInputForm.id,
HumanInputForm.expiration_time,
HumanInputForm.form_definition,
).where(HumanInputForm.id.in_(human_input_form_ids))
hitl_surface = _INVOKE_FROM_TO_HITL_SURFACE.get(self._application_generate_entity.invoke_from)
with Session(bind=db.engine) as session:
for form_id, expiration_time, form_definition in session.execute(stmt):
expiration_times_by_form_id[str(form_id)] = expiration_time
@@ -355,17 +359,17 @@ class WorkflowResponseConverter:
except (TypeError, json.JSONDecodeError):
definition_payload = {}
display_in_ui_by_form_id[str(form_id)] = bool(definition_payload.get("display_in_ui"))
form_token_by_form_id = load_form_tokens_by_form_id(
dispositions_by_form_id = load_form_dispositions_by_form_id(
human_input_form_ids,
session=session,
surface=_INVOKE_FROM_TO_HITL_SURFACE.get(self._application_generate_entity.invoke_from),
surface=hitl_surface,
)
# Reconnect paths must preserve the same pause-reason contract as live streams;
# otherwise clients see schema drift after resume.
pause_reasons = enrich_human_input_pause_reasons(
pause_reasons,
form_tokens_by_form_id=form_token_by_form_id,
dispositions_by_form_id=dispositions_by_form_id,
expiration_times_by_form_id={
form_id: int(expiration_time.timestamp())
for form_id, expiration_time in expiration_times_by_form_id.items()
@@ -379,6 +383,7 @@ class WorkflowResponseConverter:
expiration_time = expiration_times_by_form_id.get(reason.form_id)
if expiration_time is None:
raise ValueError(f"HumanInputForm not found for pause reason, form_id={reason.form_id}")
disposition = dispositions_by_form_id.get(reason.form_id)
responses.append(
HumanInputRequiredResponse(
task_id=task_id,
@@ -391,7 +396,8 @@ class WorkflowResponseConverter:
inputs=reason.inputs,
actions=reason.actions,
display_in_ui=display_in_ui_by_form_id.get(reason.form_id, False),
form_token=form_token_by_form_id.get(reason.form_id),
form_token=disposition.form_token if disposition else None,
approval_channels=list(disposition.approval_channels) if disposition else [],
resolved_default_values=reason.resolved_default_values,
expiration_time=int(expiration_time.timestamp()),
),
api/core/app/entities/task_entities.py
+3
View File
@@ -288,6 +288,7 @@ class HumanInputRequiredResponse(StreamResponse):
actions: Sequence[UserActionConfig] = Field(default_factory=list)
display_in_ui: bool = False
form_token: str | None = None
approval_channels: list[str] = Field(default_factory=list)
resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)
expiration_time: int = Field(..., description="Unix timestamp in seconds")
@@ -311,6 +312,7 @@ class HumanInputRequiredPauseReasonPayload(BaseModel):
actions: Sequence[UserActionConfig] = Field(default_factory=list)
display_in_ui: bool = False
form_token: str | None = None
approval_channels: list[str] = Field(default_factory=list)
resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)
expiration_time: int
@@ -325,6 +327,7 @@ class HumanInputRequiredPauseReasonPayload(BaseModel):
actions=data.actions,
display_in_ui=data.display_in_ui,
form_token=data.form_token,
approval_channels=data.approval_channels,
resolved_default_values=data.resolved_default_values,
expiration_time=data.expiration_time,
)
api/core/workflow/human_input_forms.py
+48 -47
View File
@@ -12,60 +12,61 @@ from collections.abc import Sequence
from sqlalchemy import select
from sqlalchemy.orm import Session
from core.workflow.human_input_policy import HumanInputSurface, get_preferred_form_token
from core.workflow.human_input_policy import (
FormDisposition,
HumanInputSurface,
disposition_for_surface,
)
from extensions.ext_database import db
from models.human_input import HumanInputFormRecipient, RecipientType
def load_form_dispositions_by_form_id(
form_ids: Sequence[str],
*,
session: Session | None = None,
surface: HumanInputSurface | None = None,
) -> dict[str, FormDisposition]:
"""Resolve each paused form's resume token and approval channels for `surface`."""
unique_form_ids = list(dict.fromkeys(form_ids))
if not unique_form_ids:
return {}
if session is not None:
return _load_form_dispositions_by_form_id(session, unique_form_ids, surface=surface)
with Session(bind=db.engine, expire_on_commit=False) as new_session:
return _load_form_dispositions_by_form_id(new_session, unique_form_ids, surface=surface)
def _load_form_dispositions_by_form_id(
session: Session,
form_ids: Sequence[str],
*,
surface: HumanInputSurface | None,
) -> dict[str, FormDisposition]:
recipients_by_form_id: dict[str, list[tuple[RecipientType, str]]] = {}
stmt = select(HumanInputFormRecipient).where(HumanInputFormRecipient.form_id.in_(form_ids))
for recipient in session.scalars(stmt):
recipients_by_form_id.setdefault(recipient.form_id, []).append(
(recipient.recipient_type, recipient.access_token or "")
)
return {
form_id: disposition_for_surface(recipients, surface=surface)
for form_id, recipients in recipients_by_form_id.items()
}
def load_form_tokens_by_form_id(
form_ids: Sequence[str],
*,
session: Session | None = None,
surface: HumanInputSurface | None = None,
) -> dict[str, str]:
"""Load the preferred access token for each human input form."""
unique_form_ids = list(dict.fromkeys(form_ids))
if not unique_form_ids:
return {}
if session is not None:
return _load_form_tokens_by_form_id(session, unique_form_ids, surface=surface)
with Session(bind=db.engine, expire_on_commit=False) as new_session:
return _load_form_tokens_by_form_id(new_session, unique_form_ids, surface=surface)
def _load_form_tokens_by_form_id(
session: Session,
form_ids: Sequence[str],
*,
surface: HumanInputSurface | None = None,
) -> dict[str, str]:
recipients_by_form_id: dict[str, list[tuple[RecipientType, str]]] = {}
stmt = select(HumanInputFormRecipient).where(HumanInputFormRecipient.form_id.in_(form_ids))
for recipient in session.scalars(stmt):
if not recipient.access_token:
continue
recipients_by_form_id.setdefault(recipient.form_id, []).append(
(recipient.recipient_type, recipient.access_token)
)
tokens_by_form_id: dict[str, str] = {}
for form_id, recipients in recipients_by_form_id.items():
token = _get_surface_form_token(recipients, surface=surface)
if token is not None:
tokens_by_form_id[form_id] = token
return tokens_by_form_id
def _get_surface_form_token(
recipients: Sequence[tuple[RecipientType, str]],
*,
surface: HumanInputSurface | None,
) -> str | None:
if surface in {HumanInputSurface.SERVICE_API, HumanInputSurface.OPENAPI}:
for recipient_type, token in recipients:
if recipient_type == RecipientType.STANDALONE_WEB_APP and token:
return token
return get_preferred_form_token(recipients)
"""Resume tokens only, for callers that don't surface approval channels."""
dispositions = load_form_dispositions_by_form_id(form_ids, session=session, surface=surface)
return {
form_id: disposition.form_token
for form_id, disposition in dispositions.items()
if disposition.form_token is not None
}
api/core/workflow/human_input_policy.py
+37 -6
View File
@@ -2,14 +2,14 @@ from __future__ import annotations
from collections.abc import Mapping, Sequence
from enum import StrEnum
from typing import Any
from typing import Any, NamedTuple
from graphon.entities.pause_reason import HumanInputRequired, PauseReason, PauseReasonType
from graphon.nodes.human_input.entities import FormInputConfig, SelectInputConfig
from graphon.nodes.human_input.enums import ValueSourceType
from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
from graphon.variables import ArrayStringSegment
from models.human_input import RecipientType
from models.human_input import ApprovalChannel, RecipientType
class HumanInputSurface(StrEnum):
@@ -20,7 +20,7 @@ class HumanInputSurface(StrEnum):
# SERVICE_API and OPENAPI are intentionally narrower than CONSOLE: token callers
# should only be able to act on end-user web forms, not internal console flows.
_ALLOWED_RECIPIENT_TYPES_BY_SURFACE: dict[HumanInputSurface, frozenset[RecipientType]] = {
ALLOWED_RECIPIENT_TYPES_BY_SURFACE: dict[HumanInputSurface, frozenset[RecipientType]] = {
HumanInputSurface.SERVICE_API: frozenset({RecipientType.STANDALONE_WEB_APP}),
HumanInputSurface.CONSOLE: frozenset({RecipientType.CONSOLE, RecipientType.BACKSTAGE}),
HumanInputSurface.OPENAPI: frozenset({RecipientType.STANDALONE_WEB_APP}),
@@ -41,7 +41,7 @@ def is_recipient_type_allowed_for_surface(
) -> bool:
if recipient_type is None:
return False
return recipient_type in _ALLOWED_RECIPIENT_TYPES_BY_SURFACE[surface]
return recipient_type in ALLOWED_RECIPIENT_TYPES_BY_SURFACE[surface]
def get_preferred_form_token(
@@ -59,10 +59,39 @@ def get_preferred_form_token(
return chosen_token
class FormDisposition(NamedTuple):
"""How a paused form resolves for one API surface.
A form's recipients split into those the surface may act on (yielding a resume
`form_token`) and those it may not (their channels named in `approval_channels`
so the caller is told where approval actually happens instead).
"""
form_token: str | None
approval_channels: list[ApprovalChannel]
def disposition_for_surface(
recipients: Sequence[tuple[RecipientType, str]],
*,
surface: HumanInputSurface | None,
) -> FormDisposition:
if surface is None:
return FormDisposition(form_token=get_preferred_form_token(recipients), approval_channels=[])
allowed = ALLOWED_RECIPIENT_TYPES_BY_SURFACE[surface]
actionable = [(recipient_type, token) for recipient_type, token in recipients if recipient_type in allowed]
return FormDisposition(
form_token=get_preferred_form_token(actionable),
approval_channels=sorted(
{recipient_type.approval_channel for recipient_type, _ in recipients if recipient_type not in allowed}
),
)
def enrich_human_input_pause_reasons(
reasons: Sequence[Mapping[str, Any]],
*,
form_tokens_by_form_id: Mapping[str, str],
dispositions_by_form_id: Mapping[str, FormDisposition],
expiration_times_by_form_id: Mapping[str, int],
) -> list[dict[str, Any]]:
enriched: list[dict[str, Any]] = []
@@ -71,7 +100,9 @@ def enrich_human_input_pause_reasons(
if updated.get("TYPE") == PauseReasonType.HUMAN_INPUT_REQUIRED:
form_id = updated.get("form_id")
if isinstance(form_id, str):
updated["form_token"] = form_tokens_by_form_id.get(form_id)
disposition = dispositions_by_form_id.get(form_id)
updated["form_token"] = disposition.form_token if disposition else None
updated["approval_channels"] = list(disposition.approval_channels) if disposition else []
expiration_time = expiration_times_by_form_id.get(form_id)
if expiration_time is not None:
updated["expiration_time"] = expiration_time