Skip to content
Merged
37 changes: 30 additions & 7 deletions api/core/workflow/nodes/agent_v2/runtime_request_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@
from graphon.file import File, FileTransferMethod
from graphon.variables.segments import Segment
from graphon.variables.template_resolution import convert_template
from libs.workflow_template_values import resolve_template_value
from models.agent import Agent, AgentConfigSnapshot, WorkflowAgentNodeBinding
from models.agent_config_entities import (
AgentKnowledgeMetadataFilteringConfig,
Expand Down Expand Up @@ -216,7 +217,7 @@ def build(self, context: WorkflowAgentRuntimeBuildContext) -> WorkflowAgentRunti
runtime_config_skills=runtime_config_skills,
)
soul_prompt = expand_prompt_mentions(agent_soul.prompt.system_prompt, soul_prompt_resolver).strip()
knowledge_config = build_knowledge_layer_config(agent_soul)
knowledge_config = build_knowledge_layer_config(agent_soul, variable_pool=context.variable_pool)
context_window_tokens = resolve_model_context_window(
run_context=context.dify_context,
provider_name=agent_soul.model.model_provider,
Expand Down Expand Up @@ -743,14 +744,18 @@ def build_shell_layer_config(agent_soul: AgentSoulConfig) -> DifyShellLayerConfi
)


def build_knowledge_layer_config(agent_soul: AgentSoulConfig) -> DifyKnowledgeBaseLayerConfig | None:
def build_knowledge_layer_config(
agent_soul: AgentSoulConfig, *, variable_pool: VariablePoolReader | None = None
) -> DifyKnowledgeBaseLayerConfig | None:
"""Map Agent Soul knowledge sets into one Dify knowledge-base layer.

Agent Soul DTO validation owns malformed set rejection. Runtime mapping is
intentionally lossless: every configured set is forwarded with its query
policy, dataset refs, retrieval controls, and metadata-filtering controls.
``score_threshold=None`` means disabled threshold filtering and maps to the
inner retrieval request's ``0.0`` default through the Agent backend DTO.
Workflow callers supply the current run's variable pool; ``None`` preserves
literal filters for standalone Agent requests, which have no workflow scope.
"""
if not agent_soul.knowledge.sets:
return None
Expand All @@ -774,7 +779,9 @@ def build_knowledge_layer_config(agent_soul: AgentSoulConfig) -> DifyKnowledgeBa
value=knowledge_set.query.value,
),
retrieval=_knowledge_retrieval_config(knowledge_set.retrieval),
metadata_filtering=_knowledge_metadata_filtering_config(knowledge_set.metadata_filtering),
metadata_filtering=_knowledge_metadata_filtering_config(
knowledge_set.metadata_filtering, variable_pool=variable_pool
),
)
for knowledge_set in agent_soul.knowledge.sets
],
Expand Down Expand Up @@ -821,17 +828,33 @@ def _knowledge_retrieval_config(retrieval: AgentKnowledgeRetrievalConfig) -> Dif

def _knowledge_metadata_filtering_config(
metadata_filtering: AgentKnowledgeMetadataFilteringConfig,
*,
variable_pool: VariablePoolReader | None,
) -> DifyKnowledgeMetadataFilteringConfig:
conditions = metadata_filtering.conditions
if conditions is not None and metadata_filtering.mode == "manual" and variable_pool is not None:
try:
conditions = conditions.model_copy(
update={
"conditions": [
condition.model_copy(update={"value": resolve_template_value(condition.value, variable_pool)})
for condition in conditions.conditions
]
}
)
except ValueError as error:
raise WorkflowAgentRuntimeRequestBuildError(
"agent_metadata_filter_variable_not_found", str(error)
) from error

return DifyKnowledgeMetadataFilteringConfig(
mode=metadata_filtering.mode,
model_config=_knowledge_model_config(metadata_filtering.metadata_model_config),
conditions=cast(
Any,
metadata_filtering.conditions.model_dump(
mode="json", exclude={"conditions": {"__all__": {"id", "metadata_id"}}}
),
conditions.model_dump(mode="json", exclude={"conditions": {"__all__": {"id", "metadata_id"}}}),
)
if metadata_filtering.conditions is not None
if conditions is not None
else None,
)

