Compare commits

..
595 changed files with 15863 additions and 23711 deletions
@@ -45,7 +45,6 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
- Inspect the target feature area.
- Reuse an existing step when wording and behavior already match.
- Add a new step only for a genuinely new user action or assertion.
- Before adding several similar steps, scan the target capability for an existing domain noun that can be parameterized without hiding behavior.
- Keep edits close to the current capability folder unless the step is broadly reusable.
2. Write behavior-first scenarios.
- Describe user-observable behavior, not DOM mechanics.
@@ -54,16 +53,12 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
3. Write step definitions in the local style.
- Keep one step to one user-visible action or one assertion.
- Prefer Cucumber Expressions such as `{string}` and `{int}`.
- Use a bounded regex only when the accepted values are a small explicit domain set and Cucumber Expressions would make the Gherkin less natural.
- Do not create one-off steps for each case variant when the same domain action or outcome applies to named surfaces, modes, or resources.
- Scope locators to stable containers when the page has repeated elements.
- Avoid page-object layers or extra helper abstractions unless repeated complexity clearly justifies them.
4. Use Playwright in the local style.
- Prefer user-facing locators: `getByRole`, `getByLabel`, `getByPlaceholder`, `getByText`, then `getByTestId` for explicit contracts.
- Use web-first `expect(...)` assertions.
- Do not use `waitForTimeout`, manual polling, or raw visibility checks when a locator action or retrying assertion already expresses the behavior.
- Use `expect.poll` for API persistence, backend eventual consistency, captured browser events, or other non-DOM state; prefer locator assertions for DOM readiness and visible UI state.
- If a product element has real user-facing semantics but no accessible name, prefer fixing that accessible contract over adding a test id.
5. Validate narrowly.
- Run the narrowest tagged scenario or flow that exercises the change.
- Run `vpr lint --fix --quiet` from the repository root and `pnpm -C e2e type-check`.
@@ -39,7 +39,6 @@ Prefer reuse when:
- the user action is genuinely the same
- the expected outcome is genuinely the same
- the wording stays natural across features
- the parameter is a real product domain value such as a named surface, mode, resource, or status
Write a new step when:
@@ -47,8 +46,6 @@ Write a new step when:
- reusing the old wording would make the scenario misleading
- a supposedly generic step would become an implementation-detail wrapper
Do not optimize for a low step count by making vague steps. Optimize for a small set of truthful, domain-owned steps.
### 4. Prefer Cucumber Expressions
Use Cucumber Expressions for parameters unless regex is clearly necessary.
@@ -62,8 +59,6 @@ Common examples:
Keep expressions readable. If a step needs complicated parsing logic, first ask whether the scenario wording should be simpler.
Use regex for a bounded natural-language alternative only when it keeps Gherkin readable, for example `/(Web app|Backend service API)/`. Avoid broad regexes that accept unowned language.
### 5. Keep step definitions thin and meaningful
Step definitions are glue between Gherkin and automation, not a second abstraction language.
@@ -41,7 +41,6 @@ Also remember:
- repeated content usually needs scoping to a stable container
- exact text matching is often too brittle when role/name or label already exists
- `getByTestId` is acceptable when semantics are weak but the contract is intentional
- when a real UI region, card, status, or icon lacks an accessible name, prefer adding that semantic contract in product code before falling back to `getByTestId`
### 3. Use web-first assertions
@@ -63,8 +62,6 @@ Avoid:
If a condition genuinely needs custom retry logic, use Playwright's polling/assertion tools deliberately and keep that choice local and explicit.
Use `expect.poll` for non-DOM truth such as API state, backend eventual consistency, generated resources, or captured browser events. For DOM state, use locator assertions so Playwright can apply actionability and web-first retry semantics.
### 4. Let actions wait for actionability
Locator actions already wait for the element to be actionable. Do not preface every click/fill with extra timing logic unless the action needs a specific visible/ready assertion for clarity.
@@ -40,13 +40,9 @@ Flag:
- JavaScript conditional class logic for visual states that the Dify UI/Base UI primitive already exposes through `data-*` attributes or CSS variables.
- Controlled props added when uncontrolled DOM state or CSS variables would be enough.
- Thin wrappers that rename Base UI parts without adding semantics.
- Generic Base UI selection primitives wrapped without preserving their value generics, such as `Select.Root<Value, Multiple>`, `RadioGroup<Value>`, or `Radio.Root<Value>`.
- Shared select/radio option components that type selected values as `string` while callers pass enums, unions, booleans, numbers, objects, or nullable placeholder values.
Prefer Base UI/Dify UI data attributes and CSS variables for visual state: `data-open`, `data-checked`, `data-disabled`, `data-highlighted`, `data-popup-open`, `group-data-*`, `peer-data-*`, `has-[:focus-visible]`, and primitive CSS variables such as anchor width or transform origin. Use JS conditional classes for product/business state that the primitive does not expose.
For non-string `Select` and `RadioGroup` values, prefer explicit domain generics at the root and at child value carriers. JSX children do not inherit the parent generic, so `RadioGroup<PromptMode>` should compose with `Radio<PromptMode>`, `RadioRoot<PromptMode>`, or option values from a typed collection. For `Select`, prefer the Base UI `items` collection pattern for typed value-to-label rendering, and flag string coercion helpers used only to recover display labels.
## Forms
Flag:
@@ -69,7 +69,6 @@ Use this as the component decision guide for Dify web. Existing code is referenc
- Avoid barrel files that only re-export secondary owners. `index.tsx` is acceptable for a route/tab entry component; import header controls, switches, sections, and row owners from their concrete owner files.
- Type simple one-off props inline. Use a named `Props` type only when reused, exported, complex, or clearer.
- Use API-generated or API-returned types at component boundaries. Keep small UI conversion helpers and one-off UI extensions beside the component that needs them.
- Preserve domain value types for selection components. Do not widen enum, union, boolean, numeric, object, or nullable select/radio values to `string`; keep wrappers and option value carriers typed from their feature option collection.
- Avoid `common.tsx` buckets for shared UI. Use a feature-local `components/` folder with concrete filenames that describe the shared role.
- Do not create type aliases that only rename another type. Use aliases only for real UI concepts, refinements, or reusable local contracts.
- Name values by their domain role and backend API contract, especially persistent IDs and route params. Normalize framework or route params at the boundary.
+2 -1
View File
@@ -6,6 +6,8 @@ on:
merge_group:
branches: ["main"]
types: [checks_requested]
push:
branches: ["main"]
permissions:
actions: write
@@ -322,7 +324,6 @@ jobs:
- check-changes
if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed == 'true'
uses: ./.github/workflows/web-e2e.yml
secrets: inherit
web-e2e-skip:
name: Skip Web Full-Stack E2E
-54
View File
@@ -1,54 +0,0 @@
name: Post-Merge Checks
on:
push:
branches: ["main"]
permissions:
contents: read
concurrency:
group: post-merge-${{ github.sha }}
cancel-in-progress: false
jobs:
check-changes:
name: Check Changed Files
runs-on: depot-ubuntu-24.04
outputs:
external-e2e-changed: ${{ steps.changes.outputs.external_e2e }}
steps:
- uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
- uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
id: changes
with:
filters: |
external_e2e:
- 'e2e/features/agent-v2/**'
- 'e2e/scripts/**'
- 'e2e/support/**'
- '.github/workflows/post-merge.yml'
- '.github/workflows/web-e2e.yml'
- '.github/actions/setup-web/**'
- 'dify-agent/**'
- 'api/clients/agent_backend/**'
- 'api/core/app/apps/agent_app/**'
- 'api/core/workflow/nodes/agent_v2/**'
- 'api/controllers/console/agent/**'
- 'api/services/agent/**'
- 'api/core/plugin/**'
- 'api/services/plugin/**'
- 'api/core/tools/**'
- 'api/services/tools/**'
- 'web/features/agent-v2/**'
- 'web/app/(commonLayout)/roster/**'
- 'web/app/components/workflow/nodes/agent-v2/**'
external-e2e:
name: External Runtime E2E
needs: check-changes
if: needs.check-changes.outputs.external-e2e-changed == 'true'
uses: ./.github/workflows/web-e2e.yml
with:
run-external-runtime: true
secrets: inherit
+2 -49
View File
@@ -2,11 +2,6 @@ name: Web Full-Stack E2E
on:
workflow_call:
inputs:
run-external-runtime:
required: false
type: boolean
default: false
permissions:
contents: read
@@ -37,9 +32,7 @@ jobs:
with:
enable-cache: true
python-version: "3.12"
cache-dependency-glob: |
api/uv.lock
dify-agent/uv.lock
cache-dependency-glob: api/uv.lock
- name: Install API dependencies
run: uv sync --project api --dev
@@ -58,52 +51,12 @@ jobs:
E2E_INIT_PASSWORD: E2eInit12345
run: vp run e2e:full
- name: Run external runtime E2E tests
if: ${{ inputs.run-external-runtime }}
working-directory: ./e2e
env:
E2E_ADMIN_EMAIL: e2e-admin@example.com
E2E_ADMIN_NAME: E2E Admin
E2E_ADMIN_PASSWORD: E2eAdmin12345
E2E_AGENT_DECISION_MODEL_NAME: ${{ vars.E2E_AGENT_DECISION_MODEL_NAME || 'gpt-5.5' }}
E2E_AGENT_DECISION_MODEL_PROVIDER: ${{ vars.E2E_AGENT_DECISION_MODEL_PROVIDER || 'openai' }}
E2E_AGENT_DECISION_MODEL_TYPE: ${{ vars.E2E_AGENT_DECISION_MODEL_TYPE || 'llm' }}
E2E_EXTERNAL_RUNTIME_SEED_SPECS: ${{ vars.E2E_EXTERNAL_RUNTIME_SEED_SPECS }}
E2E_EXTERNAL_RUNTIME_TAGS: ${{ vars.E2E_EXTERNAL_RUNTIME_TAGS }}
E2E_FORCE_WEB_BUILD: "1"
E2E_INIT_PASSWORD: E2eInit12345
E2E_MARKETPLACE_API_URL: ${{ vars.E2E_MARKETPLACE_API_URL }}
E2E_MARKETPLACE_PLUGIN_IDS: ${{ vars.E2E_MARKETPLACE_PLUGIN_IDS }}
E2E_MARKETPLACE_PLUGIN_UNIQUE_IDENTIFIERS: ${{ vars.E2E_MARKETPLACE_PLUGIN_UNIQUE_IDENTIFIERS }}
E2E_MODEL_PROVIDER_CREDENTIALS_JSON: ${{ secrets.E2E_MODEL_PROVIDER_CREDENTIALS_JSON }}
E2E_START_AGENT_BACKEND: ${{ vars.E2E_START_AGENT_BACKEND || '1' }}
E2E_STABLE_MODEL_NAME: ${{ vars.E2E_STABLE_MODEL_NAME || 'gpt-5-nano' }}
E2E_STABLE_MODEL_PROVIDER: ${{ vars.E2E_STABLE_MODEL_PROVIDER || 'openai' }}
E2E_STABLE_MODEL_TYPE: ${{ vars.E2E_STABLE_MODEL_TYPE || 'llm' }}
run: |
if [[ -z "${E2E_MODEL_PROVIDER_CREDENTIALS_JSON}" ]]; then
echo "E2E_MODEL_PROVIDER_CREDENTIALS_JSON is required for external runtime E2E." >&2
exit 1
fi
if [[ -d cucumber-report ]]; then
rm -rf cucumber-report-non-external
mv cucumber-report cucumber-report-non-external
fi
trap 'vp run e2e:middleware:down' EXIT
vp run e2e:middleware:up
vp run e2e:external:prepare
vp run e2e:external
- name: Upload Cucumber report
if: ${{ !cancelled() }}
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: cucumber-report
path: |
e2e/cucumber-report
e2e/cucumber-report-non-external
path: e2e/cucumber-report
retention-days: 7
- name: Upload E2E logs
@@ -67,7 +67,6 @@ class AgentBackendRunSucceededInternalEvent(AgentBackendInternalEventBase):
type: Literal[AgentBackendInternalEventType.RUN_SUCCEEDED] = AgentBackendInternalEventType.RUN_SUCCEEDED
output: JsonValue
session_snapshot: CompositorSessionSnapshot
usage: dict[str, JsonValue] | None = None
class AgentBackendDeferredToolCallInternalEvent(AgentBackendInternalEventBase):
@@ -77,7 +76,6 @@ class AgentBackendDeferredToolCallInternalEvent(AgentBackendInternalEventBase):
deferred_tool_call: DeferredToolCallPayload
message: str | None = None
session_snapshot: CompositorSessionSnapshot
usage: dict[str, JsonValue] | None = None
class AgentBackendRunFailedInternalEvent(AgentBackendInternalEventBase):
@@ -142,7 +140,6 @@ class AgentBackendRunEventAdapter:
deferred_tool_call=event.data.deferred_tool_call,
message=_deferred_tool_call_message(event.data.deferred_tool_call),
session_snapshot=event.data.session_snapshot,
usage=_agent_run_usage(event.data.usage),
)
]
return [
@@ -151,7 +148,6 @@ class AgentBackendRunEventAdapter:
source_event_id=event.id,
output=event.data.output,
session_snapshot=event.data.session_snapshot,
usage=_agent_run_usage(event.data.usage),
)
]
case RunFailedEvent():
@@ -188,13 +184,3 @@ def _deferred_tool_call_message(payload: DeferredToolCallPayload) -> str:
return title
return f"Agent backend requested external input via deferred tool '{payload.tool_name}'."
def _agent_run_usage(usage: object | None) -> dict[str, JsonValue] | None:
"""Return JSON-safe usage metadata from optional Agent backend usage."""
if usage is None:
return None
dumped = _EVENT_DATA_ADAPTER.dump_python(usage, mode="json")
if not isinstance(dumped, dict):
return None
return cast(dict[str, JsonValue], dumped)
+11 -18
View File
@@ -171,7 +171,7 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):
knowledge: DifyKnowledgeBaseLayerConfig | None = None
config_layer_config: DifyConfigLayerConfig | None = None
# Drive Skills & Files declaration (dify.drive) — an index the agent pulls
# through the back proxy, never inline content.
# through the back proxy, never inline content; see AGENT_DRIVE_MANIFEST_ENABLED.
drive_config: DifyDriveLayerConfig | None = None
# Human-in-the-loop ask_human deferred tool (dify.ask_human). Present only when
# the Agent Soul configures human involvement; a deferred call ends the run and
@@ -220,7 +220,7 @@ class AgentBackendAgentAppRunInput(BaseModel):
knowledge: DifyKnowledgeBaseLayerConfig | None = None
config_layer_config: DifyConfigLayerConfig | None = None
# Drive Skills & Files declaration (dify.drive) — an index the agent pulls
# through the back proxy, never inline content.
# through the back proxy, never inline content; see AGENT_DRIVE_MANIFEST_ENABLED.
drive_config: DifyDriveLayerConfig | None = None
# Human-in-the-loop ask_human deferred tool (dify.ask_human). Present only when
# the Agent Soul configures human involvement (ENG-635).
@@ -254,11 +254,10 @@ class AgentBackendRunRequestBuilder:
"""Build an Agent App conversation-turn run request.
Layer graph: optional Agent Soul system prompt → user prompt →
execution context → optional shell / config / drive / history
(multi-turn) → LLM → optional plugin-direct tools / core-routed tools /
knowledge search / ask_human / structured output. Mirrors the
workflow-node layer ordering minus the workflow-job / previous-node
prompt.
execution context → optional history (multi-turn) → LLM → optional
plugin-direct tools / core-routed tools / knowledge search →
optional structured output. Mirrors the workflow-node layer ordering
minus the workflow-job / previous-node prompt.
"""
layers: list[RunLayerSpec] = []
if run_input.agent_soul_prompt:
@@ -355,14 +354,11 @@ class AgentBackendRunRequestBuilder:
)
if run_input.tools is not None and run_input.tools.tools:
plugin_tool_deps = {"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID}
if include_shell:
plugin_tool_deps["shell"] = DIFY_SHELL_LAYER_ID
layers.append(
RunLayerSpec(
name=DIFY_PLUGIN_TOOLS_LAYER_ID,
type=DIFY_PLUGIN_TOOLS_LAYER_TYPE_ID,
deps=plugin_tool_deps,
deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
metadata=run_input.metadata,
config=run_input.tools,
)
@@ -478,9 +474,9 @@ class AgentBackendRunRequestBuilder:
"""Build a workflow Agent Node run request without defining another wire schema.
Layer graph mirrors the workflow surface: prompts → execution context →
optional shell / config / drive / history → LLM → optional
plugin-direct tools / core-routed tools / knowledge search /
ask_human / structured output.
optional drive/history → LLM → optional plugin-direct tools /
core-routed tools / knowledge search → optional auxiliary layers such
as ask_human, shell, and structured output.
"""
layers: list[RunLayerSpec] = []
if run_input.agent_soul_prompt:
@@ -585,14 +581,11 @@ class AgentBackendRunRequestBuilder:
)
if run_input.tools is not None and run_input.tools.tools:
plugin_tool_deps = {"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID}
if include_shell:
plugin_tool_deps["shell"] = DIFY_SHELL_LAYER_ID
layers.append(
RunLayerSpec(
name=DIFY_PLUGIN_TOOLS_LAYER_ID,
type=DIFY_PLUGIN_TOOLS_LAYER_TYPE_ID,
deps=plugin_tool_deps,
deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
metadata=run_input.metadata,
config=run_input.tools,
)
+13 -18
View File
@@ -10,7 +10,6 @@ import click
import sqlalchemy as sa
import yaml
from core.db.session_factory import session_factory
from extensions.ext_database import db
from models import Tenant
from models.model import App
@@ -107,8 +106,7 @@ def export_migration_data(input_file: str | None, output_file: str | None, overw
assert output_file is not None
raw_config = _load_json_object(input_file, "Export config")
selection = ExportConfigParser().parse(raw_config)
with session_factory.create_session() as session:
result = MigrationExportService().export(session, selection)
result = MigrationExportService().export(selection)
MigrationPackageService().save_package(result.package, output_file, overwrite=overwrite)
click.echo(click.style(f"Output written to {output_file}", fg="green"))
_render_report(result.report_items, context=_with_output_path(result.report_context, output_file))
@@ -155,21 +153,19 @@ def import_migration_data(
_require_options(("--input", input_file))
assert input_file is not None
package = MigrationPackageService().load_package(input_file)
with session_factory.create_session() as session:
result = MigrationImportService().import_package(
session,
ImportRequest(
package=package,
cli_target_tenant=target_tenant,
operator_email=operator_email,
options_override=_build_options_override(
package.metadata.import_options,
id_strategy=id_strategy,
conflict_strategy=conflict_strategy,
create_app_api_token_on_import=create_app_api_token_on_import,
),
result = MigrationImportService().import_package(
ImportRequest(
package=package,
cli_target_tenant=target_tenant,
operator_email=operator_email,
options_override=_build_options_override(
package.metadata.import_options,
id_strategy=id_strategy,
conflict_strategy=conflict_strategy,
create_app_api_token_on_import=create_app_api_token_on_import,
),
)
)
_render_report(result.report_items, context=result.report_context)
except MigrationDataError as exc:
raise click.ClickException(str(exc)) from exc
@@ -252,8 +248,7 @@ def migration_data_wizard() -> None:
conflict_strategy=conflict_strategy,
output_file=output_file,
)
with session_factory.create_session() as session:
result = MigrationExportService().export(session, selection)
result = MigrationExportService().export(selection)
MigrationPackageService().save_package(result.package, output_file, overwrite=overwrite)
click.echo(click.style(f"Output written to {output_file}", fg="green"))
_print_wizard_step("Report")
+2 -3
View File
@@ -13,7 +13,6 @@ from core.rag.index_processor.constant.built_in_field import BuiltInField
from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
from core.rag.models.document import ChildDocument, Document
from extensions.ext_database import db
from libs.pagination import paginate_query
from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, DatasetMetadataBinding, DocumentSegment
from models.dataset import Document as DatasetDocument
from models.enums import DatasetMetadataType, IndexingStatus, SegmentStatus
@@ -184,7 +183,7 @@ def migrate_knowledge_vector_database():
.order_by(Dataset.created_at.desc())
)
datasets = paginate_query(stmt, page=page, per_page=50, max_per_page=50)
datasets = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
if not datasets.items:
break
except SQLAlchemyError:
@@ -410,7 +409,7 @@ def old_metadata_migration():
.where(DatasetDocument.doc_metadata.is_not(None))
.order_by(DatasetDocument.created_at.desc())
)
documents = paginate_query(stmt, page=page, per_page=50, max_per_page=50)
documents = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
except SQLAlchemyError:
raise
if not documents:
+7 -5
View File
@@ -1,4 +1,4 @@
from pydantic import Field, NonNegativeFloat
from pydantic import Field
from pydantic_settings import BaseSettings
@@ -32,10 +32,12 @@ class AgentBackendConfig(BaseSettings):
default=False,
)
AGENT_APP_TEXT_DELTA_DEBOUNCE_SECONDS: NonNegativeFloat = Field(
AGENT_DRIVE_MANIFEST_ENABLED: bool = Field(
description=(
"Buffer Agent App assistant text deltas for up to this many seconds before "
"publishing SSE chunks. Set to 0 to publish each delta immediately."
"Inject the dify.drive layer (Skills & Files drive manifest declaration) "
"into Agent runs. The declaration is an index only — the agent backend "
"pulls the actual SKILL.md / files through the back proxy. Set this to "
"false only when temporarily rolling back the drive integration."
),
default=0.5,
default=True,
)
-57
View File
@@ -1,57 +0,0 @@
"""Controller session decorators.
`with_session` is an HTTP controller helper: it opens one SQLAlchemy session
for a Resource handler and injects it as the first argument after `self`.
Handlers use a transaction by default so migrated write paths keep
commit/rollback handling; pure read handlers may opt out with `write=False`.
"""
from collections.abc import Callable
from functools import wraps
from typing import Concatenate, overload
from sqlalchemy.orm import Session
from core.db.session_factory import session_factory
@overload
def with_session[T, **P, R](
view: Callable[Concatenate[T, Session, P], R],
*,
write: bool = True,
) -> Callable[Concatenate[T, P], R]: ...
@overload
def with_session[T, **P, R](
view: None = None,
*,
write: bool = True,
) -> Callable[[Callable[Concatenate[T, Session, P], R]], Callable[Concatenate[T, P], R]]: ...
def with_session[T, **P, R](
view: Callable[Concatenate[T, Session, P], R] | None = None,
*,
write: bool = True,
) -> (
Callable[Concatenate[T, P], R] | Callable[[Callable[Concatenate[T, Session, P], R]], Callable[Concatenate[T, P], R]]
):
"""Inject a request-scoped session, using a transaction only for write handlers."""
def decorator(view: Callable[Concatenate[T, Session, P], R]) -> Callable[Concatenate[T, P], R]:
@wraps(view)
def wrapper(self: T, *args: P.args, **kwargs: P.kwargs) -> R:
if write:
with session_factory.get_session_maker().begin() as session:
return view(self, session, *args, **kwargs)
with session_factory.create_session() as session:
return view(self, session, *args, **kwargs)
return wrapper
if view is None:
return decorator
return decorator(view)
+17 -13
View File
@@ -91,31 +91,33 @@ class AgentIdPath(BaseModel):
class AgentAppCreatePayload(BaseModel):
name: str = Field(..., min_length=1, description="Agent name")
description: str | None = Field(default=None, description="Agent description (max 400 chars)", max_length=400)
role: str | None = Field(default=None, description="Agent role", max_length=255)
role: str = Field(..., min_length=1, description="Agent role", max_length=255)
icon_type: IconType | None = Field(default=None, description="Icon type")
icon: str | None = Field(default=None, description="Icon")
icon_background: str | None = Field(default=None, description="Icon background color")
@field_validator("role")
@classmethod
def validate_role(cls, value: str | None) -> str | None:
if value is None:
return None
return value.strip()
def validate_role(cls, value: str) -> str:
role = value.strip()
if not role:
raise ValueError("Agent role is required.")
return role
# Keep agent-app roster DTOs agent-specific instead of reusing the shared
# /apps response/request models. The roster surface needs Agent-only fields such
# as `role`, while the generic console/apps contracts must stay unchanged.
class AgentAppUpdatePayload(GenericUpdateAppPayload):
role: str | None = Field(default=None, description="Agent role", max_length=255)
role: str = Field(..., min_length=1, description="Agent role", max_length=255)
@field_validator("role")
@classmethod
def validate_role(cls, value: str | None) -> str | None:
if value is None:
return None
return value.strip()
def validate_role(cls, value: str) -> str:
role = value.strip()
if not role:
raise ValueError("Agent role is required.")
return role
class AgentAppCopyPayload(BaseModel):
@@ -131,7 +133,10 @@ class AgentAppCopyPayload(BaseModel):
def validate_role(cls, value: str | None) -> str | None:
if value is None:
return None
return value.strip()
role = value.strip()
if not role:
raise ValueError("Agent role is required when provided.")
return role
class AgentApiStatusPayload(BaseModel):
@@ -526,7 +531,6 @@ class AgentAppListApi(Resource):
page=args.page,
limit=args.limit,
mode="agent",
sort_by=args.sort_by,
name=args.name,
tag_ids=args.tag_ids,
creator_ids=args.creator_ids,
@@ -561,7 +565,7 @@ class AgentAppListApi(Resource):
name=args.name,
description=args.description,
mode="agent",
agent_role=args.role or "",
agent_role=args.role,
icon_type=args.icon_type,
icon=args.icon,
icon_background=args.icon_background,
@@ -44,10 +44,6 @@ class AgentSandboxListQuery(BaseModel):
path: str = Field(default=".", description="Directory path relative to the sandbox workspace")
class AgentSandboxInfoQuery(BaseModel):
conversation_id: str = Field(min_length=1, description="Agent App conversation ID")
class AgentSandboxFileQuery(BaseModel):
conversation_id: str = Field(min_length=1, description="Agent App conversation ID")
path: str = Field(min_length=1, description="File path relative to the sandbox workspace")
@@ -95,11 +91,6 @@ class SandboxListResponse(ResponseModel):
truncated: bool = False
class SandboxInfoResponse(ResponseModel):
session_id: str
workspace_cwd: str
class SandboxReadResponse(ResponseModel):
path: str
size: int | None = None
@@ -123,13 +114,7 @@ register_schema_models(
AgentSandboxUploadPayload,
WorkflowAgentSandboxUploadPayload,
)
register_response_schema_models(
console_ns,
SandboxInfoResponse,
SandboxListResponse,
SandboxReadResponse,
SandboxUploadResponse,
)
register_response_schema_models(console_ns, SandboxListResponse, SandboxReadResponse, SandboxUploadResponse)
def _handle(exc: Exception) -> tuple[dict[str, object], int]:
@@ -148,30 +133,6 @@ def _handle(exc: Exception) -> tuple[dict[str, object], int]:
raise exc
@console_ns.route("/agent/<uuid:agent_id>/sandbox")
class AgentAppSandboxInfoResource(Resource):
@console_ns.doc("get_agent_app_sandbox_info")
@console_ns.doc(description="Get basic information for an Agent App conversation sandbox")
@console_ns.doc(params={"agent_id": "Agent ID", **query_params_from_model(AgentSandboxInfoQuery)})
@console_ns.response(200, "Sandbox information returned", console_ns.models[SandboxInfoResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, tenant_id: str, agent_id: UUID):
app_model = resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
query = query_params_from_request(AgentSandboxInfoQuery)
try:
result = AgentAppSandboxService().get_info(
tenant_id=tenant_id,
app_id=app_model.id,
conversation_id=query.conversation_id,
)
except Exception as exc:
return _handle(exc)
return result.model_dump()
@console_ns.route("/agent/<uuid:agent_id>/sandbox/files")
class AgentAppSandboxListResource(Resource):
@console_ns.doc("list_agent_app_sandbox_files")
+13 -80
View File
@@ -1,7 +1,7 @@
import json
import logging
from collections.abc import Generator, Iterator, Mapping
from typing import Any, Literal, Protocol, runtime_checkable
from collections.abc import Generator
from typing import Any, Literal
from uuid import UUID
from flask import request
@@ -60,11 +60,6 @@ from services.errors.llm import InvokeRateLimitError
logger = logging.getLogger(__name__)
@runtime_checkable
class _ClosableStream(Protocol):
def close(self) -> None: ...
def _resolve_debugger_chat_streaming(
*, app_mode: AppMode, response_mode: str, response_mode_provided: bool = True
) -> bool:
@@ -123,12 +118,15 @@ edit workspace files, run validation or debugging commands, make exploratory che
Use only the current Build chat message history to identify changes that need to be persisted. Do not inspect, test, or
validate old config unless the message history already shows that the old config is invalid.
Only update the build-draft config note when the current Build chat contains durable context that later runs need.
Do not create, update, delete, inspect, or fill gaps in other Agent config resources, including config files, config
skills, config env, tools, models, knowledge, or prompt settings.
Persist only the build-draft config resources that need to change, using the Agent config CLI usage provided in the
runtime prompt:
When updating the config note with the Agent config CLI usage provided in the runtime prompt, record only durable
context needed by later runs, such as:
- config files for reusable artifacts that should be available later,
- config skills for reusable procedures or tools that should be available later,
- config env when environment keys or values need to be recorded,
- config note for concise durable context when useful.
When updating the config note, record only durable context needed by later runs, such as:
- what you installed or configured outside the workspace for this agent,
- where those external updates live, including CLI tools, packages, and persistent $HOME paths,
@@ -439,6 +437,7 @@ def _drain_streaming_generate_response(response: RateLimitGenerator | Generator[
changes the HTTP boundary: it drains the SSE stream server-side and returns
success after the generated build-chat message reaches ``message_end``.
"""
close = getattr(response, "close", None)
try:
for chunk in response:
for raw_event in chunk.split("\n\n"):
@@ -468,8 +467,8 @@ def _drain_streaming_generate_response(response: RateLimitGenerator | Generator[
if payload_event == "error":
raise CompletionRequestError(str(payload.get("message") or "Build chat finalization failed."))
finally:
if isinstance(response, _ClosableStream):
response.close()
if callable(close):
close()
raise CompletionRequestError("Build chat finalization did not complete.")
@@ -532,8 +531,6 @@ def _generate_chat_message_response(
args=args,
streaming=streaming,
)
if AppMode.value_of(app_model.mode) == AppMode.AGENT and streaming:
response = _raise_agent_stream_error_before_response(response)
return helper.compact_generate_response(response)
@@ -546,67 +543,3 @@ def _stop_chat_message(*, current_user_id: str, app_model: App, task_id: str):
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
def _raise_agent_stream_error_before_response(response):
"""Surface immediate Agent App stream errors as HTTP errors before SSE starts.
The shared streaming helper always returns HTTP 200 once the SSE response is
created. Agent v2 configuration errors, such as an invalid model API key,
can be the first real stream event after the initial ping; pre-reading that
first non-ping event lets the console API return the existing 400 error
contract instead of a successful HTTP response carrying only an SSE error.
"""
if isinstance(response, Mapping):
return response
buffered: list[str] = []
iterator = iter(response)
while True:
try:
chunk = next(iterator)
except StopIteration:
return iter(buffered)
if not isinstance(chunk, str):
return _prepend_stream_chunks(buffered, chunk, iterator)
if _is_sse_ping(chunk):
buffered.append(chunk)
continue
error_payload = _extract_sse_error_payload(chunk)
if error_payload is not None:
if isinstance(response, _ClosableStream):
response.close()
message = error_payload.get("message")
raise CompletionRequestError(str(message or "Agent App chat failed."))
return _prepend_stream_chunks(buffered, chunk, iterator)
def _prepend_stream_chunks(buffered: list[Any], first: Any, iterator: Iterator[Any]) -> Generator[Any, None, None]:
yield from buffered
yield first
yield from iterator
def _is_sse_ping(chunk: str) -> bool:
return chunk.strip() == "event: ping"
def _extract_sse_error_payload(chunk: str) -> dict[str, Any] | None:
for raw_event in chunk.split("\n\n"):
data_lines: list[str] = []
for line in raw_event.splitlines():
if line.startswith("data: "):
data_lines.append(line.removeprefix("data: "))
if not data_lines:
continue
try:
payload = json.loads("\n".join(data_lines))
except json.JSONDecodeError:
continue
if isinstance(payload, dict) and payload.get("event") == "error":
return payload
return None
+2 -3
View File
@@ -41,7 +41,6 @@ from fields.conversation_fields import (
from libs.datetime_utils import naive_utc_now, parse_time_range
from libs.helper import dump_response
from libs.login import login_required
from libs.pagination import paginate_query
from models import Conversation, EndUser, Message, MessageAnnotation
from models.account import Account
from models.model import App, AppMode
@@ -157,7 +156,7 @@ class CompletionConversationApi(Resource):
query = query.order_by(Conversation.created_at.desc())
conversations = paginate_query(query, page=args.page, per_page=args.limit)
conversations = db.paginate(query, page=args.page, per_page=args.limit, error_out=False)
return dump_response(ConversationPaginationResponse, conversations)
@@ -311,7 +310,7 @@ class ChatConversationApi(Resource):
case _:
query = query.order_by(Conversation.created_at.desc())
conversations = paginate_query(query, page=args.page, per_page=args.limit)
conversations = db.paginate(query, page=args.page, per_page=args.limit, error_out=False)
return dump_response(ConversationWithSummaryPaginationResponse, conversations)
+50 -8
View File
@@ -1,26 +1,26 @@
"""Controller decorators for console app resources.
App-loading decorators prefer a session injected by
`controllers.common.session.with_session` when present, while still supporting
existing handlers that have not been migrated yet and still rely on
Flask-SQLAlchemy's scoped `db.session`.
`with_session` opens one SQLAlchemy session for a request handler and injects it
as the first argument after `self`. Handlers use a transaction by default so
migrated write paths keep commit/rollback handling; pure read handlers may opt
out with `write=False`. App-loading decorators prefer that injected session when
present, while still supporting existing handlers that have not been migrated
yet and still rely on Flask-SQLAlchemy's scoped `db.session`.
"""
from collections.abc import Callable
from functools import wraps
from typing import cast, overload
from typing import Concatenate, cast, overload
from sqlalchemy import select
from sqlalchemy.orm import Session
from controllers.common.session import with_session
from controllers.console.app.error import AppNotFoundError
from core.db.session_factory import session_factory
from extensions.ext_database import db
from libs.login import current_account_with_tenant
from models import App, AppMode
__all__ = ["get_app_model", "get_app_model_with_trial", "with_session"]
def _load_app_model(session: Session, app_id: str) -> App | None:
"""Load the tenant-scoped app row with the request session owned by `with_session`."""
@@ -45,6 +45,48 @@ def _load_app_model_with_trial(app_id: str) -> App | None:
return app_model
@overload
def with_session[T, **P, R](
view: Callable[Concatenate[T, Session, P], R],
*,
write: bool = True,
) -> Callable[Concatenate[T, P], R]: ...
@overload
def with_session[T, **P, R](
view: None = None,
*,
write: bool = True,
) -> Callable[[Callable[Concatenate[T, Session, P], R]], Callable[Concatenate[T, P], R]]: ...
def with_session[T, **P, R](
view: Callable[Concatenate[T, Session, P], R] | None = None,
*,
write: bool = True,
) -> (
Callable[Concatenate[T, P], R] | Callable[[Callable[Concatenate[T, Session, P], R]], Callable[Concatenate[T, P], R]]
):
"""Inject a request-scoped session, using a transaction only for write handlers."""
def decorator(view: Callable[Concatenate[T, Session, P], R]) -> Callable[Concatenate[T, P], R]:
@wraps(view)
def wrapper(self: T, *args: P.args, **kwargs: P.kwargs) -> R:
if write:
with session_factory.get_session_maker().begin() as session:
return view(self, session, *args, **kwargs)
with session_factory.create_session() as session:
return view(self, session, *args, **kwargs)
return wrapper
if view is None:
return decorator
return decorator(view)
def _get_injected_session(args: tuple[object, ...]) -> Session | None:
"""Return the request session inserted by `with_session`, if this handler has been migrated."""
if len(args) < 2:
@@ -46,7 +46,6 @@ from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
from libs.datetime_utils import naive_utc_now
from libs.helper import dump_response, to_timestamp
from libs.login import login_required
from libs.pagination import paginate_query
from models import Account, DatasetProcessRule, Document, DocumentSegment, UploadFile
from models.dataset import DocumentPipelineExecutionLog
from models.enums import IndexingStatus, SegmentStatus
@@ -369,7 +368,7 @@ class DatasetDocumentListApi(Resource):
desc(Document.position),
)
paginated_documents = paginate_query(query, page=page, per_page=limit, max_per_page=100)
paginated_documents = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
documents = paginated_documents.items
DocumentService.enrich_documents_with_summary_index_status(
@@ -57,7 +57,6 @@ from fields.segment_fields import (
from graphon.model_runtime.entities.model_entities import ModelType
from libs.helper import dump_response, escape_like_pattern
from libs.login import login_required
from libs.pagination import paginate_query
from models import Account
from models.dataset import Dataset, Document, DocumentSegment
from models.model import UploadFile
@@ -271,7 +270,7 @@ class DatasetDocumentSegmentListApi(Resource):
elif args.enabled.lower() == "false":
query = query.where(DocumentSegment.enabled == False)
segments = paginate_query(query, page=page, per_page=limit, max_per_page=100)
segments = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
segment_list = list(segments.items)
segment_ids = [segment.id for segment in segment_list]
@@ -5,7 +5,7 @@ from flask import request
from flask_restx import Resource
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.orm import sessionmaker
from werkzeug.exceptions import NotFound
from controllers.common.fields import SimpleDataResponse
@@ -16,7 +16,6 @@ from controllers.common.schema import (
register_schema_models,
)
from controllers.console import console_ns
from controllers.console.app.wraps import with_session
from controllers.console.wraps import (
account_initialization_required,
enterprise_license_required,
@@ -103,13 +102,10 @@ class PipelineTemplateListApi(Resource):
@account_initialization_required
@enterprise_license_required
@with_current_tenant_id
@with_session
def get(self, session: Session, current_tenant_id: str) -> JsonResponseWithStatus:
def get(self, current_tenant_id: str) -> JsonResponseWithStatus:
query = PipelineTemplateListQuery.model_validate(request.args.to_dict(flat=True))
# get pipeline templates
pipeline_templates = RagPipelineService.get_pipeline_templates(
session, query.type, query.language, current_tenant_id
)
pipeline_templates = RagPipelineService.get_pipeline_templates(query.type, query.language, current_tenant_id)
return dump_response(PipelineTemplateListResponse, pipeline_templates), 200
@@ -121,11 +117,10 @@ class PipelineTemplateDetailApi(Resource):
@login_required
@account_initialization_required
@enterprise_license_required
@with_session
def get(self, session: Session, template_id: str) -> JsonResponseWithStatus:
def get(self, template_id: str) -> JsonResponseWithStatus:
query = PipelineTemplateDetailQuery.model_validate(request.args.to_dict(flat=True))
rag_pipeline_service = RagPipelineService()
pipeline_template = rag_pipeline_service.get_pipeline_template_detail(session, template_id, query.type)
pipeline_template = rag_pipeline_service.get_pipeline_template_detail(template_id, query.type)
if pipeline_template is None:
raise NotFound("Pipeline template not found from upstream service.")
return dump_response(PipelineTemplateDetailResponse, pipeline_template), 200
@@ -6,7 +6,7 @@ from uuid import UUID
from flask import abort, request
from flask_restx import Resource
from pydantic import BaseModel, Field, RootModel, ValidationError
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.orm import sessionmaker
from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
import services
@@ -26,7 +26,6 @@ from controllers.console.app.workflow import (
WorkflowPaginationResponse,
WorkflowResponse,
)
from controllers.console.app.wraps import with_session
from controllers.console.datasets.wraps import get_rag_pipeline
from controllers.console.wraps import (
RBACPermission,
@@ -344,8 +343,7 @@ class DraftRagPipelineRunApi(Resource):
@rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
@with_current_user
@get_rag_pipeline
@with_session
def post(self, session: Session, current_user: Account, pipeline: Pipeline):
def post(self, current_user: Account, pipeline: Pipeline):
"""
Run draft workflow
"""
@@ -354,7 +352,6 @@ class DraftRagPipelineRunApi(Resource):
try:
response = PipelineGenerateService.generate(
session=session,
pipeline=pipeline,
user=current_user,
args=args,
@@ -378,8 +375,7 @@ class PublishedRagPipelineRunApi(Resource):
@rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
@with_current_user
@get_rag_pipeline
@with_session
def post(self, session: Session, current_user: Account, pipeline: Pipeline):
def post(self, current_user: Account, pipeline: Pipeline):
"""
Run published workflow
"""
@@ -389,7 +385,6 @@ class PublishedRagPipelineRunApi(Resource):
try:
response = PipelineGenerateService.generate(
session=session,
pipeline=pipeline,
user=current_user,
args=args,
@@ -1019,14 +1014,13 @@ class RagPipelineTransformApi(Resource):
@login_required
@account_initialization_required
@with_current_user
@with_session
def post(self, session: Session, current_user: Account, dataset_id: UUID):
def post(self, current_user: Account, dataset_id: UUID):
if not (current_user.has_edit_permission or current_user.is_dataset_operator):
raise Forbidden()
dataset_id_str = str(dataset_id)
rag_pipeline_transform_service = RagPipelineTransformService()
result = rag_pipeline_transform_service.transform_dataset(dataset_id_str, session)
result = rag_pipeline_transform_service.transform_dataset(dataset_id_str, db.session)
return result
@@ -7,7 +7,7 @@ from sqlalchemy.orm import Session
from werkzeug.exceptions import InternalServerError, NotFound
import services
from controllers.common.fields import SimpleResultResponse
from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console.app.error import (
AppUnavailableError,
@@ -75,7 +75,7 @@ class ChatMessagePayload(BaseModel):
register_schema_models(console_ns, CompletionMessageExplorePayload, ChatMessagePayload)
register_response_schema_models(console_ns, SimpleResultResponse)
register_response_schema_models(console_ns, GeneratedAppResponse, SimpleResultResponse)
# define completion api for user
@@ -85,7 +85,7 @@ register_response_schema_models(console_ns, SimpleResultResponse)
)
class CompletionApi(InstalledAppResource):
@console_ns.expect(console_ns.models[CompletionMessageExplorePayload.__name__])
@console_ns.response(200, "Success")
@console_ns.response(200, "Success", console_ns.models[GeneratedAppResponse.__name__])
@with_current_user
@with_session
def post(self, session: Session, current_user: Account, installed_app: InstalledApp):
@@ -114,7 +114,6 @@ class CompletionApi(InstalledAppResource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -159,7 +158,7 @@ class CompletionStopApi(InstalledAppResource):
app_mode=AppMode.value_of(app_model.mode),
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@console_ns.route(
@@ -168,7 +167,7 @@ class CompletionStopApi(InstalledAppResource):
)
class ChatApi(InstalledAppResource):
@console_ns.expect(console_ns.models[ChatMessagePayload.__name__])
@console_ns.response(200, "Success")
@console_ns.response(200, "Success", console_ns.models[GeneratedAppResponse.__name__])
@with_current_user
@with_session
def post(self, session: Session, current_user: Account, installed_app: InstalledApp):
@@ -197,7 +196,6 @@ class ChatApi(InstalledAppResource):
streaming=True,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -245,4 +243,4 @@ class ChatStopApi(InstalledAppResource):
app_mode=app_mode,
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
+3 -2
View File
@@ -8,6 +8,7 @@ from sqlalchemy.orm import Session
from werkzeug.exceptions import InternalServerError, NotFound
from controllers.common.controller_schemas import MessageFeedbackPayload, MessageListQuery
from controllers.common.fields import GeneratedAppResponse
from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
from controllers.console.app.error import (
AppMoreLikeThisDisabledError,
@@ -60,6 +61,7 @@ class MoreLikeThisQuery(BaseModel):
register_schema_models(console_ns, MessageListQuery, MessageFeedbackPayload, MoreLikeThisQuery)
register_response_schema_models(
console_ns,
GeneratedAppResponse,
ExploreMessageInfiniteScrollPagination,
ResultResponse,
SuggestedQuestionsResponse,
@@ -142,7 +144,7 @@ class MessageFeedbackApi(InstalledAppResource):
)
class MessageMoreLikeThisApi(InstalledAppResource):
@console_ns.doc(params=query_params_from_model(MoreLikeThisQuery))
@console_ns.response(200, "Success")
@console_ns.response(200, "Success", console_ns.models[GeneratedAppResponse.__name__])
@with_current_user
@with_session
def get(self, session: Session, current_user: Account, installed_app: InstalledApp, message_id: UUID):
@@ -167,7 +169,6 @@ class MessageMoreLikeThisApi(InstalledAppResource):
invoke_from=InvokeFrom.EXPLORE,
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except MessageNotExistsError:
raise NotFound("Message Not Exists.")
+72 -20
View File
@@ -1,9 +1,9 @@
import logging
from datetime import datetime
from typing import Any, Literal
from typing import Any, Literal, cast
from flask import request
from flask_restx import Resource
from flask_restx import Resource, fields, marshal, marshal_with
from pydantic import AliasChoices, BaseModel, Field, field_validator
from sqlalchemy import select
from sqlalchemy.orm import Session
@@ -19,6 +19,7 @@ from controllers.common.fields import (
from controllers.common.fields import Parameters as ParametersResponse
from controllers.common.fields import Site as SiteResponse
from controllers.common.schema import (
get_or_create_model,
query_params_from_model,
register_response_schema_models,
register_schema_models,
@@ -57,12 +58,27 @@ from core.errors.error import (
)
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from fields.app_fields import (
app_detail_fields_with_site,
deleted_tool_fields,
model_config_fields,
site_fields,
tag_fields,
)
from fields.base import ResponseModel
from fields.dataset_fields import dataset_fields
from fields.member_fields import simple_account_fields
from fields.message_fields import SuggestedQuestionsResponse
from fields.workflow_fields import (
conversation_variable_fields,
pipeline_variable_fields,
workflow_fields,
workflow_partial_fields,
)
from graphon.graph_engine.manager import GraphEngineManager
from graphon.model_runtime.errors.invoke import InvokeError
from libs import helper
from libs.helper import dump_response, to_timestamp, uuid_value
from libs.helper import to_timestamp, uuid_value
from models import Account
from models.account import TenantStatus
from models.model import AppMode, Site
@@ -90,6 +106,48 @@ from services.recommended_app_service import RecommendedAppService
logger = logging.getLogger(__name__)
model_config_model = get_or_create_model("TrialAppModelConfig", model_config_fields)
workflow_partial_model = get_or_create_model("TrialWorkflowPartial", workflow_partial_fields)
deleted_tool_model = get_or_create_model("TrialDeletedTool", deleted_tool_fields)
tag_model = get_or_create_model("TrialTag", tag_fields)
site_model = get_or_create_model("TrialSite", site_fields)
app_detail_fields_with_site_copy = app_detail_fields_with_site.copy()
app_detail_fields_with_site_copy["model_config"] = fields.Nested(
model_config_model, attribute="app_model_config", allow_null=True
)
app_detail_fields_with_site_copy["workflow"] = fields.Nested(workflow_partial_model, allow_null=True)
app_detail_fields_with_site_copy["deleted_tools"] = fields.List(fields.Nested(deleted_tool_model))
app_detail_fields_with_site_copy["tags"] = fields.List(fields.Nested(tag_model))
app_detail_fields_with_site_copy["site"] = fields.Nested(site_model)
app_detail_with_site_model = get_or_create_model("TrialAppDetailWithSite", app_detail_fields_with_site_copy)
simple_account_model = get_or_create_model("TrialSimpleAccount", simple_account_fields)
conversation_variable_model = get_or_create_model("TrialConversationVariable", conversation_variable_fields)
pipeline_variable_model = get_or_create_model("TrialPipelineVariable", pipeline_variable_fields)
workflow_fields_copy = workflow_fields.copy()
workflow_fields_copy["created_by"] = fields.Nested(simple_account_model, attribute="created_by_account")
workflow_fields_copy["updated_by"] = fields.Nested(
simple_account_model, attribute="updated_by_account", allow_null=True
)
workflow_fields_copy["conversation_variables"] = fields.List(fields.Nested(conversation_variable_model))
workflow_fields_copy["rag_pipeline_variables"] = fields.List(fields.Nested(pipeline_variable_model))
workflow_model = get_or_create_model("TrialWorkflow", workflow_fields_copy)
dataset_model = get_or_create_model("TrialDataset", dataset_fields)
dataset_list_model = get_or_create_model(
"TrialDatasetList",
{
"data": fields.List(fields.Nested(dataset_model)),
"has_more": fields.Boolean,
"limit": fields.Integer,
"total": fields.Integer,
"page": fields.Integer,
},
)
class WorkflowRunRequest(BaseModel):
inputs: dict
files: list | None = Field(default=None)
@@ -329,11 +387,6 @@ class TrialDatasetResponse(ResponseModel):
created_at: int | None = None
permission_keys: list[str] = Field(default_factory=list)
@field_validator("created_at", mode="before")
@classmethod
def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
return to_timestamp(value)
class TrialDatasetListResponse(ResponseModel):
data: list[TrialDatasetResponse]
@@ -343,7 +396,7 @@ class TrialDatasetListResponse(ResponseModel):
page: int
class TrialSimpleAccount(ResponseModel):
class TrialWorkflowAccount(ResponseModel):
id: str
name: str | None = None
email: str | None = None
@@ -357,12 +410,12 @@ class TrialWorkflowResponse(ResponseModel):
version: str | None = None
marked_name: str | None = None
marked_comment: str | None = None
created_by: TrialSimpleAccount | None = Field(
created_by: TrialWorkflowAccount | None = Field(
default=None,
validation_alias=AliasChoices("created_by_account", "created_by"),
)
created_at: int | None = None
updated_by: TrialSimpleAccount | None = Field(
updated_by: TrialWorkflowAccount | None = Field(
default=None,
validation_alias=AliasChoices("updated_by_account", "updated_by"),
)
@@ -400,8 +453,6 @@ register_response_schema_models(
TrialWorkflowResponse,
)
simple_account_model = console_ns.models[TrialSimpleAccount.__name__]
class TrialAppWorkflowRunApi(TrialAppResource):
@trial_feature_enable
@@ -789,28 +840,27 @@ class TrialAppParameterApi(Resource):
class AppApi(Resource):
@console_ns.response(200, "Success", console_ns.models[TrialAppDetailResponse.__name__])
@get_app_model_with_trial(None)
@marshal_with(app_detail_with_site_model)
def get(self, app_model):
"""Get app detail"""
app_service = AppService()
app_model = app_service.get_app(app_model)
return dump_response(TrialAppDetailResponse, app_model)
return app_model
class AppWorkflowApi(Resource):
@console_ns.response(200, "Success", console_ns.models[TrialWorkflowResponse.__name__])
@get_app_model_with_trial(None)
@marshal_with(workflow_model)
def get(self, app_model):
"""Get workflow detail"""
if not app_model.workflow_id:
raise AppUnavailableError()
workflow = db.session.get(Workflow, app_model.workflow_id)
if workflow is None:
raise AppUnavailableError()
return dump_response(TrialWorkflowResponse, workflow)
return workflow
class DatasetListApi(Resource):
@@ -828,8 +878,10 @@ class DatasetListApi(Resource):
else:
raise NeedAddIdsError()
response = {"data": datasets, "has_more": len(datasets) == limit, "limit": limit, "total": total, "page": page}
return dump_response(TrialDatasetListResponse, response)
data = cast(list[dict[str, Any]], marshal(datasets, dataset_fields))
response = {"data": data, "has_more": len(datasets) == limit, "limit": limit, "total": total, "page": page}
return response
console_ns.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
+4 -5
View File
@@ -4,7 +4,7 @@ from sqlalchemy.orm import Session
from werkzeug.exceptions import InternalServerError
from controllers.common.controller_schemas import WorkflowRunPayload
from controllers.common.fields import SimpleResultResponse
from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
from controllers.common.schema import register_response_schema_models, register_schema_model
from controllers.console.app.error import (
CompletionRequestError,
@@ -38,13 +38,13 @@ from .. import console_ns
logger = logging.getLogger(__name__)
register_schema_model(console_ns, WorkflowRunPayload)
register_response_schema_models(console_ns, SimpleResultResponse)
register_response_schema_models(console_ns, GeneratedAppResponse, SimpleResultResponse)
@console_ns.route("/installed-apps/<uuid:installed_app_id>/workflows/run")
class InstalledAppWorkflowRunApi(InstalledAppResource):
@console_ns.expect(console_ns.models[WorkflowRunPayload.__name__])
@console_ns.response(200, "Success")
@console_ns.response(200, "Success", console_ns.models[GeneratedAppResponse.__name__])
@with_current_user
@with_session
def post(self, session: Session, current_user: Account, installed_app: InstalledApp):
@@ -70,7 +70,6 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
streaming=True,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except ProviderTokenNotInitError as ex:
raise ProviderNotInitializeError(ex.description)
@@ -110,4 +109,4 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
# New graph engine command channel mechanism
GraphEngineManager(redis_client).send_stop_command(task_id)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
+43 -35
View File
@@ -4,18 +4,18 @@ from uuid import UUID
from flask import request
from flask_restx import Resource
from pydantic import BaseModel, Field, RootModel, field_validator
from pydantic import BaseModel, Field, TypeAdapter, field_validator
from constants import HIDDEN_VALUE
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import dump_response, to_timestamp
from libs.helper import to_timestamp
from libs.login import login_required
from models.api_based_extension import APIBasedExtension
from services.api_based_extension_service import APIBasedExtensionService
from services.code_based_extension_service import CodeBasedExtensionService
from ..common.schema import query_params_from_model, register_response_schema_models, register_schema_models
from ..common.schema import DEFAULT_REF_TEMPLATE_OPENAPI_3_0, query_params_from_model, register_schema_models
from . import console_ns
from .wraps import account_initialization_required, setup_required, with_current_tenant_id
@@ -61,21 +61,36 @@ class APIBasedExtensionResponse(ResponseModel):
return to_timestamp(value)
class APIBasedExtensionListResponse(RootModel[list[APIBasedExtensionResponse]]):
pass
register_schema_models(
console_ns,
CodeBasedExtensionQuery,
APIBasedExtensionPayload,
)
register_response_schema_models(
console_ns,
CodeBasedExtensionResponse,
APIBasedExtensionResponse,
APIBasedExtensionListResponse,
)
console_ns.schema_model(
"APIBasedExtensionListResponse",
TypeAdapter(list[APIBasedExtensionResponse]).json_schema(ref_template=DEFAULT_REF_TEMPLATE_OPENAPI_3_0),
)
def _serialize_api_based_extension(extension: APIBasedExtension) -> dict[str, Any]:
return APIBasedExtensionResponse.model_validate(extension, from_attributes=True).model_dump(mode="json")
def _serialize_saved_api_based_extension(extension: APIBasedExtension, api_key: str) -> dict[str, Any]:
"""Serialize a saved extension with the plaintext key used for response masking only.
APIBasedExtensionService.save mutates the ORM object to hold the encrypted token before returning it. The response
contract, however, should match list/detail responses, where api_key is masked from the decrypted token.
"""
return APIBasedExtensionResponse(
id=extension.id,
name=extension.name,
api_endpoint=extension.api_endpoint,
api_key=api_key,
created_at=to_timestamp(extension.created_at),
).model_dump(mode="json")
@console_ns.route("/code-based-extension")
@@ -104,16 +119,16 @@ class CodeBasedExtensionAPI(Resource):
class APIBasedExtensionAPI(Resource):
@console_ns.doc("get_api_based_extensions")
@console_ns.doc(description="Get all API-based extensions for current tenant")
@console_ns.response(200, "Success", console_ns.models[APIBasedExtensionListResponse.__name__])
@console_ns.response(200, "Success", console_ns.models["APIBasedExtensionListResponse"])
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, current_tenant_id: str):
return dump_response(
APIBasedExtensionListResponse,
APIBasedExtensionService.get_all_by_tenant_id(db.session(), current_tenant_id),
)
return [
_serialize_api_based_extension(extension)
for extension in APIBasedExtensionService.get_all_by_tenant_id(db.session(), current_tenant_id)
]
@console_ns.doc("create_api_based_extension")
@console_ns.doc(description="Create a new API-based extension")
@@ -133,14 +148,12 @@ class APIBasedExtensionAPI(Resource):
api_key=payload.api_key,
)
extension = APIBasedExtensionService.save(db.session(), extension_data)
return APIBasedExtensionResponse(
id=extension.id,
name=extension.name,
api_endpoint=extension.api_endpoint,
api_key=payload.api_key,
created_at=to_timestamp(extension.created_at),
).model_dump(mode="json"), 201
return (
_serialize_saved_api_based_extension(
APIBasedExtensionService.save(db.session(), extension_data), payload.api_key
),
201,
)
@console_ns.route("/api-based-extension/<uuid:id>")
@@ -156,9 +169,8 @@ class APIBasedExtensionDetailAPI(Resource):
def get(self, current_tenant_id: str, id: UUID):
api_based_extension_id = str(id)
return dump_response(
APIBasedExtensionResponse,
APIBasedExtensionService.get_with_tenant_id(db.session(), current_tenant_id, api_based_extension_id),
return _serialize_api_based_extension(
APIBasedExtensionService.get_with_tenant_id(db.session(), current_tenant_id, api_based_extension_id)
)
@console_ns.doc("update_api_based_extension")
@@ -187,14 +199,10 @@ class APIBasedExtensionDetailAPI(Resource):
extension_data_from_db.api_key = payload.api_key
api_key_for_response = payload.api_key
APIBasedExtensionService.save(db.session(), extension_data_from_db)
return APIBasedExtensionResponse(
id=extension_data_from_db.id,
name=extension_data_from_db.name,
api_endpoint=extension_data_from_db.api_endpoint,
api_key=api_key_for_response,
created_at=to_timestamp(extension_data_from_db.created_at),
).model_dump(mode="json")
return _serialize_saved_api_based_extension(
APIBasedExtensionService.save(db.session(), extension_data_from_db),
api_key_for_response,
)
@console_ns.doc("delete_api_based_extension")
@console_ns.doc(description="Delete API-based extension")
+19 -17
View File
@@ -3,7 +3,7 @@ from uuid import UUID
from flask import request
from flask_restx import Resource
from pydantic import BaseModel, Field, RootModel, field_validator
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from werkzeug.exceptions import Forbidden
@@ -23,7 +23,6 @@ from controllers.console.wraps import (
)
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import dump_response
from libs.login import current_account_with_tenant, login_required
from models import Account
from models.enums import TagType
@@ -86,10 +85,6 @@ class TagResponse(ResponseModel):
return str(value)
class TagListResponse(RootModel[list[TagResponse]]):
pass
register_schema_models(
console_ns,
TagBasePayload,
@@ -97,8 +92,9 @@ register_schema_models(
TagBindingPayload,
TagBindingRemovePayload,
TagListQueryParam,
TagResponse,
)
register_response_schema_models(console_ns, SimpleResultResponse, TagResponse, TagListResponse)
register_response_schema_models(console_ns, SimpleResultResponse)
def _enforce_snippet_tag_rbac_if_needed(tag_type: TagType | str | None) -> None:
@@ -132,14 +128,18 @@ class TagListApi(Resource):
@login_required
@account_initialization_required
@console_ns.doc(params=query_params_from_model(TagListQueryParam))
@console_ns.response(200, "Success", console_ns.models[TagListResponse.__name__])
@console_ns.doc(responses={200: ("Success", [console_ns.models[TagResponse.__name__]])})
@with_current_tenant_id
def get(self, current_tenant_id: str):
raw_args = request.args.to_dict()
param = TagListQueryParam.model_validate(raw_args)
tags = TagService.get_tags(db.session(), param.type, current_tenant_id, param.keyword)
return dump_response(TagListResponse, tags), 200
serialized_tags = [
TagResponse.model_validate(tag, from_attributes=True).model_dump(mode="json") for tag in tags
]
return serialized_tags, 200
@console_ns.expect(console_ns.models[TagBasePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TagResponse.__name__])
@@ -156,7 +156,11 @@ class TagListApi(Resource):
_enforce_snippet_tag_rbac_if_needed(payload.type)
tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=payload.type), db.session)
return dump_response(TagResponse, {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}), 200
response = TagResponse.model_validate(
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
).model_dump(mode="json")
return response, 200
@console_ns.route("/tags/<uuid:tag_id>")
@@ -179,13 +183,11 @@ class TagUpdateDeleteApi(Resource):
binding_count = TagService.get_tag_binding_count(tag_id_str, db.session)
return (
dump_response(
TagResponse,
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count},
),
200,
)
response = TagResponse.model_validate(
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
).model_dump(mode="json")
return response, 200
@setup_required
@login_required
@@ -33,7 +33,6 @@ from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import OptionalTimestampField, TimestampField, dump_response, to_timestamp
from libs.login import login_required
from libs.pagination import paginate_query
from models.account import Account, Tenant, TenantAccountJoin, TenantCustomConfigDict, TenantStatus
from services.account_service import TenantService
from services.billing_service import BillingService, SubscriptionPlan
@@ -295,7 +294,7 @@ class WorkspaceListApi(Resource):
args = WorkspaceListQuery.model_validate(payload)
stmt = select(Tenant).order_by(Tenant.created_at.desc())
tenants = paginate_query(stmt, page=args.page, per_page=args.limit)
tenants = db.paginate(select=stmt, page=args.page, per_page=args.limit, error_out=False)
has_more = False
if tenants.has_next:
+1 -3
View File
@@ -28,7 +28,6 @@ class PluginUploadQuery(BaseModel):
sign: str = Field(..., description="HMAC signature")
tenant_id: str = Field(..., description="Tenant identifier")
user_id: str | None = Field(default=None, description="User identifier")
conversation_id: str | None = Field(default=None, description="Conversation identifier")
register_schema_models(files_ns, PluginUploadQuery)
@@ -93,7 +92,6 @@ class PluginUploadFileApi(Resource):
mimetype=mimetype,
tenant_id=tenant_id,
user_id=user.id,
conversation_id=args.conversation_id,
timestamp=timestamp,
nonce=nonce,
sign=sign,
@@ -107,7 +105,7 @@ class PluginUploadFileApi(Resource):
file_binary=file.stream.read(),
mimetype=mimetype,
filename=filename,
conversation_id=args.conversation_id,
conversation_id=None,
)
extension = guess_extension(tool_file.mimetype) or ".bin"
@@ -434,7 +434,6 @@ class PluginUploadFileRequestApi(Resource):
mimetype=payload.mimetype,
tenant_id=tenant_model.id,
user_id=user_model.id,
conversation_id=payload.conversation_id,
)
return BaseBackwardsInvocationResponse(data={"url": url}).model_dump()
@@ -59,7 +59,6 @@ from fields.document_fields import (
)
from libs.helper import dump_response
from libs.login import current_user
from libs.pagination import paginate_query
from models.dataset import Dataset, Document, DocumentSegment
from models.enums import SegmentStatus
from services.dataset_service import DatasetService, DocumentService
@@ -946,8 +945,8 @@ class DocumentListApi(DatasetApiResource):
query = query.order_by(desc(Document.created_at), desc(Document.position))
paginated_documents = paginate_query(
query, page=query_params.page, per_page=query_params.limit, max_per_page=100
paginated_documents = db.paginate(
select=query, page=query_params.page, per_page=query_params.limit, max_per_page=100, error_out=False
)
documents = paginated_documents.items
@@ -5,7 +5,6 @@ from uuid import UUID
from flask import request
from pydantic import BaseModel, Field, RootModel
from sqlalchemy import select
from sqlalchemy.orm import Session
from werkzeug.exceptions import Forbidden, NotFound
import services
@@ -17,7 +16,6 @@ from controllers.common.schema import (
register_schema_model,
register_schema_models,
)
from controllers.console.app.wraps import with_session
from controllers.service_api import service_api_ns
from controllers.service_api.dataset.error import PipelineRunError
from controllers.service_api.dataset.rag_pipeline.serializers import serialize_upload_file
@@ -266,8 +264,7 @@ class PipelineRunApi(DatasetApiResource):
"Pipeline run successfully",
service_api_ns.models[GeneratedAppResponse.__name__],
)
@with_session
def post(self, session: Session, tenant_id: str, dataset_id: UUID):
def post(self, tenant_id: str, dataset_id: UUID):
"""Resource for running a rag pipeline."""
dataset_id_str = str(dataset_id)
# Verify dataset ownership
@@ -285,7 +282,6 @@ class PipelineRunApi(DatasetApiResource):
pipeline: Pipeline = rag_pipeline_service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset_id_str)
try:
response: dict[Any, Any] | Generator[str, Any, None] = PipelineGenerateService.generate(
session=session,
pipeline=pipeline,
user=current_user,
args=payload.model_dump(),
@@ -507,7 +507,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
),
event_adapter=AgentBackendRunEventAdapter(),
session_store=AgentAppRuntimeSessionStore(),
text_delta_debounce_seconds=dify_config.AGENT_APP_TEXT_DELTA_DEBOUNCE_SECONDS,
)
def _run_input_guards(
+9 -102
View File
@@ -16,8 +16,6 @@ from __future__ import annotations
import json
import logging
import time
from collections.abc import Mapping
from decimal import Decimal
from typing import Any, Literal
@@ -76,23 +74,12 @@ def _prompt_messages_from_query(user_query: str | None) -> list[PromptMessage]:
return [UserPromptMessage(content=user_query)]
def _llm_usage_from_agent_backend(usage: Mapping[str, Any] | None) -> LLMUsage | None:
if usage is None:
return None
try:
return LLMUsage.from_metadata(usage)
except (TypeError, ValueError):
logger.warning("Failed to parse Agent backend usage metadata: %s", usage, exc_info=True)
return None
def publish_text_answer(
*,
queue_manager: AppQueueManager,
model_name: str,
answer: str,
user_query: str | None = None,
usage: LLMUsage | None = None,
) -> None:
"""Publish a complete assistant answer as one chunk + message-end.
@@ -112,7 +99,6 @@ def publish_text_answer(
model_name=model_name,
answer=answer,
user_query=user_query,
usage=usage,
)
@@ -141,7 +127,6 @@ def publish_message_end(
model_name: str,
answer: str,
user_query: str | None = None,
usage: LLMUsage | None = None,
) -> None:
"""Publish the terminal assistant result without emitting another delta."""
prompt_messages = _prompt_messages_from_query(user_query)
@@ -151,46 +136,13 @@ def publish_message_end(
model=model_name,
prompt_messages=prompt_messages,
message=AssistantPromptMessage(content=answer),
usage=usage or LLMUsage.empty_usage(),
usage=LLMUsage.empty_usage(),
),
),
PublishFrom.APPLICATION_MANAGER,
)
class _TextDeltaDebouncer:
"""Batch assistant text deltas on stream-event boundaries for final SSE output."""
def __init__(self, *, debounce_seconds: float) -> None:
self._debounce_seconds = debounce_seconds
self._parts: list[str] = []
self._first_pending_at: float | None = None
def push(self, delta: str) -> str | None:
if not delta:
return None
if self._debounce_seconds <= 0:
return delta
now = time.monotonic()
if not self._parts:
self._first_pending_at = now
self._parts.append(delta)
if self._first_pending_at is not None and now - self._first_pending_at >= self._debounce_seconds:
return self.flush()
return None
def flush(self) -> str | None:
if not self._parts:
return None
text = "".join(self._parts)
self._parts = []
self._first_pending_at = None
return text
class _AgentProcessRecorder:
"""Persist Agent v2 thinking/tool process events through the legacy thought model."""
@@ -478,13 +430,11 @@ class AgentAppRunner:
agent_backend_client: AgentBackendRunClient,
event_adapter: AgentBackendRunEventAdapter,
session_store: AgentAppRuntimeSessionStore,
text_delta_debounce_seconds: float,
) -> None:
self._request_builder = request_builder
self._agent_backend_client = agent_backend_client
self._event_adapter = event_adapter
self._session_store = session_store
self._text_delta_debounce_seconds = text_delta_debounce_seconds
def run(
self,
@@ -562,7 +512,6 @@ class AgentAppRunner:
answer=answer,
query=query,
streamed_answer=streamed_answer,
usage=_llm_usage_from_agent_backend(terminal.usage),
)
self._save_session(
scope=scope,
@@ -775,39 +724,19 @@ class AgentAppRunner:
model_name: str,
query: str | None,
):
"""Consume backend events while preserving raw recorder granularity.
Process events are recorded immediately for observability. Only the
final assistant text deltas sent through the EasyUI queue are debounced,
with flushes happening on later stream events or terminal boundaries.
"""
terminal = None
streamed_answer_parts: list[str] = []
text_delta_debouncer = _TextDeltaDebouncer(debounce_seconds=self._text_delta_debounce_seconds)
process_recorder = _AgentProcessRecorder(
dify_context=dify_context,
message_id=message_id,
queue_manager=queue_manager,
)
def flush_pending_text() -> None:
pending_text = text_delta_debouncer.flush()
if pending_text:
publish_text_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=pending_text,
user_query=query,
)
for public_event in self._agent_backend_client.stream_events(run_id):
if queue_manager.is_stopped():
flush_pending_text()
self._cancel_run(run_id)
raise GenerateTaskStoppedError()
for internal_event in self._event_adapter.adapt(public_event):
if queue_manager.is_stopped():
flush_pending_text()
self._cancel_run(run_id)
raise GenerateTaskStoppedError()
if internal_event.type in (
@@ -829,22 +758,18 @@ class AgentAppRunner:
text_delta = self._extract_stream_text_delta(internal_event)
if text_delta:
streamed_answer_parts.append(text_delta)
debounced_delta = text_delta_debouncer.push(text_delta)
if debounced_delta:
publish_text_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=debounced_delta,
user_query=query,
)
publish_text_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=text_delta,
user_query=query,
)
continue
continue
flush_pending_text()
terminal = internal_event
break
if terminal is not None:
break
flush_pending_text()
return terminal, "".join(streamed_answer_parts)
def _cancel_run(self, run_id: str) -> None:
@@ -868,20 +793,10 @@ class AgentAppRunner:
answer: str,
query: str | None,
streamed_answer: str,
usage: LLMUsage | None,
) -> None:
"""Finish a successful streamed turn without duplicating the final text."""
if not answer and streamed_answer:
answer = streamed_answer
if not streamed_answer:
publish_text_answer(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
user_query=query,
usage=usage,
)
self._publish_answer(queue_manager=queue_manager, model_name=model_name, answer=answer, query=query)
return
if answer.startswith(streamed_answer):
@@ -897,13 +812,7 @@ class AgentAppRunner:
"using terminal output for message persistence."
)
publish_message_end(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
user_query=query,
usage=usage,
)
publish_message_end(queue_manager=queue_manager, model_name=model_name, answer=answer, user_query=query)
def _save_session(
self,
@@ -943,8 +852,6 @@ class AgentAppRunner:
configured the value is a JSON object, which we serialize so the chat
message always has a string body.
"""
if output is None:
return ""
if isinstance(output, str):
return output
if isinstance(output, dict):
@@ -46,7 +46,7 @@ from core.workflow.nodes.agent_v2.runtime_request_builder import (
)
from models.agent_config_entities import AgentSoulConfig, AgentSoulToolsConfig
from models.provider_ids import ModelProviderID
from services.agent.prompt_mentions import expand_prompt_mentions
from services.agent.prompt_mentions import build_soul_mention_resolver, expand_prompt_mentions
class AgentAppRuntimeRequestBuildError(ValueError):
@@ -124,14 +124,17 @@ class AgentAppRuntimeRequestBuilder:
"cli_tool_count": len(agent_soul.tools.cli_tools),
}
config_layer_config, config_warnings = build_config_layer_config(
agent_soul,
agent_id=context.agent_id,
config_version_id=context.agent_config_snapshot_id,
config_version_kind=context.agent_config_version_kind,
)
append_runtime_warnings(metadata, config_warnings)
soul_prompt_resolver = build_config_aware_soul_mention_resolver(agent_soul)
config_layer_config = None
soul_prompt_resolver = build_soul_mention_resolver(agent_soul)
if dify_config.AGENT_DRIVE_MANIFEST_ENABLED:
config_layer_config, config_warnings = build_config_layer_config(
agent_soul,
agent_id=context.agent_id,
config_version_id=context.agent_config_snapshot_id,
config_version_kind=context.agent_config_version_kind,
)
append_runtime_warnings(metadata, config_warnings)
soul_prompt_resolver = build_config_aware_soul_mention_resolver(agent_soul)
knowledge_config = build_knowledge_layer_config(agent_soul)
request = self._request_builder.build_for_agent_app(
+1 -3
View File
@@ -1,6 +1,5 @@
import logging
from collections.abc import Sequence
from urllib.parse import urlencode
import httpx
from yarl import URL
@@ -17,8 +16,7 @@ MARKETPLACE_TIMEOUT = 30
def get_plugin_pkg_url(plugin_unique_identifier: str) -> str:
query = urlencode({"unique_identifier": plugin_unique_identifier})
return f"{marketplace_api_url / 'api/v1/plugins/download'}?{query}"
return str((marketplace_api_url / "api/v1/plugins/download").with_query(unique_identifier=plugin_unique_identifier))
def download_plugin_pkg(plugin_unique_identifier: str):
-1
View File
@@ -230,7 +230,6 @@ class RequestRequestUploadFile(BaseModel):
filename: str
mimetype: str
conversation_id: str | None = None
class RequestDownloadFileMapping(BaseModel):
+3 -35
View File
@@ -4,8 +4,6 @@ This module owns plugin daemon management calls that are shared by API services
and core runtimes. Plugin model provider discovery is cached here, alongside
plugin install, uninstall, and upgrade invalidation, so all cache mutations for
plugin-owned provider metadata stay tenant-scoped and in one place.
Provider cache payloads may be stored as prefixed zstd bytes; readers also
accept legacy plain JSON payloads for rolling upgrades and existing Redis keys.
The console plugin list also normalizes endpoint setup counters against live
endpoint records. Some plugin daemon builds return stale ``endpoints_*``
@@ -21,7 +19,6 @@ from contextlib import contextmanager
from mimetypes import guess_type
from typing import Literal, Protocol
import zstandard
from pydantic import BaseModel, TypeAdapter, ValidationError
from redis import RedisError
from redis.exceptions import LockError
@@ -95,8 +92,6 @@ class PluginService:
PLUGIN_MODEL_PROVIDERS_LOCK_TTL = 30
PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT = 2.0
PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL = 0.05
PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX = b"\x00dify-plugin-model-providers-zstd-v1:"
PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES = 64 * 1024
PLUGIN_INSTALL_TASK_TERMINAL_STATUSES = (PluginInstallTaskStatus.Success, PluginInstallTaskStatus.Failed)
# Mirror the detail-panel endpoint query size so list reconciliation and
# the visible endpoint drawer exercise the same daemon pagination path.
@@ -147,27 +142,6 @@ class PluginService:
declaration.provider_name = cls._get_provider_short_name_alias(provider)
return declaration
@classmethod
def _encode_plugin_model_providers_cache_payload(cls, payload: bytes) -> bytes:
if len(payload) < cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES:
return payload
return cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + zstandard.compress(payload, level=1)
@classmethod
def _decode_plugin_model_providers_cache_payload(cls, payload: bytes | bytearray | str) -> bytes | bytearray | str:
if isinstance(payload, str):
return payload
prefix = cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX
if not payload.startswith(prefix):
return payload
try:
return zstandard.decompress(payload[len(prefix) :])
except zstandard.ZstdError as exc:
raise ValueError("Invalid compressed plugin model providers cache payload.") from exc
@classmethod
def _load_plugin_model_providers_generation(cls, tenant_id: str) -> int | None:
cache_key = cls._get_plugin_model_providers_generation_cache_key(tenant_id)
@@ -225,8 +199,7 @@ class PluginService:
continue
try:
payload = cls._decode_plugin_model_providers_cache_payload(cached_providers)
providers = tuple(_provider_entities_adapter.validate_json(payload))
providers = tuple(_provider_entities_adapter.validate_json(cached_providers))
return providers, True
except (TypeError, ValueError, ValidationError):
logger.warning(
@@ -252,9 +225,7 @@ class PluginService:
) -> None:
cache_key = cls._get_plugin_model_providers_cache_key(tenant_id, generation)
try:
payload = cls._encode_plugin_model_providers_cache_payload(
_provider_entities_adapter.dump_json(list(providers))
)
payload = _provider_entities_adapter.dump_json(list(providers))
redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL, payload)
except (RedisError, RuntimeError):
logger.warning("Failed to cache plugin model providers for tenant %s.", tenant_id, exc_info=True)
@@ -902,10 +873,7 @@ class PluginService:
tenant_id,
plugin_unique_identifiers,
PluginInstallationSource.Package,
[
{"plugin_unique_identifier": plugin_unique_identifier}
for plugin_unique_identifier in plugin_unique_identifiers
],
[{}],
)
PluginService.invalidate_plugin_model_providers_cache(tenant_id)
return result
+1 -36
View File
@@ -1159,39 +1159,6 @@ class DatasetRetrieval:
all_documents.extend(documents)
def _run_retriever_thread(
self,
*,
flask_app: Flask,
dataset_id: str,
query: str | None,
top_k: int,
all_documents: list[Document],
document_ids_filter: list[str] | None,
metadata_condition: MetadataFilteringCondition | None,
attachment_ids: list[str] | None,
cancel_event: threading.Event | None,
thread_exceptions: list[Exception] | None,
) -> None:
try:
with session_factory.create_session() as session:
self._retriever(
flask_app=flask_app,
session=session,
dataset_id=dataset_id,
query=query or "",
top_k=top_k,
all_documents=all_documents,
document_ids_filter=document_ids_filter,
metadata_condition=metadata_condition,
attachment_ids=attachment_ids,
)
except Exception as e:
if cancel_event:
cancel_event.set()
if thread_exceptions is not None:
thread_exceptions.append(e)
def to_dataset_retriever_tool(
self,
session: Session,
@@ -1830,7 +1797,7 @@ class DatasetRetrieval:
else:
continue
retrieval_thread = threading.Thread(
target=self._run_retriever_thread,
target=self._retriever,
kwargs={
"flask_app": flask_app,
"dataset_id": dataset.id,
@@ -1840,8 +1807,6 @@ class DatasetRetrieval:
"document_ids_filter": document_ids_filter,
"metadata_condition": metadata_condition,
"attachment_ids": [attachment_id] if attachment_id else None,
"cancel_event": cancel_event,
"thread_exceptions": thread_exceptions,
},
)
threads.append(retrieval_thread)
+13 -24
View File
@@ -64,45 +64,34 @@ def verify_tool_file_signature(file_id: str, timestamp: str, nonce: str, sign: s
return current_time - int(timestamp) <= dify_config.FILES_ACCESS_TIMEOUT
def get_signed_file_url_for_plugin(
filename: str, mimetype: str, tenant_id: str, user_id: str, conversation_id: str | None = None
) -> str:
def get_signed_file_url_for_plugin(filename: str, mimetype: str, tenant_id: str, user_id: str) -> str:
"""Build the signed upload URL used by the plugin-facing file upload endpoint."""
base_url = dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL
upload_url = f"{base_url}/files/upload/for-plugin"
timestamp = str(int(time.time()))
nonce = os.urandom(16).hex()
data_to_sign = f"upload|{filename}|{mimetype}|{tenant_id}|{user_id}|{conversation_id or ''}|{timestamp}|{nonce}"
data_to_sign = f"upload|{filename}|{mimetype}|{tenant_id}|{user_id}|{timestamp}|{nonce}"
sign = hmac.new(_secret_key(), data_to_sign.encode(), hashlib.sha256).digest()
encoded_sign = base64.urlsafe_b64encode(sign).decode()
query_params = {
"timestamp": timestamp,
"nonce": nonce,
"sign": encoded_sign,
"user_id": user_id,
"tenant_id": tenant_id,
}
if conversation_id:
query_params["conversation_id"] = conversation_id
query = urllib.parse.urlencode(query_params)
query = urllib.parse.urlencode(
{
"timestamp": timestamp,
"nonce": nonce,
"sign": encoded_sign,
"user_id": user_id,
"tenant_id": tenant_id,
}
)
return f"{upload_url}?{query}"
def verify_plugin_file_signature(
*,
filename: str,
mimetype: str,
tenant_id: str,
user_id: str,
conversation_id: str | None = None,
timestamp: str,
nonce: str,
sign: str,
*, filename: str, mimetype: str, tenant_id: str, user_id: str, timestamp: str, nonce: str, sign: str
) -> bool:
"""Verify the signature used by the plugin-facing file upload endpoint."""
data_to_sign = f"upload|{filename}|{mimetype}|{tenant_id}|{user_id}|{conversation_id or ''}|{timestamp}|{nonce}"
data_to_sign = f"upload|{filename}|{mimetype}|{tenant_id}|{user_id}|{timestamp}|{nonce}"
recalculated_sign = hmac.new(_secret_key(), data_to_sign.encode(), hashlib.sha256).digest()
recalculated_encoded_sign = base64.urlsafe_b64encode(recalculated_sign).decode()
@@ -11,7 +11,6 @@ from dify_agent.layers.dify_plugin import (
DifyPluginToolCredentialType,
DifyPluginToolParameter,
DifyPluginToolParameterForm,
DifyPluginToolParameterType,
DifyPluginToolsLayerConfig,
)
from sqlalchemy import select
@@ -352,9 +351,7 @@ class WorkflowAgentDifyToolsBuilder:
@staticmethod
def _tool_layer_destination(tool_config: AgentSoulDifyToolConfig) -> Literal["plugin", "core"]:
provider_type = ToolProviderType.value_of(tool_config.provider_type)
if provider_type is ToolProviderType.PLUGIN or (
provider_type is ToolProviderType.BUILT_IN and _is_plugin_provider_id(tool_config.provider_id)
):
if provider_type is ToolProviderType.PLUGIN:
return "plugin"
if provider_type in {
ToolProviderType.BUILT_IN,
@@ -407,7 +404,7 @@ class WorkflowAgentDifyToolsBuilder:
credentials=self._normalize_credentials(runtime.credentials, tool_name=exposed_name),
runtime_parameters=runtime_parameters,
parameters=parameters,
parameters_json_schema=self._plugin_parameters_json_schema(tool_runtime, parameters),
parameters_json_schema=tool_runtime.get_llm_parameters_json_schema(),
)
def _to_core_backend_tool_config(
@@ -459,41 +456,6 @@ class WorkflowAgentDifyToolsBuilder:
description = tool_runtime.entity.description.llm
return description
@staticmethod
def _plugin_parameters_json_schema(
tool_runtime: Tool,
parameters: list[DifyPluginToolParameter],
) -> dict[str, Any]:
schema = tool_runtime.get_llm_parameters_json_schema()
properties = schema.setdefault("properties", {})
required = schema.setdefault("required", [])
if not isinstance(properties, dict) or not isinstance(required, list):
raise WorkflowAgentDifyToolsBuildError(
"agent_tool_declaration_invalid",
f"Dify Plugin Tool {tool_runtime.entity.identity.name!r} has invalid parameter schema.",
)
for parameter in parameters:
if parameter.form is not DifyPluginToolParameterForm.LLM:
continue
if parameter.type is DifyPluginToolParameterType.FILE:
properties[parameter.name] = _plugin_file_input_schema(parameter.llm_description or "")
elif parameter.type in {
DifyPluginToolParameterType.FILES,
DifyPluginToolParameterType.SYSTEM_FILES,
}:
properties[parameter.name] = {
"type": "array",
"items": _plugin_file_input_schema(parameter.llm_description or ""),
"description": parameter.llm_description or "",
}
else:
continue
if parameter.required and parameter.name not in required:
required.append(parameter.name)
return schema
@staticmethod
def _runtime_parameters(
tool_runtime: Tool,
@@ -536,44 +498,3 @@ class WorkflowAgentDifyToolsBuilder:
),
)
return normalized
def _is_plugin_provider_id(provider_id: str | None) -> bool:
if not provider_id:
return False
parts = provider_id.split("/")
return len(parts) == 3 and all(parts)
def _plugin_file_input_schema(description: str) -> dict[str, Any]:
return {
"description": description,
"anyOf": [
{
"type": "string",
"minLength": 1,
"description": "HTTP(S) URL or sandbox-local file path.",
},
{
"type": "object",
"additionalProperties": False,
"required": ["transfer_method", "url"],
"properties": {
"transfer_method": {"type": "string", "enum": ["remote_url"]},
"url": {"type": "string", "minLength": 1},
},
},
{
"type": "object",
"additionalProperties": False,
"required": ["transfer_method", "reference"],
"properties": {
"transfer_method": {
"type": "string",
"enum": ["local_file", "tool_file", "datasource_file"],
},
"reference": {"type": "string", "minLength": 1},
},
},
],
}
@@ -333,8 +333,6 @@ class WorkflowAgentOutputAdapter:
session_snapshot = None
if isinstance(event, AgentBackendRunSucceededInternalEvent | AgentBackendDeferredToolCallInternalEvent):
session_snapshot = event.session_snapshot
if event.usage is not None:
agent_backend["usage"] = dict(event.usage)
if session_snapshot is not None:
agent_backend["session_snapshot"] = {
"layer_count": len(session_snapshot.layers),
@@ -206,14 +206,17 @@ class WorkflowAgentRuntimeRequestBuilder:
"cli_tool_count": len(agent_soul.tools.cli_tools),
}
config_layer_config, config_warnings = build_config_layer_config(
agent_soul,
agent_id=context.agent.id,
config_version_id=context.snapshot.id,
config_version_kind="snapshot",
)
append_runtime_warnings(metadata, config_warnings)
soul_prompt_resolver = build_config_aware_soul_mention_resolver(agent_soul)
config_layer_config: DifyConfigLayerConfig | None = None
soul_prompt_resolver = build_soul_mention_resolver(agent_soul)
if dify_config.AGENT_DRIVE_MANIFEST_ENABLED:
config_layer_config, config_warnings = build_config_layer_config(
agent_soul,
agent_id=context.agent.id,
config_version_id=context.snapshot.id,
config_version_kind="snapshot",
)
append_runtime_warnings(metadata, config_warnings)
soul_prompt_resolver = build_config_aware_soul_mention_resolver(agent_soul)
soul_prompt = expand_prompt_mentions(agent_soul.prompt.system_prompt, soul_prompt_resolver).strip()
knowledge_config = build_knowledge_layer_config(agent_soul)
@@ -714,7 +717,7 @@ def build_shell_layer_config(agent_soul: AgentSoulConfig) -> DifyShellLayerConfi
for tool in (_shell_cli_tool(item) for item in agent_soul.tools.cli_tools if _cli_tool_enabled(item))
if tool is not None
],
env=_shell_env_vars(agent_soul.env.variables, agent_soul.env.secret_refs),
env=[env for env in (_shell_env_var(item) for item in agent_soul.env.variables) if env is not None],
secret_refs=[
secret for secret in (_shell_secret_ref(item) for item in agent_soul.env.secret_refs) if secret is not None
],
@@ -861,13 +864,8 @@ def build_config_layer_config(
agent_id: str | None = None,
config_version_id: str | None = None,
config_version_kind: Literal["snapshot", "draft", "build_draft"] = "snapshot",
) -> tuple[DifyConfigLayerConfig, list[dict[str, str]]]:
"""Build the always-present Agent config layer from Agent Soul state.
The ``dify.config`` layer must exist for every Agent v2 runtime request so
the backend can expose the config CLI/help surface even when the current
Agent Soul has no config assets, note, or prompt mentions.
"""
) -> tuple[DifyConfigLayerConfig | None, list[dict[str, str]]]:
"""Derive prompt-mentioned eager-pull names from Agent Soul."""
ordered_mentions = list(
dict.fromkeys(
@@ -876,6 +874,14 @@ def build_config_layer_config(
if mention.kind in {MentionKind.SKILL, MentionKind.FILE} and mention.ref_id
)
)
if (
not agent_soul.config_skills
and not agent_soul.config_files
and not agent_soul.config_note
and not ordered_mentions
):
return None, []
skill_names = {skill.name for skill in agent_soul.config_skills}
file_names = {file_ref.name for file_ref in agent_soul.config_files}
warnings: list[dict[str, str]] = []
@@ -962,7 +968,11 @@ def _shell_cli_tool(item: object) -> DifyShellCliToolConfig | None:
if not commands and not isinstance(name, str):
return None
tool_env = data.get("env") if isinstance(data.get("env"), Mapping) else {}
env = _shell_env_vars(_env_entries(tool_env, "variables"), _env_entries(tool_env, "secret_refs"))
env = [
env_var
for env_var in (_shell_env_var(item) for item in _env_entries(tool_env, "variables"))
if env_var is not None
]
secret_refs = [
secret_ref
for secret_ref in (_shell_secret_ref(item) for item in _env_entries(tool_env, "secret_refs"))
@@ -985,12 +995,6 @@ def _env_entries(env: object, key: str) -> list[object]:
return entries
def _shell_env_vars(variables: Sequence[object], secret_refs: Sequence[object]) -> list[DifyShellEnvVarConfig]:
env_vars = [_shell_env_var(item) for item in variables]
secret_env_vars = [_shell_env_var(item) for item in secret_refs if _has_secret_value(item)]
return [env for env in [*env_vars, *secret_env_vars] if env is not None]
def _shell_env_var(item: object) -> DifyShellEnvVarConfig | None:
data = _plain_mapping(item)
name = _name_from_mapping(data)
@@ -1007,15 +1011,13 @@ def _shell_secret_ref(item: object) -> DifyShellSecretRefConfig | None:
name = _name_from_mapping(data)
if name is None:
return None
# Inline Composer values are passed as env vars because the agent-backend
# secret ref schema only accepts short backend-managed reference IDs.
if _has_secret_value(item):
return None
ref = data.get("ref") or data.get("credential_id") or data.get("provider_credential_id")
if ref is None:
ref = data.get("id")
if ref is None:
return None
ref = (
data.get("ref")
or data.get("value")
or data.get("id")
or data.get("credential_id")
or data.get("provider_credential_id")
)
return DifyShellSecretRefConfig(name=name, ref=str(ref) if ref is not None else None)
@@ -1027,12 +1029,6 @@ def _plain_mapping(item: object) -> dict[str, Any]:
return {}
def _has_secret_value(item: object) -> bool:
data = _plain_mapping(item)
value = data.get("value")
return isinstance(value, str) and bool(value)
def _name_from_mapping(item: Mapping[str, Any]) -> str | None:
for key in ("name", "key", "env_name", "variable"):
value = item.get(key)
+6 -12
View File
@@ -15,13 +15,13 @@ simple_account_fields = {
}
class SimpleAccountResponse(ResponseModel):
class SimpleAccount(ResponseModel):
id: str
name: str
email: str
class _AccountAvatarResponseMixin(ResponseModel):
class _AccountAvatar(ResponseModel):
avatar: str | None = None
@computed_field(return_type=str | None) # type: ignore[prop-decorator]
@@ -30,7 +30,7 @@ class _AccountAvatarResponseMixin(ResponseModel):
return build_avatar_url(self.avatar)
class AccountResponse(_AccountAvatarResponseMixin):
class Account(_AccountAvatar):
id: str
name: str
email: str
@@ -48,7 +48,7 @@ class AccountResponse(_AccountAvatarResponseMixin):
return to_timestamp(value)
class AccountWithRoleResponse(_AccountAvatarResponseMixin):
class AccountWithRole(_AccountAvatar):
id: str
name: str
email: str
@@ -65,11 +65,5 @@ class AccountWithRoleResponse(_AccountAvatarResponseMixin):
return to_timestamp(value)
class AccountWithRoleListResponse(ResponseModel):
accounts: list[AccountWithRoleResponse]
SimpleAccount = SimpleAccountResponse
Account = AccountResponse
AccountWithRole = AccountWithRoleResponse
AccountWithRoleList = AccountWithRoleListResponse
class AccountWithRoleList(ResponseModel):
accounts: list[AccountWithRole]
-86
View File
@@ -1,86 +0,0 @@
from __future__ import annotations
import math
from dataclasses import dataclass
from sqlalchemy import Select, func, select
from sqlalchemy.orm import Session, scoped_session
@dataclass
class PaginatedResult[T]:
"""Minimal pagination container backed by plain SQLAlchemy queries.
Drop-in replacement for Flask-SQLAlchemy's ``db.paginate`` return value.
Only the attributes actually consumed across the codebase are exposed:
``items``, ``total``, ``page``, ``per_page``, ``pages``, ``has_next``.
"""
items: list[T]
total: int
page: int
per_page: int
@property
def pages(self) -> int:
if self.per_page == 0:
return 0
return max(1, math.ceil(self.total / self.per_page))
@property
def has_next(self) -> bool:
return self.page < self.pages
def __iter__(self):
return iter(self.items)
def paginate_query(
stmt: Select,
*,
page: int = 1,
per_page: int = 20,
max_per_page: int | None = None,
session: Session | scoped_session | None = None,
) -> PaginatedResult:
"""Execute *stmt* as a paginated query using plain SQLAlchemy.
Parameters
----------
stmt:
A SQLAlchemy ``select()`` statement.
page:
1-based page number.
per_page:
Number of items per page.
max_per_page:
Hard ceiling for *per_page*; ``None`` means no cap.
session:
The session to use. Falls back to ``db.session`` when omitted.
"""
if session is None:
from extensions.ext_database import db
session = db.session
if max_per_page is not None:
per_page = min(per_page, max_per_page)
page = max(1, page)
per_page = max(1, per_page)
# total count — wrap in a scalar subquery so arbitrary selects work
count_stmt = select(func.count()).select_from(stmt.subquery())
total: int = session.scalar(count_stmt) or 0 # type: ignore[assignment]
# fetch the page
offset = (page - 1) * per_page
page_stmt = stmt.limit(per_page).offset(offset)
items = list(session.scalars(page_stmt).all())
return PaginatedResult(
items=items,
total=total,
page=page,
per_page=per_page,
)
+4 -14
View File
@@ -250,10 +250,9 @@ class AgentSecretRefConfig(AgentFlexibleConfig):
env_name: str | None = Field(default=None, max_length=255)
variable: str | None = Field(default=None, max_length=255)
type: str | None = Field(default=None, max_length=64)
# User-provided secret value. Long API tokens are valid here; runtime maps
# this field into a shell env var, while ref/id/credential_id fields keep the
# backend-managed secret reference path.
value: str | None = None
# UI-facing selected secret reference. This is a credential/ref id, not the
# plaintext secret value; runtime maps it to the shell-layer ``ref``.
value: str | None = Field(default=None, max_length=255)
id: str | None = Field(default=None, max_length=255)
ref: str | None = Field(default=None, max_length=255)
credential_id: str | None = Field(default=None, max_length=255)
@@ -516,18 +515,9 @@ class AgentTextToSpeechFeatureConfig(AgentFeatureToggleConfig):
autoPlay: str | None = None
class AgentSuggestedQuestionsAfterAnswerModelConfig(AgentFlexibleConfig):
"""Legacy Chat App model config used only for follow-up question generation."""
provider: str = Field(min_length=1, max_length=255)
name: str = Field(min_length=1, max_length=255)
mode: str | None = Field(default=None, max_length=64)
completion_params: dict[str, Any] | None = None
class AgentSuggestedQuestionsAfterAnswerFeatureConfig(AgentFeatureToggleConfig):
prompt: str | None = None
model: AgentSuggestedQuestionsAfterAnswerModelConfig | None = None
model: AgentSoulModelConfig | None = None
class AgentModerationIOConfig(AgentFlexibleConfig):
+6 -6
View File
@@ -125,7 +125,7 @@ class TriggerSubscription(TypeBase):
DateTime,
nullable=False,
server_default=func.current_timestamp(),
onupdate=func.current_timestamp(),
server_onupdate=func.current_timestamp(),
init=False,
)
@@ -180,7 +180,7 @@ class TriggerOAuthSystemClient(TypeBase):
DateTime,
nullable=False,
server_default=func.current_timestamp(),
onupdate=func.current_timestamp(),
server_onupdate=func.current_timestamp(),
init=False,
)
@@ -210,7 +210,7 @@ class TriggerOAuthTenantClient(TypeBase):
DateTime,
nullable=False,
server_default=func.current_timestamp(),
onupdate=func.current_timestamp(),
server_onupdate=func.current_timestamp(),
init=False,
)
@@ -370,7 +370,7 @@ class WorkflowWebhookTrigger(TypeBase):
DateTime,
nullable=False,
server_default=func.current_timestamp(),
onupdate=func.current_timestamp(),
server_onupdate=func.current_timestamp(),
init=False,
)
@@ -430,7 +430,7 @@ class WorkflowPluginTrigger(TypeBase):
DateTime,
nullable=False,
server_default=func.current_timestamp(),
onupdate=func.current_timestamp(),
server_onupdate=func.current_timestamp(),
init=False,
)
@@ -479,7 +479,7 @@ class AppTrigger(TypeBase):
DateTime,
nullable=False,
default=naive_utc_now(),
onupdate=naive_utc_now(),
server_onupdate=func.current_timestamp(),
init=False,
)
+258 -112
View File
@@ -38,7 +38,7 @@ Get account avatar url
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [POST] /account/change-email
#### Request Body
@@ -77,7 +77,7 @@ Get account avatar url
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [POST] /account/change-email/validity
#### Request Body
@@ -198,7 +198,7 @@ Get account avatar url
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [POST] /account/interface-theme
#### Request Body
@@ -211,7 +211,7 @@ Get account avatar url
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [POST] /account/name
#### Request Body
@@ -224,7 +224,7 @@ Get account avatar url
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [POST] /account/password
#### Request Body
@@ -237,14 +237,14 @@ Get account avatar url
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [GET] /account/profile
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [POST] /account/timezone
#### Request Body
@@ -257,7 +257,7 @@ Get account avatar url
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |
| 200 | Success | **application/json**: [Account](#account)<br> |
### [POST] /activate
Activate account with invitation token
@@ -1217,22 +1217,6 @@ List workflow apps that reference this Agent App's bound Agent (read-only)
| 200 | Referencing workflows listed successfully | **application/json**: [AgentReferencingWorkflowsResponse](#agentreferencingworkflowsresponse)<br> |
| 404 | Agent not found | |
### [GET] /agent/{agent_id}/sandbox
Get basic information for an Agent App conversation sandbox
#### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| agent_id | path | Agent ID | Yes | string (uuid) |
| conversation_id | query | Agent App conversation ID | Yes | string |
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Sandbox information returned | **application/json**: [SandboxInfoResponse](#sandboxinforesponse)<br> |
### [GET] /agent/{agent_id}/sandbox/files
List a directory in an Agent App conversation sandbox
@@ -7067,9 +7051,9 @@ Request body:
#### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br> |
### [POST] /installed-apps/{installed_app_id}/chat-messages/{task_id}/stop
#### Parameters
@@ -7100,9 +7084,9 @@ Request body:
#### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br> |
### [POST] /installed-apps/{installed_app_id}/completion-messages/{task_id}/stop
#### Parameters
@@ -7243,9 +7227,9 @@ Request body:
#### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br> |
### [GET] /installed-apps/{installed_app_id}/messages/{message_id}/suggested-questions
#### Parameters
@@ -7375,9 +7359,9 @@ Request body:
#### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br> |
### [POST] /installed-apps/{installed_app_id}/workflows/tasks/{task_id}/stop
**Stop workflow task**
@@ -9327,7 +9311,7 @@ Remove one or more tag bindings from a target.
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [TagListResponse](#taglistresponse)<br> |
| 200 | Success | **application/json**: [ [TagResponse](#tagresponse) ]<br> |
### [POST] /tags
#### Request Body
@@ -9942,7 +9926,7 @@ Increment snippet use count by 1
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountWithRoleListResponse](#accountwithrolelistresponse)<br> |
| 200 | Success | **application/json**: [AccountWithRoleList](#accountwithrolelist)<br> |
### [GET] /workspaces/current/default-model
#### Parameters
@@ -10151,7 +10135,7 @@ Update a plugin endpoint
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [AccountWithRoleListResponse](#accountwithrolelistresponse)<br> |
| 200 | Success | **application/json**: [AccountWithRoleList](#accountwithrolelist)<br> |
### [POST] /workspaces/current/members/invite-email
#### Request Body
@@ -12660,6 +12644,23 @@ Default namespace
| role_name | string | | No |
| tenant_id | string | | No |
#### Account
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| avatar | string | | No |
| avatar_url | string | | Yes |
| created_at | integer | | No |
| email | string | | Yes |
| id | string | | Yes |
| interface_language | string | | No |
| interface_theme | string | | No |
| is_password_set | boolean | | Yes |
| last_login_at | integer | | No |
| last_login_ip | string | | No |
| name | string | | Yes |
| timezone | string | | No |
#### AccountAvatarPayload
| Name | Type | Description | Required |
@@ -12735,36 +12736,13 @@ Default namespace
| password | string | | No |
| repeat_new_password | string | | Yes |
#### AccountResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| avatar | string | | No |
| avatar_url | string | | Yes |
| created_at | integer | | No |
| email | string | | Yes |
| id | string | | Yes |
| interface_language | string | | No |
| interface_theme | string | | No |
| is_password_set | boolean | | Yes |
| last_login_at | integer | | No |
| last_login_ip | string | | No |
| name | string | | Yes |
| timezone | string | | No |
#### AccountTimezonePayload
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| timezone | string | | Yes |
#### AccountWithRoleListResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| accounts | [ [AccountWithRoleResponse](#accountwithroleresponse) ] | | Yes |
#### AccountWithRoleResponse
#### AccountWithRole
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
@@ -12780,6 +12758,12 @@ Default namespace
| roles | [ object ] | | No |
| status | string | | Yes |
#### AccountWithRoleList
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| accounts | [ [AccountWithRole](#accountwithrole) ] | | Yes |
#### ActivateCheckQuery
| Name | Type | Description | Required |
@@ -12828,7 +12812,7 @@ Default namespace
| ---- | ---- | ----------- | -------- |
| conversation_id | string | | No |
| created_at | integer | | No |
| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| elapsed_time | number | | No |
| exceptions_count | integer | | No |
| finished_at | integer | | No |
@@ -12935,7 +12919,7 @@ Default namespace
| icon_background | string | Icon background color | No |
| icon_type | [IconType](#icontype) | Icon type | No |
| name | string | Agent name | Yes |
| role | string | Agent role | No |
| role | string | Agent role | Yes |
#### AgentAppDetailWithSite
@@ -13062,7 +13046,7 @@ default (the config form sends the full desired feature state on save).
| icon_type | [IconType](#icontype) | Icon type | No |
| max_active_requests | integer | Maximum active requests | No |
| name | string | App name | Yes |
| role | string | Agent role | No |
| role | string | Agent role | Yes |
| use_icon_as_answer_icon | boolean | Use icon as answer icon | No |
#### AgentAverageResponseTimeStatisticResponse
@@ -14569,20 +14553,9 @@ Soft lifecycle state for Agent records.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| enabled | boolean | | No |
| model | [AgentSuggestedQuestionsAfterAnswerModelConfig](#agentsuggestedquestionsafteranswermodelconfig) | | No |
| model | [AgentSoulModelConfig](#agentsoulmodelconfig) | | No |
| prompt | string | | No |
#### AgentSuggestedQuestionsAfterAnswerModelConfig
Legacy Chat App model config used only for follow-up question generation.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| completion_params | object | | No |
| mode | string | | No |
| name | string | | Yes |
| provider | string | | Yes |
#### AgentTextToSpeechFeatureConfig
| Name | Type | Description | Required |
@@ -20412,13 +20385,6 @@ Whitelist scopes accepted by RBAC app and dataset access config APIs.
| size | integer | | No |
| type | string, <br>**Available values:** "dir", "file", "other", "symlink" | *Enum:* `"dir"`, `"file"`, `"other"`, `"symlink"` | Yes |
#### SandboxInfoResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| session_id | string | | Yes |
| workspace_cwd | string | | Yes |
#### SandboxListResponse
| Name | Type | Description | Required |
@@ -20609,14 +20575,6 @@ Whitelist scopes accepted by RBAC app and dataset access config APIs.
| id | string | | Yes |
| name | string | | Yes |
#### SimpleAccountResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| email | string | | Yes |
| id | string | | Yes |
| name | string | | Yes |
#### SimpleConversation
| Name | Type | Description | Required |
@@ -20954,7 +20912,7 @@ Query parameters for listing snippet published workflows.
| ---- | ---- | ----------- | -------- |
| conversation_variables | [ [WorkflowConversationVariableResponse](#workflowconversationvariableresponse) ] | | Yes |
| created_at | integer | | Yes |
| created_by | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by | [SimpleAccount](#simpleaccount) | | No |
| environment_variables | [ [WorkflowEnvironmentVariableResponse](#workflowenvironmentvariableresponse) ] | | Yes |
| features | object | | Yes |
| graph | object | | Yes |
@@ -20966,7 +20924,7 @@ Query parameters for listing snippet published workflows.
| rag_pipeline_variables | [ [PipelineVariableResponse](#pipelinevariableresponse) ] | | Yes |
| tool_published | boolean | | Yes |
| updated_at | integer | | Yes |
| updated_by | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| updated_by | [SimpleAccount](#simpleaccount) | | No |
| version | string | | Yes |
#### StarredAppListQuery
@@ -21188,12 +21146,6 @@ Model class for provider system configuration response.
| keyword | string | Search keyword | No |
| type | string, <br>**Available values:** "", "app", "knowledge", "snippet" | Tag type filter<br>*Enum:* `""`, `"app"`, `"knowledge"`, `"snippet"` | No |
#### TagListResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| TagListResponse | array | | |
#### TagResponse
| Name | Type | Description | Required |
@@ -21447,6 +21399,35 @@ Enum class for tool provider
| use_icon_as_answer_icon | boolean | | No |
| workflow | [TrialWorkflowPartialResponse](#trialworkflowpartialresponse) | | No |
#### TrialAppDetailWithSite
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| access_mode | string | | No |
| api_base_url | string | | No |
| created_at | long | | No |
| created_by | string | | No |
| deleted_tools | [ [TrialDeletedTool](#trialdeletedtool) ] | | No |
| description | string | | No |
| enable_api | boolean | | No |
| enable_site | boolean | | No |
| icon | string | | No |
| icon_background | string | | No |
| icon_type | string | | No |
| icon_url | string | | No |
| id | string | | No |
| max_active_requests | integer | | No |
| mode | string | | No |
| model_config | [TrialAppModelConfig](#trialappmodelconfig) | | No |
| name | string | | No |
| permission_keys | [ string ] | | No |
| site | [TrialSite](#trialsite) | | No |
| tags | [ [TrialTag](#trialtag) ] | | No |
| updated_at | long | | No |
| updated_by | string | | No |
| use_icon_as_answer_icon | boolean | | No |
| workflow | [TrialWorkflowPartial](#trialworkflowpartial) | | No |
#### TrialAppMode
| Name | Type | Description | Required |
@@ -21462,6 +21443,35 @@ Enum class for tool provider
| name | string | | Yes |
| provider | string | | Yes |
#### TrialAppModelConfig
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| agent_mode | object | | No |
| annotation_reply | object | | No |
| chat_prompt_config | object | | No |
| completion_prompt_config | object | | No |
| created_at | long | | No |
| created_by | string | | No |
| dataset_configs | object | | No |
| dataset_query_variable | string | | No |
| external_data_tools | [ object ] | | No |
| file_upload | object | | No |
| model | object | | No |
| more_like_this | object | | No |
| opening_statement | string | | No |
| pre_prompt | string | | No |
| prompt_type | string | | No |
| retriever_resource | object | | No |
| sensitive_word_avoidance | object | | No |
| speech_to_text | object | | No |
| suggested_questions | [ string ] | | No |
| suggested_questions_after_answer | object | | No |
| text_to_speech | object | | No |
| updated_at | long | | No |
| updated_by | string | | No |
| user_input_form | [ object ] | | No |
#### TrialAppModelConfigResponse
| Name | Type | Description | Required |
@@ -21491,6 +21501,40 @@ Enum class for tool provider
| updated_by | string | | No |
| user_input_form | [ [JsonObject](#jsonobject) ] | | No |
#### TrialConversationVariable
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| description | string | | No |
| id | string | | No |
| name | string | | No |
| value | string<br>integer<br>number<br>boolean<br>object<br>[ object ] | | No |
| value_type | string | | No |
#### TrialDataset
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | long | | No |
| created_by | string | | No |
| data_source_type | string | | No |
| description | string | | No |
| id | string | | No |
| indexing_technique | string | | No |
| name | string | | No |
| permission | string | | No |
| permission_keys | [ string ] | | No |
#### TrialDatasetList
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data | [ [TrialDataset](#trialdataset) ] | | No |
| has_more | boolean | | No |
| limit | integer | | No |
| page | integer | | No |
| total | integer | | No |
#### TrialDatasetListQuery
| Name | Type | Description | Required |
@@ -21523,6 +21567,14 @@ Enum class for tool provider
| permission | string | | No |
| permission_keys | [ string ] | | No |
#### TrialDeletedTool
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| provider_id | string | | No |
| tool_name | string | | No |
| type | string | | No |
#### TrialDeletedToolResponse
| Name | Type | Description | Required |
@@ -21543,14 +21595,62 @@ Enum class for tool provider
| ---- | ---- | ----------- | -------- |
| trial_models | [ string ] | | Yes |
#### TrialPipelineVariable
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| allow_file_extension | [ string ] | | No |
| allow_file_upload_methods | [ string ] | | No |
| allowed_file_types | [ string ] | | No |
| belong_to_node_id | string | | No |
| default_value | string<br>integer<br>number<br>boolean<br>object<br>[ object ] | | No |
| label | string | | No |
| max_length | integer | | No |
| options | [ string ] | | No |
| placeholder | string | | No |
| required | boolean | | No |
| tooltips | string | | No |
| type | string | | No |
| unit | string | | No |
| variable | string | | No |
#### TrialSimpleAccount
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| email | string | | No |
| id | string | | Yes |
| id | string | | No |
| name | string | | No |
#### TrialSite
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| access_token | string | | No |
| app_base_url | string | | No |
| chat_color_theme | string | | No |
| chat_color_theme_inverted | boolean | | No |
| code | string | | No |
| copyright | string | | No |
| created_at | long | | No |
| created_by | string | | No |
| custom_disclaimer | string | | No |
| customize_domain | string | | No |
| customize_token_strategy | string | | No |
| default_language | string | | No |
| description | string | | No |
| icon | string | | No |
| icon_background | string | | No |
| icon_type | string | | No |
| icon_url | string | | No |
| privacy_policy | string | | No |
| prompt_public | boolean | | No |
| show_workflow_steps | boolean | | No |
| title | string | | No |
| updated_at | long | | No |
| updated_by | string | | No |
| use_icon_as_answer_icon | boolean | | No |
#### TrialSiteResponse
| Name | Type | Description | Required |
@@ -21581,6 +21681,14 @@ Enum class for tool provider
| updated_by | string | | No |
| use_icon_as_answer_icon | boolean | | No |
#### TrialTag
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| id | string | | No |
| name | string | | No |
| type | string | | No |
#### TrialTagResponse
| Name | Type | Description | Required |
@@ -21589,6 +21697,44 @@ Enum class for tool provider
| name | string | | Yes |
| type | string | | Yes |
#### TrialWorkflow
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| conversation_variables | [ [TrialConversationVariable](#trialconversationvariable) ] | | No |
| created_at | long | | No |
| created_by | [TrialSimpleAccount](#trialsimpleaccount) | | No |
| environment_variables | [ object ] | | No |
| features | object | | No |
| graph | object | | No |
| hash | string | | No |
| id | string | | No |
| marked_comment | string | | No |
| marked_name | string | | No |
| rag_pipeline_variables | [ [TrialPipelineVariable](#trialpipelinevariable) ] | | No |
| tool_published | boolean | | No |
| updated_at | long | | No |
| updated_by | [TrialSimpleAccount](#trialsimpleaccount) | | No |
| version | string | | No |
#### TrialWorkflowAccount
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| email | string | | No |
| id | string | | Yes |
| name | string | | No |
#### TrialWorkflowPartial
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | long | | No |
| created_by | string | | No |
| id | string | | No |
| updated_at | long | | No |
| updated_by | string | | No |
#### TrialWorkflowPartialResponse
| Name | Type | Description | Required |
@@ -21605,7 +21751,7 @@ Enum class for tool provider
| ---- | ---- | ----------- | -------- |
| conversation_variables | [ [JsonObject](#jsonobject) ] | | No |
| created_at | integer | | No |
| created_by | [TrialSimpleAccount](#trialsimpleaccount) | | No |
| created_by | [TrialWorkflowAccount](#trialworkflowaccount) | | No |
| environment_variables | [ [JsonObject](#jsonobject) ] | | No |
| features | [JsonObject](#jsonobject) | | No |
| graph | [JsonObject](#jsonobject) | | Yes |
@@ -21616,7 +21762,7 @@ Enum class for tool provider
| rag_pipeline_variables | [ [JsonObject](#jsonobject) ] | | No |
| tool_published | boolean | | No |
| updated_at | integer | | No |
| updated_by | [TrialSimpleAccount](#trialsimpleaccount) | | No |
| updated_by | [TrialWorkflowAccount](#trialworkflowaccount) | | No |
| version | string | | No |
#### TriggerCreationMethod
@@ -22035,7 +22181,7 @@ How a workflow node is bound to an Agent.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | integer | | No |
| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| created_by_end_user | [SimpleEndUser](#simpleenduser) | | No |
| created_by_role | string | | No |
| created_from | string | | No |
@@ -22072,7 +22218,7 @@ How a workflow node is bound to an Agent.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | integer | | No |
| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| created_by_end_user | [SimpleEndUser](#simpleenduser) | | No |
| id | string | | Yes |
| trigger_metadata | | | No |
@@ -22173,7 +22319,7 @@ How a workflow node is bound to an Agent.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| users | [ [AccountWithRoleResponse](#accountwithroleresponse) ] | | Yes |
| users | [ [AccountWithRole](#accountwithrole) ] | | Yes |
#### WorkflowCommentReply
@@ -22596,7 +22742,7 @@ tenant's default model. The underlying generator never raises — an empty
| ---- | ---- | ----------- | -------- |
| conversation_variables | [ [WorkflowConversationVariableResponse](#workflowconversationvariableresponse) ] | | Yes |
| created_at | integer | | Yes |
| created_by | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by | [SimpleAccount](#simpleaccount) | | No |
| environment_variables | [ [WorkflowEnvironmentVariableResponse](#workflowenvironmentvariableresponse) ] | | Yes |
| features | object | | Yes |
| graph | object | | Yes |
@@ -22607,7 +22753,7 @@ tenant's default model. The underlying generator never raises — an empty
| rag_pipeline_variables | [ [PipelineVariableResponse](#pipelinevariableresponse) ] | | Yes |
| tool_published | boolean | | Yes |
| updated_at | integer | | Yes |
| updated_by | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| updated_by | [SimpleAccount](#simpleaccount) | | No |
| version | string | | Yes |
#### WorkflowRestoreResponse
@@ -22642,7 +22788,7 @@ tenant's default model. The underlying generator never raises — an empty
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | integer | | No |
| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| created_by_end_user | [SimpleEndUser](#simpleenduser) | | No |
| created_by_role | string | | No |
| elapsed_time | number | | No |
@@ -22681,7 +22827,7 @@ tenant's default model. The underlying generator never raises — an empty
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | integer | | No |
| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| elapsed_time | number | | No |
| exceptions_count | integer | | No |
| finished_at | integer | | No |
@@ -22728,7 +22874,7 @@ tenant's default model. The underlying generator never raises — an empty
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | integer | | No |
| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| created_by_end_user | [SimpleEndUser](#simpleenduser) | | No |
| created_by_role | string | | No |
| elapsed_time | number | | No |
+2 -2
View File
@@ -3937,7 +3937,7 @@ Model class for provider with models response.
| output_variable_name | string | | Yes |
| type | string | | No |
#### SimpleAccountResponse
#### SimpleAccount
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
@@ -4148,7 +4148,7 @@ in form definiton, or a variable while the workflow is running.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | integer | | No |
| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| created_by_end_user | [SimpleEndUser](#simpleenduser) | | No |
| created_by_role | string | | No |
| created_from | string | | No |
-1
View File
@@ -42,7 +42,6 @@ dependencies = [
"opentelemetry-propagator-b3>=1.41.1,<2.0.0",
"readabilipy==0.3.0",
"resend>=2.27.0,<3.0.0",
"zstandard==0.25.0",
# Emerging: newer and fast-moving, use compatible pins
"fastopenapi[flask]==0.7.0",
"graphon==0.6.0",
+1 -2
View File
@@ -12,7 +12,6 @@ from core.rag.index_processor.index_processor_factory import IndexProcessorFacto
from enums.cloud_plan import CloudPlan
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from libs.pagination import paginate_query
from models.dataset import Dataset, DatasetAutoDisableLog, DatasetQuery, Document
from services.feature_service import FeatureService
@@ -89,7 +88,7 @@ def clean_unused_datasets_task():
.order_by(Dataset.created_at.desc())
)
datasets = paginate_query(stmt, page=page, per_page=50)
datasets = db.paginate(stmt, page=page, per_page=50, error_out=False)
except SQLAlchemyError:
raise
+3 -1
View File
@@ -619,7 +619,9 @@ WHERE
@staticmethod
def _statistics_message_scope_sql(source_filter: AgentSourceFilter) -> str:
app_scope = "m.app_id = :app_id"
if source_filter.invoke_from is not None:
if source_filter.invoke_from is None:
app_scope += " AND m.invoke_from != :debugger"
else:
app_scope += " AND m.invoke_from = :source"
workflow_binding_filters = []
if source_filter.app_id:
+1 -4
View File
@@ -175,10 +175,7 @@ class SkillPackageService:
continue
raise SkillPackageError(
"files_outside_skill_root",
(
"skill package must contain exactly one skill; "
"multiple skill folders in one archive are not supported"
),
"skill archive contains files outside the selected skill root",
status_code=400,
)
normalized_path = safe_path.removeprefix(skill_root_prefix)
+3 -38
View File
@@ -13,7 +13,7 @@ from collections.abc import Callable
from agenton.compositor import CompositorSessionSnapshot
from dify_agent.client import Client
from dify_agent.protocol import RuntimeLayerSpec, SandboxLocator, build_sandbox_locator_from_layer_specs
from pydantic import BaseModel, TypeAdapter
from pydantic import TypeAdapter
from sqlalchemy import select
from configs import dify_config
@@ -38,15 +38,8 @@ class AgentSandboxInspectorError(Exception):
self.status_code = status_code
class AgentSandboxInfo(BaseModel):
"""Basic Agent App sandbox metadata returned after a successful availability probe."""
session_id: str
workspace_cwd: str
class AgentAppSandboxService:
"""Inspect and proxy file access for an Agent App conversation sandbox."""
"""List/read/upload files in an Agent App conversation sandbox."""
def __init__(
self,
@@ -57,18 +50,6 @@ class AgentAppSandboxService:
self._session_store = session_store or AgentAppRuntimeSessionStore()
self._client_factory = client_factory or _default_client_factory
def get_info(self, *, tenant_id: str, app_id: str, conversation_id: str) -> AgentSandboxInfo:
locator = self._resolve_locator(tenant_id=tenant_id, app_id=app_id, conversation_id=conversation_id)
session_id, workspace_cwd = _extract_shell_workspace_or_raise(
snapshot=locator.session_snapshot,
not_found_message="this conversation's agent has no sandbox workspace",
)
return AgentSandboxInfo(
session_id=session_id,
workspace_cwd=workspace_cwd,
)
def list_files(self, *, tenant_id: str, app_id: str, conversation_id: str, path: str):
locator = self._resolve_locator(tenant_id=tenant_id, app_id=app_id, conversation_id=conversation_id)
return self._client_factory().list_sandbox_files_sync(locator, path)
@@ -224,22 +205,6 @@ def _build_locator_or_raise(
raise AgentSandboxInspectorError("no_sandbox", not_found_message, status_code=404) from exc
def _extract_shell_workspace_or_raise(
*,
snapshot: CompositorSessionSnapshot,
not_found_message: str,
) -> tuple[str, str]:
shell_layer = next((layer for layer in snapshot.layers if layer.name == "shell"), None)
if shell_layer is None:
raise AgentSandboxInspectorError("no_sandbox", not_found_message, status_code=404)
session_id = shell_layer.runtime_state.get("session_id")
workspace_cwd = shell_layer.runtime_state.get("workspace_cwd")
if not isinstance(session_id, str) or not isinstance(workspace_cwd, str):
raise AgentSandboxInspectorError("no_sandbox", not_found_message, status_code=404)
return session_id, workspace_cwd
def _deserialize_runtime_layer_specs(value: str | None) -> list[RuntimeLayerSpec]:
if not value:
return []
@@ -257,4 +222,4 @@ def _default_client_factory() -> Client:
return Client(base_url=base_url)
__all__ = ["AgentAppSandboxService", "AgentSandboxInfo", "AgentSandboxInspectorError", "WorkflowAgentSandboxService"]
__all__ = ["AgentAppSandboxService", "AgentSandboxInspectorError", "WorkflowAgentSandboxService"]
+7 -10
View File
@@ -496,16 +496,13 @@ class AgentConfigService:
user_id=user_id,
)
self._require_writable(target, surface=surface)
try:
skill_ref, _package = self._skill_normalizer.normalize(
content=content,
filename=filename,
requested_name=None,
tenant_id=tenant_id,
user_id=user_id,
)
except SkillPackageError as exc:
raise AgentConfigServiceError(exc.code, exc.message, status_code=exc.status_code) from exc
skill_ref, _package = self._skill_normalizer.normalize(
content=content,
filename=filename,
requested_name=None,
tenant_id=tenant_id,
user_id=user_id,
)
agent_soul = target.agent_soul.model_copy(deep=True)
existing = {item.name: item for item in agent_soul.config_skills}
order = [item.name for item in agent_soul.config_skills]
+4 -3
View File
@@ -13,7 +13,6 @@ from extensions.ext_database import db
from extensions.ext_redis import redis_client
from libs.datetime_utils import naive_utc_now
from libs.login import current_account_with_tenant
from libs.pagination import paginate_query
from models.model import App, AppAnnotationHitHistory, AppAnnotationSetting, Message, MessageAnnotation
from services.app_ref_service import AnnotationRef, AppRef
from services.feature_service import FeatureService
@@ -243,7 +242,7 @@ class AppAnnotationService:
.where(MessageAnnotation.app_id == app_id)
.order_by(MessageAnnotation.created_at.desc(), MessageAnnotation.id.desc())
)
annotations = paginate_query(stmt, page=page, per_page=limit, max_per_page=100)
annotations = db.paginate(select=stmt, page=page, per_page=limit, max_per_page=100, error_out=False)
return annotations.items, annotations.total or 0
@classmethod
@@ -574,7 +573,9 @@ class AppAnnotationService:
)
.order_by(AppAnnotationHitHistory.created_at.desc())
)
annotation_hit_histories = paginate_query(stmt, page=page, per_page=limit, max_per_page=100)
annotation_hit_histories = db.paginate(
select=stmt, page=page, per_page=limit, max_per_page=100, error_out=False
)
return annotation_hit_histories.items, annotation_hit_histories.total or 0
@classmethod
+7 -5
View File
@@ -5,6 +5,7 @@ from datetime import datetime
from typing import Any, Literal, NotRequired, TypedDict, cast, override
import sqlalchemy as sa
from flask_sqlalchemy.pagination import Pagination
from pydantic import BaseModel, Field
from sqlalchemy import ColumnElement, select
from sqlalchemy.exc import IntegrityError
@@ -23,7 +24,6 @@ from graphon.model_runtime.entities.model_entities import ModelPropertyKey, Mode
from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
from libs.datetime_utils import naive_utc_now
from libs.login import current_user
from libs.pagination import PaginatedResult, paginate_query
from models import Account, AppStar
from models.agent import Agent, AgentIconType, AgentScope, AgentSource, AgentStatus
from models.model import App, AppMode, AppModelConfig, IconType, Site
@@ -220,7 +220,7 @@ class AppService:
def get_paginate_apps(
self, user_id: str, tenant_id: str, params: AppListParams, session: scoped_session
) -> PaginatedResult | None:
) -> Pagination | None:
"""
Get app list with pagination, filters, and explicit sort order.
:param user_id: user id
@@ -234,10 +234,11 @@ class AppService:
order_by = self._build_app_list_order_by(params.sort_by)
app_models = paginate_query(
app_models = db.paginate(
sa.select(App).where(*filters).order_by(order_by),
page=params.page,
per_page=params.limit,
error_out=False,
)
app_ids = [str(app.id) for app in app_models.items]
@@ -254,7 +255,7 @@ class AppService:
def get_paginate_starred_apps(
self, user_id: str, tenant_id: str, params: StarredAppListParams, session: scoped_session
) -> PaginatedResult | None:
) -> Pagination | None:
"""
Get apps starred by the current account with pagination, filters, and explicit sort order.
"""
@@ -263,7 +264,7 @@ class AppService:
return None
order_by = self._build_app_list_order_by(params.sort_by)
app_models = paginate_query(
app_models = db.paginate(
sa.select(App)
.join(
AppStar,
@@ -277,6 +278,7 @@ class AppService:
.order_by(order_by),
page=params.page,
per_page=params.limit,
error_out=False,
)
for app in app_models.items:
+16 -20
View File
@@ -6,9 +6,9 @@ from uuid import UUID
import sqlalchemy as sa
import yaml
from sqlalchemy.orm import Session
from core.tools.tool_manager import ToolManager
from extensions.ext_database import db
from graphon.model_runtime.utils.encoders import jsonable_encoder
from models import Account, Tenant
from models.account import TenantAccountJoin
@@ -120,8 +120,8 @@ class MigrationExportService:
self.package_service = package_service or MigrationPackageService()
self.dependency_discovery_service = dependency_discovery_service or DependencyDiscoveryService()
def export(self, session: Session, selection: ExportSelection) -> ExportResult:
tenant = self._get_tenant(session, selection)
def export(self, selection: ExportSelection) -> ExportResult:
tenant = self._get_tenant(selection)
package = self.package_service.build_empty_package(
source_tenant_id=tenant.id,
source_tenant_name=tenant.name,
@@ -131,7 +131,7 @@ class MigrationExportService:
report_items: list[ResourceReportItem] = []
discovered_dependencies: list[DiscoveredDependency] = []
apps = self._selected_apps(session, tenant.id, selection)
apps = self._selected_apps(tenant.id, selection)
exported_app_ids = {app.id for app in apps}
for app in apps:
dsl_content = AppDslService.export_dsl(app_model=app, include_secret=selection.include_secrets)
@@ -157,7 +157,6 @@ class MigrationExportService:
report_items=report_items,
)
self._export_workflow_tools(
session,
tenant,
self._provider_ids(
selection.additional_workflow_tools, discovered_dependencies, DependencyKind.WORKFLOW_TOOL
@@ -168,7 +167,6 @@ class MigrationExportService:
report_items=report_items,
)
self._export_mcp_tools(
session,
tenant_id=tenant.id,
provider_ids=self._provider_ids(
selection.additional_mcp_tools,
@@ -195,9 +193,9 @@ class MigrationExportService:
),
)
def _get_tenant(self, session: Session, selection: ExportSelection) -> Tenant:
def _get_tenant(self, selection: ExportSelection) -> Tenant:
if selection.source_tenant_id:
tenant = session.get(Tenant, selection.source_tenant_id)
tenant = db.session.get(Tenant, selection.source_tenant_id)
if tenant is None:
raise MigrationDataError(f"Source tenant not found: {selection.source_tenant_id}")
if tenant.name != selection.source_tenant_name:
@@ -205,7 +203,7 @@ class MigrationExportService:
f"Source tenant id/name mismatch: {selection.source_tenant_id} / {selection.source_tenant_name}"
)
return tenant
tenants = list(session.scalars(sa.select(Tenant).where(Tenant.name == selection.source_tenant_name)).all())
tenants = list(db.session.scalars(sa.select(Tenant).where(Tenant.name == selection.source_tenant_name)).all())
if not tenants:
raise MigrationDataError(f"Source tenant not found: {selection.source_tenant_name}")
if len(tenants) > 1:
@@ -214,13 +212,13 @@ class MigrationExportService:
)
return tenants[0]
def _selected_apps(self, session: Session, tenant_id: str, selection: ExportSelection) -> list[App]:
def _selected_apps(self, tenant_id: str, selection: ExportSelection) -> list[App]:
query = sa.select(App).where(App.tenant_id == tenant_id, App.mode.in_(SUPPORTED_APP_MODES))
if not selection.export_all_apps:
if not selection.app_ids:
return []
query = query.where(App.id.in_(selection.app_ids))
apps = list(session.scalars(query).all())
apps = list(db.session.scalars(query).all())
if not selection.export_all_apps and len(apps) != len(set(selection.app_ids)):
found_ids = {app.id for app in apps}
missing_ids = [app_id for app_id in selection.app_ids if app_id not in found_ids]
@@ -267,7 +265,6 @@ class MigrationExportService:
def _export_workflow_tools(
self,
session: Session,
tenant: Tenant,
provider_ids: Iterable[str],
*,
@@ -279,7 +276,7 @@ class MigrationExportService:
provider_ids = self._dedupe(provider_ids)
if not provider_ids:
return
owner = self._get_tenant_owner(session, tenant.id)
owner = self._get_tenant_owner(tenant.id)
if owner is None:
for provider_id in provider_ids:
report_items.append(
@@ -309,7 +306,7 @@ class MigrationExportService:
exported_workflow_tools.append(tool_info)
if tool_info.get("app_id") not in exported_app_ids:
workflow_app_id = str(tool_info.get("app_id") or "")
workflow_app = session.get(App, workflow_app_id) if workflow_app_id else None
workflow_app = db.session.get(App, workflow_app_id) if workflow_app_id else None
self._record_dependency_metadata(
[
DiscoveredDependency(
@@ -330,8 +327,8 @@ class MigrationExportService:
ResourceReportItem(ResourceType.WORKFLOW_TOOL, provider_id, provider_id, "unresolved", str(exc))
)
def _get_tenant_owner(self, session: Session, tenant_id: str) -> Account | None:
return session.scalar(
def _get_tenant_owner(self, tenant_id: str) -> Account | None:
return db.session.scalar(
sa.select(Account)
.join(TenantAccountJoin, Account.id == TenantAccountJoin.account_id)
.where(TenantAccountJoin.tenant_id == tenant_id, TenantAccountJoin.role == "owner")
@@ -341,7 +338,6 @@ class MigrationExportService:
def _export_mcp_tools(
self,
session: Session,
*,
tenant_id: str,
provider_ids: Iterable[str],
@@ -359,7 +355,7 @@ class MigrationExportService:
)
continue
try:
provider = self._get_mcp_provider(session, tenant_id, provider_id)
provider = self._get_mcp_provider(tenant_id, provider_id)
exported_mcp_tools.append(self._serialize_mcp_provider(provider))
report_items.append(ResourceReportItem(ResourceType.MCP_TOOL, provider_id, provider.name, "exported"))
except Exception as exc:
@@ -367,11 +363,11 @@ class MigrationExportService:
ResourceReportItem(ResourceType.MCP_TOOL, provider_id, provider_id, "unresolved", str(exc))
)
def _get_mcp_provider(self, session: Session, tenant_id: str, provider_id: str) -> MCPToolProvider:
def _get_mcp_provider(self, tenant_id: str, provider_id: str) -> MCPToolProvider:
predicates = [MCPToolProvider.server_identifier == provider_id]
if self._is_uuid_string(provider_id):
predicates.append(MCPToolProvider.id == provider_id)
provider = session.scalar(
provider = db.session.scalar(
sa.select(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant_id, sa.or_(*predicates))
)
if provider is None:
+49 -64
View File
@@ -82,24 +82,24 @@ class ImportTargetResolver:
"Target tenant must be provided by --target-tenant, import config, or package metadata."
)
def resolve(self, session: Session, request: ImportRequest) -> ImportTarget:
def resolve(self, request: ImportRequest) -> ImportTarget:
target_tenant_name = self.select_target_tenant_name(request)
package_target = request.package.metadata.target_tenant or {}
if request.cli_target_tenant or request.config_target_tenant:
tenant = self._resolve_tenant_by_id_or_name(session, target_tenant_name)
tenant = self._resolve_tenant_by_id_or_name(target_tenant_name)
elif package_target.get("id") and self._is_uuid(package_target["id"]):
tenant = session.get(Tenant, package_target["id"])
tenant = db.session.get(Tenant, package_target["id"])
if tenant is not None and package_target.get("name") and tenant.name != package_target.get("name"):
raise MigrationDataError(
f"Target tenant id/name mismatch: {package_target['id']} / {package_target['name']}"
)
else:
tenant = self._resolve_tenant_by_id_or_name(session, target_tenant_name)
tenant = self._resolve_tenant_by_id_or_name(target_tenant_name)
if tenant is None:
raise MigrationDataError(f"Target tenant not found: {target_tenant_name}")
account_query = (
session.query(Account)
db.session.query(Account)
.join(TenantAccountJoin, Account.id == TenantAccountJoin.account_id)
.filter(TenantAccountJoin.tenant_id == tenant.id)
)
@@ -123,12 +123,12 @@ class ImportTargetResolver:
operator_email=account.email,
)
def _resolve_tenant_by_id_or_name(self, session: Session, value: str) -> Tenant | None:
def _resolve_tenant_by_id_or_name(self, value: str) -> Tenant | None:
if self._is_uuid(value):
tenant = session.get(Tenant, value)
tenant = db.session.get(Tenant, value)
if tenant is not None:
return tenant
tenants = list(session.scalars(sa.select(Tenant).where(Tenant.name == value)).all())
tenants = list(db.session.scalars(sa.select(Tenant).where(Tenant.name == value)).all())
if len(tenants) > 1:
raise MigrationDataError(f"Target tenant name is ambiguous; use target_tenant.id: {value}")
return tenants[0] if tenants else None
@@ -149,8 +149,8 @@ class MigrationImportService:
def __init__(self, *, target_resolver: ImportTargetResolver | None = None) -> None:
self.target_resolver = target_resolver or ImportTargetResolver()
def import_package(self, session: Session, request: ImportRequest) -> ImportResult:
target = self.target_resolver.resolve(session, request)
def import_package(self, request: ImportRequest) -> ImportResult:
target = self.target_resolver.resolve(request)
options = request.options_override or request.package.metadata.import_options
report_items = [
ResourceReportItem(
@@ -165,7 +165,6 @@ class MigrationImportService:
id_mapping_details: list[ResourceIdMapping] = []
self._import_api_tools(
session,
request.package,
target,
options,
@@ -174,13 +173,12 @@ class MigrationImportService:
id_mapping_details,
self._source_api_provider_ids_by_name(request.package),
)
self._import_mcp_tools(session, request.package, target, options, report_items, id_mapping, id_mapping_details)
self._preflight_dependency_only_mcp(session, request.package, target, report_items)
self._import_mcp_tools(request.package, target, options, report_items, id_mapping, id_mapping_details)
self._preflight_dependency_only_mcp(request.package, target, report_items)
workflow_tool_app_ids = self._workflow_tool_source_app_ids(request.package)
imported_workflow_ids: set[str] = set()
if workflow_tool_app_ids:
self._import_workflows(
session,
request.package,
target,
options,
@@ -190,11 +188,8 @@ class MigrationImportService:
imported_workflow_ids=imported_workflow_ids,
only_app_ids=workflow_tool_app_ids,
)
self._import_workflow_tools(
session, request.package, target, options, id_mapping, id_mapping_details, report_items
)
self._import_workflow_tools(request.package, target, options, id_mapping, id_mapping_details, report_items)
self._import_workflows(
session,
request.package,
target,
options,
@@ -218,7 +213,6 @@ class MigrationImportService:
def _import_workflows(
self,
session: Session,
package: MigrationPackage,
target: ImportTarget,
options: ImportOptions,
@@ -229,8 +223,8 @@ class MigrationImportService:
only_app_ids: set[str] | None = None,
skip_app_ids: set[str] | None = None,
) -> None:
account = session.get(Account, target.operator_id)
tenant = session.get(Tenant, target.tenant_id)
account = db.session.get(Account, target.operator_id)
tenant = db.session.get(Tenant, target.tenant_id)
if account is None:
raise MigrationDataError(f"Operator account not found: {target.operator_id}")
if tenant is None:
@@ -248,7 +242,7 @@ class MigrationImportService:
id_mapping,
)
existing_app = (
self._find_existing_app(session, app_id, target.tenant_id)
self._find_existing_app(app_id, target.tenant_id)
if options.id_strategy == IdStrategy.PRESERVE_ID
else None
)
@@ -270,7 +264,6 @@ class MigrationImportService:
continue
imported_app_id = self._import_workflow_app(
session=session,
account=account,
workflow_data=workflow_data,
dsl_content=dsl_content,
@@ -290,7 +283,7 @@ class MigrationImportService:
if imported_workflow_ids is not None:
imported_workflow_ids.add(app_id)
if options.create_app_api_token_on_import:
self._create_or_reuse_app_api_token(session, imported_app_id, target.tenant_id)
self._create_or_reuse_app_api_token(imported_app_id, target.tenant_id)
report_items.append(
ResourceReportItem(
ResourceType.WORKFLOW,
@@ -311,7 +304,6 @@ class MigrationImportService:
def _import_workflow_app(
self,
*,
session: Session,
account: Account,
workflow_data: dict[str, object],
dsl_content: str,
@@ -319,7 +311,7 @@ class MigrationImportService:
existing_app: App | None,
options: ImportOptions,
) -> str:
import_service = AppDslService(session)
import_service = AppDslService(cast(Session, db.session))
if existing_app is not None:
import_result = import_service.import_app(
account=account,
@@ -340,7 +332,7 @@ class MigrationImportService:
raise MigrationDataError(f"Workflow import failed: {error}")
if import_result.app_id is None:
raise MigrationDataError(f"Workflow import did not return an app id: {workflow_data.get('name')}")
session.commit()
db.session.commit()
return import_result.app_id
def _rewrite_workflow_dsl_provider_ids(self, dsl_content: str, id_mapping: dict[str, str]) -> str:
@@ -408,13 +400,13 @@ class MigrationImportService:
def _should_preserve_source_app_id(self, options: ImportOptions) -> bool:
return options.id_strategy == IdStrategy.PRESERVE_ID
def _find_existing_app(self, session: Session, app_id: str | None, tenant_id: str) -> App | None:
def _find_existing_app(self, app_id: str | None, tenant_id: str) -> App | None:
if not self._is_uuid_string(app_id):
return None
return session.scalar(sa.select(App).where(App.id == app_id, App.tenant_id == tenant_id))
return db.session.scalar(sa.select(App).where(App.id == app_id, App.tenant_id == tenant_id))
def _create_or_reuse_app_api_token(self, session: Session, app_id: str, tenant_id: str) -> None:
existing = session.scalar(
def _create_or_reuse_app_api_token(self, app_id: str, tenant_id: str) -> None:
existing = db.session.scalar(
sa.select(ApiToken).where(
ApiToken.type == ApiTokenType.APP,
ApiToken.app_id == app_id,
@@ -428,12 +420,11 @@ class MigrationImportService:
api_token.tenant_id = tenant_id
api_token.token = ApiToken.generate_api_key("app", 24)
api_token.type = ApiTokenType.APP
session.add(api_token)
session.commit()
db.session.add(api_token)
db.session.commit()
def _import_api_tools(
self,
session: Session,
package: MigrationPackage,
target: ImportTarget,
options: ImportOptions,
@@ -445,7 +436,7 @@ class MigrationImportService:
for tool_data in package.tools:
provider_name = self._required_string(tool_data, "provider_name", "api_tool")
schema = self._required_string(tool_data, "schema", "api_tool")
existing = session.scalar(
existing = db.session.scalar(
sa.select(ApiToolProvider).where(
ApiToolProvider.tenant_id == target.tenant_id,
ApiToolProvider.name == provider_name,
@@ -510,7 +501,7 @@ class MigrationImportService:
icon=icon,
)
status = "created"
target_provider = self._find_api_tool_provider(session, target.tenant_id, provider_name)
target_provider = self._find_api_tool_provider(target.tenant_id, provider_name)
if target_provider is not None:
self._record_id_mappings(
id_mapping,
@@ -522,8 +513,8 @@ class MigrationImportService:
)
report_items.append(ResourceReportItem(ResourceType.API_TOOL, provider_name, provider_name, status))
def _find_api_tool_provider(self, session: Session, tenant_id: str, provider_name: str) -> ApiToolProvider | None:
return session.scalar(
def _find_api_tool_provider(self, tenant_id: str, provider_name: str) -> ApiToolProvider | None:
return db.session.scalar(
sa.select(ApiToolProvider).where(
ApiToolProvider.tenant_id == tenant_id,
ApiToolProvider.name == provider_name,
@@ -558,7 +549,6 @@ class MigrationImportService:
def _import_workflow_tools(
self,
session: Session,
package: MigrationPackage,
target: ImportTarget,
options: ImportOptions,
@@ -568,13 +558,13 @@ class MigrationImportService:
) -> None:
if not package.workflow_tools:
return
account = session.get(Account, target.operator_id)
account = db.session.get(Account, target.operator_id)
if account is None:
raise MigrationDataError(f"Operator account not found: {target.operator_id}")
for workflow_tool_data in package.workflow_tools:
app_id = self._optional_string(workflow_tool_data.get("app_id"))
resolved_app_id = id_mapping.get(app_id or "", app_id)
if not resolved_app_id or self._find_existing_app(session, resolved_app_id, target.tenant_id) is None:
if not resolved_app_id or self._find_existing_app(resolved_app_id, target.tenant_id) is None:
report_items.append(
ResourceReportItem(
ResourceType.WORKFLOW_TOOL,
@@ -586,7 +576,7 @@ class MigrationImportService:
)
continue
try:
self._ensure_workflow_app_is_published(session, target, account, resolved_app_id)
self._ensure_workflow_app_is_published(target, account, resolved_app_id)
except Exception as exc:
report_items.append(
ResourceReportItem(
@@ -602,7 +592,7 @@ class MigrationImportService:
tool_name = self._required_string(workflow_tool_data, "name", "workflow_tool")
lookup_workflow_tool_id = workflow_tool_id if options.id_strategy == IdStrategy.PRESERVE_ID else None
existing = self._find_existing_workflow_tool(
session, target.tenant_id, lookup_workflow_tool_id, tool_name, resolved_app_id
target.tenant_id, lookup_workflow_tool_id, tool_name, resolved_app_id
)
if existing is not None and options.conflict_strategy == ConflictStrategy.FAIL:
raise MigrationDataError(f"Workflow tool already exists and conflict_strategy=fail: {tool_name}")
@@ -669,7 +659,7 @@ class MigrationImportService:
)
status = "created"
target_provider = self._find_existing_workflow_tool(
session, target.tenant_id, import_id or None, tool_name, resolved_app_id
target.tenant_id, import_id or None, tool_name, resolved_app_id
)
if target_provider is None:
raise MigrationDataError(f"Workflow tool was not created: {tool_name}")
@@ -685,10 +675,8 @@ class MigrationImportService:
)
report_items.append(ResourceReportItem(ResourceType.WORKFLOW_TOOL, identifier, tool_name, status))
def _ensure_workflow_app_is_published(
self, session: Session, target: ImportTarget, account: Account, app_id: str
) -> None:
app = self._find_existing_app(session, app_id, target.tenant_id)
def _ensure_workflow_app_is_published(self, target: ImportTarget, account: Account, app_id: str) -> None:
app = self._find_existing_app(app_id, target.tenant_id)
if app is None:
raise MigrationDataError(f"Referenced workflow app was not found in target tenant: {app_id}")
if app.workflow_id:
@@ -714,7 +702,6 @@ class MigrationImportService:
def _import_mcp_tools(
self,
session: Session,
package: MigrationPackage,
target: ImportTarget,
options: ImportOptions,
@@ -727,7 +714,7 @@ class MigrationImportService:
server_identifier = self._required_string(mcp_data, "server_identifier", "mcp_tool")
provider_id = self._optional_string(mcp_data.get("id"))
lookup_provider_id = provider_id if options.id_strategy == IdStrategy.PRESERVE_ID else None
existing = self._find_existing_mcp_tool(session, target.tenant_id, lookup_provider_id, server_identifier)
existing = self._find_existing_mcp_tool(target.tenant_id, lookup_provider_id, server_identifier)
if existing is not None and options.conflict_strategy == ConflictStrategy.FAIL:
raise MigrationDataError(f"MCP tool already exists and conflict_strategy=fail: {name}")
if existing is not None and options.conflict_strategy == ConflictStrategy.SKIP:
@@ -743,7 +730,7 @@ class MigrationImportService:
report_items.append(ResourceReportItem(ResourceType.MCP_TOOL, existing.id, name, "skipped"))
continue
service = MCPToolManageService(session=session)
service = MCPToolManageService(session=cast(Session, db.session))
configuration = MCPConfiguration.model_validate(mcp_data.get("configuration") or {})
authentication = (
MCPAuthentication.model_validate(mcp_data["authentication"]) if mcp_data.get("authentication") else None
@@ -765,7 +752,7 @@ class MigrationImportService:
# stored mode (update_provider now defaults to OFF when omitted).
identity_mode=IdentityMode(existing.identity_mode),
)
session.commit()
db.session.commit()
status = "updated"
identifier = existing.id
provider = existing
@@ -783,16 +770,14 @@ class MigrationImportService:
configuration=configuration,
authentication=authentication,
)
created_provider = self._find_existing_mcp_tool(
session, target.tenant_id, lookup_provider_id, server_identifier
)
created_provider = self._find_existing_mcp_tool(target.tenant_id, lookup_provider_id, server_identifier)
if created_provider is None:
raise MigrationDataError(f"MCP provider was not created: {name}")
status = "created"
provider = created_provider
identifier = provider.id
self._restore_mcp_provider_tools(provider, mcp_data)
session.commit()
db.session.commit()
if provider_id:
self._record_id_mappings(
id_mapping,
@@ -812,12 +797,12 @@ class MigrationImportService:
provider.authed = True
def _find_existing_mcp_tool(
self, session: Session, tenant_id: str, provider_id: str | None, server_identifier: str
self, tenant_id: str, provider_id: str | None, server_identifier: str
) -> MCPToolProvider | None:
predicates = [MCPToolProvider.server_identifier == server_identifier]
if self._is_uuid_string(provider_id):
predicates.append(MCPToolProvider.id == provider_id)
return session.scalar(
return db.session.scalar(
sa.select(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant_id, or_(*predicates)).limit(1)
)
@@ -831,26 +816,26 @@ class MigrationImportService:
return True
def _find_existing_workflow_tool(
self, session: Session, tenant_id: str, workflow_tool_id: str | None, tool_name: str, app_id: str
self, tenant_id: str, workflow_tool_id: str | None, tool_name: str, app_id: str
) -> WorkflowToolProvider | None:
predicates = [WorkflowToolProvider.name == tool_name, WorkflowToolProvider.app_id == app_id]
if self._is_uuid_string(workflow_tool_id):
predicates.append(WorkflowToolProvider.id == workflow_tool_id)
return session.scalar(
return db.session.scalar(
sa.select(WorkflowToolProvider)
.where(WorkflowToolProvider.tenant_id == tenant_id, or_(*predicates))
.limit(1)
)
def _preflight_dependency_only_mcp(
self, session: Session, package: MigrationPackage, target: ImportTarget, report_items: list[ResourceReportItem]
self, package: MigrationPackage, target: ImportTarget, report_items: list[ResourceReportItem]
) -> None:
for dependency in package.dependencies:
if dependency.get("kind") != DependencyKind.MCP_TOOL.value:
continue
provider_id = str(dependency.get("provider_id", dependency.get("id", "")))
provider_name = self._optional_string(dependency.get("provider_name") or dependency.get("name"))
existing = self._find_dependency_only_mcp_provider(session, target.tenant_id, provider_id, provider_name)
existing = self._find_dependency_only_mcp_provider(target.tenant_id, provider_id, provider_name)
report_name = f"mcp_tool {provider_name or getattr(existing, 'name', None) or provider_id}"
if existing is not None:
report_items.append(
@@ -879,12 +864,12 @@ class MigrationImportService:
)
def _find_dependency_only_mcp_provider(
self, session: Session, tenant_id: str, provider_id: str, provider_name: str | None
self, tenant_id: str, provider_id: str, provider_name: str | None
) -> MCPToolProvider | None:
predicates = [MCPToolProvider.server_identifier == provider_id]
if self._is_uuid_string(provider_id):
predicates.append(MCPToolProvider.id == provider_id)
return session.scalar(
return db.session.scalar(
sa.select(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant_id, or_(*predicates)).limit(1)
)
+6 -14
View File
@@ -34,7 +34,6 @@ from graphon.model_runtime.model_providers.base.text_embedding_model import Text
from libs import helper
from libs.datetime_utils import naive_utc_now
from libs.login import current_user
from libs.pagination import paginate_query
from models import Account, TenantAccountRole
from models.dataset import (
AppDatasetJoin,
@@ -356,7 +355,7 @@ class DatasetService:
else:
return [], 0
datasets = paginate_query(query, page=page, per_page=per_page, max_per_page=100)
datasets = db.paginate(select=query, page=page, per_page=per_page, max_per_page=100, error_out=False)
return datasets.items, datasets.total
@@ -400,7 +399,7 @@ class DatasetService:
accessible_filter = sa.or_(Dataset.maintainer == user.id, accessible_filter)
stmt = stmt.where(accessible_filter)
datasets = paginate_query(stmt, page=1, per_page=len(ids), max_per_page=len(ids))
datasets = db.paginate(select=stmt, page=1, per_page=len(ids), max_per_page=len(ids), error_out=False)
return datasets.items, datasets.total
@@ -1404,7 +1403,7 @@ class DatasetService:
def get_dataset_queries(dataset_id: str, page: int, per_page: int):
stmt = select(DatasetQuery).filter_by(dataset_id=dataset_id).order_by(db.desc(DatasetQuery.created_at))
dataset_queries = paginate_query(stmt, page=page, per_page=per_page, max_per_page=100)
dataset_queries = db.paginate(select=stmt, page=page, per_page=per_page, max_per_page=100, error_out=False)
return dataset_queries.items, dataset_queries.total
@@ -3872,14 +3871,7 @@ class SegmentService:
session.add(document)
# Delete database records
session.execute(
delete(DocumentSegment).where(
DocumentSegment.id.in_(segment_db_ids),
DocumentSegment.dataset_id == dataset.id,
DocumentSegment.document_id == document.id,
DocumentSegment.tenant_id == current_user.current_tenant_id,
)
)
session.execute(delete(DocumentSegment).where(DocumentSegment.id.in_(segment_ids)))
session.commit()
@classmethod
@@ -4128,7 +4120,7 @@ class SegmentService:
if keyword:
escaped_keyword = helper.escape_like_pattern(keyword)
query = query.where(ChildChunk.content.ilike(f"%{escaped_keyword}%", escape="\\"))
return paginate_query(query, page=page, per_page=limit, max_per_page=100)
return db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
@classmethod
def get_child_chunk_by_id(
@@ -4180,7 +4172,7 @@ class SegmentService:
query = query.where(DocumentSegment.content.ilike(f"%{escaped_keyword}%", escape="\\"))
query = query.order_by(DocumentSegment.position.asc(), DocumentSegment.id.asc())
paginated_segments = paginate_query(query, page=page, per_page=limit, max_per_page=100)
paginated_segments = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
return paginated_segments.items, paginated_segments.total
+4 -2
View File
@@ -10,9 +10,9 @@ from sqlalchemy.orm import Session
from constants import HIDDEN_VALUE
from core.helper import ssrf_proxy
from core.rag.entities import MetadataFilteringCondition
from extensions.ext_database import db
from graphon.nodes.http_request.exc import InvalidHttpMethodError
from libs.datetime_utils import naive_utc_now
from libs.pagination import paginate_query
from models.dataset import (
Dataset,
ExternalKnowledgeApis,
@@ -42,7 +42,9 @@ class ExternalDatasetService:
escaped_search = escape_like_pattern(search)
query = query.where(ExternalKnowledgeApis.name.ilike(f"%{escaped_search}%", escape="\\"))
external_knowledge_apis = paginate_query(query, page=page, per_page=per_page, max_per_page=100)
external_knowledge_apis = db.paginate(
select=query, page=page, per_page=per_page, max_per_page=100, error_out=False
)
return external_knowledge_apis.items, external_knowledge_apis.total
+1 -2
View File
@@ -25,7 +25,6 @@ from core.plugin.impl.plugin import PluginInstaller
from core.plugin.plugin_service import PluginService
from core.tools.entities.tool_entities import ToolProviderType
from extensions.ext_database import db
from libs.pagination import paginate_query
from models.account import Tenant
from models.model import App, AppMode, AppModelConfig
from models.provider_ids import ModelProviderID, ToolProviderID
@@ -500,7 +499,7 @@ class PluginMigration:
total_failed_tenant = 0
while True:
# paginate
tenants = paginate_query(sa.select(Tenant).order_by(Tenant.created_at.desc()), page=page, per_page=100)
tenants = db.paginate(sa.select(Tenant).order_by(Tenant.created_at.desc()), page=page, per_page=100)
if tenants.items is None or len(tenants.items) == 0:
break
@@ -1,11 +1,10 @@
from collections.abc import Mapping
from typing import Any
from sqlalchemy.orm import Session
from configs import dify_config
from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
from core.app.entities.app_invoke_entities import InvokeFrom
from extensions.ext_database import db
from models.dataset import Document, Pipeline
from models.enums import IndexingStatus
from models.model import Account, App, EndUser
@@ -17,7 +16,6 @@ class PipelineGenerateService:
@classmethod
def generate(
cls,
session: Session,
pipeline: Pipeline,
user: Account | EndUser,
args: Mapping[str, Any],
@@ -37,7 +35,7 @@ class PipelineGenerateService:
workflow = cls._get_workflow(pipeline, invoke_from)
if original_document_id := args.get("original_document_id"):
# update document status to waiting
cls.update_document_status(original_document_id, session)
cls.update_document_status(original_document_id)
return PipelineGenerator.convert_to_event_stream(
PipelineGenerator().generate(
pipeline=pipeline,
@@ -107,12 +105,13 @@ class PipelineGenerateService:
return workflow
@classmethod
def update_document_status(cls, document_id: str, session: Session):
def update_document_status(cls, document_id: str):
"""
Update document status to waiting
:param document_id: document id
"""
document = session.get(Document, document_id)
document = db.session.get(Document, document_id)
if document:
document.indexing_status = IndexingStatus.WAITING
session.add(document)
db.session.add(document)
db.session.commit()
@@ -4,7 +4,6 @@ from pathlib import Path
from typing import Any, override
from flask import current_app
from sqlalchemy.orm import Session
from services.rag_pipeline.pipeline_template.pipeline_template_base import PipelineTemplateRetrievalBase
from services.rag_pipeline.pipeline_template.pipeline_template_type import PipelineTemplateType
@@ -22,15 +21,13 @@ class BuiltInPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
return PipelineTemplateType.BUILTIN
@override
def get_pipeline_templates(
self, session: Session, language: str, current_tenant_id: str | None = None
) -> dict[str, Any]:
def get_pipeline_templates(self, language: str, current_tenant_id: str | None = None) -> dict[str, Any]:
del current_tenant_id
result = self.fetch_pipeline_templates_from_builtin(language)
return result
@override
def get_pipeline_template_detail(self, session: Session, template_id: str) -> dict[str, Any] | None:
def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
result = self.fetch_pipeline_template_detail_from_builtin(template_id)
return result
@@ -2,8 +2,8 @@ from typing import Any, TypedDict, override
import yaml
from sqlalchemy import select
from sqlalchemy.orm import Session
from extensions.ext_database import db
from libs.login import resolve_tenant_id_fallback
from models.dataset import PipelineCustomizedTemplate
from services.rag_pipeline.pipeline_template.pipeline_template_base import PipelineTemplateRetrievalBase
@@ -40,38 +40,29 @@ class CustomizedPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
"""
@override
def get_pipeline_templates(
self, session: Session, language: str, current_tenant_id: str | None = None
) -> dict[str, Any]:
def get_pipeline_templates(self, language: str, current_tenant_id: str | None = None) -> dict[str, Any]:
current_tenant_id = resolve_tenant_id_fallback(current_tenant_id)
return self.fetch_pipeline_templates_from_customized(
session=session, tenant_id=current_tenant_id, language=language
)
return self.fetch_pipeline_templates_from_customized(tenant_id=current_tenant_id, language=language)
@override
def get_pipeline_template_detail(self, session: Session, template_id: str) -> dict[str, Any] | None:
return self.fetch_pipeline_template_detail_from_db(session, template_id)
def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
return self.fetch_pipeline_template_detail_from_db(template_id)
@override
def get_type(self) -> str:
return PipelineTemplateType.CUSTOMIZED
@classmethod
def fetch_pipeline_templates_from_customized(
cls, session: Session, tenant_id: str, language: str
) -> dict[str, Any]:
def fetch_pipeline_templates_from_customized(cls, tenant_id: str, language: str) -> dict[str, Any]:
"""
Fetch pipeline templates from db.
:param tenant_id: tenant id
:param language: language
:return:
"""
pipeline_customized_templates = session.scalars(
pipeline_customized_templates = db.session.scalars(
select(PipelineCustomizedTemplate)
.where(
PipelineCustomizedTemplate.tenant_id == tenant_id,
PipelineCustomizedTemplate.language == language,
)
.where(PipelineCustomizedTemplate.tenant_id == tenant_id, PipelineCustomizedTemplate.language == language)
.order_by(PipelineCustomizedTemplate.position.asc(), PipelineCustomizedTemplate.created_at.desc())
).all()
recommended_pipelines_results: list[CustomizedTemplateItemDict] = []
@@ -89,13 +80,13 @@ class CustomizedPipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
return {"pipeline_templates": recommended_pipelines_results}
@classmethod
def fetch_pipeline_template_detail_from_db(cls, session: Session, template_id: str) -> dict[str, Any] | None:
def fetch_pipeline_template_detail_from_db(cls, template_id: str) -> dict[str, Any] | None:
"""
Fetch pipeline template detail from db.
:param template_id: Template ID
:return:
"""
pipeline_template = session.get(PipelineCustomizedTemplate, template_id)
pipeline_template = db.session.get(PipelineCustomizedTemplate, template_id)
if not pipeline_template:
return None
@@ -2,8 +2,8 @@ from typing import Any, TypedDict, override
import yaml
from sqlalchemy import select
from sqlalchemy.orm import Session
from extensions.ext_database import db
from models.dataset import PipelineBuiltInTemplate
from services.rag_pipeline.pipeline_template.pipeline_template_base import PipelineTemplateRetrievalBase
from services.rag_pipeline.pipeline_template.pipeline_template_type import PipelineTemplateType
@@ -40,22 +40,20 @@ class DatabasePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
"""
@override
def get_pipeline_templates(
self, session: Session, language: str, current_tenant_id: str | None = None
) -> dict[str, Any]:
def get_pipeline_templates(self, language: str, current_tenant_id: str | None = None) -> dict[str, Any]:
del current_tenant_id
return self.fetch_pipeline_templates_from_db(session, language)
return self.fetch_pipeline_templates_from_db(language)
@override
def get_pipeline_template_detail(self, session: Session, template_id: str) -> dict[str, Any] | None:
return self.fetch_pipeline_template_detail_from_db(session, template_id)
def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
return self.fetch_pipeline_template_detail_from_db(template_id)
@override
def get_type(self) -> str:
return PipelineTemplateType.DATABASE
@classmethod
def fetch_pipeline_templates_from_db(cls, session: Session, language: str) -> dict[str, Any]:
def fetch_pipeline_templates_from_db(cls, language: str) -> dict[str, Any]:
"""
Fetch pipeline templates from db.
:param language: language
@@ -63,7 +61,9 @@ class DatabasePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
"""
pipeline_built_in_templates = list(
session.scalars(select(PipelineBuiltInTemplate).where(PipelineBuiltInTemplate.language == language)).all()
db.session.scalars(
select(PipelineBuiltInTemplate).where(PipelineBuiltInTemplate.language == language)
).all()
)
recommended_pipelines_results: list[PipelineTemplateItemDict] = []
@@ -83,14 +83,14 @@ class DatabasePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
return {"pipeline_templates": recommended_pipelines_results}
@classmethod
def fetch_pipeline_template_detail_from_db(cls, session: Session, template_id: str) -> dict[str, Any] | None:
def fetch_pipeline_template_detail_from_db(cls, template_id: str) -> dict[str, Any] | None:
"""
Fetch pipeline template detail from db.
:param pipeline_id: Pipeline ID
:return:
"""
# is in public recommended list
pipeline_template = session.get(PipelineBuiltInTemplate, template_id)
pipeline_template = db.session.get(PipelineBuiltInTemplate, template_id)
if not pipeline_template:
return None
@@ -1,15 +1,11 @@
from typing import Any, Protocol
from sqlalchemy.orm import Session
class PipelineTemplateRetrievalBase(Protocol):
"""Interface for pipeline template retrieval."""
def get_pipeline_templates(
self, session: Session, language: str, current_tenant_id: str | None = None
) -> dict[str, Any]: ...
def get_pipeline_templates(self, language: str, current_tenant_id: str | None = None) -> dict[str, Any]: ...
def get_pipeline_template_detail(self, session: Session, template_id: str) -> dict[str, Any] | None: ...
def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None: ...
def get_type(self) -> str: ...
@@ -2,7 +2,6 @@ import logging
from typing import Any, override
import httpx
from sqlalchemy.orm import Session
from configs import dify_config
from services.rag_pipeline.pipeline_template.database.database_retrieval import DatabasePipelineTemplateRetrieval
@@ -18,23 +17,21 @@ class RemotePipelineTemplateRetrieval(PipelineTemplateRetrievalBase):
"""
@override
def get_pipeline_template_detail(self, session: Session, template_id: str) -> dict[str, Any] | None:
def get_pipeline_template_detail(self, template_id: str) -> dict[str, Any] | None:
try:
return self.fetch_pipeline_template_detail_from_dify_official(template_id)
except Exception as e:
logger.warning("fetch recommended app detail from dify official failed: %r, switch to database.", e)
return DatabasePipelineTemplateRetrieval.fetch_pipeline_template_detail_from_db(session, template_id)
return DatabasePipelineTemplateRetrieval.fetch_pipeline_template_detail_from_db(template_id)
@override
def get_pipeline_templates(
self, session: Session, language: str, current_tenant_id: str | None = None
) -> dict[str, Any]:
def get_pipeline_templates(self, language: str, current_tenant_id: str | None = None) -> dict[str, Any]:
del current_tenant_id
try:
return self.fetch_pipeline_templates_from_dify_official(language)
except Exception as e:
logger.warning("fetch pipeline templates from dify official failed: %r, switch to database.", e)
return DatabasePipelineTemplateRetrieval.fetch_pipeline_templates_from_db(session, language)
return DatabasePipelineTemplateRetrieval.fetch_pipeline_templates_from_db(language)
@override
def get_type(self) -> str:
+291 -388
View File
@@ -27,7 +27,6 @@ from core.datasource.entities.datasource_entities import (
from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
from core.datasource.website_crawl.website_crawl_plugin import WebsiteCrawlDatasourcePlugin
from core.db.session_factory import session_factory
from core.helper import marketplace
from core.rag.entities import DatasourceCompletedEvent, DatasourceErrorEvent, DatasourceProcessingEvent
from core.repositories.factory import DifyCoreRepositoryFactory, OrderConfig
@@ -99,8 +98,7 @@ class RagPipelineService:
def __init__(self, session_maker: sessionmaker | None = None):
"""Initialize RagPipelineService with repository dependencies."""
if session_maker is None:
session_maker = session_factory.get_session_maker()
self._session_maker = session_maker
session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
self._node_execution_service_repo = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository(
session_maker
)
@@ -109,7 +107,6 @@ class RagPipelineService:
@classmethod
def get_pipeline_templates(
cls,
session: Session,
type: str = "built-in",
language: str = "en-US",
current_tenant_id: str | None = None,
@@ -117,7 +114,7 @@ class RagPipelineService:
if type == "built-in":
mode = dify_config.HOSTED_FETCH_PIPELINE_TEMPLATES_MODE
retrieval_instance = PipelineTemplateRetrievalFactory.get_pipeline_template_factory(mode)()
result = retrieval_instance.get_pipeline_templates(session, language, current_tenant_id)
result = retrieval_instance.get_pipeline_templates(language, current_tenant_id)
if not result.get("pipeline_templates") and language != "en-US":
template_retrieval = PipelineTemplateRetrievalFactory.get_built_in_pipeline_template_retrieval()
result = template_retrieval.fetch_pipeline_templates_from_builtin("en-US")
@@ -125,13 +122,11 @@ class RagPipelineService:
else:
mode = "customized"
retrieval_instance = PipelineTemplateRetrievalFactory.get_pipeline_template_factory(mode)()
result = retrieval_instance.get_pipeline_templates(session, language, current_tenant_id)
result = retrieval_instance.get_pipeline_templates(language, current_tenant_id)
return result
@classmethod
def get_pipeline_template_detail(
cls, session: Session, template_id: str, type: str = "built-in"
) -> dict[str, Any] | None:
def get_pipeline_template_detail(cls, template_id: str, type: str = "built-in") -> dict[str, Any] | None:
"""
Get pipeline template detail.
@@ -142,9 +137,7 @@ class RagPipelineService:
if type == "built-in":
mode = dify_config.HOSTED_FETCH_PIPELINE_TEMPLATES_MODE
retrieval_instance = PipelineTemplateRetrievalFactory.get_pipeline_template_factory(mode)()
built_in_result: dict[str, Any] | None = retrieval_instance.get_pipeline_template_detail(
session, template_id
)
built_in_result: dict[str, Any] | None = retrieval_instance.get_pipeline_template_detail(template_id)
if built_in_result is None:
logger.warning(
"pipeline template retrieval returned empty result, template_id: %s, mode: %s",
@@ -155,9 +148,7 @@ class RagPipelineService:
else:
mode = "customized"
retrieval_instance = PipelineTemplateRetrievalFactory.get_pipeline_template_factory(mode)()
customized_result: dict[str, Any] | None = retrieval_instance.get_pipeline_template_detail(
session, template_id
)
customized_result: dict[str, Any] | None = retrieval_instance.get_pipeline_template_detail(template_id)
return customized_result
@classmethod
@@ -167,7 +158,6 @@ class RagPipelineService:
template_info: PipelineTemplateInfoEntity,
current_user: Account | None = None,
current_tenant_id: str | None = None,
session: Session | None = None,
):
"""
Update pipeline template.
@@ -175,17 +165,7 @@ class RagPipelineService:
:param template_info: template info
"""
current_user, current_tenant_id = resolve_account_fallback(current_user, current_tenant_id)
if session is None:
with session_factory.get_session_maker().begin() as new_session:
return cls.update_customized_pipeline_template(
template_id,
template_info,
current_user,
current_tenant_id,
session=new_session,
)
customized_template: PipelineCustomizedTemplate | None = session.scalar(
customized_template: PipelineCustomizedTemplate | None = db.session.scalar(
select(PipelineCustomizedTemplate)
.where(
PipelineCustomizedTemplate.id == template_id,
@@ -198,7 +178,7 @@ class RagPipelineService:
# check template name is exist
template_name = template_info.name
if template_name:
template = session.scalar(
template = db.session.scalar(
select(PipelineCustomizedTemplate)
.where(
PipelineCustomizedTemplate.name == template_name,
@@ -213,22 +193,16 @@ class RagPipelineService:
customized_template.description = template_info.description
customized_template.icon = template_info.icon_info.model_dump()
customized_template.updated_by = current_user.id
db.session.commit()
return customized_template
@classmethod
def delete_customized_pipeline_template(
cls, template_id: str, current_tenant_id: str | None = None, session: Session | None = None
):
def delete_customized_pipeline_template(cls, template_id: str, current_tenant_id: str | None = None):
"""
Delete customized pipeline template.
"""
current_tenant_id = resolve_tenant_id_fallback(current_tenant_id)
if session is None:
with session_factory.get_session_maker().begin() as new_session:
cls.delete_customized_pipeline_template(template_id, current_tenant_id, session=new_session)
return
customized_template: PipelineCustomizedTemplate | None = session.scalar(
customized_template: PipelineCustomizedTemplate | None = db.session.scalar(
select(PipelineCustomizedTemplate)
.where(
PipelineCustomizedTemplate.id == template_id,
@@ -238,23 +212,23 @@ class RagPipelineService:
)
if not customized_template:
raise ValueError("Customized pipeline template not found.")
session.delete(customized_template)
db.session.delete(customized_template)
db.session.commit()
def get_draft_workflow(self, pipeline: Pipeline) -> Workflow | None:
"""
Get draft workflow
"""
# fetch draft workflow by rag pipeline
with self._session_maker() as session:
workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.version == "draft",
)
.limit(1)
workflow = db.session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.version == "draft",
)
.limit(1)
)
# return draft workflow
return workflow
@@ -268,31 +242,29 @@ class RagPipelineService:
return None
# fetch published workflow by workflow_id
with self._session_maker() as session:
workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.id == pipeline.workflow_id,
)
.limit(1)
workflow = db.session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.id == pipeline.workflow_id,
)
.limit(1)
)
return workflow
def get_published_workflow_by_id(self, pipeline: Pipeline, workflow_id: str) -> Workflow | None:
"""Fetch a published workflow snapshot by ID for restore operations."""
with self._session_maker() as session:
workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.id == workflow_id,
)
.limit(1)
workflow = db.session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.id == workflow_id,
)
.limit(1)
)
if workflow and workflow.version == Workflow.VERSION_DRAFT:
raise IsDraftWorkflowError("source workflow must be published")
return workflow
@@ -350,51 +322,39 @@ class RagPipelineService:
Sync draft workflow
:raises WorkflowHashNotEqualError
"""
with self._session_maker.begin() as session:
managed_pipeline = session.get(Pipeline, pipeline.id)
if not managed_pipeline:
raise ValueError("Pipeline not found")
# fetch draft workflow by app_model
workflow = self.get_draft_workflow(pipeline=pipeline)
# fetch draft workflow by app_model
workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == managed_pipeline.tenant_id,
Workflow.app_id == managed_pipeline.id,
Workflow.version == "draft",
)
.limit(1)
if workflow and workflow.unique_hash != unique_hash:
raise WorkflowHashNotEqualError()
# create draft workflow if not found
if not workflow:
workflow = Workflow(
tenant_id=pipeline.tenant_id,
app_id=pipeline.id,
features="{}",
type=WorkflowType.RAG_PIPELINE.value,
version="draft",
graph=json.dumps(graph),
created_by=account.id,
environment_variables=environment_variables,
conversation_variables=conversation_variables,
rag_pipeline_variables=rag_pipeline_variables,
)
if workflow and workflow.unique_hash != unique_hash:
raise WorkflowHashNotEqualError()
# create draft workflow if not found
if not workflow:
workflow = Workflow(
tenant_id=managed_pipeline.tenant_id,
app_id=managed_pipeline.id,
features="{}",
type=WorkflowType.RAG_PIPELINE.value,
version="draft",
graph=json.dumps(graph),
created_by=account.id,
environment_variables=environment_variables,
conversation_variables=conversation_variables,
rag_pipeline_variables=rag_pipeline_variables,
)
session.add(workflow)
session.flush()
managed_pipeline.workflow_id = workflow.id
pipeline.workflow_id = workflow.id
# update draft workflow if found
else:
workflow.graph = json.dumps(graph)
workflow.updated_by = account.id
workflow.updated_at = datetime.now(UTC).replace(tzinfo=None)
workflow.environment_variables = environment_variables
workflow.conversation_variables = conversation_variables
workflow.rag_pipeline_variables = rag_pipeline_variables
db.session.add(workflow)
db.session.flush()
pipeline.workflow_id = workflow.id
# update draft workflow if found
else:
workflow.graph = json.dumps(graph)
workflow.updated_by = account.id
workflow.updated_at = datetime.now(UTC).replace(tzinfo=None)
workflow.environment_variables = environment_variables
workflow.conversation_variables = conversation_variables
workflow.rag_pipeline_variables = rag_pipeline_variables
# commit db session changes
db.session.commit()
# trigger workflow events TODO
# app_draft_workflow_was_synced.send(pipeline, synced_draft_workflow=workflow)
@@ -415,48 +375,26 @@ class RagPipelineService:
the pipeline-specific flush/link step that wires a newly created draft
back onto ``pipeline.workflow_id``.
"""
with self._session_maker.begin() as session:
managed_pipeline = session.get(Pipeline, pipeline.id)
if not managed_pipeline:
raise ValueError("Pipeline not found")
source_workflow = self.get_published_workflow_by_id(pipeline=pipeline, workflow_id=workflow_id)
if not source_workflow:
raise WorkflowNotFoundError("Workflow not found.")
source_workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == managed_pipeline.tenant_id,
Workflow.app_id == managed_pipeline.id,
Workflow.id == workflow_id,
)
.limit(1)
)
if source_workflow and source_workflow.version == Workflow.VERSION_DRAFT:
raise IsDraftWorkflowError("source workflow must be published")
if not source_workflow:
raise WorkflowNotFoundError("Workflow not found.")
draft_workflow = self.get_draft_workflow(pipeline=pipeline)
draft_workflow, is_new_draft = apply_published_workflow_snapshot_to_draft(
tenant_id=pipeline.tenant_id,
app_id=pipeline.id,
source_workflow=source_workflow,
draft_workflow=draft_workflow,
account=account,
updated_at_factory=lambda: datetime.now(UTC).replace(tzinfo=None),
)
draft_workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == managed_pipeline.tenant_id,
Workflow.app_id == managed_pipeline.id,
Workflow.version == Workflow.VERSION_DRAFT,
)
.limit(1)
)
draft_workflow, is_new_draft = apply_published_workflow_snapshot_to_draft(
tenant_id=managed_pipeline.tenant_id,
app_id=managed_pipeline.id,
source_workflow=source_workflow,
draft_workflow=draft_workflow,
account=account,
updated_at_factory=lambda: datetime.now(UTC).replace(tzinfo=None),
)
if is_new_draft:
db.session.add(draft_workflow)
db.session.flush()
pipeline.workflow_id = draft_workflow.id
if is_new_draft:
session.add(draft_workflow)
session.flush()
managed_pipeline.workflow_id = draft_workflow.id
pipeline.workflow_id = draft_workflow.id
db.session.commit()
return draft_workflow
@@ -633,7 +571,7 @@ class RagPipelineService:
workflow_node_execution.id
)
with self._session_maker.begin() as session:
with sessionmaker(bind=db.engine).begin() as session:
draft_var_saver = DraftVariableSaver(
session=session,
app_id=pipeline.id,
@@ -1050,22 +988,23 @@ class RagPipelineService:
dataset_id = get_system_segment(variable_pool, SystemVariableKey.DATASET_ID)
pipeline_id = get_system_segment(variable_pool, SystemVariableKey.APP_ID)
if document_id and dataset_id and pipeline_id:
with self._session_maker.begin() as session:
document = session.scalar(
select(Document)
.join(Dataset, Dataset.id == Document.dataset_id)
.where(
Document.id == document_id.value,
Document.tenant_id == tenant_id,
Document.dataset_id == dataset_id.value,
Dataset.tenant_id == tenant_id,
Dataset.pipeline_id == pipeline_id.value,
)
.limit(1)
document = db.session.scalar(
select(Document)
.join(Dataset, Dataset.id == Document.dataset_id)
.where(
Document.id == document_id.value,
Document.tenant_id == tenant_id,
Document.dataset_id == dataset_id.value,
Dataset.tenant_id == tenant_id,
Dataset.pipeline_id == pipeline_id.value,
)
if document:
document.indexing_status = IndexingStatus.ERROR
document.error = error
.limit(1)
)
if document:
document.indexing_status = IndexingStatus.ERROR
document.error = error
db.session.add(document)
db.session.commit()
return workflow_node_execution
@@ -1281,81 +1220,82 @@ class RagPipelineService:
Publish customized pipeline template
"""
current_user, _ = resolve_account_fallback(current_user, current_tenant_id)
with session_factory.get_session_maker().begin() as session:
pipeline = session.get(Pipeline, pipeline_id)
if not pipeline:
raise ValueError("Pipeline not found")
if not pipeline.workflow_id:
raise ValueError("Pipeline workflow not found")
workflow = session.get(Workflow, pipeline.workflow_id)
if not workflow:
raise ValueError("Workflow not found")
pipeline = db.session.get(Pipeline, pipeline_id)
if not pipeline:
raise ValueError("Pipeline not found")
if not pipeline.workflow_id:
raise ValueError("Pipeline workflow not found")
workflow = db.session.get(Workflow, pipeline.workflow_id)
if not workflow:
raise ValueError("Workflow not found")
with sessionmaker(db.engine).begin() as session:
dataset = pipeline.retrieve_dataset(session=session)
if not dataset:
raise ValueError("Dataset not found")
# check template name is exist
template_name = args.get("name")
if template_name:
template = session.scalar(
select(PipelineCustomizedTemplate)
.where(
PipelineCustomizedTemplate.name == template_name,
PipelineCustomizedTemplate.tenant_id == pipeline.tenant_id,
)
.limit(1)
)
if template:
raise ValueError("Template name is already exists")
max_position = session.scalar(
select(func.max(PipelineCustomizedTemplate.position)).where(
PipelineCustomizedTemplate.tenant_id == pipeline.tenant_id
# check template name is exist
template_name = args.get("name")
if template_name:
template = db.session.scalar(
select(PipelineCustomizedTemplate)
.where(
PipelineCustomizedTemplate.name == template_name,
PipelineCustomizedTemplate.tenant_id == pipeline.tenant_id,
)
.limit(1)
)
if template:
raise ValueError("Template name is already exists")
from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
max_position = db.session.scalar(
select(func.max(PipelineCustomizedTemplate.position)).where(
PipelineCustomizedTemplate.tenant_id == pipeline.tenant_id
)
)
from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
with sessionmaker(db.engine).begin() as session:
rag_pipeline_dsl_service = RagPipelineDslService(session)
dsl = rag_pipeline_dsl_service.export_rag_pipeline_dsl(pipeline=pipeline, include_secret=True)
if args.get("icon_info") is None:
args["icon_info"] = {}
if args.get("description") is None:
raise ValueError("Description is required")
if args.get("name") is None:
raise ValueError("Name is required")
pipeline_customized_template = PipelineCustomizedTemplate(
name=args.get("name") or "",
description=args.get("description") or "",
icon=args.get("icon_info") or {},
tenant_id=pipeline.tenant_id,
yaml_content=dsl,
install_count=0,
position=max_position + 1 if max_position else 1,
chunk_structure=dataset.chunk_structure,
language="en-US",
created_by=current_user.id,
)
session.add(pipeline_customized_template)
if args.get("icon_info") is None:
args["icon_info"] = {}
if args.get("description") is None:
raise ValueError("Description is required")
if args.get("name") is None:
raise ValueError("Name is required")
pipeline_customized_template = PipelineCustomizedTemplate(
name=args.get("name") or "",
description=args.get("description") or "",
icon=args.get("icon_info") or {},
tenant_id=pipeline.tenant_id,
yaml_content=dsl,
install_count=0,
position=max_position + 1 if max_position else 1,
chunk_structure=dataset.chunk_structure,
language="en-US",
created_by=current_user.id,
)
db.session.add(pipeline_customized_template)
db.session.commit()
def is_workflow_exist(self, pipeline: Pipeline) -> bool:
with self._session_maker() as session:
return (
session.scalar(
select(func.count(Workflow.id)).where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.version == Workflow.VERSION_DRAFT,
)
return (
db.session.scalar(
select(func.count(Workflow.id)).where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.version == Workflow.VERSION_DRAFT,
)
or 0
) > 0
)
or 0
) > 0
def get_node_last_run(
self, pipeline: Pipeline, workflow: Workflow, node_id: str
) -> WorkflowNodeExecutionModel | None:
node_execution_service_repo = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository(
self._session_maker
sessionmaker(db.engine)
)
node_exec = node_execution_service_repo.get_node_last_execution(
@@ -1431,7 +1371,7 @@ class RagPipelineService:
# Convert node_execution to WorkflowNodeExecution after save
workflow_node_execution_db_model = repository._to_db_model(workflow_node_execution) # type: ignore
with self._session_maker.begin() as session:
with sessionmaker(bind=db.engine).begin() as session:
draft_var_saver = DraftVariableSaver(
session=session,
app_id=pipeline.id,
@@ -1465,10 +1405,7 @@ class RagPipelineService:
if type and type != "all":
stmt = stmt.where(PipelineRecommendedPlugin.type == type)
with self._session_maker() as session:
pipeline_recommended_plugins = session.scalars(
stmt.order_by(PipelineRecommendedPlugin.position.asc())
).all()
pipeline_recommended_plugins = db.session.scalars(stmt.order_by(PipelineRecommendedPlugin.position.asc())).all()
if not pipeline_recommended_plugins:
return {
@@ -1507,173 +1444,41 @@ class RagPipelineService:
"""
Retry error document
"""
with self._session_maker() as session:
document_pipeline_execution_log = session.scalar(
select(DocumentPipelineExecutionLog)
.where(DocumentPipelineExecutionLog.document_id == document.id)
.limit(1)
)
if not document_pipeline_execution_log:
raise ValueError("Document pipeline execution log not found")
pipeline = session.get(Pipeline, document_pipeline_execution_log.pipeline_id)
if not pipeline:
raise ValueError("Pipeline not found")
# convert to app config
workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.id == pipeline.workflow_id,
)
.limit(1)
)
if not workflow:
raise ValueError("Workflow not found")
PipelineGenerator().generate(
pipeline=pipeline,
workflow=workflow,
user=user,
args={
"inputs": document_pipeline_execution_log.input_data,
"start_node_id": document_pipeline_execution_log.datasource_node_id,
"datasource_type": document_pipeline_execution_log.datasource_type,
"datasource_info_list": [json.loads(document_pipeline_execution_log.datasource_info)],
"original_document_id": document.id,
},
invoke_from=InvokeFrom.PUBLISHED_PIPELINE,
streaming=False,
call_depth=0,
workflow_thread_pool_id=None,
is_retry=True,
)
document_pipeline_execution_log = db.session.scalar(
select(DocumentPipelineExecutionLog).where(DocumentPipelineExecutionLog.document_id == document.id).limit(1)
)
if not document_pipeline_execution_log:
raise ValueError("Document pipeline execution log not found")
pipeline = db.session.get(Pipeline, document_pipeline_execution_log.pipeline_id)
if not pipeline:
raise ValueError("Pipeline not found")
# convert to app config
workflow = self.get_published_workflow(pipeline)
if not workflow:
raise ValueError("Workflow not found")
PipelineGenerator().generate(
pipeline=pipeline,
workflow=workflow,
user=user,
args={
"inputs": document_pipeline_execution_log.input_data,
"start_node_id": document_pipeline_execution_log.datasource_node_id,
"datasource_type": document_pipeline_execution_log.datasource_type,
"datasource_info_list": [json.loads(document_pipeline_execution_log.datasource_info)],
"original_document_id": document.id,
},
invoke_from=InvokeFrom.PUBLISHED_PIPELINE,
streaming=False,
call_depth=0,
workflow_thread_pool_id=None,
is_retry=True,
)
def get_datasource_plugins(self, tenant_id: str, dataset_id: str, is_published: bool) -> list[dict]:
"""
Get datasource plugins
"""
with self._session_maker() as session:
dataset: Dataset | None = session.scalar(
select(Dataset)
.where(
Dataset.id == dataset_id,
Dataset.tenant_id == tenant_id,
)
.limit(1)
)
if not dataset:
raise ValueError("Dataset not found")
pipeline: Pipeline | None = session.scalar(
select(Pipeline)
.where(
Pipeline.id == dataset.pipeline_id,
Pipeline.tenant_id == tenant_id,
)
.limit(1)
)
if not pipeline:
raise ValueError("Pipeline not found")
if is_published:
workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.id == pipeline.workflow_id,
)
.limit(1)
)
else:
workflow = session.scalar(
select(Workflow)
.where(
Workflow.tenant_id == pipeline.tenant_id,
Workflow.app_id == pipeline.id,
Workflow.version == Workflow.VERSION_DRAFT,
)
.limit(1)
)
if not pipeline or not workflow:
raise ValueError("Pipeline or workflow not found")
datasource_nodes = workflow.graph_dict.get("nodes", [])
datasource_plugins = []
for datasource_node in datasource_nodes:
if datasource_node.get("data", {}).get("type") == "datasource":
datasource_node_data = datasource_node["data"]
if not datasource_node_data:
continue
variables = workflow.rag_pipeline_variables
if variables:
variables_map = {item["variable"]: item for item in variables}
else:
variables_map = {}
datasource_parameters = datasource_node_data.get("datasource_parameters", {})
user_input_variables_keys = []
user_input_variables = []
for _, value in datasource_parameters.items():
if value.get("value") and isinstance(value.get("value"), str):
pattern = (
r"\{\{#([a-zA-Z0-9_]{1,50}"
r"(?:\.[a-zA-Z0-9_][a-zA-Z0-9_]{0,29}){1,10})#\}\}"
)
match = re.match(pattern, value["value"])
if match:
full_path = match.group(1)
last_part = full_path.split(".")[-1]
user_input_variables_keys.append(last_part)
elif value.get("value") and isinstance(value.get("value"), list):
last_part = value.get("value")[-1]
user_input_variables_keys.append(last_part)
for key, value in variables_map.items():
if key in user_input_variables_keys:
user_input_variables.append(value)
# get credentials
datasource_provider_service: DatasourceProviderService = DatasourceProviderService()
credentials: list[dict[Any, Any]] = datasource_provider_service.list_datasource_credentials(
tenant_id=tenant_id,
provider=datasource_node_data.get("provider_name"),
plugin_id=datasource_node_data.get("plugin_id"),
)
credential_info_list: list[Any] = []
for credential in credentials:
credential_info_list.append(
{
"id": credential.get("id"),
"name": credential.get("name"),
"type": credential.get("type"),
"is_default": credential.get("is_default"),
}
)
datasource_plugins.append(
{
"node_id": datasource_node.get("id"),
"plugin_id": datasource_node_data.get("plugin_id"),
"provider_name": datasource_node_data.get("provider_name"),
"datasource_type": datasource_node_data.get("provider_type"),
"title": datasource_node_data.get("title"),
"user_input_variables": user_input_variables,
"credentials": credential_info_list,
}
)
return datasource_plugins
def get_pipeline(self, tenant_id: str, dataset_id: str, session: Session | None = None) -> Pipeline:
"""
Get pipeline
"""
if session is None:
with self._session_maker() as new_session:
return self.get_pipeline(tenant_id, dataset_id, session=new_session)
dataset: Dataset | None = session.scalar(
dataset: Dataset | None = db.session.scalar(
select(Dataset)
.where(
Dataset.id == dataset_id,
@@ -1683,7 +1488,105 @@ class RagPipelineService:
)
if not dataset:
raise ValueError("Dataset not found")
pipeline: Pipeline | None = session.scalar(
pipeline: Pipeline | None = db.session.scalar(
select(Pipeline)
.where(
Pipeline.id == dataset.pipeline_id,
Pipeline.tenant_id == tenant_id,
)
.limit(1)
)
if not pipeline:
raise ValueError("Pipeline not found")
workflow: Workflow | None = None
if is_published:
workflow = self.get_published_workflow(pipeline=pipeline)
else:
workflow = self.get_draft_workflow(pipeline=pipeline)
if not pipeline or not workflow:
raise ValueError("Pipeline or workflow not found")
datasource_nodes = workflow.graph_dict.get("nodes", [])
datasource_plugins = []
for datasource_node in datasource_nodes:
if datasource_node.get("data", {}).get("type") == "datasource":
datasource_node_data = datasource_node["data"]
if not datasource_node_data:
continue
variables = workflow.rag_pipeline_variables
if variables:
variables_map = {item["variable"]: item for item in variables}
else:
variables_map = {}
datasource_parameters = datasource_node_data.get("datasource_parameters", {})
user_input_variables_keys = []
user_input_variables = []
for _, value in datasource_parameters.items():
if value.get("value") and isinstance(value.get("value"), str):
pattern = r"\{\{#([a-zA-Z0-9_]{1,50}(?:\.[a-zA-Z0-9_][a-zA-Z0-9_]{0,29}){1,10})#\}\}"
match = re.match(pattern, value["value"])
if match:
full_path = match.group(1)
last_part = full_path.split(".")[-1]
user_input_variables_keys.append(last_part)
elif value.get("value") and isinstance(value.get("value"), list):
last_part = value.get("value")[-1]
user_input_variables_keys.append(last_part)
for key, value in variables_map.items():
if key in user_input_variables_keys:
user_input_variables.append(value)
# get credentials
datasource_provider_service: DatasourceProviderService = DatasourceProviderService()
credentials: list[dict[Any, Any]] = datasource_provider_service.list_datasource_credentials(
tenant_id=tenant_id,
provider=datasource_node_data.get("provider_name"),
plugin_id=datasource_node_data.get("plugin_id"),
)
credential_info_list: list[Any] = []
for credential in credentials:
credential_info_list.append(
{
"id": credential.get("id"),
"name": credential.get("name"),
"type": credential.get("type"),
"is_default": credential.get("is_default"),
}
)
datasource_plugins.append(
{
"node_id": datasource_node.get("id"),
"plugin_id": datasource_node_data.get("plugin_id"),
"provider_name": datasource_node_data.get("provider_name"),
"datasource_type": datasource_node_data.get("provider_type"),
"title": datasource_node_data.get("title"),
"user_input_variables": user_input_variables,
"credentials": credential_info_list,
}
)
return datasource_plugins
def get_pipeline(self, tenant_id: str, dataset_id: str) -> Pipeline:
"""
Get pipeline
"""
dataset: Dataset | None = db.session.scalar(
select(Dataset)
.where(
Dataset.id == dataset_id,
Dataset.tenant_id == tenant_id,
)
.limit(1)
)
if not dataset:
raise ValueError("Dataset not found")
pipeline: Pipeline | None = db.session.scalar(
select(Pipeline)
.where(
Pipeline.id == dataset.pipeline_id,
@@ -8,7 +8,7 @@ from uuid import uuid4
import yaml
from flask_login import current_user
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlalchemy.orm import scoped_session
from configs import dify_config
from constants import DOCUMENT_EXTENSIONS
@@ -16,6 +16,7 @@ from core.plugin.impl.plugin import PluginInstaller
from core.plugin.plugin_service import PluginService
from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
from core.rag.retrieval.retrieval_methods import RetrievalMethod
from extensions.ext_database import db
from factories import variable_factory
from models.dataset import Dataset, Document, DocumentPipelineExecutionLog, Pipeline
from models.enums import DatasetRuntimeMode, DataSourceType
@@ -28,7 +29,7 @@ logger = logging.getLogger(__name__)
class RagPipelineTransformService:
def transform_dataset(self, dataset_id: str, session: Session):
def transform_dataset(self, dataset_id: str, session: scoped_session):
dataset = session.get(Dataset, dataset_id)
if not dataset:
raise ValueError("Dataset not found")
@@ -44,11 +45,11 @@ class RagPipelineTransformService:
indexing_technique = dataset.indexing_technique
if not datasource_type and not indexing_technique:
return self._transform_to_empty_pipeline(dataset, session=session)
return self._transform_to_empty_pipeline(dataset)
doc_form = dataset.doc_form
if not doc_form:
return self._transform_to_empty_pipeline(dataset, session=session)
return self._transform_to_empty_pipeline(dataset)
retrieval_model = RetrievalSetting.model_validate(dataset.retrieval_model) if dataset.retrieval_model else None
pipeline_yaml = self._get_transform_yaml(doc_form, datasource_type, indexing_technique)
# deal dependencies
@@ -80,7 +81,7 @@ class RagPipelineTransformService:
workflow_data["graph"] = graph
pipeline_yaml["workflow"] = workflow_data
# create pipeline
pipeline = self._create_pipeline(pipeline_yaml, session=session)
pipeline = self._create_pipeline(pipeline_yaml)
# save chunk structure to dataset
if doc_form == IndexStructureType.PARENT_CHILD_INDEX:
@@ -96,7 +97,7 @@ class RagPipelineTransformService:
# deal document data
self._deal_document_data(dataset, session)
session.flush()
session.commit()
return {
"pipeline_id": pipeline.id,
"dataset_id": dataset_id,
@@ -194,7 +195,6 @@ class RagPipelineTransformService:
def _create_pipeline(
self,
data: dict[str, Any],
session: Session,
) -> Pipeline:
"""Create a new app or update an existing one."""
pipeline_data = data.get("rag_pipeline", {})
@@ -227,8 +227,8 @@ class RagPipelineTransformService:
)
pipeline.id = str(uuid4())
session.add(pipeline)
session.flush()
db.session.add(pipeline)
db.session.flush()
# create draft workflow
draft_workflow = Workflow(
tenant_id=pipeline.tenant_id,
@@ -254,11 +254,11 @@ class RagPipelineTransformService:
conversation_variables=conversation_variables,
rag_pipeline_variables=rag_pipeline_variables_list,
)
session.add(draft_workflow)
session.add(published_workflow)
session.flush()
db.session.add(draft_workflow)
db.session.add(published_workflow)
db.session.flush()
pipeline.workflow_id = published_workflow.id
session.add(pipeline)
db.session.add(pipeline)
return pipeline
def _deal_dependencies(self, pipeline_yaml: dict[str, Any], tenant_id: str):
@@ -289,29 +289,29 @@ class RagPipelineTransformService:
logger.debug("Installing missing pipeline plugins %s", need_install_plugin_unique_identifiers)
PluginService.install_from_marketplace_pkg(tenant_id, need_install_plugin_unique_identifiers)
def _transform_to_empty_pipeline(self, dataset: Dataset, session: Session):
def _transform_to_empty_pipeline(self, dataset: Dataset):
pipeline = Pipeline(
tenant_id=dataset.tenant_id,
name=dataset.name,
description=dataset.description,
created_by=current_user.id,
)
session.add(pipeline)
session.flush()
db.session.add(pipeline)
db.session.flush()
dataset.pipeline_id = pipeline.id
dataset.runtime_mode = DatasetRuntimeMode.RAG_PIPELINE
dataset.updated_by = current_user.id
dataset.updated_at = datetime.now(UTC).replace(tzinfo=None)
session.add(dataset)
session.flush()
db.session.add(dataset)
db.session.commit()
return {
"pipeline_id": pipeline.id,
"dataset_id": dataset.id,
"status": "success",
}
def _deal_document_data(self, dataset: Dataset, session: Session):
def _deal_document_data(self, dataset: Dataset, session: scoped_session):
file_node_id = "1752479895761"
notion_node_id = "1752489759475"
jina_node_id = "1752491761974"
@@ -54,7 +54,7 @@ class TestPipelineTemplateListApi:
return_value=templates,
),
):
response, status = method(api, MagicMock(), str(uuid4()))
response, status = method(api, str(uuid4()))
assert status == 200
assert response == {
@@ -100,7 +100,7 @@ class TestPipelineTemplateDetailApi:
return_value=service,
),
):
response, status = method(api, MagicMock(), "tpl-1")
response, status = method(api, "tpl-1")
assert status == 200
assert response == {**template, "created_by": None}
@@ -120,7 +120,7 @@ class TestPipelineTemplateDetailApi:
),
):
with pytest.raises(NotFound):
method(api, MagicMock(), "non-existent-id")
method(api, "non-existent-id")
def test_get_returns_404_for_customized_type_not_found(self, app: Flask) -> None:
api = PipelineTemplateDetailApi()
@@ -137,7 +137,7 @@ class TestPipelineTemplateDetailApi:
),
):
with pytest.raises(NotFound):
method(api, MagicMock(), "non-existent-id")
method(api, "non-existent-id")
class TestCustomizedPipelineTemplateApi:
@@ -470,7 +470,7 @@ class TestPipelineRunApis:
return_value={"ok": True},
),
):
assert method(api, MagicMock(), user, pipeline) == {"ok": True}
assert method(api, user, pipeline) == {"ok": True}
def test_draft_run_rate_limit(self, app: Flask) -> None:
api = DraftRagPipelineRunApi()
@@ -498,7 +498,7 @@ class TestPipelineRunApis:
),
):
with pytest.raises(InvokeRateLimitHttpError):
method(api, MagicMock(), user, pipeline)
method(api, user, pipeline)
class TestDraftNodeRun:
@@ -600,7 +600,7 @@ class TestMiscApis:
app.test_request_context("/"),
):
with pytest.raises(Forbidden):
method(api, MagicMock(spec=Session), user, "ds1")
method(api, user, "ds1")
def test_recommended_plugins(self, app: Flask) -> None:
api = RagPipelineRecommendedPluginApi()
@@ -655,7 +655,7 @@ class TestPublishedRagPipelineRunApi:
return_value={"ok": True},
),
):
result = method(api, MagicMock(), user, pipeline)
result = method(api, user, pipeline)
assert result == {"ok": True}
def test_published_run_rate_limit(self, app: Flask) -> None:
@@ -681,7 +681,7 @@ class TestPublishedRagPipelineRunApi:
),
):
with pytest.raises(InvokeRateLimitHttpError):
method(api, MagicMock(), user, pipeline)
method(api, user, pipeline)
class TestDefaultBlockConfigApi:
@@ -101,8 +101,8 @@ class TestRagPipelineServiceGetPipeline:
service = self._make_service(flask_app_with_containers)
with pytest.raises(ValueError, match="Pipeline not found"):
service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset.id, session=db_session_with_containers)
with pytest.raises(ValueError, match="(Dataset not found|Pipeline not found)"):
service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset.id)
def test_get_pipeline_returns_pipeline_when_found(
self, db_session_with_containers: Session, flask_app_with_containers: Flask
@@ -117,7 +117,7 @@ class TestRagPipelineServiceGetPipeline:
service = self._make_service(flask_app_with_containers)
result = service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset.id, session=db_session_with_containers)
result = service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset.id)
assert result.id == pipeline.id
@@ -165,9 +165,7 @@ class TestUpdateCustomizedPipelineTemplate:
description="Updated description",
icon_info=IconInfo(icon="🔥"),
)
result = RagPipelineService.update_customized_pipeline_template(
template.id, info, account, tenant_id, session=db_session_with_containers
)
result = RagPipelineService.update_customized_pipeline_template(template.id, info, account, tenant_id)
assert result.name == "Updated Name"
assert result.description == "Updated description"
@@ -205,9 +203,7 @@ class TestUpdateCustomizedPipelineTemplate:
icon_info=IconInfo(icon="📄"),
)
with pytest.raises(ValueError, match="Template name is already exists"):
RagPipelineService.update_customized_pipeline_template(
template1.id, info, account, tenant_id, session=db_session_with_containers
)
RagPipelineService.update_customized_pipeline_template(template1.id, info, account, tenant_id)
class TestDeleteCustomizedPipelineTemplate:
@@ -245,14 +241,14 @@ class TestDeleteCustomizedPipelineTemplate:
template_id = template.id
db_session_with_containers.flush()
RagPipelineService.delete_customized_pipeline_template(
template_id, tenant_id, session=db_session_with_containers
)
RagPipelineService.delete_customized_pipeline_template(template_id, tenant_id)
# Verify the record is deleted within the same context
from sqlalchemy import select
remaining = db_session_with_containers.scalar(
from extensions.ext_database import db as ext_db
remaining = ext_db.session.scalar(
select(PipelineCustomizedTemplate).where(PipelineCustomizedTemplate.id == template_id)
)
assert remaining is None
@@ -1,7 +1,6 @@
import pytest
from agenton.compositor import CompositorSessionSnapshot
from dify_agent.protocol import (
AgentRunUsage,
DeferredToolCallPayload,
PydanticAIStreamRunEvent,
RunCancelledEvent,
@@ -60,11 +59,7 @@ def test_event_adapter_maps_run_succeeded_to_final_output():
RunSucceededEvent(
id="3-0",
run_id="run-1",
data=RunSucceededEventData(
output={"summary": "done"},
session_snapshot=snapshot,
usage=AgentRunUsage(prompt_tokens=2, completion_tokens=3),
),
data=RunSucceededEventData(output={"summary": "done"}, session_snapshot=snapshot),
)
)
@@ -74,7 +69,6 @@ def test_event_adapter_maps_run_succeeded_to_final_output():
source_event_id="3-0",
output={"summary": "done"},
session_snapshot=snapshot,
usage={"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5},
)
]
@@ -1,41 +1,13 @@
import json
from pathlib import Path
from click.testing import CliRunner
from commands import data_migration
from commands.data_migration import (
ID_STRATEGY_CHOICES,
export_migration_data,
export_migration_data_template,
import_migration_data,
)
from services.data_migration.entities import (
ConflictStrategy,
ExportResult,
ImportOptions,
ImportResult,
MigrationPackage,
ReportContext,
)
class FakeSessionContext:
session: object
entered: bool
exited: bool
def __init__(self, session: object) -> None:
self.session = session
self.entered = False
self.exited = False
def __enter__(self) -> object:
self.entered = True
return self.session
def __exit__(self, *_args: object) -> None:
self.exited = True
def test_export_command_requires_input_and_output():
@@ -97,89 +69,3 @@ def test_export_template_command_requires_overwrite_for_existing_output(tmp_path
assert result.exit_code != 0
assert "already exists" in result.output
def test_export_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
session = object()
session_context = FakeSessionContext(session)
captured: dict[str, object] = {}
input_file = tmp_path / "export-config.json"
output_file = tmp_path / "migration-package.json"
input_file.write_text(json.dumps({"source_tenant": {"name": "source"}, "apps": {"all": True}}))
package = MigrationPackage.from_mapping({"metadata": {"version": "1", "source_scope": "single"}})
class FakeMigrationExportService:
def export(self, export_session, selection):
captured["session"] = export_session
captured["selection"] = selection
return ExportResult(package=package, report_items=[], report_context=ReportContext())
class FakeMigrationPackageService:
def save_package(self, package_to_save, path, *, overwrite):
captured["package"] = package_to_save
captured["path"] = path
captured["overwrite"] = overwrite
monkeypatch.setattr(data_migration.session_factory, "create_session", lambda: session_context)
monkeypatch.setattr(data_migration, "MigrationExportService", FakeMigrationExportService)
monkeypatch.setattr(data_migration, "MigrationPackageService", FakeMigrationPackageService)
result = CliRunner().invoke(
export_migration_data,
["--input", str(input_file), "--output", str(output_file)],
)
assert result.exit_code == 0
assert captured["session"] is session
assert captured["package"] is package
assert captured["path"] == str(output_file)
assert captured["overwrite"] is False
assert session_context.entered
assert session_context.exited
def test_import_command_uses_cli_owned_session(monkeypatch, tmp_path: Path):
session = object()
session_context = FakeSessionContext(session)
captured: dict[str, object] = {}
input_file = tmp_path / "migration-package.json"
input_file.write_text("{}")
package = MigrationPackage.from_mapping(
{
"metadata": {
"version": "1",
"source_scope": "single",
"target_tenant": {"name": "target"},
"import_options": {"conflict_strategy": "fail"},
}
}
)
class FakeMigrationImportService:
def import_package(self, import_session, request):
captured["session"] = import_session
captured["request"] = request
return ImportResult(report_items=[], report_context=ReportContext(target_tenant="target"))
class FakeMigrationPackageService:
def load_package(self, path):
captured["path"] = path
return package
monkeypatch.setattr(data_migration.session_factory, "create_session", lambda: session_context)
monkeypatch.setattr(data_migration, "MigrationImportService", FakeMigrationImportService)
monkeypatch.setattr(data_migration, "MigrationPackageService", FakeMigrationPackageService)
result = CliRunner().invoke(
import_migration_data,
["--input", str(input_file), "--conflict-strategy", "skip"],
)
assert result.exit_code == 0
assert captured["session"] is session
assert captured["path"] == str(input_file)
request = captured["request"]
assert request.package is package
assert request.options_override == ImportOptions(conflict_strategy=ConflictStrategy.SKIP)
assert session_context.entered
assert session_context.exited
@@ -261,7 +261,7 @@ def test_generate_specs_include_console_contract_shapes_for_schema_migration(tmp
assert {"document", "audio", "video", "custom", "preview_config"} <= set(file_upload)
assert "detail" in schemas["WorkflowFileUploadImagePayload"]["properties"]
assert {"mode", "file_type_list"} <= set(schemas["WorkflowFileUploadPreviewConfigPayload"]["properties"])
assert schemas["AccountWithRoleResponse"]["properties"]["avatar_url"]["readOnly"] is True
assert schemas["AccountWithRole"]["properties"]["avatar_url"]["readOnly"] is True
def test_checked_in_agent_v2_knowledge_openapi_and_generated_contracts_are_in_sync():
@@ -1,174 +0,0 @@
from __future__ import annotations
import pytest
from controllers.common import session as session_module
class FakeSession:
committed: bool
rolled_back: bool
closed: bool
def __init__(self) -> None:
self.committed = False
self.rolled_back = False
self.closed = False
def commit(self) -> None:
self.committed = True
def rollback(self) -> None:
self.rolled_back = True
class FakeSessionBegin:
session: FakeSession
entered: bool
exited: bool
exc_type: object | None
def __init__(self, session: FakeSession) -> None:
self.session = session
self.entered = False
self.exited = False
self.exc_type = None
def __enter__(self) -> FakeSession:
self.entered = True
return self.session
def __exit__(self, exc_type: object | None, *_args: object) -> None:
self.exited = True
self.exc_type = exc_type
if exc_type is None:
self.session.commit()
else:
self.session.rollback()
self.session.closed = True
class FakeSessionContext:
session: FakeSession
entered: bool
exited: bool
exc_type: object | None
def __init__(self, session: FakeSession) -> None:
self.session = session
self.entered = False
self.exited = False
self.exc_type = None
def __enter__(self) -> FakeSession:
self.entered = True
return self.session
def __exit__(self, exc_type: object | None, *_args: object) -> None:
self.exited = True
self.exc_type = exc_type
self.session.closed = True
class FakeSessionMaker:
begin_context: FakeSessionBegin
def __init__(self, session: FakeSession) -> None:
self.begin_context = FakeSessionBegin(session)
def begin(self) -> FakeSessionBegin:
return self.begin_context
def test_with_session_write_commits_on_success(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_maker = FakeSessionMaker(session)
monkeypatch.setattr(session_module.session_factory, "get_session_maker", lambda: session_maker)
class Handler:
@session_module.with_session(write=True)
def post(self, injected_session):
assert injected_session is session
return "ok"
assert Handler().post() == "ok"
assert session.closed
assert session.committed
assert not session.rolled_back
assert session_maker.begin_context.entered
assert session_maker.begin_context.exited
assert session_maker.begin_context.exc_type is None
def test_with_session_default_write_commits_on_success(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_maker = FakeSessionMaker(session)
monkeypatch.setattr(session_module.session_factory, "get_session_maker", lambda: session_maker)
class Handler:
@session_module.with_session
def post(self, injected_session):
assert injected_session is session
return "ok"
assert Handler().post() == "ok"
assert session.committed
assert not session.rolled_back
def test_with_session_write_rolls_back_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_maker = FakeSessionMaker(session)
monkeypatch.setattr(session_module.session_factory, "get_session_maker", lambda: session_maker)
class Handler:
@session_module.with_session(write=True)
def get(self, _session):
raise RuntimeError("boom")
with pytest.raises(RuntimeError, match="boom"):
Handler().get()
assert session.closed
assert not session.committed
assert session.rolled_back
assert session_maker.begin_context.entered
assert session_maker.begin_context.exited
assert session_maker.begin_context.exc_type is RuntimeError
def test_with_session_read_mode_does_not_commit(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_context = FakeSessionContext(session)
monkeypatch.setattr(session_module.session_factory, "create_session", lambda: session_context)
class Handler:
@session_module.with_session(write=False)
def get(self, injected_session):
assert injected_session is session
return "ok"
assert Handler().get() == "ok"
assert session.closed
assert not session.committed
assert not session.rolled_back
assert session_context.entered
assert session_context.exited
assert session_context.exc_type is None
def test_with_session_preserves_wrapped_metadata(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_maker = FakeSessionMaker(session)
monkeypatch.setattr(session_module.session_factory, "get_session_maker", lambda: session_maker)
class Handler:
@session_module.with_session
def get(self, _session):
"""handler docs"""
return "ok"
assert Handler.get.__name__ == "get"
assert Handler.get.__doc__ == "handler docs"
@@ -318,9 +318,7 @@ def test_agent_app_list_and_create_use_agent_route(
lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)),
)
with app.test_request_context(
"/console/api/agent?page=1&limit=10&mode=workflow&sort_by=recently_created&is_created_by_me=true"
):
with app.test_request_context("/console/api/agent?page=1&limit=10&mode=workflow"):
listed = unwrap(AgentAppListApi.get)(AgentAppListApi(), "tenant-1", SimpleNamespace(id=account_id))
assert listed["page"] == 1
@@ -345,8 +343,6 @@ def test_agent_app_list_and_create_use_agent_route(
list_call = cast(dict[str, object], captured["list"])
list_params = cast(Any, list_call["params"])
assert list_params.mode == "agent"
assert list_params.sort_by == "recently_created"
assert list_params.is_created_by_me is True
assert list_params.status == "normal"
with app.test_request_context(
@@ -374,54 +370,20 @@ def test_agent_app_list_and_create_use_agent_route(
assert create_params.agent_role == "Coordinator"
def test_agent_app_create_payload_allows_optional_role() -> None:
omitted = roster_controller.AgentAppCreatePayload.model_validate(
{"name": "Iris", "description": "Agent app", "icon_type": "emoji", "icon": "robot"}
)
blank = roster_controller.AgentAppCreatePayload.model_validate(
{"name": "Iris", "description": "Agent app", "role": " ", "icon_type": "emoji", "icon": "robot"}
)
assert omitted.role is None
assert blank.role == ""
def test_agent_app_create_omits_optional_role_as_empty_string(
app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
) -> None:
captured: dict[str, object] = {}
class FakeAppService:
def create_app(self, tenant_id: str, params: object, account: object) -> object:
captured["create"] = {"tenant_id": tenant_id, "params": params, "account": account}
return _app_detail_obj(id="app-created", bound_agent_id="agent-created")
monkeypatch.setattr(roster_controller, "AppService", FakeAppService)
monkeypatch.setattr(
roster_controller,
"_serialize_agent_app_detail",
lambda app_model, **_kwargs: {"id": "agent-created", "app_id": app_model.id},
)
current_user = SimpleNamespace(id=account_id)
def test_agent_app_create_requires_role(app: Flask, account_id: str) -> None:
with app.test_request_context(
"/console/api/agent",
json={
"name": "No-role Iris",
"description": "Agent app",
"icon_type": "emoji",
"icon": "robot",
},
json={"name": "Iris", "description": "Agent app", "icon_type": "emoji", "icon": "robot"},
):
created, status = unwrap(AgentAppListApi.post)(AgentAppListApi(), "tenant-1", current_user)
with pytest.raises(ValueError, match="Field required"):
unwrap(AgentAppListApi.post)(AgentAppListApi(), "tenant-1", SimpleNamespace(id=account_id))
assert status == 201
assert created == {"id": "agent-created", "app_id": "app-created"}
create_call = cast(dict[str, object], captured["create"])
create_params = cast(Any, create_call["params"])
assert create_call["tenant_id"] == "tenant-1"
assert create_call["account"] is current_user
assert create_params.agent_role == ""
with app.test_request_context(
"/console/api/agent",
json={"name": "Iris", "description": "Agent app", "role": " ", "icon_type": "emoji", "icon": "robot"},
):
with pytest.raises(ValueError, match="Agent role is required"):
unwrap(AgentAppListApi.post)(AgentAppListApi(), "tenant-1", SimpleNamespace(id=account_id))
def test_agent_app_detail_update_delete_resolve_app_from_agent_id(
@@ -843,7 +805,7 @@ def test_agent_api_status_and_key_routes_resolve_backing_app(
}
def test_agent_app_update_allows_empty_role(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
def test_agent_app_update_rejects_empty_role(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
agent_id = "00000000-0000-0000-0000-000000000001"
app_model = _app_detail_obj(id="app-1", bound_agent_id=agent_id)
captured: dict[str, object] = {}
@@ -858,28 +820,11 @@ def test_agent_app_update_allows_empty_role(app: Flask, monkeypatch: pytest.Monk
"get_app_backing_agent",
lambda _self, **kwargs: SimpleNamespace(
id=agent_id,
app_id="app-1",
backing_app_id=None,
role="",
debug_conversation_id="debug-conversation-detail",
active_config_snapshot_id=None,
),
)
monkeypatch.setattr(
roster_controller.AgentRosterService,
"get_or_create_agent_app_debug_conversation_id",
lambda _self, **kwargs: "debug-conversation-detail",
)
monkeypatch.setattr(
roster_controller.AgentRosterService,
"count_agent_app_debug_conversation_messages",
lambda _self, **kwargs: 0,
)
monkeypatch.setattr(
roster_controller.AgentRosterService,
"active_config_is_published",
lambda _self, **kwargs: False,
)
monkeypatch.setattr(
roster_controller.FeatureService,
"get_system_features",
@@ -900,11 +845,8 @@ def test_agent_app_update_allows_empty_role(app: Flask, monkeypatch: pytest.Monk
"/console/api/agent/00000000-0000-0000-0000-000000000001",
json={"name": "Renamed", "description": "", "role": "", "icon_type": "emoji", "icon": "R"},
):
updated = unwrap(AgentAppApi.put)(AgentAppApi(), "tenant-1", SimpleNamespace(id="account-1"), agent_id)
assert updated["role"] == ""
update_call = cast(dict[str, object], captured["update"])
assert cast(dict[str, object], update_call["args"])["role"] == ""
with pytest.raises(ValueError, match="String should have at least 1 character"):
unwrap(AgentAppApi.put)(AgentAppApi(), "tenant-1", SimpleNamespace(id="account-1"), agent_id)
def test_invite_options_get_parses_app_id(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -1427,56 +1369,6 @@ def test_agent_chat_generate_and_stop_routes_resolve_app_from_agent_id(
assert stop_call == {"current_user_id": account_id, "app_model": app_model, "task_id": "task-1"}
def test_agent_chat_stream_preflight_raises_first_error_event() -> None:
class ClosableStream:
def __init__(self) -> None:
self.closed = False
self._chunks = iter(
[
"event: ping\n\n",
(
'data: {"event":"error","message":"Incorrect API key provided",'
'"code":"completion_request_error","status":400}\n\n'
),
]
)
def __iter__(self):
return self
def __next__(self) -> str:
return next(self._chunks)
def close(self) -> None:
self.closed = True
stream = ClosableStream()
with pytest.raises(CompletionRequestError) as exc_info:
completion_controller._raise_agent_stream_error_before_response(stream)
assert "Incorrect API key provided" in exc_info.value.description
assert stream.closed is True
def test_agent_chat_stream_preflight_preserves_first_normal_event() -> None:
stream = iter(
[
"event: ping\n\n",
'data: {"event":"message","answer":"hello"}\n\n',
'data: {"event":"message_end"}\n\n',
]
)
wrapped = completion_controller._raise_agent_stream_error_before_response(stream)
assert list(wrapped) == [
"event: ping\n\n",
'data: {"event":"message","answer":"hello"}\n\n',
'data: {"event":"message_end"}\n\n',
]
def test_agent_build_chat_finalize_route_resolves_app_from_agent_id(
app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
) -> None:
@@ -9,17 +9,13 @@ from dify_agent.protocol import SandboxListResponse, SandboxReadResponse, Sandbo
from controllers.console import agent_app_sandbox as module
from models.model import App, AppMode, IconType
from services.agent_app_sandbox_service import AgentSandboxInfo, AgentSandboxInspectorError
from services.agent_app_sandbox_service import AgentSandboxInspectorError
class _AgentAppService:
def __init__(self) -> None:
self.calls: list[tuple[str, str, str, str, str]] = []
def get_info(self, *, tenant_id: str, app_id: str, conversation_id: str) -> AgentSandboxInfo:
self.calls.append(("info", tenant_id, app_id, conversation_id, ""))
return AgentSandboxInfo(session_id="abc1234", workspace_cwd="~/workspace/abc1234")
def list_files(self, *, tenant_id: str, app_id: str, conversation_id: str, path: str) -> SandboxListResponse:
self.calls.append(("list", tenant_id, app_id, conversation_id, path))
return SandboxListResponse(path=path, entries=[], truncated=False)
@@ -135,17 +131,14 @@ def test_agent_app_sandbox_resources_proxy_service(monkeypatch: pytest.MonkeyPat
SimpleNamespace(get_json=lambda silent=True: {"conversation_id": "conv-1", "path": "report.txt"}),
)
info = unwrap(module.AgentAppSandboxInfoResource.get)(object(), "tenant-1", "agent-1")
listing = unwrap(module.AgentAppSandboxListResource.get)(object(), "tenant-1", "agent-1")
preview = unwrap(module.AgentAppSandboxReadResource.get)(object(), "tenant-1", "agent-1")
upload = unwrap(module.AgentAppSandboxUploadResource.post)(object(), "tenant-1", "agent-1")
assert info == {"session_id": "abc1234", "workspace_cwd": "~/workspace/abc1234"}
assert listing["path"] == "sub/report.txt"
assert preview["text"] == "hello"
assert upload["file"]["reference"] == "dify-file-ref:file-1"
assert service.calls == [
("info", "tenant-1", "app-1", "conv-1", ""),
("list", "tenant-1", "app-1", "conv-1", "sub/report.txt"),
("read", "tenant-1", "app-1", "conv-1", "sub/report.txt"),
("upload", "tenant-1", "app-1", "conv-1", "report.txt"),
@@ -154,9 +147,6 @@ def test_agent_app_sandbox_resources_proxy_service(monkeypatch: pytest.MonkeyPat
def test_agent_app_sandbox_resource_returns_normalized_errors(monkeypatch: pytest.MonkeyPatch) -> None:
class FailingService:
def get_info(self, **kwargs):
raise AgentSandboxInspectorError("no_active_session", "no active session", status_code=404)
def list_files(self, **kwargs):
raise AgentSandboxInspectorError("no_active_session", "no active session", status_code=404)
@@ -166,10 +156,6 @@ def test_agent_app_sandbox_resource_returns_normalized_errors(monkeypatch: pytes
module, "query_params_from_request", lambda model: SimpleNamespace(conversation_id="conv-1", path=".")
)
assert unwrap(module.AgentAppSandboxInfoResource.get)(object(), "tenant-1", "agent-1") == (
{"code": "no_active_session", "message": "no active session"},
404,
)
assert unwrap(module.AgentAppSandboxListResource.get)(object(), "tenant-1", "agent-1") == (
{"code": "no_active_session", "message": "no active session"},
404,
@@ -30,7 +30,7 @@ def test_completion_conversation_list_returns_paginated_result(app: Flask, monke
paginate_result.total = 0
paginate_result.has_next = False
paginate_result.items = []
monkeypatch.setattr(conversation_module, "paginate_query", lambda *_args, **_kwargs: paginate_result)
monkeypatch.setattr(conversation_module.db, "paginate", lambda *_args, **_kwargs: paginate_result)
with app.test_request_context("/console/api/apps/app-1/completion-conversations", method="GET"):
response = method(api, account, app_model=SimpleNamespace(id="app-1"))
@@ -71,7 +71,7 @@ def test_chat_conversation_list_advanced_chat_calls_paginate(app: Flask, monkeyp
paginate_result.total = 0
paginate_result.has_next = False
paginate_result.items = []
monkeypatch.setattr(conversation_module, "paginate_query", lambda *_args, **_kwargs: paginate_result)
monkeypatch.setattr(conversation_module.db, "paginate", lambda *_args, **_kwargs: paginate_result)
with app.test_request_context("/console/api/apps/app-1/chat-conversations", method="GET"):
response = method(api, account, app_model=SimpleNamespace(id="app-1", mode=AppMode.ADVANCED_CHAT))
@@ -4,7 +4,6 @@ from types import SimpleNamespace
import pytest
from controllers.common.session import with_session
from controllers.console.app import wraps as wraps_module
from controllers.console.app.error import AppNotFoundError
from models.model import AppMode
@@ -12,10 +11,16 @@ from models.model import AppMode
class FakeSession:
app_model: object | None
committed: bool
rolled_back: bool
closed: bool
scalar_called: bool
def __init__(self, app_model: object | None = None) -> None:
self.app_model = app_model
self.committed = False
self.rolled_back = False
self.closed = False
self.scalar_called = False
def scalar(self, *_args: object, **_kwargs: object) -> object | None:
@@ -23,10 +28,68 @@ class FakeSession:
return self.app_model
def commit(self) -> None:
pass
self.committed = True
def rollback(self) -> None:
pass
self.rolled_back = True
class FakeSessionBegin:
session: FakeSession
entered: bool
exited: bool
exc_type: object | None
def __init__(self, session: FakeSession) -> None:
self.session = session
self.entered = False
self.exited = False
self.exc_type = None
def __enter__(self) -> FakeSession:
self.entered = True
return self.session
def __exit__(self, exc_type: object | None, *_args: object) -> None:
self.exited = True
self.exc_type = exc_type
if exc_type is None:
self.session.commit()
else:
self.session.rollback()
self.session.closed = True
class FakeSessionContext:
session: FakeSession
entered: bool
exited: bool
exc_type: object | None
def __init__(self, session: FakeSession) -> None:
self.session = session
self.entered = False
self.exited = False
self.exc_type = None
def __enter__(self) -> FakeSession:
self.entered = True
return self.session
def __exit__(self, exc_type: object | None, *_args: object) -> None:
self.exited = True
self.exc_type = exc_type
self.session.closed = True
class FakeSessionMaker:
begin_context: FakeSessionBegin
def __init__(self, session: FakeSession) -> None:
self.begin_context = FakeSessionBegin(session)
def begin(self) -> FakeSessionBegin:
return self.begin_context
def test_get_app_model_injects_model(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -63,13 +126,11 @@ def test_get_app_model_requires_app_id() -> None:
handler()
def test_wraps_with_session_reexports_common_session_decorator() -> None:
assert wraps_module.with_session is with_session
def test_get_app_model_prefers_injected_session(monkeypatch: pytest.MonkeyPatch) -> None:
def test_with_session_defaults_to_write_session_for_get_app_model(monkeypatch: pytest.MonkeyPatch) -> None:
app_model = SimpleNamespace(id="app-1", mode=AppMode.CHAT.value, status="normal", tenant_id="t1")
session = FakeSession(app_model)
session_maker = FakeSessionMaker(session)
monkeypatch.setattr(wraps_module.session_factory, "get_session_maker", lambda: session_maker)
monkeypatch.setattr(wraps_module, "current_account_with_tenant", lambda: (None, "t1"))
monkeypatch.setattr(
wraps_module.db,
@@ -78,9 +139,80 @@ def test_get_app_model_prefers_injected_session(monkeypatch: pytest.MonkeyPatch)
)
class Handler:
@wraps_module.with_session
@wraps_module.get_app_model
def get(self, _injected_session, app_model):
def get(self, injected_session, app_model):
assert injected_session is session
return app_model.id
assert Handler().get(session, app_id="app-1") == "app-1"
assert Handler().get(app_id="app-1") == "app-1"
assert session.scalar_called
assert session.committed
assert not session.rolled_back
assert session.closed
assert session_maker.begin_context.entered
assert session_maker.begin_context.exited
assert session_maker.begin_context.exc_type is None
def test_with_session_read_mode_does_not_commit(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_context = FakeSessionContext(session)
monkeypatch.setattr(wraps_module.session_factory, "create_session", lambda: session_context)
class Handler:
@wraps_module.with_session(write=False)
def get(self, injected_session):
assert injected_session is session
return "ok"
assert Handler().get() == "ok"
assert session.closed
assert not session.committed
assert not session.rolled_back
assert session_context.entered
assert session_context.exited
assert session_context.exc_type is None
def test_with_session_write_commits_on_success(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_maker = FakeSessionMaker(session)
monkeypatch.setattr(wraps_module.session_factory, "get_session_maker", lambda: session_maker)
class Handler:
@wraps_module.with_session(write=True)
def post(self, injected_session):
assert injected_session is session
return "ok"
assert Handler().post() == "ok"
assert session.closed
assert session.committed
assert not session.rolled_back
assert session_maker.begin_context.entered
assert session_maker.begin_context.exited
assert session_maker.begin_context.exc_type is None
def test_with_session_write_rolls_back_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
session = FakeSession()
session_maker = FakeSessionMaker(session)
monkeypatch.setattr(wraps_module.session_factory, "get_session_maker", lambda: session_maker)
class Handler:
@wraps_module.with_session(write=True)
def get(self, _session):
raise RuntimeError("boom")
with pytest.raises(RuntimeError, match="boom"):
Handler().get()
assert session.closed
assert not session.committed
assert session.rolled_back
assert session_maker.begin_context.entered
assert session_maker.begin_context.exited
assert session_maker.begin_context.exc_type is RuntimeError
@@ -1,7 +1,7 @@
from __future__ import annotations
from inspect import unwrap
from unittest.mock import Mock, PropertyMock, patch
from unittest.mock import PropertyMock, patch
import pytest
from flask import Flask
@@ -64,9 +64,7 @@ class TestPipelineTemplateListApi:
tenant_id = "tenant-1"
service_calls: list[tuple[str, str, str]] = []
def get_pipeline_templates(
session: Mock, template_type: str, language: str, current_tenant_id: str
) -> dict[str, object]:
def get_pipeline_templates(template_type: str, language: str, current_tenant_id: str) -> dict[str, object]:
service_calls.append((template_type, language, current_tenant_id))
return {"pipeline_templates": [_template_item()]}
@@ -74,7 +72,7 @@ class TestPipelineTemplateListApi:
app.test_request_context("/rag/pipeline/templates"),
patch.object(module.RagPipelineService, "get_pipeline_templates", side_effect=get_pipeline_templates),
):
response, status = method(api, Mock(), tenant_id)
response, status = method(api, tenant_id)
assert status == 200
assert service_calls == [("built-in", "en-US", tenant_id)]
@@ -94,9 +92,7 @@ class TestPipelineTemplateListApi:
tenant_id = "tenant-1"
service_calls: list[tuple[str, str, str]] = []
def get_pipeline_templates(
session: Mock, template_type: str, language: str, current_tenant_id: str
) -> dict[str, object]:
def get_pipeline_templates(template_type: str, language: str, current_tenant_id: str) -> dict[str, object]:
service_calls.append((template_type, language, current_tenant_id))
return {"pipeline_templates": []}
@@ -104,7 +100,7 @@ class TestPipelineTemplateListApi:
app.test_request_context("/rag/pipeline/templates?type=customized&language=ja-JP"),
patch.object(module.RagPipelineService, "get_pipeline_templates", side_effect=get_pipeline_templates),
):
response, status = method(api, Mock(), tenant_id)
response, status = method(api, tenant_id)
assert status == 200
assert response == {"pipeline_templates": []}
@@ -118,9 +114,7 @@ class TestPipelineTemplateDetailApi:
service_calls: list[tuple[str, str]] = []
class Service:
def get_pipeline_template_detail(
self, session: Mock, template_id: str, template_type: str
) -> dict[str, object]:
def get_pipeline_template_detail(self, template_id: str, template_type: str) -> dict[str, object]:
service_calls.append((template_id, template_type))
return _template_detail()
@@ -128,7 +122,7 @@ class TestPipelineTemplateDetailApi:
app.test_request_context("/rag/pipeline/templates/template-1?type=customized"),
patch.object(module, "RagPipelineService", Service),
):
response, status = method(api, Mock(), "template-1")
response, status = method(api, "template-1")
assert status == 200
assert response == {**_template_detail(), "created_by": None}
@@ -139,7 +133,7 @@ class TestPipelineTemplateDetailApi:
method = unwrap(api.get)
class Service:
def get_pipeline_template_detail(self, session: Mock, template_id: str, template_type: str) -> None:
def get_pipeline_template_detail(self, template_id: str, template_type: str) -> None:
return None
with (
@@ -147,7 +141,7 @@ class TestPipelineTemplateDetailApi:
patch.object(module, "RagPipelineService", Service),
):
with pytest.raises(NotFound):
method(api, Mock(), "missing")
method(api, "missing")
class TestCustomizedPipelineTemplateApi:
@@ -207,7 +207,7 @@ class TestDatasetDocumentListApi:
with (
app.test_request_context("/?fetch=true"),
patch(
"controllers.console.datasets.datasets_document.paginate_query",
"controllers.console.datasets.datasets_document.db.paginate",
return_value=pagination,
),
patch(
@@ -237,7 +237,7 @@ class TestDatasetDocumentListApi:
with (
app.test_request_context("/?keyword=test&status=enabled&sort=created_at"),
patch(
"controllers.console.datasets.datasets_document.paginate_query",
"controllers.console.datasets.datasets_document.db.paginate",
return_value=pagination,
),
patch(
@@ -263,7 +263,7 @@ class TestDatasetDocumentListApi:
with (
app.test_request_context("/"),
patch(
"controllers.console.datasets.datasets_document.paginate_query",
"controllers.console.datasets.datasets_document.db.paginate",
return_value=pagination,
),
patch(
@@ -341,7 +341,7 @@ class TestDatasetDocumentListApi:
with (
app.test_request_context("/?fetch=maybe"),
patch(
"controllers.console.datasets.datasets_document.paginate_query",
"controllers.console.datasets.datasets_document.db.paginate",
return_value=pagination,
),
patch(
@@ -363,7 +363,7 @@ class TestDatasetDocumentListApi:
with (
app.test_request_context("/?sort=hit_count"),
patch(
"controllers.console.datasets.datasets_document.paginate_query",
"controllers.console.datasets.datasets_document.db.paginate",
return_value=pagination,
),
patch(
@@ -1537,7 +1537,7 @@ class TestDocumentListAdvancedCases:
with (
app.test_request_context("/?sort=updated_at"),
patch(
"controllers.console.datasets.datasets_document.paginate_query",
"controllers.console.datasets.datasets_document.db.paginate",
return_value=pagination,
),
patch(
@@ -161,7 +161,7 @@ class TestDatasetDocumentSegmentListApi:
return_value=document,
),
patch(
"controllers.console.datasets.datasets_segments.paginate_query",
"controllers.console.datasets.datasets_segments.db.paginate",
return_value=pagination,
),
patch(
@@ -1207,7 +1207,7 @@ class TestSegmentListAdvancedCases:
return_value=document,
),
patch(
"controllers.console.datasets.datasets_segments.paginate_query",
"controllers.console.datasets.datasets_segments.db.paginate",
return_value=pagination,
),
patch(
@@ -1255,7 +1255,7 @@ class TestSegmentListAdvancedCases:
SimpleNamespace(SQLALCHEMY_DATABASE_URI_SCHEME="postgresql"),
),
patch(
"controllers.console.datasets.datasets_segments.paginate_query",
"controllers.console.datasets.datasets_segments.db.paginate",
return_value=pagination,
) as paginate_mock,
):
@@ -1267,7 +1267,7 @@ class TestSegmentListAdvancedCases:
"33333333-3333-3333-3333-333333333333",
)
query = paginate_mock.call_args.args[0]
query = paginate_mock.call_args.kwargs["select"]
sql = str(query.compile(compile_kwargs={"literal_binds": True}))
assert "jsonb_array_elements_text(CASE" in sql
assert "ELSE CAST('[]' AS JSONB)" in sql
@@ -1,7 +1,5 @@
from datetime import UTC, datetime
from inspect import unwrap as inspect_unwrap
from io import BytesIO
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock, patch
from uuid import uuid4
@@ -97,59 +95,7 @@ def valid_parameters() -> dict[str, object]:
def test_trial_workflow_uses_trial_scoped_simple_account_model() -> None:
assert module.simple_account_model.name == "TrialSimpleAccount"
assert module.simple_account_model.__schema__["properties"].keys() >= {"id", "name", "email"}
def test_trial_dataset_list_preserves_slim_dataset_fields(app: Flask):
class DatasetListItem:
id = "dataset-1"
name = "Dataset"
description = "description"
permission = "only_me"
data_source_type = "upload_file"
indexing_technique = "high_quality"
created_by = "user-1"
created_at = datetime(2024, 1, 1, tzinfo=UTC)
permission_keys = ["dataset.acl.readonly"]
@property
def app_count(self):
raise AssertionError("trial dataset list should not serialize detail-only computed fields")
api = module.DatasetListApi()
method = unwrap(api.get)
app_model = SimpleNamespace(tenant_id="tenant-1")
with (
app.test_request_context("/?page=1&limit=20&ids=dataset-1"),
patch.object(
module.DatasetService,
"get_datasets_by_ids",
return_value=([DatasetListItem()], 1),
) as get_datasets,
):
result = method(api, app_model)
get_datasets.assert_called_once_with(["dataset-1"], "tenant-1")
assert result == {
"data": [
{
"id": "dataset-1",
"name": "Dataset",
"description": "description",
"permission": "only_me",
"data_source_type": "upload_file",
"indexing_technique": "high_quality",
"created_by": "user-1",
"created_at": 1704067200,
"permission_keys": ["dataset.acl.readonly"],
}
],
"has_more": False,
"limit": 20,
"total": 1,
"page": 1,
}
assert hasattr(module.simple_account_model, "items")
class TestTrialAppWorkflowRunApi:
@@ -291,7 +291,7 @@ class TestWorkspaceListApi:
with (
app.test_request_context("/all-workspaces", query_string={"page": 1, "limit": 20}),
patch("controllers.console.workspace.workspace.paginate_query", return_value=paginate_result),
patch("controllers.console.workspace.workspace.db.paginate", return_value=paginate_result),
):
result, status = method(api)
@@ -308,7 +308,7 @@ class TestWorkspaceListApi:
with (
app.test_request_context("/all-workspaces", query_string={"page": 1, "limit": 1}),
patch("controllers.console.workspace.workspace.paginate_query", return_value=paginate_result),
patch("controllers.console.workspace.workspace.db.paginate", return_value=paginate_result),
):
result, status = method(api)
@@ -65,7 +65,6 @@ class TestPluginUploadFileApi:
"sign": "sig",
"tenant_id": "tenant-1",
"user_id": "user-1",
"conversation_id": "conversation-1",
},
file=dummy_file,
)
@@ -84,10 +83,6 @@ class TestPluginUploadFileApi:
assert result["id"] == "file-id"
assert result["reference"] == build_file_reference(record_id="file-id")
assert result["preview_url"] == "signed-url"
mock_verify_signature.assert_called_once()
assert mock_verify_signature.call_args.kwargs["conversation_id"] == "conversation-1"
tool_file_manager_instance.create_file_by_raw.assert_called_once()
assert tool_file_manager_instance.create_file_by_raw.call_args.kwargs["conversation_id"] == "conversation-1"
def test_missing_file(self):
module.request = fake_request(
@@ -272,7 +272,6 @@ class TestPluginUploadFileRequestApi:
mock_payload = MagicMock()
mock_payload.filename = "test.pdf"
mock_payload.mimetype = "application/pdf"
mock_payload.conversation_id = "conversation-id"
# Act
raw_post = _extract_raw_post(PluginUploadFileRequestApi)
@@ -280,11 +279,7 @@ class TestPluginUploadFileRequestApi:
# Assert
mock_get_url.assert_called_once_with(
filename="test.pdf",
mimetype="application/pdf",
tenant_id="tenant-id",
user_id="user-id",
conversation_id="conversation-id",
filename="test.pdf", mimetype="application/pdf", tenant_id="tenant-id", user_id="user-id"
)
assert result["data"]["url"] == "https://storage.example.com/signed-upload-url"
@@ -851,10 +851,9 @@ class TestDocumentApiDelete:
class TestDocumentListApi:
"""Test suite for DocumentListApi endpoint."""
@patch("controllers.service_api.dataset.document.paginate_query")
@patch("controllers.service_api.dataset.document.DocumentService")
@patch("controllers.service_api.dataset.document.db")
def test_list_documents_success(self, mock_db, mock_doc_svc, mock_paginate, app: Flask, mock_tenant, mock_dataset):
def test_list_documents_success(self, mock_db, mock_doc_svc, app: Flask, mock_tenant, mock_dataset):
"""Test successful document list retrieval."""
# Arrange
mock_db.session.scalar.return_value = mock_dataset
@@ -869,7 +868,7 @@ class TestDocumentListApi:
make_serializable_document(id="doc-2", name="Document 2"),
]
mock_pagination.total = 2
mock_paginate.return_value = mock_pagination
mock_db.paginate.return_value = mock_pagination
mock_doc_svc.enrich_documents_with_summary_index_status.return_value = None
@@ -14,7 +14,6 @@ import pytest
from agenton.compositor import CompositorSessionSnapshot
from dify_agent.layers.ask_human import AskHumanToolResult
from dify_agent.protocol import (
AgentRunUsage,
CancelRunRequest,
CancelRunResponse,
PydanticAIStreamRunEvent,
@@ -60,6 +59,13 @@ class _FakeCredentialsProvider:
return {"openai_api_key": "sk-test"}
@pytest.fixture(autouse=True)
def _disable_drive_manifest_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
"core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", False
)
class _NoToolsBuilder:
def build_layers(self, **kwargs):
del kwargs
@@ -69,16 +75,12 @@ class _NoToolsBuilder:
class _FakeQueueManager:
def __init__(self) -> None:
self.events: list[Any] = []
self._stop_requested = False
def publish(self, event: Any, _from: Any) -> None:
self.events.append(event)
def is_stopped(self) -> bool:
return self._stop_requested
def request_stop(self) -> None:
self._stop_requested = True
return False
class _StoppedQueueManager(_FakeQueueManager):
@@ -141,65 +143,10 @@ class _StreamingFakeAgentBackendRunClient(FakeAgentBackendRunClient):
data=RunSucceededEventData(
output={"text": "hello agent"},
session_snapshot=CompositorSessionSnapshot(layers=[]),
usage=AgentRunUsage(prompt_tokens=3, completion_tokens=5),
),
)
class _StreamingRecordingFakeAgentBackendRunClient(_RecordingFakeAgentBackendRunClient):
@override
def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
del after
created_at = datetime(2026, 1, 1, tzinfo=UTC)
yield RunStartedEvent(id="1-0", run_id=run_id, created_at=created_at)
yield PydanticAIStreamRunEvent(
id="2-0",
run_id=run_id,
created_at=created_at,
data=PartDeltaEvent(index=0, delta=TextPartDelta(content_delta="hello ")),
)
yield PydanticAIStreamRunEvent(
id="3-0",
run_id=run_id,
created_at=created_at,
data=PartDeltaEvent(index=0, delta=TextPartDelta(content_delta="agent")),
)
yield RunSucceededEvent(
id="4-0",
run_id=run_id,
created_at=created_at,
data=RunSucceededEventData(
output={"text": "hello agent"},
session_snapshot=CompositorSessionSnapshot(layers=[]),
),
)
class _StreamingStopAfterFirstDeltaFakeAgentBackendRunClient(_RecordingFakeAgentBackendRunClient):
def __init__(self, *, queue_manager: _FakeQueueManager, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._queue_manager = queue_manager
@override
def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
del after
created_at = datetime(2026, 1, 1, tzinfo=UTC)
yield RunStartedEvent(id="1-0", run_id=run_id, created_at=created_at)
yield PydanticAIStreamRunEvent(
id="2-0",
run_id=run_id,
created_at=created_at,
data=PartDeltaEvent(index=0, delta=TextPartDelta(content_delta="hello ")),
)
self._queue_manager.request_stop()
yield PydanticAIStreamRunEvent(
id="3-0",
run_id=run_id,
created_at=created_at,
data=PartDeltaEvent(index=0, delta=TextPartDelta(content_delta="agent")),
)
class _StreamingPartStartFakeAgentBackendRunClient(FakeAgentBackendRunClient):
@override
def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
@@ -223,46 +170,6 @@ class _StreamingPartStartFakeAgentBackendRunClient(FakeAgentBackendRunClient):
)
class _NullOutputFakeAgentBackendRunClient(FakeAgentBackendRunClient):
@override
def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
del after
created_at = datetime(2026, 1, 1, tzinfo=UTC)
yield RunStartedEvent(id="1-0", run_id=run_id, created_at=created_at)
yield RunSucceededEvent(
id="2-0",
run_id=run_id,
created_at=created_at,
data=RunSucceededEventData(
output=None,
session_snapshot=CompositorSessionSnapshot(layers=[]),
),
)
class _StreamingTextNullOutputFakeAgentBackendRunClient(FakeAgentBackendRunClient):
@override
def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
del after
created_at = datetime(2026, 1, 1, tzinfo=UTC)
yield RunStartedEvent(id="1-0", run_id=run_id, created_at=created_at)
yield PydanticAIStreamRunEvent(
id="2-0",
run_id=run_id,
created_at=created_at,
data=PartDeltaEvent(index=0, delta=TextPartDelta(content_delta="streamed answer")),
)
yield RunSucceededEvent(
id="3-0",
run_id=run_id,
created_at=created_at,
data=RunSucceededEventData(
output=None,
session_snapshot=CompositorSessionSnapshot(layers=[]),
),
)
class _ProcessStreamingFakeAgentBackendRunClient(FakeAgentBackendRunClient):
@override
def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
@@ -369,19 +276,6 @@ class _FakeSessionStore:
)
class _MonotonicClock:
def __init__(self, *values: float) -> None:
self._values = list(values)
self._index = 0
def __call__(self) -> float:
if self._index >= len(self._values):
return self._values[-1]
value = self._values[self._index]
self._index += 1
return value
def _soul() -> AgentSoulConfig:
return AgentSoulConfig.model_validate(
{
@@ -405,12 +299,7 @@ def _dify_ctx() -> Any:
)
def _runner(
client: FakeAgentBackendRunClient,
store: _FakeSessionStore,
*,
text_delta_debounce_seconds: float | None = 0,
) -> AgentAppRunner:
def _runner(client: FakeAgentBackendRunClient, store: _FakeSessionStore) -> AgentAppRunner:
return AgentAppRunner(
request_builder=AgentAppRuntimeRequestBuilder(
credentials_provider=_FakeCredentialsProvider(),
@@ -419,7 +308,6 @@ def _runner(
agent_backend_client=client,
event_adapter=AgentBackendRunEventAdapter(),
session_store=store, # type: ignore[arg-type]
text_delta_debounce_seconds=text_delta_debounce_seconds,
)
@@ -486,10 +374,15 @@ def test_successful_turn_publishes_chunk_and_message_end_and_saves_session():
assert saved_scope.agent_config_snapshot_id == "snap-1"
assert saved_run_id == "fake-run-1"
assert saved_snapshot is not None
assert saved_specs
# A successful turn carries no ask_human pause correlation.
assert pending_form_id is None
assert pending_tool_call_id is None
assert [spec.name for spec in saved_specs] == [
"agent_soul_prompt",
"agent_app_user_prompt",
"execution_context",
"history",
]
def test_successful_turn_forwards_agent_backend_stream_text_deltas_without_duplicate_terminal_chunk():
@@ -497,16 +390,13 @@ def test_successful_turn_forwards_agent_backend_stream_text_deltas_without_dupli
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store, text_delta_debounce_seconds=0), qm)
_run(_runner(client, store), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
end_events = [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
assert [event.chunk.delta.message.content for event in chunk_events] == ["hello ", "agent"]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == "hello agent"
assert end_events[0].llm_result.usage.prompt_tokens == 3
assert end_events[0].llm_result.usage.completion_tokens == 5
assert end_events[0].llm_result.usage.total_tokens == 8
assert store.saved
@@ -515,7 +405,7 @@ def test_successful_turn_forwards_part_start_text_and_publishes_missing_terminal
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store, text_delta_debounce_seconds=0), qm)
_run(_runner(client, store), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
end_events = [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
@@ -524,34 +414,6 @@ def test_successful_turn_forwards_part_start_text_and_publishes_missing_terminal
assert end_events[0].llm_result.message.content == "hello agent"
def test_successful_turn_with_null_terminal_output_publishes_empty_answer_not_literal_null():
client = _NullOutputFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
end_events = [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
assert chunk_events == []
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == ""
def test_successful_turn_with_streamed_text_and_null_terminal_output_keeps_streamed_answer():
client = _StreamingTextNullOutputFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store, text_delta_debounce_seconds=0), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
end_events = [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
assert [event.chunk.delta.message.content for event in chunk_events] == ["streamed answer"]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == "streamed answer"
def test_successful_turn_persists_thinking_and_tool_process_events(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
@@ -559,7 +421,7 @@ def test_successful_turn_persists_thinking_and_tool_process_events(monkeypatch):
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store, text_delta_debounce_seconds=0), qm)
_run(_runner(client, store), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
assert [event.chunk.delta.message.content for event in chunk_events] == ["final answer"]
@@ -574,50 +436,6 @@ def test_successful_turn_persists_thinking_and_tool_process_events(monkeypatch):
assert rows[1].observation == "ok"
def test_streaming_turn_batches_text_deltas_within_debounce_window(monkeypatch):
monkeypatch.setattr(app_runner_module.time, "monotonic", _MonotonicClock(0.0, 0.2))
client = _StreamingFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store, text_delta_debounce_seconds=0.5), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
end_events = [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
assert [event.chunk.delta.message.content for event in chunk_events] == ["hello agent"]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == "hello agent"
def test_streaming_turn_flushes_pending_text_before_terminal_success(monkeypatch):
monkeypatch.setattr(app_runner_module.time, "monotonic", _MonotonicClock(0.0))
client = _StreamingPartStartFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store, text_delta_debounce_seconds=0.5), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
end_events = [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
assert [event.chunk.delta.message.content for event in chunk_events] == ["hello", " agent"]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == "hello agent"
def test_streaming_turn_flushes_pending_text_before_stop_and_cancel(monkeypatch):
monkeypatch.setattr(app_runner_module.time, "monotonic", _MonotonicClock(0.0))
store = _FakeSessionStore()
qm = _FakeQueueManager()
client = _StreamingStopAfterFirstDeltaFakeAgentBackendRunClient(queue_manager=qm)
with pytest.raises(GenerateTaskStoppedError):
_run(_runner(client, store, text_delta_debounce_seconds=0.5), qm)
chunk_events = [e for e in qm.events if isinstance(e, QueueLLMChunkEvent)]
assert [event.chunk.delta.message.content for event in chunk_events] == ["hello "]
assert client.cancelled_run_ids == ["fake-run-1"]
def test_tool_result_without_identity_does_not_attach_to_previous_tool(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
@@ -793,7 +611,6 @@ def test_stopped_task_cancels_agent_backend_run_and_skips_session_save():
def test_extract_answer_handles_plain_string_and_dict():
assert AgentAppRunner._extract_answer(None) == ""
assert AgentAppRunner._extract_answer("plain text") == "plain text"
assert AgentAppRunner._extract_answer({"text": "hi"}) == "hi"
assert AgentAppRunner._extract_answer({"a": 1}) == '{"a": 1}'
@@ -191,8 +191,6 @@ class TestAgentAppRuntimeRequestBuilder:
"agent_soul_prompt",
"agent_app_user_prompt",
"execution_context",
DIFY_SHELL_LAYER_ID,
DIFY_CONFIG_LAYER_ID,
"history",
"llm",
]
@@ -396,7 +394,10 @@ def _soul_with_model_and_skill() -> AgentSoulConfig:
class TestAgentAppConfigLayer:
def test_config_layer_injected(self):
def test_config_layer_injected_when_flag_enabled(self, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
builder = AgentAppRuntimeRequestBuilder(
credentials_provider=_FakeCredentialsProvider(),
dify_tools_builder=_NoToolsBuilder(), # type: ignore[arg-type]
@@ -422,7 +423,10 @@ class TestAgentAppConfigLayer:
assert names.index(DIFY_SHELL_LAYER_ID) == names.index("execution_context") + 1
assert names.index(DIFY_CONFIG_LAYER_ID) == names.index(DIFY_SHELL_LAYER_ID) + 1
def test_config_layer_present_when_agent_soul_has_no_config_assets(self, monkeypatch: pytest.MonkeyPatch):
def test_no_config_layer_when_agent_soul_has_no_config_assets(self, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
monkeypatch.setattr("core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_SHELL_ENABLED", True)
builder = AgentAppRuntimeRequestBuilder(
credentials_provider=_FakeCredentialsProvider(),
@@ -432,20 +436,14 @@ class TestAgentAppConfigLayer:
result = builder.build(_ctx(_soul_with_model()))
layers = {layer.name: layer for layer in result.request.composition.layers}
assert layers[DIFY_CONFIG_LAYER_ID].config.model_dump(mode="json") == {
"agent_id": "agent-1",
"config_version": {"id": "snap-1", "kind": "snapshot", "writable": False},
"skills": [],
"files": [],
"env_keys": [],
"note": "",
"mentioned_skill_names": [],
"mentioned_file_names": [],
}
assert DIFY_CONFIG_LAYER_ID not in layers
assert layers[DIFY_SHELL_LAYER_ID].deps == {"execution_context": "execution_context"}
assert layers[DIFY_SHELL_LAYER_ID].config.agent_stub_drive_ref is None
def test_config_layer_for_build_draft_marks_config_writable(self):
def test_config_layer_for_build_draft_marks_config_writable(self, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
builder = AgentAppRuntimeRequestBuilder(
credentials_provider=_FakeCredentialsProvider(),
dify_tools_builder=_NoToolsBuilder(), # type: ignore[arg-type]
@@ -472,6 +470,17 @@ class TestAgentAppConfigLayer:
"mentioned_file_names": [],
}
def test_no_config_layer_when_flag_disabled(self, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", False
)
builder = AgentAppRuntimeRequestBuilder(
credentials_provider=_FakeCredentialsProvider(),
dify_tools_builder=_NoToolsBuilder(), # type: ignore[arg-type]
)
result = builder.build(_ctx(_soul_with_model_and_skill()))
assert all(layer.name != DIFY_CONFIG_LAYER_ID for layer in result.request.composition.layers)
@pytest.mark.parametrize(
("system_prompt", "expected_prefix"),
[
@@ -487,9 +496,13 @@ class TestAgentAppConfigLayer:
)
def test_agent_app_runtime_expands_config_mentions_in_agent_soul_prompt(
self,
monkeypatch: pytest.MonkeyPatch,
system_prompt: str,
expected_prefix: str,
):
monkeypatch.setattr(
"core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
soul = _soul_with_model_and_skill()
soul.prompt.system_prompt = system_prompt
builder = AgentAppRuntimeRequestBuilder(
@@ -505,7 +518,11 @@ class TestAgentAppConfigLayer:
def test_agent_app_runtime_missing_config_mentions_fall_back_without_marker_leak(
self,
monkeypatch: pytest.MonkeyPatch,
):
monkeypatch.setattr(
"core.app.apps.agent_app.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
soul = _soul_with_model()
soul.prompt.system_prompt = (
"Use [§skill:ghost-skill:Ghost Skill§], [§file:ghost.txt:Ghost File§], and [§file:no-label.txt§]."
@@ -520,8 +537,3 @@ class TestAgentAppConfigLayer:
prompt_layer = next(layer for layer in result.request.composition.layers if layer.name == "agent_soul_prompt")
assert prompt_layer.config.prefix == "Use Ghost Skill, Ghost File, and no-label.txt."
assert "" not in prompt_layer.config.prefix
assert [warning["code"] for warning in result.metadata["runtime_support"]["unsupported_runtime_warnings"]] == [
"mention_target_missing",
"mention_target_missing",
"mention_target_missing",
]
@@ -14,22 +14,22 @@ from core.helper.marketplace import (
def test_get_plugin_pkg_url_contains_unique_identifier() -> None:
url = get_plugin_pkg_url("langgenius/openai:0.4.2@checksum")
url = get_plugin_pkg_url("plugin@1.0.0")
assert "api/v1/plugins/download" in url
assert "unique_identifier=langgenius%2Fopenai%3A0.4.2%40checksum" in url
assert "unique_identifier=plugin@1.0.0" in url
def test_download_plugin_pkg_delegates_with_configured_size(mocker: MockerFixture) -> None:
mocked_download = mocker.patch("core.helper.marketplace.download_with_size_limit", return_value=b"pkg")
mocker.patch("core.helper.marketplace.dify_config.PLUGIN_MAX_PACKAGE_SIZE", 1234)
result = download_plugin_pkg("langgenius/openai:0.4.2@checksum")
result = download_plugin_pkg("plugin.a.b")
assert result == b"pkg"
mocked_download.assert_called_once()
called_url, called_limit = mocked_download.call_args.args
assert "unique_identifier=langgenius%2Fopenai%3A0.4.2%40checksum" in called_url
assert "unique_identifier=plugin.a.b" in called_url
assert called_limit == 1234
@@ -95,16 +95,6 @@ def _metadata_condition() -> AppMetadataFilteringCondition:
return AppMetadataFilteringCondition(logical_operator="and", conditions=[])
@contextmanager
def _patched_retriever_session():
session = MagicMock()
session_ctx = MagicMock()
session_ctx.__enter__.return_value = session
session_ctx.__exit__.return_value = None
with patch("core.rag.retrieval.dataset_retrieval.session_factory.create_session", return_value=session_ctx):
yield session
def create_side_effect_for_search(documents: list[Document]):
"""
Create a side effect function for mocking search methods.
@@ -1692,15 +1682,7 @@ class TestRetrievalService:
# Mock _retriever to return documents
def side_effect_retriever(
flask_app,
session,
dataset_id,
query,
top_k,
all_documents,
document_ids_filter,
metadata_condition,
attachment_ids,
flask_app, dataset_id, query, top_k, all_documents, document_ids_filter, metadata_condition, attachment_ids
):
all_documents.extend([doc1, doc2])
@@ -1712,24 +1694,23 @@ class TestRetrievalService:
all_documents = []
# Act - Call with dataset_count = 1
with _patched_retriever_session():
dataset_retrieval._multiple_retrieve_thread(
flask_app=mock_flask_app,
available_datasets=[mock_dataset],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True,
reranking_mode="reranking_model",
reranking_model={"reranking_provider_name": "cohere", "reranking_model_name": "rerank-v2"},
weights=None,
top_k=5,
score_threshold=0.5,
query="test query",
attachment_id=None,
dataset_count=1, # Single dataset - should skip second reranking
)
dataset_retrieval._multiple_retrieve_thread(
flask_app=mock_flask_app,
available_datasets=[mock_dataset],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True,
reranking_mode="reranking_model",
reranking_model={"reranking_provider_name": "cohere", "reranking_model_name": "rerank-v2"},
weights=None,
top_k=5,
score_threshold=0.5,
query="test query",
attachment_id=None,
dataset_count=1, # Single dataset - should skip second reranking
)
# Assert
# DataPostProcessor should NOT be called (second reranking skipped)
@@ -1776,15 +1757,7 @@ class TestRetrievalService:
# Mock _retriever to return documents
def side_effect_retriever(
flask_app,
session,
dataset_id,
query,
top_k,
all_documents,
document_ids_filter,
metadata_condition,
attachment_ids,
flask_app, dataset_id, query, top_k, all_documents, document_ids_filter, metadata_condition, attachment_ids
):
all_documents.extend([doc1, doc2])
@@ -1820,24 +1793,23 @@ class TestRetrievalService:
mock_dataset2.provider = "dify"
# Act - Call with dataset_count = 2
with _patched_retriever_session():
dataset_retrieval._multiple_retrieve_thread(
flask_app=mock_flask_app,
available_datasets=[mock_dataset, mock_dataset2],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True,
reranking_mode="reranking_model",
reranking_model={"reranking_provider_name": "cohere", "reranking_model_name": "rerank-v2"},
weights=None,
top_k=5,
score_threshold=0.5,
query="test query",
attachment_id=None,
dataset_count=2, # Multiple datasets - should perform second reranking
)
dataset_retrieval._multiple_retrieve_thread(
flask_app=mock_flask_app,
available_datasets=[mock_dataset, mock_dataset2],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True,
reranking_mode="reranking_model",
reranking_model={"reranking_provider_name": "cohere", "reranking_model_name": "rerank-v2"},
weights=None,
top_k=5,
score_threshold=0.5,
query="test query",
attachment_id=None,
dataset_count=2, # Multiple datasets - should perform second reranking
)
# Assert
# DataPostProcessor SHOULD be called (second reranking performed)
@@ -1895,15 +1867,7 @@ class TestRetrievalService:
# Mock _retriever to return documents
def side_effect_retriever(
flask_app,
session,
dataset_id,
query,
top_k,
all_documents,
document_ids_filter,
metadata_condition,
attachment_ids,
flask_app, dataset_id, query, top_k, all_documents, document_ids_filter, metadata_condition, attachment_ids
):
all_documents.extend([doc1, doc2])
@@ -1925,24 +1889,23 @@ class TestRetrievalService:
all_documents = []
# Act - Call with dataset_count = 1
with _patched_retriever_session():
dataset_retrieval._multiple_retrieve_thread(
flask_app=mock_flask_app,
available_datasets=[mock_dataset],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True, # Reranking enabled but should be skipped for single dataset
reranking_mode="reranking_model",
reranking_model={"reranking_provider_name": "cohere", "reranking_model_name": "rerank-v2"},
weights=None,
top_k=5,
score_threshold=0.5,
query="test query",
attachment_id=None,
dataset_count=1,
)
dataset_retrieval._multiple_retrieve_thread(
flask_app=mock_flask_app,
available_datasets=[mock_dataset],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True, # Reranking enabled but should be skipped for single dataset
reranking_mode="reranking_model",
reranking_model={"reranking_provider_name": "cohere", "reranking_model_name": "rerank-v2"},
weights=None,
top_k=5,
score_threshold=0.5,
query="test query",
attachment_id=None,
dataset_count=1,
)
# Assert
# DataPostProcessor should NOT be called
@@ -3757,15 +3720,7 @@ class TestKnowledgeRetrievalRegression:
)
def fake_retriever(
flask_app,
session,
dataset_id,
query,
top_k,
all_documents,
document_ids_filter,
metadata_condition,
attachment_ids,
flask_app, dataset_id, query, top_k, all_documents, document_ids_filter, metadata_condition, attachment_ids
):
all_documents.append(document)
@@ -3789,35 +3744,32 @@ class TestKnowledgeRetrievalRegression:
thread_exceptions: list[Exception] = []
def target():
with (
patch.object(dataset_retrieval, "_retriever", side_effect=fake_retriever),
patch(
with patch.object(dataset_retrieval, "_retriever", side_effect=fake_retriever):
with patch(
"core.rag.retrieval.dataset_retrieval.DataPostProcessor",
ContextRequiredPostProcessor,
),
_patched_retriever_session(),
):
dataset_retrieval._multiple_retrieve_thread(
flask_app=flask_app,
available_datasets=[mock_dataset, secondary_dataset],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True,
reranking_mode="reranking_model",
reranking_model={
"reranking_provider_name": "cohere",
"reranking_model_name": "rerank-v2",
},
weights=None,
top_k=3,
score_threshold=0.0,
query="test query",
attachment_id=None,
dataset_count=2, # force reranking branch
thread_exceptions=thread_exceptions,
)
):
dataset_retrieval._multiple_retrieve_thread(
flask_app=flask_app,
available_datasets=[mock_dataset, secondary_dataset],
metadata_condition=None,
metadata_filter_document_ids=None,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=True,
reranking_mode="reranking_model",
reranking_model={
"reranking_provider_name": "cohere",
"reranking_model_name": "rerank-v2",
},
weights=None,
top_k=3,
score_threshold=0.0,
query="test query",
attachment_id=None,
dataset_count=2, # force reranking branch
thread_exceptions=thread_exceptions, # ✅ key
)
t = threading.Thread(target=target)
t.start()
@@ -3829,53 +3781,6 @@ class TestKnowledgeRetrievalRegression:
# Current buggy code should record an exception (not raise it)
assert not thread_exceptions, thread_exceptions
def test_run_retriever_thread_provides_session_to_retriever(self):
dataset_retrieval = DatasetRetrieval()
all_documents: list[Document] = []
with _patched_retriever_session() as session:
with patch.object(dataset_retrieval, "_retriever") as mock_retriever:
dataset_retrieval._run_retriever_thread(
flask_app=_FakeFlaskApp(),
dataset_id="dataset-1",
query="test query",
top_k=3,
all_documents=all_documents,
document_ids_filter=None,
metadata_condition=None,
attachment_ids=None,
cancel_event=None,
thread_exceptions=[],
)
mock_retriever.assert_called_once()
assert mock_retriever.call_args.kwargs["session"] is session
def test_run_retriever_thread_records_retriever_exception(self):
dataset_retrieval = DatasetRetrieval()
all_documents: list[Document] = []
cancel_event = threading.Event()
thread_exceptions: list[Exception] = []
expected_error = RuntimeError("retrieval failed")
with _patched_retriever_session():
with patch.object(dataset_retrieval, "_retriever", side_effect=expected_error):
dataset_retrieval._run_retriever_thread(
flask_app=_FakeFlaskApp(),
dataset_id="dataset-1",
query="test query",
top_k=3,
all_documents=all_documents,
document_ids_filter=None,
metadata_condition=None,
attachment_ids=None,
cancel_event=cancel_event,
thread_exceptions=thread_exceptions,
)
assert cancel_event.is_set()
assert thread_exceptions == [expected_error]
class _FakeFlaskApp:
def app_context(self):
@@ -138,7 +138,6 @@ def test_get_signed_file_url_for_plugin_and_verify_roundtrip(monkeypatch: pytest
mimetype="application/pdf",
tenant_id="tenant-id",
user_id="user-id",
conversation_id="conversation-id",
)
parsed = urlparse(url)
query = parse_qs(parsed.query)
@@ -147,14 +146,12 @@ def test_get_signed_file_url_for_plugin_and_verify_roundtrip(monkeypatch: pytest
assert parsed.path == "/files/upload/for-plugin"
assert query["tenant_id"] == ["tenant-id"]
assert query["user_id"] == ["user-id"]
assert query["conversation_id"] == ["conversation-id"]
assert (
verify_plugin_file_signature(
filename="report.pdf",
mimetype="application/pdf",
tenant_id="tenant-id",
user_id="user-id",
conversation_id="conversation-id",
timestamp=query["timestamp"][0],
nonce=query["nonce"][0],
sign=query["sign"][0],
@@ -2,6 +2,7 @@ from types import SimpleNamespace
from typing import cast
from unittest.mock import MagicMock, patch
import pytest
from agenton.compositor import CompositorSessionSnapshot
from dify_agent.layers.ask_human import AskHumanToolResult
from dify_agent.protocol import RunStartedEvent, RunSucceededEvent, RunSucceededEventData
@@ -50,6 +51,13 @@ class FakeCredentialsProvider:
return {"api_key": "secret-key"}
@pytest.fixture(autouse=True)
def _disable_drive_manifest_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", False
)
def _restored_file(*, transfer_method: FileTransferMethod, reference: str) -> File:
return File(
type=FileType.DOCUMENT,
@@ -273,8 +281,7 @@ def test_agent_node_run_maps_successful_agent_backend_run_to_node_result():
assert agent_log["agent_backend"]["run_id"] == "fake-run-1"
assert agent_log["agent_backend"]["status"] == "succeeded"
assert result.process_data["agent_id"] == "agent-1"
layers = {layer["name"]: layer for layer in result.inputs["agent_backend_request"]["composition"]["layers"]}
assert layers["llm"]["config"]["credentials"] == "[REDACTED]"
assert result.inputs["agent_backend_request"]["composition"]["layers"][5]["config"]["credentials"] == "[REDACTED]"
def test_agent_node_run_normalizes_declared_file_output_with_canonical_mapping():
@@ -148,31 +148,6 @@ def _file_tool() -> FakeTool:
return FakeTool(entity=entity, runtime=runtime)
def _files_tool() -> FakeTool:
parameters = [
ToolParameter(
name="documents",
label=I18nObject(en_US="Documents"),
type=ToolParameter.ToolParameterType.FILES,
form=ToolParameter.ToolParameterForm.LLM,
required=True,
llm_description="The documents to inspect.",
)
]
entity = ToolEntity(
identity=ToolIdentity(
author="langgenius",
name="inspect",
label=I18nObject(en_US="Inspect"),
provider="documents",
),
description=ToolDescription(human=I18nObject(en_US="Inspect"), llm="Inspect documents."),
parameters=parameters,
)
runtime = ToolRuntime(tenant_id="tenant-1", user_id="user-1", credentials={}, runtime_parameters={})
return FakeTool(entity=entity, runtime=runtime)
def _tts_tool() -> FakeTool:
parameters = [
ToolParameter(
@@ -299,103 +274,6 @@ def test_builds_core_tool_with_file_llm_parameter():
assert runtime_provider.last_use_default_for_missing_form_parameters is True
def test_builds_plugin_tool_with_file_llm_parameter_schema():
runtime_provider = FakeRuntimeProvider(_file_tool())
builder = WorkflowAgentDifyToolsBuilder(tool_runtime_provider=runtime_provider)
tools = AgentSoulToolsConfig.model_validate(
{
"dify_tools": [
{
"provider_id": "langgenius/audio/audio",
"provider_type": "plugin",
"tool_name": "asr",
"credential_type": "unauthorized",
}
]
}
)
result = _build(builder, tools)
assert result is not None
schema = result.tools[0].parameters_json_schema
file_schema = schema["properties"]["audio_file"]
assert file_schema["anyOf"][0]["type"] == "string"
assert file_schema["anyOf"][1]["properties"]["transfer_method"]["enum"] == ["remote_url"]
assert file_schema["anyOf"][2]["properties"]["transfer_method"]["enum"] == [
"local_file",
"tool_file",
"datasource_file",
]
assert schema["required"] == ["audio_file"]
def test_builds_plugin_tool_with_files_llm_parameter_schema():
runtime_provider = FakeRuntimeProvider(_files_tool())
builder = WorkflowAgentDifyToolsBuilder(tool_runtime_provider=runtime_provider)
tools = AgentSoulToolsConfig.model_validate(
{
"dify_tools": [
{
"provider_id": "langgenius/documents/documents",
"provider_type": "plugin",
"tool_name": "inspect",
"credential_type": "unauthorized",
}
]
}
)
result = _build(builder, tools)
assert result is not None
schema = result.tools[0].parameters_json_schema
files_schema = schema["properties"]["documents"]
assert files_schema["type"] == "array"
assert files_schema["items"]["anyOf"][0]["description"] == "HTTP(S) URL or sandbox-local file path."
assert schema["required"] == ["documents"]
def test_builds_builtin_compat_plugin_tool_with_files_llm_parameter_schema():
runtime_provider = FakeRuntimeProvider(_files_tool())
builder = WorkflowAgentDifyToolsBuilder(tool_runtime_provider=runtime_provider)
tools = AgentSoulToolsConfig.model_validate(
{
"dify_tools": [
{
"provider_id": "langgenius/dify-gmail/dify-gmail",
"provider_type": "builtin",
"provider": "langgenius/dify-gmail/dify-gmail",
"tool_name": "add_attachment_to_draft",
"credential_type": "api-key",
"credential_id": "credential-1",
}
]
}
)
result = builder.build_layers(
tenant_id="tenant-1",
app_id="app-1",
user_id="user-1",
tools=tools,
invoke_from=InvokeFrom.DEBUGGER,
)
assert result.plugin_tools is not None
assert result.core_tools is None
prepared = result.plugin_tools.tools[0]
assert prepared.plugin_id == "langgenius/dify-gmail"
assert prepared.provider == "dify-gmail"
assert prepared.tool_name == "add_attachment_to_draft"
files_schema = prepared.parameters_json_schema["properties"]["documents"]
assert files_schema["type"] == "array"
assert files_schema["items"]["anyOf"][0]["description"] == "HTTP(S) URL or sandbox-local file path."
assert prepared.parameters_json_schema["required"] == ["documents"]
assert runtime_provider.last_agent_tool is not None
assert runtime_provider.last_agent_tool.provider_type.value == "builtin"
def test_build_layers_routes_plugin_direct_and_builtin_via_core() -> None:
runtime_provider = FakeRuntimeProvider(_tool())
builder = WorkflowAgentDifyToolsBuilder(tool_runtime_provider=runtime_provider)
@@ -47,6 +47,13 @@ class FakeCredentialsProvider:
return {"api_key": "secret-key"}
@pytest.fixture(autouse=True)
def _disable_drive_manifest_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", False
)
class CapturingCredentialsProvider:
def __init__(self) -> None:
self.provider_name: str | None = None
@@ -484,8 +491,7 @@ def test_build_shell_layer_config_accepts_legacy_fallback_keys():
"secret_refs": [
{"variable": "TOKEN", "credential_id": "credential-1"},
{"name": "API_KEY", "provider_credential_id": "credential-2"},
{"name": "EDITABLE_TOKEN", "value": "inline-secret-value"},
{"name": "LEGACY_SECRET_REF", "id": "credential-3"},
{"name": "EDITABLE_TOKEN", "value": "credential-3"},
{"ref": "missing-name"},
],
},
@@ -502,12 +508,11 @@ def test_build_shell_layer_config_accepts_legacy_fallback_keys():
assert config["env"] == [
{"name": "PROJECT_NAME", "value": "demo"},
{"name": "RETRY_COUNT", "value": "3"},
{"name": "EDITABLE_TOKEN", "value": "inline-secret-value"},
]
assert config["secret_refs"] == [
{"name": "TOKEN", "ref": "credential-1"},
{"name": "API_KEY", "ref": "credential-2"},
{"name": "LEGACY_SECRET_REF", "ref": "credential-3"},
{"name": "EDITABLE_TOKEN", "ref": "credential-3"},
]
assert config["sandbox"] is None
@@ -608,37 +613,7 @@ def test_build_shell_layer_config_maps_cli_tool_scoped_env():
]
def test_build_shell_layer_config_maps_cli_tool_inline_secret_value_to_env():
agent_soul = AgentSoulConfig.model_validate(
{
"tools": {
"cli_tools": [
{
"name": "github",
"command": "apt-get install -y gh",
"env": {
"secret_refs": [{"name": "GITHUB_TOKEN", "value": "ghp_" + "x" * 300}],
},
}
]
}
}
)
config = build_shell_layer_config(agent_soul).model_dump(mode="json")
assert config["cli_tools"] == [
{
"name": "github",
"install_commands": ["apt-get install -y gh"],
"env": [{"name": "GITHUB_TOKEN", "value": "ghp_" + "x" * 300}],
"secret_refs": [],
}
]
def test_builds_workflow_run_request_with_dify_plugin_tools_layer(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr("core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_SHELL_ENABLED", True)
def test_builds_workflow_run_request_with_dify_plugin_tools_layer():
context = _context()
snapshot = AgentConfigSnapshot(
id="snapshot-1",
@@ -674,10 +649,7 @@ def test_builds_workflow_run_request_with_dify_plugin_tools_layer(monkeypatch: p
dumped = result.request.model_dump(mode="json")
layers = {layer["name"]: layer for layer in dumped["composition"]["layers"]}
assert layers[DIFY_PLUGIN_TOOLS_LAYER_ID]["type"] == "dify.plugin.tools"
assert layers[DIFY_PLUGIN_TOOLS_LAYER_ID]["deps"] == {
"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID,
"shell": DIFY_SHELL_LAYER_ID,
}
assert layers[DIFY_PLUGIN_TOOLS_LAYER_ID]["deps"] == {"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID}
assert layers[DIFY_PLUGIN_TOOLS_LAYER_ID]["config"]["tools"][0]["tool_name"] == "current_time"
assert result.metadata["agent_tools"] == {
"dify_tool_count": 1,
@@ -1328,7 +1300,7 @@ def test_build_config_layer_config_includes_soul_context_and_mentions():
assert warnings == []
def test_build_config_layer_config_returns_empty_config_for_empty_agent_soul():
def test_build_config_layer_config_returns_none_for_empty_agent_soul():
from core.workflow.nodes.agent_v2.runtime_request_builder import build_config_layer_config
soul = AgentSoulConfig(
@@ -1336,43 +1308,30 @@ def test_build_config_layer_config_returns_empty_config_for_empty_agent_soul():
)
config, warnings = build_config_layer_config(soul)
assert config is not None
assert config.model_dump(mode="json") == {
"agent_id": None,
"config_version": {"id": None, "kind": "snapshot", "writable": False},
"skills": [],
"files": [],
"env_keys": [],
"note": "",
"mentioned_skill_names": [],
"mentioned_file_names": [],
}
assert config is None
assert warnings == []
def test_workflow_run_request_has_config_layer_with_empty_agent_soul(monkeypatch: pytest.MonkeyPatch):
def test_workflow_run_request_has_no_config_layer_with_empty_agent_soul(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
monkeypatch.setattr("core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_SHELL_ENABLED", True)
result = WorkflowAgentRuntimeRequestBuilder(credentials_provider=FakeCredentialsProvider()).build(_context())
dumped = result.request.model_dump(mode="json")
layers = {layer["name"]: layer for layer in dumped["composition"]["layers"]}
assert layers[DIFY_CONFIG_LAYER_ID]["config"] == {
"agent_id": "agent-1",
"config_version": {"id": "snapshot-1", "kind": "snapshot", "writable": False},
"skills": [],
"files": [],
"env_keys": [],
"note": "",
"mentioned_skill_names": [],
"mentioned_file_names": [],
}
assert DIFY_CONFIG_LAYER_ID not in layers
assert layers[DIFY_SHELL_LAYER_ID]["deps"] == {"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID}
assert layers[DIFY_SHELL_LAYER_ID]["config"]["agent_stub_drive_ref"] is None
def test_workflow_run_request_contains_config_layer():
def test_workflow_run_request_contains_config_layer_when_flag_enabled(monkeypatch: pytest.MonkeyPatch):
"""Contract test: locks the dify.config composition shape against cross-package drift."""
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
context = _context()
context.snapshot.config_snapshot = _soul_with_config_assets()
@@ -1413,7 +1372,10 @@ def test_workflow_run_request_contains_config_layer():
assert any(spec.name == DIFY_CONFIG_LAYER_ID and spec.type == "dify.config" for spec in specs)
def test_workflow_runtime_expands_config_mentions_in_agent_soul_prompt():
def test_workflow_runtime_expands_config_mentions_in_agent_soul_prompt(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
context = _context()
context.snapshot.config_snapshot = _soul_with_config_assets()
@@ -1424,7 +1386,10 @@ def test_workflow_runtime_expands_config_mentions_in_agent_soul_prompt():
assert "" not in soul_prompt.config.prefix
def test_workflow_runtime_missing_config_mentions_fall_back_to_label_then_name():
def test_workflow_runtime_missing_config_mentions_fall_back_to_label_then_name(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", True
)
context = _context()
context.snapshot.config_snapshot = AgentSoulConfig(
prompt={
@@ -1440,11 +1405,20 @@ def test_workflow_runtime_missing_config_mentions_fall_back_to_label_then_name()
soul_prompt = next(layer for layer in result.request.composition.layers if layer.name == "agent_soul_prompt")
assert soul_prompt.config.prefix == "Use Ghost Skill, Ghost File, and no-label.txt."
assert "" not in soul_prompt.config.prefix
assert [warning["code"] for warning in result.metadata["runtime_support"]["unsupported_runtime_warnings"]] == [
"mention_target_missing",
"mention_target_missing",
"mention_target_missing",
]
def test_workflow_run_request_has_no_config_layer_when_flag_disabled(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.runtime_request_builder.dify_config.AGENT_DRIVE_MANIFEST_ENABLED", False
)
context = _context()
context.snapshot.config_snapshot = _soul_with_config_assets()
result = WorkflowAgentRuntimeRequestBuilder(credentials_provider=FakeCredentialsProvider()).build(context)
dumped = result.request.model_dump(mode="json")
assert all(layer["name"] != DIFY_CONFIG_LAYER_ID for layer in dumped["composition"]["layers"])
assert result.metadata["runtime_support"]["unsupported_runtime_warnings"] == []
def test_build_config_layer_config_missing_mentions_warn_without_catalog():
@@ -283,7 +283,6 @@ class TestExtractFilename:
result = extract_filename("http://example.com/path/file%20name%GG.txt?x=1", None)
# %GG is invalid, should be replaced with replacement character
assert result is not None
assert "file" in result
assert ".txt" in result

Some files were not shown because too many files have changed in this diff Show More