Expand Down
26 changes: 26 additions & 0 deletions api/core/workflow/workflow_entry.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
is_start_node_type,
resolve_workflow_node_class,
)
from core.workflow.nodes.agent_v2.binding_resolver import WorkflowAgentBindingResolver
from core.workflow.nodes.agent_v2.discriminator import is_dify_agent_node_data
from core.workflow.nodes.human_input.boundary import HumanInputFormEventFilter
from core.workflow.system_variables import (
default_system_variables,
Expand All @@ -47,6 +49,8 @@
from graphon.nodes.container_effects import ContainerAwaitRequest
from graphon.runtime import GraphRuntimeState, ReadOnlyGraphRuntimeStateWrapper, VariablePool
from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
from libs.workflow_template_values import template_value_selectors
from models.agent_config_entities import AgentSoulConfig
from models.workflow import Workflow

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -271,6 +275,28 @@ def single_step_run(
)
except NotImplementedError:
variable_mapping = {}
if is_dify_agent_node_data(node_config_data):
# Agent Soul lives in its owned snapshot, outside graph node data.
# Single-step execution must also load references used only by its
# retrieval filters, which are absent from the task prompt.
bundle = WorkflowAgentBindingResolver().resolve(
tenant_id=workflow.tenant_id,
app_id=workflow.app_id,
workflow_id=workflow.id,
node_id=node_id,
)
agent_soul = AgentSoulConfig.model_validate(bundle.snapshot.config_snapshot_dict)
variable_mapping = {
**variable_mapping,
**{
f"{node_id}.metadata_filter.{'.'.join(selector)}": selector
for knowledge_set in agent_soul.knowledge.sets
if knowledge_set.metadata_filtering.mode == "manual"
and knowledge_set.metadata_filtering.conditions is not None
for condition in knowledge_set.metadata_filtering.conditions.conditions
for selector in template_value_selectors(condition.value)
},
}
variable_mapping = inject_default_system_variable_mappings(
node_id=node_id,
node_type=node_type,
Expand Down
52 changes: 52 additions & 0 deletions api/libs/workflow_template_values.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
from collections.abc import Iterator, Sequence
from typing import Protocol

from graphon.nodes.base.variable_template_parser import extract_selectors_from_template
from graphon.variables.segments import Segment
from graphon.variables.template_resolution import VARIABLE_PATTERN

type TemplateValue = str | Sequence[str] | int | float | None


class TemplateVariablePool(Protocol):
def get(self, selector: Sequence[str], /) -> Segment | None: ...


def template_value_selectors(value: TemplateValue) -> Iterator[list[str]]:
"""Extract workflow references from a scalar string or string sequence."""
values = [value] if isinstance(value, str) else value
if isinstance(values, Sequence):
for template in values:
for reference in extract_selectors_from_template(template):
yield list(reference.value_selector)


def resolve_template_value(value: TemplateValue, variable_pool: TemplateVariablePool) -> TemplateValue:
"""Resolve scalar values and string sequences against a workflow variable pool.

A whole scalar reference retains its string or numeric type;
interpolated text and elements of string lists remain strings. Missing
references fail explicitly rather than becoming literal selector text.
"""

def get_segment(selector: str) -> Segment:
segment = variable_pool.get(selector.split("."))
if segment is None:
raise ValueError(f"Workflow variable {selector} is not available.")
return segment

def resolve_template(template: str) -> str | int | float | None:
if match := VARIABLE_PATTERN.fullmatch(template):
resolved = get_segment(match.group(1)).to_object()
if resolved is None or isinstance(resolved, str | float):
return resolved
if isinstance(resolved, int) and not isinstance(resolved, bool):
return resolved
return str(resolved)
return VARIABLE_PATTERN.sub(lambda match: get_segment(match.group(1)).text, template)

if isinstance(value, str):
return resolve_template(value)
if isinstance(value, Sequence):
return [str(resolve_template(item)) for item in value]
return value
11 changes: 9 additions & 2 deletions api/tests/unit_tests/commands/test_fix_app_site_missing.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,13 @@ def _bind_command_database(
return command_sessions, command_session


def _run_fix_app_site_missing() -> None:
"""Invoke the command callback through a narrowed, non-optional reference."""
callback = system_commands.fix_app_site_missing.callback
assert callback is not None
callback()


def test_fix_app_site_missing_passes_loaded_session_to_signal(
monkeypatch: pytest.MonkeyPatch,
sqlite_engine: Engine,
Expand All @@ -89,7 +96,7 @@ def create_site(sender: App, *, account: Account, session: Session) -> None:
monkeypatch.setattr(system_commands.app_was_created, "send", send)

try:
system_commands.fix_app_site_missing.callback()
_run_fix_app_site_missing()
finally:
command_sessions.remove()

Expand Down Expand Up @@ -123,7 +130,7 @@ def fail_signal(sender: App, **_kwargs: object) -> None:
monkeypatch.setattr(system_commands.app_was_created, "send", MagicMock(side_effect=fail_signal))

try:
system_commands.fix_app_site_missing.callback()
_run_fix_app_site_missing()
finally:
command_sessions.remove()

Expand Down
14 changes: 10 additions & 4 deletions api/tests/unit_tests/commands/test_reset_encrypt_key_pair.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,10 @@


def _invoke_reset() -> int:
callback = commands.reset_encrypt_key_pair.callback
assert callback is not None
try:
commands.reset_encrypt_key_pair.callback()
callback()
except SystemExit as e:
return int(e.code or 0)
return 0
Expand Down Expand Up @@ -87,7 +89,7 @@ def _bind_command_to_sqlite(monkeypatch: pytest.MonkeyPatch, session: Session) -
monkeypatch.setattr(system_commands, "db", SimpleNamespace(engine=session.get_bind()))


def test_reset_aborts_when_not_self_hosted(monkeypatch, capsys):
def test_reset_aborts_when_not_self_hosted(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.CLOUD)

exit_code = _invoke_reset()
Expand Down Expand Up @@ -128,8 +130,12 @@ def test_reset_purges_provider_and_tool_tables_for_each_tenant(
assert TENANT_ID in captured.out

sqlite_session.expire_all()
assert sqlite_session.get(Tenant, TENANT_ID).encrypt_public_key == f"new-key-{TENANT_ID}"
assert sqlite_session.get(Tenant, OTHER_TENANT_ID).encrypt_public_key == f"new-key-{OTHER_TENANT_ID}"
tenant_after = sqlite_session.get(Tenant, TENANT_ID)
other_tenant_after = sqlite_session.get(Tenant, OTHER_TENANT_ID)
assert tenant_after is not None
assert other_tenant_after is not None
assert tenant_after.encrypt_public_key == f"new-key-{TENANT_ID}"
assert other_tenant_after.encrypt_public_key == f"new-key-{OTHER_TENANT_ID}"
assert sqlite_session.scalars(select(Provider).where(Provider.provider_type == ProviderType.CUSTOM)).all() == []
assert sqlite_session.scalars(select(ProviderModel)).all() == []
assert sqlite_session.scalars(select(BuiltinToolProvider)).all() == []
Expand Down
Loading
Loading