mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-26 19:04:02 +08:00
fix: fallback to sync call in lfx run when stream=true (#12906)
* fix(lfx): unblock streaming flows in lfx run
LCModelComponent._handle_stream tried to persist a partial Message via
send_message whenever the LM was wired to ChatOutput, but the message-
store path requires session_id and the chunk-consumption path requires
an EventManager. lfx run had neither, so flows with stream=True crashed
in astore_message ("session_id, sender, sender_name must be provided").
- run_flow now auto-generates a session_id when none is supplied so the
message-store validator passes; an explicit value still wins.
- lfx run gains a --session-id flag for memory continuity across runs
(Memory / MessageHistory components keyed on session_id).
- _handle_stream now also requires an EventManager before taking the
streaming branch — without one, the chunk iterator would be stored
but never drained, surfacing as an empty result downstream. Falls
back to ainvoke and returns the full text instead.
Tests cover the autogen, caller-precedence, and uniqueness cases on
run_flow plus the four _handle_stream branches (no session_id, no
event_manager, both present, not connected to chat output).
* fix(lfx): autogen session_id in CUGA agent when graph has none
Matches the pattern already used by base/agents/agent.py and
base/agents/altk_base_agent.py. Without this, calling the CUGA agent
outside run_flow (which now autogens a session_id) would still fail
astore_message validation.
* [autofix.ci] apply automated fixes
* fix(lfx): plumb session_id through serve /run and /stream
StreamRequest already declared a session_id field but it was never
applied to the graph; RunRequest didn't have the field at all. Both
endpoints called execute_graph_with_capture, which executed against
an empty graph.session_id, so message-store paths skipped storage
silently and Memory components could not maintain continuity.
- Add session_id to RunRequest.
- Have execute_graph_with_capture accept session_id, autogen if empty
(matches run_flow), and apply it to graph.session_id before
execution.
- Forward session_id from both /run and /stream handlers.
* fix(lfx): propagate session_id and user_id so memory works on lfx run
The lfx run path uses graph.async_start instead of graph.arun, bypassing
the has_session_id_vertices propagation loop in Graph._run that the
playground hits via build_graph_from_data. Memory/MessageHistory
components reading session_id from their input field would see "" even
when --session-id was passed. Replicate the loop in run_flow and
execute_graph_with_capture, with the same precedence as the playground:
hardcoded values on the component win.
AgentComponent's variable lookup precheck blocks any flow that resolves
variables (e.g. api_key) when graph.user_id is empty. Auto-generate a
ceremonial UUID; lfx's env-fallback variable service ignores user_id, so
this only satisfies the precheck — env vars remain process-global with
no per-user scoping.
Make VariableService.get_variable async to match the langflow call site
(`await variable_service.get_variable(...)` in custom_component.get_variable).
The signature accepts and ignores user_id/field/session kwargs so flows
behave identically under either backend.
Tests cover propagation precedence (empty input filled, hardcoded value
preserved, missing vertex skipped), user_id auto-gen and caller-takes-
precedence, and the async signature with kwarg absorption.
* [autofix.ci] apply automated fixes
* [autofix.ci] apply automated fixes (attempt 2/3)
* [autofix.ci] apply automated fixes (attempt 3/3)
* fix(lfx): plumb fallback_to_env_vars so DatabaseVariableService works in lfx run
The lfx run path uses graph.async_start, which never propagated
fallback_to_env_vars to vertex builds — the flag defaulted to False all
the way down to update_params_with_load_from_db_fields. Result: if a user
swapped lfx's env-fallback VariableService for langflow's
DatabaseVariableService (via lfx.toml), every load_from_db variable
(e.g. api_key=OPENAI_API_KEY) would raise "variable not found" because
the random ceremonial user_id has no DB rows, with no env fallback.
Add fallback_to_env_vars kwarg to async_start and astep (default False,
non-breaking for existing callers). run_flow and execute_graph_with_capture
read settings.fallback_to_env_var (defaults True, settable via
LANGFLOW_FALLBACK_TO_ENV_VAR=false) and pass it through. Mirrors what
processing.process.run_graph_internal does for the langflow API path.
Make memory.stubs.astore_message tolerant of non-UUID flow_ids: the stub
is the no-op fallback when no real database is registered, so it should
not crash on synthetic identifiers (e.g. test fixtures, lfx callers
passing string flow ids). UUID parsing only normalizes format; an
invalid string is preserved verbatim.
Tests:
- TestRunFlowFallbackToEnvVars: confirms run_flow forwards
fallback_to_env_vars from settings (default and disabled).
- TestGraphExecution: same for execute_graph_with_capture.
- Existing mock_async_start signatures updated to accept **kwargs.
- Test fixtures using flow_id="test-flow-id" replaced with a real UUID
so the now-active ChatInput storage path doesn't trip stubs.py's
UUID parse — aligning fixtures with production semantics.
* test(lfx): pass session_id=None when invoking the typer run() directly
The lfx test suite calls ``run(...)`` (the typer command) without going through
typer's CLI parser. Any parameter with a ``typer.Option(...)`` default in the
signature (e.g. ``session_id``) evaluates to a ``typer.models.OptionInfo``
sentinel under that invocation pattern, not None. The session_id propagation
loop then writes that sentinel into ``vertex.raw_params["session_id"]``, which
fails ``MessageTextInput`` validation with
``Invalid value type <class 'typer.models.OptionInfo'>``.
Fix at the call site: pass ``session_id=None`` explicitly. Mirrors how typer
would resolve the option after parsing CLI args. Two tests affected;
test_run_command.py now reflects the constraint that calls bypassing typer
must pass all option-shaped args.
* fix(lfx): harden session_id/user_id handling on lfx run path
Addresses review findings on the streaming-fix PR:
- Reject empty/whitespace --session-id and --user-id up-front so a shell
quirk or empty env var surfaces a clear error instead of silently
auto-generating a fresh session and breaking Memory continuity.
- Extract a shared helper (lfx/run/_defaults.py) for session_id/user_id
auto-gen, vertex propagation, and fallback_to_env_vars resolution; both
run_flow and execute_graph_with_capture now delegate to it.
- Warn when settings_service is None (silently flipped fallback_to_env_vars
to False before).
- CUGA agent: wrap uuid.uuid4() with str() to match the rest of the file
and the codebase's expectation that Message.session_id renders as a hex
string. Add focused tests with module-level skip when the cuga import
side-effect (MODELS_METADATA["OpenAI"]) isn't satisfiable.
- Streaming-fallback warning now names the component (display_name + _id)
so users can identify which model fell back to ainvoke.
- memory/stubs.py: replace contextlib.suppress(ValueError) with explicit
try/except + warning so malformed flow_ids leave a breadcrumb.
- Improve --session-id help text to explain WHEN to set it.
- Tune session_id auto-gen log level from warning to info (less noisy on
default CLI runs; visible at -v).
- Add tests: --session-id CLI plumbing, multi-call continuity, empty/
whitespace rejection, and isinstance(str) on the streaming fallback path.
* [autofix.ci] apply automated fixes
---------
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Jordan Frazier
@ -1319,7 +1319,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
|
||||
@ -1521,7 +1521,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "lfx",
|
||||
|
||||
@ -2084,7 +2084,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -1192,7 +1192,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -545,7 +545,7 @@
|
||||
},
|
||||
{
|
||||
"name": "cryptography",
|
||||
"version": "46.0.7"
|
||||
"version": "47.0.0"
|
||||
},
|
||||
{
|
||||
"name": "langchain_chroma",
|
||||
@ -565,7 +565,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_openai",
|
||||
"version": "1.2.0"
|
||||
"version": "1.2.1"
|
||||
},
|
||||
{
|
||||
"name": "langchain_huggingface",
|
||||
@ -585,7 +585,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_ibm",
|
||||
"version": "1.0.6"
|
||||
"version": "1.0.7"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 12
|
||||
|
||||
@ -1204,7 +1204,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -1186,7 +1186,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -813,7 +813,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
@ -2105,7 +2105,7 @@
|
||||
"dependencies": [
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
|
||||
@ -1251,7 +1251,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -941,7 +941,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
|
||||
@ -1620,7 +1620,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -2823,7 +2823,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -905,7 +905,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -952,7 +952,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -370,7 +370,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
@ -958,7 +958,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
@ -2403,7 +2403,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
@ -2976,7 +2976,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
|
||||
@ -951,7 +951,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -160,7 +160,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
@ -392,7 +392,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
@ -1301,7 +1301,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -2633,7 +2633,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
|
||||
@ -1717,7 +1717,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
@ -2300,7 +2300,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
@ -2883,7 +2883,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
@ -1635,7 +1635,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
},
|
||||
{
|
||||
"name": "pydantic",
|
||||
@ -2687,7 +2687,7 @@
|
||||
},
|
||||
{
|
||||
"name": "cryptography",
|
||||
"version": "46.0.7"
|
||||
"version": "47.0.0"
|
||||
},
|
||||
{
|
||||
"name": "langchain_chroma",
|
||||
@ -3057,7 +3057,7 @@
|
||||
},
|
||||
{
|
||||
"name": "cryptography",
|
||||
"version": "46.0.7"
|
||||
"version": "47.0.0"
|
||||
},
|
||||
{
|
||||
"name": "langchain_chroma",
|
||||
@ -3077,7 +3077,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_openai",
|
||||
"version": "1.2.0"
|
||||
"version": "1.2.1"
|
||||
},
|
||||
{
|
||||
"name": "langchain_huggingface",
|
||||
@ -3097,7 +3097,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_ibm",
|
||||
"version": "1.0.6"
|
||||
"version": "1.0.7"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 12
|
||||
|
||||
@ -513,7 +513,7 @@
|
||||
},
|
||||
{
|
||||
"name": "langchain_core",
|
||||
"version": "1.3.1"
|
||||
"version": "1.3.2"
|
||||
}
|
||||
],
|
||||
"total_dependencies": 3
|
||||
|
||||
File diff suppressed because one or more lines are too long
@ -13,6 +13,7 @@ from lfx.base.constants import STREAM_INFO_TEXT
|
||||
from lfx.custom.custom_component.component import Component
|
||||
from lfx.field_typing import LanguageModel
|
||||
from lfx.inputs.inputs import BoolInput, InputTypes, MessageInput, MultilineInput
|
||||
from lfx.log.logger import logger
|
||||
from lfx.schema.message import Message
|
||||
from lfx.schema.properties import Usage
|
||||
from lfx.schema.token_usage import extract_usage_from_message
|
||||
@ -340,16 +341,40 @@ class LCModelComponent(Component):
|
||||
session_id = self._session_id
|
||||
else:
|
||||
session_id = None
|
||||
model_message = Message(
|
||||
text=runnable.astream(inputs),
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="AI",
|
||||
properties={"icon": self.icon, "state": "partial"},
|
||||
session_id=session_id,
|
||||
)
|
||||
model_message.properties.source = self._build_source(self._id, self.display_name, self)
|
||||
lf_message = await self.send_message(model_message)
|
||||
result = lf_message.text or ""
|
||||
# Streaming requires both a session_id and an event_manager:
|
||||
# - session_id is required so astore_message validation passes when send_message
|
||||
# persists the placeholder Message.
|
||||
# - event_manager is required so the chunk iterator that backs Message.text gets
|
||||
# drained; without one, no consumer iterates astream(), the iterator is stored
|
||||
# verbatim, and downstream readers see empty text.
|
||||
# If either is missing, fall back to a non-streaming ainvoke.
|
||||
event_manager = getattr(self, "_event_manager", None)
|
||||
if session_id and event_manager:
|
||||
model_message = Message(
|
||||
text=runnable.astream(inputs),
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="AI",
|
||||
properties={"icon": self.icon, "state": "partial"},
|
||||
session_id=session_id,
|
||||
)
|
||||
model_message.properties.source = self._build_source(self._id, self.display_name, self)
|
||||
lf_message = await self.send_message(model_message)
|
||||
result = lf_message.text or ""
|
||||
else:
|
||||
missing = []
|
||||
if not session_id:
|
||||
missing.append("session_id")
|
||||
if not event_manager:
|
||||
missing.append("event_manager")
|
||||
component_label = getattr(self, "display_name", None) or getattr(self, "_id", "<unknown>")
|
||||
logger.warning(
|
||||
f"Streaming fallback to ainvoke for component '{component_label}' "
|
||||
f"(id={getattr(self, '_id', '<unknown>')}): missing {', '.join(missing)}. "
|
||||
"UI will not see token-by-token streaming for this run."
|
||||
)
|
||||
ai_message = await runnable.ainvoke(inputs)
|
||||
result = ai_message.content if hasattr(ai_message, "content") else ai_message
|
||||
result = _normalize_message_content(result)
|
||||
else:
|
||||
ai_message = await runnable.ainvoke(inputs)
|
||||
result = ai_message.content if hasattr(ai_message, "content") else ai_message
|
||||
|
||||
@ -60,6 +60,13 @@ def register(app: typer.Typer) -> None:
|
||||
show_default=True,
|
||||
help="Include detailed timing information in output",
|
||||
),
|
||||
session_id: str | None = typer.Option(
|
||||
None,
|
||||
"--session-id",
|
||||
help=(
|
||||
"Session ID to attach to the run. Agent and Memory Components will use this to track conversation history."
|
||||
),
|
||||
),
|
||||
) -> None:
|
||||
"""Run a flow directly (lazy-loaded)."""
|
||||
from pathlib import Path
|
||||
@ -81,6 +88,7 @@ def register(app: typer.Typer) -> None:
|
||||
verbose_detailed=verbose_detailed,
|
||||
verbose_full=verbose_full,
|
||||
timing=timing,
|
||||
session_id=session_id,
|
||||
)
|
||||
|
||||
@app.command(name="serve", help="Serve a flow as an API", no_args_is_help=True, rich_help_panel="Running")
|
||||
|
||||
@ -29,6 +29,7 @@ from lfx.cli.script_loader import (
|
||||
load_graph_from_script,
|
||||
)
|
||||
from lfx.load import load_flow_from_json
|
||||
from lfx.run._defaults import apply_run_defaults, resolve_fallback_to_env_vars
|
||||
from lfx.schema.schema import InputValueRequest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -299,12 +300,16 @@ def prepare_graph(graph, verbose_print):
|
||||
raise typer.Exit(1) from e
|
||||
|
||||
|
||||
async def execute_graph_with_capture(graph, input_value: str | None):
|
||||
async def execute_graph_with_capture(graph, input_value: str | None, session_id: str | None = None):
|
||||
"""Execute a graph and capture output.
|
||||
|
||||
Args:
|
||||
graph: Graph object to execute
|
||||
input_value: Input value to pass to the graph
|
||||
session_id: Optional session ID. ``None`` auto-generates one so that
|
||||
message-store paths (which validate session_id) succeed; an empty or
|
||||
whitespace-only string is rejected with ``ValueError`` to surface
|
||||
shell/env-var typos (see ``lfx.run._defaults.validate_provided_id``).
|
||||
|
||||
Returns:
|
||||
Tuple of (results, captured_logs)
|
||||
@ -312,6 +317,11 @@ async def execute_graph_with_capture(graph, input_value: str | None):
|
||||
Raises:
|
||||
Exception: Re-raises any exception that occurs during graph execution
|
||||
"""
|
||||
# Apply session_id, user_id, and Memory-vertex propagation defaults via the
|
||||
# shared helper (same logic as run_flow). user_id is not exposed in this
|
||||
# entry point, so any pre-existing graph.user_id is preserved.
|
||||
apply_run_defaults(graph, session_id=session_id, user_id=None, overwrite_user_id=False)
|
||||
|
||||
# Create input request
|
||||
inputs = InputValueRequest(input_value=input_value) if input_value else None
|
||||
|
||||
@ -323,10 +333,12 @@ async def execute_graph_with_capture(graph, input_value: str | None):
|
||||
original_stdout = sys.stdout
|
||||
original_stderr = sys.stderr
|
||||
|
||||
fallback_to_env_vars = resolve_fallback_to_env_vars()
|
||||
|
||||
try:
|
||||
sys.stdout = captured_stdout
|
||||
sys.stderr = captured_stderr
|
||||
results = [result async for result in graph.async_start(inputs)]
|
||||
results = [result async for result in graph.async_start(inputs, fallback_to_env_vars=fallback_to_env_vars)]
|
||||
except Exception as exc:
|
||||
# Capture any error output that was written to stderr
|
||||
error_output = captured_stderr.getvalue()
|
||||
|
||||
@ -102,6 +102,13 @@ async def run(
|
||||
show_default=True,
|
||||
help="Include detailed timing information in output",
|
||||
),
|
||||
session_id: str | None = typer.Option(
|
||||
None,
|
||||
"--session-id",
|
||||
help=(
|
||||
"Session ID to attach to the run. Agent and Memory Components will use this to track conversation history."
|
||||
),
|
||||
),
|
||||
) -> None:
|
||||
"""Execute a Langflow graph script or JSON flow and return the result.
|
||||
|
||||
@ -121,6 +128,7 @@ async def run(
|
||||
stdin: Read JSON flow content from stdin
|
||||
check_variables: Check global variables for environment compatibility
|
||||
timing: Include detailed timing information in output
|
||||
session_id: Optional session ID; auto-generated if not supplied
|
||||
"""
|
||||
# Determine verbosity for output formatting
|
||||
verbosity = 3 if verbose_full else (2 if verbose_detailed else (1 if verbose else 0))
|
||||
@ -139,6 +147,7 @@ async def run(
|
||||
verbose_full=verbose_full,
|
||||
timing=timing,
|
||||
global_variables=None,
|
||||
session_id=session_id,
|
||||
)
|
||||
|
||||
# Output based on format
|
||||
|
||||
@ -226,6 +226,7 @@ class RunRequest(BaseModel):
|
||||
"""Request model for executing a LFX flow."""
|
||||
|
||||
input_value: str = Field(..., description="Input value passed to the flow")
|
||||
session_id: str | None = Field(default=None, description="Session ID for maintaining conversation state")
|
||||
|
||||
|
||||
class StreamRequest(BaseModel):
|
||||
@ -329,7 +330,9 @@ async def run_flow_generator_for_serve(
|
||||
# For the serve app, we'll use execute_graph_with_capture with streaming
|
||||
# Note: This is a simplified version. In a full implementation, you might want
|
||||
# to integrate with the full LFX streaming pipeline from endpoints.py
|
||||
results, logs = await execute_graph_with_capture(graph, input_request.input_value)
|
||||
results, logs = await execute_graph_with_capture(
|
||||
graph, input_request.input_value, session_id=input_request.session_id
|
||||
)
|
||||
result_data = extract_result_data(results, logs)
|
||||
|
||||
# Send the final result
|
||||
@ -422,7 +425,9 @@ def create_multi_serve_app(
|
||||
try:
|
||||
validate_flow_for_current_settings(graph)
|
||||
graph_copy = deepcopy(graph)
|
||||
results, logs = await execute_graph_with_capture(graph_copy, request.input_value)
|
||||
results, logs = await execute_graph_with_capture(
|
||||
graph_copy, request.input_value, session_id=request.session_id
|
||||
)
|
||||
result_data = extract_result_data(results, logs)
|
||||
|
||||
# Debug logging
|
||||
|
||||
@ -368,7 +368,7 @@ class CugaComponent(ToolCallingAgentComponent):
|
||||
sender_name="Cuga",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
session_id=self.graph.session_id,
|
||||
session_id=self.graph.session_id or str(uuid.uuid4()),
|
||||
)
|
||||
|
||||
# Pre-assign an ID for event processing, following the base agent pattern
|
||||
|
||||
@ -361,6 +361,7 @@ class Graph:
|
||||
event_manager: EventManager | None = None,
|
||||
*,
|
||||
reset_output_values: bool = True,
|
||||
fallback_to_env_vars: bool = False,
|
||||
):
|
||||
# Preserve start_component_id from constructor if available
|
||||
start_component_id = self._start.get_id() if self._start else None
|
||||
@ -379,7 +380,9 @@ class Graph:
|
||||
yielded_counts: dict[str, int] = defaultdict(int)
|
||||
|
||||
while should_continue(yielded_counts, max_iterations):
|
||||
result = await self.astep(event_manager=event_manager, inputs=inputs)
|
||||
result = await self.astep(
|
||||
event_manager=event_manager, inputs=inputs, fallback_to_env_vars=fallback_to_env_vars
|
||||
)
|
||||
yield result
|
||||
if isinstance(result, Finish):
|
||||
return
|
||||
@ -1441,6 +1444,8 @@ class Graph:
|
||||
files: list[str] | None = None,
|
||||
user_id: str | None = None,
|
||||
event_manager: EventManager | None = None,
|
||||
*,
|
||||
fallback_to_env_vars: bool = False,
|
||||
):
|
||||
if not self._prepared:
|
||||
msg = "Graph not prepared. Call prepare() first."
|
||||
@ -1479,6 +1484,7 @@ class Graph:
|
||||
get_cache=get_cache_func,
|
||||
set_cache=set_cache_func,
|
||||
event_manager=event_manager,
|
||||
fallback_to_env_vars=fallback_to_env_vars,
|
||||
)
|
||||
|
||||
next_runnable_vertices = await self.get_next_runnable_vertices(
|
||||
|
||||
@ -41,10 +41,20 @@ async def astore_message(
|
||||
)
|
||||
raise ValueError(msg)
|
||||
|
||||
# Set flow_id if provided
|
||||
# Set flow_id if provided. The stub is the fallback when no real database is
|
||||
# registered, so be tolerant of non-UUID flow_ids (e.g. synthetic IDs from tests
|
||||
# or callers that pass a string identifier). UUID parsing here only normalizes
|
||||
# format; an invalid string is preserved verbatim, but logged so downstream
|
||||
# UUID-expecting code paths have a breadcrumb if they later fail.
|
||||
if flow_id:
|
||||
if isinstance(flow_id, str):
|
||||
flow_id = UUID(flow_id)
|
||||
try:
|
||||
flow_id = UUID(flow_id)
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
f"flow_id {flow_id!r} is not a valid UUID; preserving verbatim. "
|
||||
"Downstream code that expects a UUID may surface a confusing error."
|
||||
)
|
||||
message.flow_id = str(flow_id)
|
||||
|
||||
# In lfx, we use the service architecture - this is a simplified implementation
|
||||
|
||||
132
src/lfx/src/lfx/run/_defaults.py
Normal file
132
src/lfx/src/lfx/run/_defaults.py
Normal file
@ -0,0 +1,132 @@
|
||||
"""Shared helpers for applying run-time defaults to a Graph.
|
||||
|
||||
The CLI's ``lfx run`` (``run_flow``) and ``lfx serve``/``execute_graph_with_capture``
|
||||
both need to:
|
||||
|
||||
1. Reject empty/whitespace-only ``session_id`` / ``user_id`` (so a typo or
|
||||
empty env var doesn't silently produce a random session and break Memory
|
||||
continuity).
|
||||
2. Auto-generate a UUID when neither is supplied (so component prechecks and
|
||||
``astore_message`` validation don't fail).
|
||||
3. Propagate ``session_id`` to ``Memory``/``MessageHistory`` vertices the way
|
||||
``langflow.api.utils.flow_utils.build_graph_from_data`` does — preserving
|
||||
any value already pinned in the flow JSON.
|
||||
4. Resolve ``fallback_to_env_vars`` from settings (mirrors the langflow API
|
||||
path in ``processing.process.run_graph_internal``).
|
||||
|
||||
Both call sites used to inline these steps; consolidating them here prevents
|
||||
behavior drift between ``lfx run`` and ``lfx serve``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from lfx.log.logger import logger
|
||||
from lfx.services.deps import get_settings_service
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from lfx.graph.graph.base import Graph
|
||||
|
||||
|
||||
def validate_provided_id(name: str, value: str | None) -> None:
|
||||
"""Reject empty/whitespace strings while allowing None (which means "auto-generate").
|
||||
|
||||
Empty strings reach this code via shell quirks (``--session-id ""``), env-var
|
||||
expansion that resolved to empty, or callers that defaulted a missing arg to
|
||||
``""`` instead of ``None``. Treating them the same as ``None`` would silently
|
||||
fall through to UUID generation — and the user would only learn their session
|
||||
didn't carry over by noticing Memory looks empty on the next run.
|
||||
"""
|
||||
if value is None:
|
||||
return
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
msg = (
|
||||
f"{name} was provided but is empty or whitespace. "
|
||||
f"Pass a non-empty value, or omit {name} to auto-generate one."
|
||||
)
|
||||
raise ValueError(msg)
|
||||
|
||||
|
||||
def apply_run_defaults(
|
||||
graph: Graph,
|
||||
*,
|
||||
session_id: str | None,
|
||||
user_id: str | None,
|
||||
overwrite_user_id: bool = True,
|
||||
) -> tuple[str, str]:
|
||||
"""Apply session_id, user_id, and vertex-propagation defaults to *graph*.
|
||||
|
||||
Validates non-None values are non-empty (raises ``ValueError`` otherwise).
|
||||
Auto-generates UUID hex strings for any value left as ``None``. Sets the
|
||||
resolved values on the graph and propagates ``session_id`` to vertices in
|
||||
``graph.has_session_id_vertices`` that don't already have a hardcoded value.
|
||||
|
||||
Args:
|
||||
graph: The graph to mutate.
|
||||
session_id: Caller-supplied session_id. ``None`` -> auto-generate.
|
||||
user_id: Caller-supplied user_id. ``None`` -> auto-generate.
|
||||
overwrite_user_id: When False, an existing ``graph.user_id`` is preserved
|
||||
(matches the prior ``execute_graph_with_capture`` behavior). When True,
|
||||
the resolved user_id always wins (matches the prior ``run_flow``
|
||||
behavior, which intentionally re-stamps the graph).
|
||||
|
||||
Returns:
|
||||
``(session_id, user_id)`` — the values actually applied to the graph.
|
||||
"""
|
||||
validate_provided_id("session_id", session_id)
|
||||
validate_provided_id("user_id", user_id)
|
||||
|
||||
if not user_id:
|
||||
user_id = uuid.uuid4().hex
|
||||
logger.debug(
|
||||
f"No user_id provided; auto-generated {user_id} to satisfy component prechecks. "
|
||||
"lfx's variable service is env-backed, so user_id is not used for variable scoping."
|
||||
)
|
||||
if overwrite_user_id or not getattr(graph, "user_id", None):
|
||||
graph.user_id = user_id
|
||||
else:
|
||||
# Caller-supplied None plus an existing graph.user_id: preserve the existing.
|
||||
user_id = graph.user_id
|
||||
|
||||
if not session_id:
|
||||
session_id = uuid.uuid4().hex
|
||||
# info, not warning: this is the normal case for one-shot CLI runs and
|
||||
# would otherwise spam stderr on every invocation. Visible at -v.
|
||||
logger.info(
|
||||
f"No session_id provided; auto-generated {session_id}. "
|
||||
"Memory/MessageHistory components will not see prior conversation state across runs."
|
||||
)
|
||||
graph.session_id = session_id
|
||||
|
||||
# Mirror langflow's build_graph_from_data: only write when raw_params has no
|
||||
# session_id (hardcoded values pinned in the flow JSON win). graph.async_start
|
||||
# bypasses the equivalent loop in Graph._run, so without this Memory components
|
||||
# would read "" from their input field even when --session-id is set.
|
||||
for vertex_id in graph.has_session_id_vertices:
|
||||
vertex = graph.get_vertex(vertex_id)
|
||||
if vertex is None:
|
||||
continue
|
||||
if not vertex.raw_params.get("session_id"):
|
||||
vertex.update_raw_params({"session_id": session_id}, overwrite=True)
|
||||
|
||||
return session_id, user_id
|
||||
|
||||
|
||||
def resolve_fallback_to_env_vars() -> bool:
|
||||
"""Read ``fallback_to_env_var`` from settings, defaulting to True if the service is missing.
|
||||
|
||||
Mirrors the langflow API path (``processing.process.run_graph_internal``):
|
||||
when a ``load_from_db`` variable misses (e.g. the ceremonial UUID has no
|
||||
Variable row), fall through to ``os.environ`` instead of failing the build.
|
||||
A missing settings_service is unusual — log a warning so the user can debug.
|
||||
"""
|
||||
settings_service = get_settings_service()
|
||||
if settings_service is None:
|
||||
logger.warning(
|
||||
"settings_service is unavailable; defaulting fallback_to_env_vars=True. "
|
||||
"If load_from_db variables are misbehaving, verify the settings service initialized."
|
||||
)
|
||||
return True
|
||||
return bool(settings_service.settings.fallback_to_env_var)
|
||||
@ -16,6 +16,7 @@ from lfx.cli.script_loader import (
|
||||
)
|
||||
from lfx.cli.validation import validate_global_variables_for_env
|
||||
from lfx.log.logger import logger
|
||||
from lfx.run._defaults import apply_run_defaults, resolve_fallback_to_env_vars, validate_provided_id
|
||||
from lfx.schema.schema import InputValueRequest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -116,6 +117,18 @@ async def run_flow(
|
||||
configure(log_level="CRITICAL", output_file=sys.stderr)
|
||||
verbosity = 0
|
||||
|
||||
# Validate caller-supplied IDs up-front so users get a clear error before any
|
||||
# graph loading work happens. None means "auto-generate later"; "" / whitespace
|
||||
# is rejected so a typo or empty env var doesn't silently produce a fresh
|
||||
# session and break Memory continuity.
|
||||
try:
|
||||
validate_provided_id("session_id", session_id)
|
||||
validate_provided_id("user_id", user_id)
|
||||
except ValueError as e:
|
||||
error_msg = str(e)
|
||||
output_error(error_msg, verbose=verbose, exception=e)
|
||||
raise RunError(error_msg, e) from e
|
||||
|
||||
start_time = time.time() if timing else None
|
||||
|
||||
# Use either positional input_value or --input-value option
|
||||
@ -221,17 +234,14 @@ async def run_flow(
|
||||
error_msg = "No input source provided"
|
||||
raise ValueError(error_msg)
|
||||
|
||||
# Set user_id on graph if provided (required for some components like AgentComponent)
|
||||
if user_id:
|
||||
graph.user_id = user_id
|
||||
if verbosity > 0:
|
||||
logger.info(f"Set graph user_id: {user_id}")
|
||||
|
||||
# Set session_id on graph if provided (isolates memory between requests)
|
||||
if session_id:
|
||||
graph.session_id = session_id
|
||||
if verbosity > 0:
|
||||
logger.info(f"Set graph session_id: {session_id}")
|
||||
# Apply session_id, user_id, and Memory-vertex propagation defaults. Both
|
||||
# are auto-generated when not supplied so component prechecks (custom_component
|
||||
# .get_variable for user_id) and astore_message validation (for session_id)
|
||||
# don't fail. See lfx.run._defaults.apply_run_defaults for the full rationale.
|
||||
session_id, user_id = apply_run_defaults(graph, session_id=session_id, user_id=user_id)
|
||||
if verbosity > 0:
|
||||
logger.info(f"Set graph user_id: {user_id}")
|
||||
logger.info(f"Set graph session_id: {session_id}")
|
||||
|
||||
# Inject global variables into graph context
|
||||
if global_variables:
|
||||
@ -348,7 +358,14 @@ async def run_flow(
|
||||
|
||||
logger.info("Starting graph execution...", level="DEBUG")
|
||||
|
||||
async for result in graph.async_start(inputs, event_manager=event_manager):
|
||||
# See lfx.run._defaults.resolve_fallback_to_env_vars for why this flag is
|
||||
# plumbed through (mirrors langflow's API path so load_from_db variables
|
||||
# fall through to os.environ on miss instead of erroring the build).
|
||||
fallback_to_env_vars = resolve_fallback_to_env_vars()
|
||||
|
||||
async for result in graph.async_start(
|
||||
inputs, event_manager=event_manager, fallback_to_env_vars=fallback_to_env_vars
|
||||
):
|
||||
result_count += 1
|
||||
if verbosity > 0:
|
||||
logger.debug(f"Processing result #{result_count}")
|
||||
|
||||
@ -22,14 +22,23 @@ class VariableService(Service):
|
||||
self.set_ready()
|
||||
logger.debug("Variable service initialized (env vars only)")
|
||||
|
||||
def get_variable(self, name: str, **kwargs) -> str | None: # noqa: ARG002
|
||||
async def get_variable(self, name: str, **kwargs) -> str | None: # noqa: ARG002
|
||||
"""Get a variable value.
|
||||
|
||||
First checks in-memory cache, then environment variables.
|
||||
|
||||
Async to match the call signature in custom_component.get_variable
|
||||
(`await variable_service.get_variable(...)`), which is the path used
|
||||
by component variable resolution. The lookup itself is sync — no I/O —
|
||||
but the coroutine wrapper is required so callers can `await` it
|
||||
regardless of which variable service implementation is registered
|
||||
(lfx env-fallback vs langflow DB-backed).
|
||||
|
||||
Args:
|
||||
name: Variable name
|
||||
**kwargs: Additional arguments (ignored in minimal implementation)
|
||||
**kwargs: Additional arguments (ignored; user_id/field/session
|
||||
from langflow's call signature are absorbed and not used,
|
||||
since this implementation has no per-user scope).
|
||||
|
||||
Returns:
|
||||
Variable value or None if not found
|
||||
|
||||
120
src/lfx/tests/unit/base/models/test_handle_stream.py
Normal file
120
src/lfx/tests/unit/base/models/test_handle_stream.py
Normal file
@ -0,0 +1,120 @@
|
||||
"""Tests for LCModelComponent._handle_stream session_id handling.
|
||||
|
||||
When a streaming LLM is wired to ChatOutput, the streaming path tries to persist
|
||||
a placeholder Message via send_message. With no session_id (e.g. ``lfx run`` with
|
||||
NoopSession and no ``--session-id``), astore_message rejects the empty value.
|
||||
The fallback in _handle_stream invokes the model non-streamingly so the run can
|
||||
still complete.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from lfx.base.models.model import LCModelComponent
|
||||
|
||||
|
||||
class _StreamableProbe(LCModelComponent):
|
||||
"""Minimal LCModelComponent subclass usable without going through Component init."""
|
||||
|
||||
display_name = "Probe"
|
||||
description = "test"
|
||||
|
||||
def build_model(self): # pragma: no cover - abstract stub
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def _make_probe(*, connected: bool, session_id, event_manager=None):
|
||||
probe = _StreamableProbe.__new__(_StreamableProbe)
|
||||
probe.is_connected_to_chat_output = MagicMock(return_value=connected)
|
||||
# ``graph`` is a read-only property on Component (-> self._vertex.graph),
|
||||
# so wire the underlying _vertex instead of trying to assign graph directly.
|
||||
probe._vertex = SimpleNamespace(graph=SimpleNamespace(session_id=session_id, flow_id=None))
|
||||
probe.icon = "brain"
|
||||
probe._id = "probe-1"
|
||||
probe._event_manager = event_manager
|
||||
probe.send_message = AsyncMock()
|
||||
probe._build_source = MagicMock(return_value=None)
|
||||
return probe
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_stream_falls_back_to_invoke_when_no_session_id():
|
||||
"""Empty graph.session_id must not call send_message; falls back to ainvoke."""
|
||||
probe = _make_probe(connected=True, session_id="", event_manager=MagicMock())
|
||||
runnable = SimpleNamespace(
|
||||
astream=MagicMock(),
|
||||
ainvoke=AsyncMock(return_value=SimpleNamespace(content="hi from invoke")),
|
||||
)
|
||||
|
||||
lf_message, result, ai_message = await probe._handle_stream(runnable, "input")
|
||||
|
||||
runnable.ainvoke.assert_awaited_once_with("input")
|
||||
runnable.astream.assert_not_called()
|
||||
probe.send_message.assert_not_awaited()
|
||||
assert lf_message is None
|
||||
assert result == "hi from invoke"
|
||||
assert isinstance(result, str)
|
||||
assert ai_message is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_stream_falls_back_to_invoke_when_no_event_manager():
|
||||
"""Without an event_manager (e.g. lfx run) the chunk iterator would never be drained.
|
||||
|
||||
The fallback prevents send_message from storing a Message whose text is an unconsumed
|
||||
AsyncIterator, which previously surfaced as an empty result downstream.
|
||||
"""
|
||||
probe = _make_probe(connected=True, session_id="sess-123", event_manager=None)
|
||||
runnable = SimpleNamespace(
|
||||
astream=MagicMock(),
|
||||
ainvoke=AsyncMock(return_value=SimpleNamespace(content="batched")),
|
||||
)
|
||||
|
||||
lf_message, result, ai_message = await probe._handle_stream(runnable, "input")
|
||||
|
||||
runnable.ainvoke.assert_awaited_once_with("input")
|
||||
runnable.astream.assert_not_called()
|
||||
probe.send_message.assert_not_awaited()
|
||||
assert lf_message is None
|
||||
assert result == "batched"
|
||||
# Lock in that the fallback returns plain text, not the unconsumed AsyncIterator
|
||||
# that astream would have produced — the original bug surfaced as empty downstream
|
||||
# text because the iterator was stored verbatim.
|
||||
assert isinstance(result, str)
|
||||
assert ai_message is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_stream_uses_send_message_when_session_id_and_event_manager_present():
|
||||
"""Both session_id and event_manager present -> original streaming + persistence path."""
|
||||
probe = _make_probe(connected=True, session_id="sess-123", event_manager=MagicMock())
|
||||
probe.send_message.return_value = SimpleNamespace(text="streamed text")
|
||||
runnable = SimpleNamespace(astream=MagicMock(), ainvoke=AsyncMock())
|
||||
|
||||
lf_message, result, ai_message = await probe._handle_stream(runnable, "input")
|
||||
|
||||
probe.send_message.assert_awaited_once()
|
||||
runnable.ainvoke.assert_not_awaited()
|
||||
assert lf_message is not None
|
||||
assert result == "streamed text"
|
||||
assert ai_message is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_stream_invokes_directly_when_not_connected_to_chat_output():
|
||||
"""Pre-existing branch: not connected -> ainvoke regardless of session_id."""
|
||||
probe = _make_probe(connected=False, session_id="sess-123", event_manager=MagicMock())
|
||||
runnable = SimpleNamespace(
|
||||
astream=MagicMock(),
|
||||
ainvoke=AsyncMock(return_value=SimpleNamespace(content="direct")),
|
||||
)
|
||||
|
||||
lf_message, result, _ = await probe._handle_stream(runnable, "input")
|
||||
|
||||
runnable.ainvoke.assert_awaited_once_with("input")
|
||||
probe.send_message.assert_not_awaited()
|
||||
assert lf_message is None
|
||||
assert result == "direct"
|
||||
@ -208,7 +208,7 @@ class TestGraphExecution:
|
||||
# Mock graph and async iterator
|
||||
mock_result = MagicMock(results={"text": "Test result"})
|
||||
|
||||
async def mock_async_start(inputs): # noqa: ARG001
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield mock_result
|
||||
|
||||
mock_graph = MagicMock()
|
||||
@ -229,7 +229,7 @@ class TestGraphExecution:
|
||||
# Ensure results attribute doesn't exist
|
||||
delattr(mock_result, "results")
|
||||
|
||||
async def mock_async_start(inputs): # noqa: ARG001
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield mock_result
|
||||
|
||||
mock_graph = MagicMock()
|
||||
@ -244,7 +244,7 @@ class TestGraphExecution:
|
||||
async def test_execute_graph_with_capture_error(self):
|
||||
"""Test graph execution with error."""
|
||||
|
||||
async def mock_async_start_error(inputs): # noqa: ARG001
|
||||
async def mock_async_start_error(inputs, **kwargs): # noqa: ARG001
|
||||
msg = "Execution failed"
|
||||
raise RuntimeError(msg)
|
||||
yield # This line never executes but makes it an async generator
|
||||
@ -255,6 +255,160 @@ class TestGraphExecution:
|
||||
with pytest.raises(RuntimeError, match="Execution failed"):
|
||||
await execute_graph_with_capture(mock_graph, "test input")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_with_capture_autogenerates_session_id(self):
|
||||
"""Auto-generate a session_id when none is provided.
|
||||
|
||||
Message-store validators reject empty session_id, so the helper assigns one
|
||||
to keep streaming/persistence paths functional in lfx serve.
|
||||
"""
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
|
||||
await execute_graph_with_capture(mock_graph, "test input")
|
||||
|
||||
assert mock_graph.session_id, "session_id should be auto-generated"
|
||||
assert isinstance(mock_graph.session_id, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_with_capture_preserves_caller_session_id(self):
|
||||
"""An explicit session_id wins over auto-generation."""
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
|
||||
await execute_graph_with_capture(mock_graph, "test input", session_id="fixed-session")
|
||||
|
||||
assert mock_graph.session_id == "fixed-session"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_propagates_session_id_to_vertices(self):
|
||||
"""Session_id must reach Memory/MessageHistory inputs on the lfx serve path.
|
||||
|
||||
``execute_graph_with_capture`` uses ``graph.async_start``, which bypasses
|
||||
the propagation loop in ``Graph._run``. The helper has to replicate it
|
||||
so served ``/run`` and ``/stream`` requests behave like the playground.
|
||||
"""
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
memory_vertex = MagicMock()
|
||||
memory_vertex.raw_params = {}
|
||||
memory_vertex.update_raw_params = MagicMock()
|
||||
mock_graph.has_session_id_vertices = ["memory-1"]
|
||||
mock_graph.get_vertex = MagicMock(return_value=memory_vertex)
|
||||
|
||||
await execute_graph_with_capture(mock_graph, "test input", session_id="my-conversation")
|
||||
|
||||
memory_vertex.update_raw_params.assert_called_once_with({"session_id": "my-conversation"}, overwrite=True)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_does_not_overwrite_hardcoded_session_id(self):
|
||||
"""Hardcoded session_id on a Memory component (set in flow JSON) wins over the request value.
|
||||
|
||||
Mirrors Langflow's playground precedence in ``build_graph_from_data``.
|
||||
"""
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
pinned_vertex = MagicMock()
|
||||
pinned_vertex.raw_params = {"session_id": "hardcoded-in-flow"}
|
||||
pinned_vertex.update_raw_params = MagicMock()
|
||||
mock_graph.has_session_id_vertices = ["memory-pinned"]
|
||||
mock_graph.get_vertex = MagicMock(return_value=pinned_vertex)
|
||||
|
||||
await execute_graph_with_capture(mock_graph, "test input", session_id="from-request")
|
||||
|
||||
pinned_vertex.update_raw_params.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_autogenerates_user_id_when_unset(self):
|
||||
"""When the graph arrives without a user_id (typical for lfx serve), assign a UUID.
|
||||
|
||||
AgentComponent's variable lookup precheck requires a non-empty user_id; the
|
||||
env-fallback variable service does not use it for scoping, so a random UUID is
|
||||
ceremonial but necessary.
|
||||
"""
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
mock_graph.user_id = None
|
||||
|
||||
await execute_graph_with_capture(mock_graph, "test input")
|
||||
|
||||
assert mock_graph.user_id, "user_id should be auto-assigned when graph has none"
|
||||
assert isinstance(mock_graph.user_id, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_preserves_existing_user_id(self):
|
||||
"""A user_id already set on the graph (e.g., by an upstream caller) is left alone."""
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
mock_graph.user_id = "preset-user-uuid"
|
||||
|
||||
await execute_graph_with_capture(mock_graph, "test input")
|
||||
|
||||
assert mock_graph.user_id == "preset-user-uuid"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_passes_fallback_from_settings_default(self):
|
||||
"""Default settings (fallback_to_env_var=True) reach async_start.
|
||||
|
||||
Lets components fall through to os.environ when a load_from_db variable
|
||||
has no DB row — matching langflow's API path behavior.
|
||||
"""
|
||||
captured: dict = {}
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
captured.update(kwargs)
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
|
||||
await execute_graph_with_capture(mock_graph, "test input")
|
||||
|
||||
assert captured.get("fallback_to_env_vars") is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_graph_respects_disabled_fallback_setting(self):
|
||||
"""When the user opts out of env fallback in settings, the flag is False."""
|
||||
captured: dict = {}
|
||||
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
captured.update(kwargs)
|
||||
yield MagicMock(results={"text": "ok"})
|
||||
|
||||
mock_graph = MagicMock()
|
||||
mock_graph.async_start = mock_async_start
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.fallback_to_env_var = False
|
||||
|
||||
with patch("lfx.run._defaults.get_settings_service", return_value=mock_settings):
|
||||
await execute_graph_with_capture(mock_graph, "test input")
|
||||
|
||||
assert captured.get("fallback_to_env_vars") is False
|
||||
|
||||
|
||||
class TestResultExtraction:
|
||||
"""Test result data extraction."""
|
||||
|
||||
@ -4,7 +4,7 @@ import contextlib
|
||||
import json
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
import typer
|
||||
@ -155,6 +155,10 @@ chat_input = ChatInput(
|
||||
def test_execute_python_script_success(self, simple_chat_script, capsys):
|
||||
"""Test executing a valid Python script."""
|
||||
# Test that Python script execution either succeeds or fails gracefully
|
||||
# ``run`` is a typer command, so any parameter with a ``typer.Option(...)`` default
|
||||
# (e.g. session_id) needs to be passed explicitly when invoking outside typer's
|
||||
# parser — otherwise the default evaluates to a ``typer.models.OptionInfo``
|
||||
# sentinel and propagates downstream.
|
||||
with contextlib.suppress(typer.Exit):
|
||||
run(
|
||||
script_path=simple_chat_script,
|
||||
@ -164,6 +168,7 @@ chat_input = ChatInput(
|
||||
output_format="json",
|
||||
flow_json=None,
|
||||
stdin=False,
|
||||
session_id=None,
|
||||
)
|
||||
|
||||
# Test passes as long as no unhandled exceptions occur
|
||||
@ -181,6 +186,7 @@ chat_input = ChatInput(
|
||||
def test_execute_python_script_verbose(self, simple_chat_script, capsys):
|
||||
"""Test executing a Python script with verbose output."""
|
||||
# Test that verbose mode execution either succeeds or fails gracefully
|
||||
# See note in ``test_execute_python_script_success`` re: session_id=None.
|
||||
with contextlib.suppress(typer.Exit):
|
||||
run(
|
||||
script_path=simple_chat_script,
|
||||
@ -190,6 +196,7 @@ chat_input = ChatInput(
|
||||
output_format="json",
|
||||
flow_json=None,
|
||||
stdin=False,
|
||||
session_id=None,
|
||||
)
|
||||
|
||||
# Test passes as long as no unhandled exceptions occur
|
||||
@ -320,7 +327,8 @@ chat_input = ChatInput(
|
||||
flow_json_str = json.dumps(simple_json_flow)
|
||||
mock_stdin.read.return_value = flow_json_str
|
||||
|
||||
# Test that stdin execution either succeeds or fails gracefully
|
||||
# Test that stdin execution either succeeds or fails gracefully.
|
||||
# See note in ``test_execute_python_script_success`` re: session_id=None.
|
||||
with pytest.raises(typer.Exit) as exc_info:
|
||||
run(
|
||||
script_path=None,
|
||||
@ -330,6 +338,7 @@ chat_input = ChatInput(
|
||||
output_format="json",
|
||||
flow_json=None,
|
||||
stdin=True,
|
||||
session_id=None,
|
||||
)
|
||||
|
||||
# Check that stdin was read and function exited cleanly
|
||||
@ -436,6 +445,7 @@ chat_input = ChatInput(
|
||||
|
||||
def test_execute_verbose_error_output(self, invalid_script, capsys):
|
||||
"""Test that verbose mode shows error details."""
|
||||
# See note in ``test_execute_python_script_success`` re: session_id=None.
|
||||
with pytest.raises(typer.Exit) as exc_info:
|
||||
run(
|
||||
script_path=invalid_script,
|
||||
@ -445,6 +455,7 @@ chat_input = ChatInput(
|
||||
output_format="json",
|
||||
flow_json=None,
|
||||
stdin=False,
|
||||
session_id=None,
|
||||
)
|
||||
|
||||
assert exc_info.value.exit_code == 1
|
||||
@ -453,6 +464,55 @@ chat_input = ChatInput(
|
||||
error_output = captured.out + captured.err
|
||||
assert "graph" in error_output.lower() or "variable" in error_output.lower()
|
||||
|
||||
def test_session_id_cli_flag_plumbs_through_to_run_flow(self, simple_chat_script):
|
||||
"""--session-id is wired from typer through `run` into `run_flow(session_id=...)`.
|
||||
|
||||
Locks in the typer Option name so a rename or wiring typo at lfx.cli.run.run
|
||||
is caught here (the unit tests on run_flow itself only cover the function
|
||||
contract, not this CLI layer).
|
||||
"""
|
||||
captured = {}
|
||||
|
||||
async def fake_run_flow(**kwargs):
|
||||
captured.update(kwargs)
|
||||
return {"success": True, "result": "ok", "logs": ""}
|
||||
|
||||
with patch("lfx.cli.run.run_flow", new=AsyncMock(side_effect=fake_run_flow)):
|
||||
run(
|
||||
script_path=simple_chat_script,
|
||||
input_value="hi",
|
||||
input_value_option=None,
|
||||
verbose=False,
|
||||
output_format="json",
|
||||
flow_json=None,
|
||||
stdin=False,
|
||||
session_id="my-fixed-session",
|
||||
)
|
||||
|
||||
assert captured.get("session_id") == "my-fixed-session"
|
||||
|
||||
def test_session_id_cli_flag_omitted_passes_none(self, simple_chat_script):
|
||||
"""No --session-id => `run_flow` receives session_id=None (auto-gen happens downstream)."""
|
||||
captured = {}
|
||||
|
||||
async def fake_run_flow(**kwargs):
|
||||
captured.update(kwargs)
|
||||
return {"success": True, "result": "ok", "logs": ""}
|
||||
|
||||
with patch("lfx.cli.run.run_flow", new=AsyncMock(side_effect=fake_run_flow)):
|
||||
run(
|
||||
script_path=simple_chat_script,
|
||||
input_value="hi",
|
||||
input_value_option=None,
|
||||
verbose=False,
|
||||
output_format="json",
|
||||
flow_json=None,
|
||||
stdin=False,
|
||||
session_id=None,
|
||||
)
|
||||
|
||||
assert captured.get("session_id") is None
|
||||
|
||||
def test_execute_without_input_value(self, simple_chat_script, capsys):
|
||||
"""Test executing without providing input value."""
|
||||
# Test that execution without input either succeeds or fails gracefully
|
||||
|
||||
@ -118,13 +118,13 @@ class TestCreateServeApp:
|
||||
def real_graph(self, simple_chat_json):
|
||||
"""Create a real graph using Graph.from_payload to match serve_app expectations."""
|
||||
# Create graph using from_payload with real test data
|
||||
return Graph.from_payload(simple_chat_json, flow_id="test-flow-id")
|
||||
return Graph.from_payload(simple_chat_json, flow_id="00000000-0000-0000-0000-000000000001")
|
||||
|
||||
@pytest.fixture
|
||||
def mock_meta(self):
|
||||
"""Create mock flow metadata."""
|
||||
return FlowMeta(
|
||||
id="test-flow-id",
|
||||
id="00000000-0000-0000-0000-000000000001",
|
||||
relative_path="test.json",
|
||||
title="Test Flow",
|
||||
description="A test flow",
|
||||
@ -132,8 +132,8 @@ class TestCreateServeApp:
|
||||
|
||||
def test_create_multi_serve_app_single_flow(self, real_graph, mock_meta):
|
||||
"""Test creating app with single flow."""
|
||||
graphs = {"test-flow-id": real_graph}
|
||||
metas = {"test-flow-id": mock_meta}
|
||||
graphs = {"00000000-0000-0000-0000-000000000001": real_graph}
|
||||
metas = {"00000000-0000-0000-0000-000000000001": mock_meta}
|
||||
verbose_print = Mock()
|
||||
|
||||
app = create_multi_serve_app(
|
||||
@ -150,7 +150,7 @@ class TestCreateServeApp:
|
||||
routes = [route.path for route in app.routes]
|
||||
assert "/health" in routes
|
||||
assert "/flows" in routes # Multi-flow always has this
|
||||
assert "/flows/test-flow-id/run" in routes # Flow-specific endpoint
|
||||
assert "/flows/00000000-0000-0000-0000-000000000001/run" in routes # Flow-specific endpoint
|
||||
|
||||
def test_create_multi_serve_app_multiple_flows(self, real_graph, mock_meta, simple_chat_json):
|
||||
"""Test creating app with multiple flows."""
|
||||
@ -164,8 +164,8 @@ class TestCreateServeApp:
|
||||
description="Second flow",
|
||||
)
|
||||
|
||||
graphs = {"test-flow-id": real_graph, "flow-2": graph2}
|
||||
metas = {"test-flow-id": mock_meta, "flow-2": meta2}
|
||||
graphs = {"00000000-0000-0000-0000-000000000001": real_graph, "flow-2": graph2}
|
||||
metas = {"00000000-0000-0000-0000-000000000001": mock_meta, "flow-2": meta2}
|
||||
verbose_print = Mock()
|
||||
|
||||
app = create_multi_serve_app(
|
||||
@ -182,14 +182,14 @@ class TestCreateServeApp:
|
||||
routes = [route.path for route in app.routes]
|
||||
assert "/health" in routes
|
||||
assert "/flows" in routes
|
||||
assert "/flows/test-flow-id/run" in routes
|
||||
assert "/flows/test-flow-id/info" in routes
|
||||
assert "/flows/00000000-0000-0000-0000-000000000001/run" in routes
|
||||
assert "/flows/00000000-0000-0000-0000-000000000001/info" in routes
|
||||
assert "/flows/flow-2/run" in routes
|
||||
assert "/flows/flow-2/info" in routes
|
||||
|
||||
def test_create_multi_serve_app_mismatched_keys(self, real_graph, mock_meta):
|
||||
"""Test error when graphs and metas have different keys."""
|
||||
graphs = {"test-flow-id": real_graph}
|
||||
graphs = {"00000000-0000-0000-0000-000000000001": real_graph}
|
||||
metas = {"different-id": mock_meta}
|
||||
verbose_print = Mock()
|
||||
|
||||
@ -217,13 +217,13 @@ class TestServeAppEndpoints:
|
||||
def real_graph_with_async(self, simple_chat_json):
|
||||
"""Create a real graph with async execution capability."""
|
||||
# Create graph using from_payload with real test data
|
||||
graph = Graph.from_payload(simple_chat_json, flow_id="test-flow-id")
|
||||
graph = Graph.from_payload(simple_chat_json, flow_id="00000000-0000-0000-0000-000000000001")
|
||||
|
||||
# Store original async_start to restore later if needed
|
||||
original_async_start = graph.async_start
|
||||
|
||||
# Mock successful execution with real ResultData
|
||||
async def mock_async_start(inputs): # noqa: ARG001
|
||||
async def mock_async_start(inputs, **kwargs): # noqa: ARG001
|
||||
# Create real Message and ResultData objects
|
||||
message = Message(text="Hello from flow")
|
||||
result_data = ResultData(
|
||||
@ -251,14 +251,14 @@ class TestServeAppEndpoints:
|
||||
from lfx.services.deps import get_settings_service
|
||||
|
||||
meta = FlowMeta(
|
||||
id="test-flow-id",
|
||||
id="00000000-0000-0000-0000-000000000001",
|
||||
relative_path="test.json",
|
||||
title="Test Flow",
|
||||
description="A test flow",
|
||||
)
|
||||
|
||||
graphs = {"test-flow-id": real_graph_with_async}
|
||||
metas = {"test-flow-id": meta}
|
||||
graphs = {"00000000-0000-0000-0000-000000000001": real_graph_with_async}
|
||||
metas = {"00000000-0000-0000-0000-000000000001": meta}
|
||||
verbose_print = Mock()
|
||||
|
||||
app = create_multi_serve_app(
|
||||
@ -282,14 +282,14 @@ class TestServeAppEndpoints:
|
||||
# Create second real graph using the same JSON structure
|
||||
graph2 = Graph.from_payload(simple_chat_json, flow_id="flow-2")
|
||||
|
||||
async def mock_async_start2(inputs): # noqa: ARG001
|
||||
async def mock_async_start2(inputs, **kwargs): # noqa: ARG001
|
||||
# Return empty results for this test
|
||||
yield MagicMock(outputs=[])
|
||||
|
||||
graph2.async_start = mock_async_start2
|
||||
|
||||
meta1 = FlowMeta(
|
||||
id="test-flow-id",
|
||||
id="00000000-0000-0000-0000-000000000001",
|
||||
relative_path="test.json",
|
||||
title="Test Flow",
|
||||
description="First flow",
|
||||
@ -301,8 +301,8 @@ class TestServeAppEndpoints:
|
||||
description="Second flow",
|
||||
)
|
||||
|
||||
graphs = {"test-flow-id": real_graph_with_async, "flow-2": graph2}
|
||||
metas = {"test-flow-id": meta1, "flow-2": meta2}
|
||||
graphs = {"00000000-0000-0000-0000-000000000001": real_graph_with_async, "flow-2": graph2}
|
||||
metas = {"00000000-0000-0000-0000-000000000001": meta1, "flow-2": meta2}
|
||||
verbose_print = Mock()
|
||||
|
||||
app = create_multi_serve_app(
|
||||
@ -340,7 +340,9 @@ class TestServeAppEndpoints:
|
||||
"type": "message",
|
||||
"component": "TestComponent",
|
||||
}
|
||||
response = app_client.post("/flows/test-flow-id/run", json=request_data, headers=headers)
|
||||
response = app_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run", json=request_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
@ -353,7 +355,7 @@ class TestServeAppEndpoints:
|
||||
request_data = {"input_value": "Test input"}
|
||||
|
||||
with patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}): # pragma: allowlist secret
|
||||
response = app_client.post("/flows/test-flow-id/run", json=request_data)
|
||||
response = app_client.post("/flows/00000000-0000-0000-0000-000000000001/run", json=request_data)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert response.json()["detail"] == "API key required"
|
||||
@ -364,7 +366,9 @@ class TestServeAppEndpoints:
|
||||
headers = {"x-api-key": "wrong-key"}
|
||||
|
||||
with patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}): # pragma: allowlist secret
|
||||
response = app_client.post("/flows/test-flow-id/run", json=request_data, headers=headers)
|
||||
response = app_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run", json=request_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert response.json()["detail"] == "Invalid API key"
|
||||
@ -376,15 +380,15 @@ class TestServeAppEndpoints:
|
||||
"""Test that /run fails closed before execution when custom components are blocked."""
|
||||
real_graph_with_async.raw_graph_data = _blocked_raw_graph()
|
||||
meta = FlowMeta(
|
||||
id="test-flow-id",
|
||||
id="00000000-0000-0000-0000-000000000001",
|
||||
relative_path="test.json",
|
||||
title="Test Flow",
|
||||
description="A test flow",
|
||||
)
|
||||
app = create_multi_serve_app(
|
||||
root_dir=Path("/test"),
|
||||
graphs={"test-flow-id": real_graph_with_async},
|
||||
metas={"test-flow-id": meta},
|
||||
graphs={"00000000-0000-0000-0000-000000000001": real_graph_with_async},
|
||||
metas={"00000000-0000-0000-0000-000000000001": meta},
|
||||
verbose_print=Mock(),
|
||||
)
|
||||
headers = {"x-api-key": "test-api-key"}
|
||||
@ -411,7 +415,7 @@ class TestServeAppEndpoints:
|
||||
):
|
||||
client = TestClient(app)
|
||||
response = client.post(
|
||||
"/flows/test-flow-id/run",
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run",
|
||||
json={"input_value": "Test input"},
|
||||
headers=headers,
|
||||
)
|
||||
@ -435,7 +439,9 @@ class TestServeAppEndpoints:
|
||||
"type": "message",
|
||||
"component": "TestComponent",
|
||||
}
|
||||
response = app_client.post("/flows/test-flow-id/run?x-api-key=test-api-key", json=request_data)
|
||||
response = app_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run?x-api-key=test-api-key", json=request_data
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["success"] is True
|
||||
@ -446,7 +452,7 @@ class TestServeAppEndpoints:
|
||||
headers = {"x-api-key": "test-api-key"}
|
||||
|
||||
# Mock execute_graph_with_capture to raise an error
|
||||
async def mock_execute_error(graph, input_value): # noqa: ARG001
|
||||
async def mock_execute_error(graph, input_value, session_id=None): # noqa: ARG001
|
||||
msg = "Flow execution failed"
|
||||
raise RuntimeError(msg)
|
||||
|
||||
@ -454,7 +460,9 @@ class TestServeAppEndpoints:
|
||||
patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}), # pragma: allowlist secret
|
||||
patch("lfx.cli.serve_app.execute_graph_with_capture", mock_execute_error),
|
||||
):
|
||||
response = app_client.post("/flows/test-flow-id/run", json=request_data, headers=headers)
|
||||
response = app_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run", json=request_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200 # Returns 200 with error in response body
|
||||
data = response.json()
|
||||
@ -471,14 +479,16 @@ class TestServeAppEndpoints:
|
||||
headers = {"x-api-key": "test-api-key"}
|
||||
|
||||
# Mock execute_graph_with_capture to return empty results
|
||||
async def mock_execute_empty(graph, input_value): # noqa: ARG001
|
||||
async def mock_execute_empty(graph, input_value, session_id=None): # noqa: ARG001
|
||||
return [], "" # Empty results and logs
|
||||
|
||||
with (
|
||||
patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}), # pragma: allowlist secret
|
||||
patch("lfx.cli.serve_app.execute_graph_with_capture", mock_execute_empty),
|
||||
):
|
||||
response = app_client.post("/flows/test-flow-id/run", json=request_data, headers=headers)
|
||||
response = app_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run", json=request_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
@ -486,6 +496,53 @@ class TestServeAppEndpoints:
|
||||
assert data["success"] is False
|
||||
assert data["type"] == "error"
|
||||
|
||||
def test_run_endpoint_forwards_session_id(self, app_client):
|
||||
"""The /run endpoint must forward session_id from RunRequest to the executor."""
|
||||
captured: dict = {}
|
||||
|
||||
async def mock_execute_capture(graph, input_value, session_id=None): # noqa: ARG001
|
||||
captured["session_id"] = session_id
|
||||
return [], ""
|
||||
|
||||
request_data = {"input_value": "Test input", "session_id": "my-conversation"}
|
||||
headers = {"x-api-key": "test-api-key"}
|
||||
|
||||
with (
|
||||
patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}), # pragma: allowlist secret
|
||||
patch("lfx.cli.serve_app.execute_graph_with_capture", mock_execute_capture),
|
||||
):
|
||||
response = app_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run", json=request_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert captured["session_id"] == "my-conversation"
|
||||
|
||||
def test_stream_endpoint_forwards_session_id(self, app_client):
|
||||
"""The /stream endpoint must forward session_id from StreamRequest to the executor."""
|
||||
captured: dict = {}
|
||||
|
||||
async def mock_execute_capture(graph, input_value, session_id=None): # noqa: ARG001
|
||||
captured["session_id"] = session_id
|
||||
return [], ""
|
||||
|
||||
request_data = {"input_value": "Test input", "session_id": "my-stream-conversation"}
|
||||
headers = {"x-api-key": "test-api-key"}
|
||||
|
||||
with (
|
||||
patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}), # pragma: allowlist secret
|
||||
patch("lfx.cli.serve_app.execute_graph_with_capture", mock_execute_capture),
|
||||
# Drain the streaming response so the background task completes before we assert.
|
||||
app_client.stream(
|
||||
"POST", "/flows/00000000-0000-0000-0000-000000000001/stream", json=request_data, headers=headers
|
||||
) as response,
|
||||
):
|
||||
assert response.status_code == 200
|
||||
for _ in response.iter_bytes():
|
||||
pass
|
||||
|
||||
assert captured["session_id"] == "my-stream-conversation"
|
||||
|
||||
def test_list_flows_endpoint(self, multi_flow_client):
|
||||
"""Test listing flows in multi-flow mode."""
|
||||
response = multi_flow_client.get("/flows")
|
||||
@ -493,7 +550,7 @@ class TestServeAppEndpoints:
|
||||
assert response.status_code == 200
|
||||
flows = response.json()
|
||||
assert len(flows) == 2
|
||||
assert any(f["id"] == "test-flow-id" for f in flows)
|
||||
assert any(f["id"] == "00000000-0000-0000-0000-000000000001" for f in flows)
|
||||
assert any(f["id"] == "flow-2" for f in flows)
|
||||
|
||||
def test_flow_info_endpoint(self, multi_flow_client):
|
||||
@ -501,11 +558,11 @@ class TestServeAppEndpoints:
|
||||
headers = {"x-api-key": "test-api-key"}
|
||||
|
||||
with patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}): # pragma: allowlist secret
|
||||
response = multi_flow_client.get("/flows/test-flow-id/info", headers=headers)
|
||||
response = multi_flow_client.get("/flows/00000000-0000-0000-0000-000000000001/info", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
info = response.json()
|
||||
assert info["id"] == "test-flow-id"
|
||||
assert info["id"] == "00000000-0000-0000-0000-000000000001"
|
||||
assert info["title"] == "Test Flow"
|
||||
assert info["description"] == "First flow"
|
||||
|
||||
@ -524,7 +581,9 @@ class TestServeAppEndpoints:
|
||||
"type": "message",
|
||||
"component": "TestComponent",
|
||||
}
|
||||
response = multi_flow_client.post("/flows/test-flow-id/run", json=request_data, headers=headers)
|
||||
response = multi_flow_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run", json=request_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
@ -536,7 +595,7 @@ class TestServeAppEndpoints:
|
||||
headers = {"x-api-key": "test-api-key"}
|
||||
|
||||
with patch.dict(os.environ, {"LANGFLOW_API_KEY": "test-api-key"}): # pragma: allowlist secret
|
||||
response = app_client.post("/flows/test-flow-id/run", json={}, headers=headers)
|
||||
response = app_client.post("/flows/00000000-0000-0000-0000-000000000001/run", json={}, headers=headers)
|
||||
|
||||
assert response.status_code == 422 # Validation error
|
||||
|
||||
@ -544,7 +603,7 @@ class TestServeAppEndpoints:
|
||||
"""Test flow execution with message-type output."""
|
||||
|
||||
# Create a real message output scenario
|
||||
async def mock_async_start_message(inputs): # noqa: ARG001
|
||||
async def mock_async_start_message(inputs, **kwargs): # noqa: ARG001
|
||||
# Create real Message and ResultData objects
|
||||
message = Message(text="Message output")
|
||||
result_data = ResultData(
|
||||
@ -576,7 +635,9 @@ class TestServeAppEndpoints:
|
||||
"type": "message",
|
||||
"component": "Chat Output",
|
||||
}
|
||||
response = app_client.post("/flows/test-flow-id/run", json=request_data, headers=headers)
|
||||
response = app_client.post(
|
||||
"/flows/00000000-0000-0000-0000-000000000001/run", json=request_data, headers=headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
@ -249,7 +249,7 @@ def create_real_graph():
|
||||
"""Helper function to create a real LFX graph with nodes/edges for serve_app."""
|
||||
# Load real JSON data and create graph using from_payload
|
||||
json_data = simple_chat_json()
|
||||
return Graph.from_payload(json_data, flow_id="test-flow-id")
|
||||
return Graph.from_payload(json_data, flow_id="00000000-0000-0000-0000-000000000001")
|
||||
|
||||
|
||||
class TestFastAPIAppCreation:
|
||||
|
||||
0
src/lfx/tests/unit/components/cuga/__init__.py
Normal file
0
src/lfx/tests/unit/components/cuga/__init__.py
Normal file
@ -0,0 +1,97 @@
|
||||
"""Regression tests for CugaAgent session_id fallback (cuga_agent.py).
|
||||
|
||||
The agent's ``message_response`` builds an internal ``Message`` with
|
||||
``session_id=self.graph.session_id or str(uuid.uuid4())``. The fallback exists
|
||||
because some flows (notably ``lfx run`` without --session-id, before run_flow
|
||||
auto-generates one) had a graph with an empty session_id, which crashed
|
||||
``astore_message`` validation. The ``str(...)`` wrap matches the rest of the
|
||||
file (every other ``uuid.uuid4()`` is wrapped) and the codebase's expectation
|
||||
that Message.session_id renders as a hex string in logs / persistence.
|
||||
|
||||
These tests halt the agent's execution at the Message construction site so we
|
||||
can verify the constructed kwargs without mocking the rest of the agent.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
# CugaComponent's class body looks up MODELS_METADATA["OpenAI"], so it can only be
|
||||
# imported when an OpenAI provider is registered (depends on optional deps in the env).
|
||||
# Skip the whole module if the import side-effect fails.
|
||||
try:
|
||||
from lfx.components.cuga import cuga_agent
|
||||
except Exception as exc:
|
||||
pytest.skip(f"cuga_agent module not importable in this env: {exc}", allow_module_level=True)
|
||||
|
||||
|
||||
class _Halt(Exception): # noqa: N818
|
||||
"""Sentinel raised by the patched Message constructor to stop further execution."""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def captured_message_kwargs():
|
||||
"""Patch cuga_agent.Message to capture constructor kwargs and halt execution."""
|
||||
captured: dict = {}
|
||||
|
||||
def fake_message(*_args, **kwargs):
|
||||
captured.update(kwargs)
|
||||
raise _Halt
|
||||
|
||||
with patch.object(cuga_agent, "Message", side_effect=fake_message):
|
||||
yield captured
|
||||
|
||||
|
||||
def _make_agent(*, graph_session_id):
|
||||
"""Build a minimally-initialized CugaComponent that reaches Message construction."""
|
||||
agent = cuga_agent.CugaComponent.__new__(cuga_agent.CugaComponent)
|
||||
agent.input_value = "hello"
|
||||
# ``graph`` is a read-only property on Component (-> self._vertex.graph),
|
||||
# so wire the underlying _vertex instead of trying to assign graph directly.
|
||||
agent._vertex = SimpleNamespace(graph=SimpleNamespace(session_id=graph_session_id))
|
||||
agent.is_connected_to_chat_output = MagicMock(return_value=True)
|
||||
agent.get_agent_requirements = AsyncMock(return_value=(MagicMock(), [], []))
|
||||
return agent
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_id_falls_back_to_string_uuid_when_graph_session_id_empty(captured_message_kwargs):
|
||||
"""Empty graph.session_id must produce a non-empty string session_id (not a UUID object)."""
|
||||
agent = _make_agent(graph_session_id="")
|
||||
|
||||
with pytest.raises(_Halt):
|
||||
await agent.message_response()
|
||||
|
||||
assert "session_id" in captured_message_kwargs
|
||||
session_id = captured_message_kwargs["session_id"]
|
||||
assert isinstance(session_id, str), (
|
||||
f"Expected str (use `str(uuid.uuid4())` not bare `uuid.uuid4()`), got {type(session_id).__name__}"
|
||||
)
|
||||
assert session_id, "Generated session_id should be non-empty"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_id_falls_back_when_graph_session_id_none(captured_message_kwargs):
|
||||
"""None graph.session_id is also covered by the `or` fallback."""
|
||||
agent = _make_agent(graph_session_id=None)
|
||||
|
||||
with pytest.raises(_Halt):
|
||||
await agent.message_response()
|
||||
|
||||
session_id = captured_message_kwargs["session_id"]
|
||||
assert isinstance(session_id, str)
|
||||
assert session_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_id_preserved_when_graph_session_id_set(captured_message_kwargs):
|
||||
"""Explicit graph.session_id wins over the auto-generated fallback."""
|
||||
agent = _make_agent(graph_session_id="caller-supplied")
|
||||
|
||||
with pytest.raises(_Halt):
|
||||
await agent.message_response()
|
||||
|
||||
assert captured_message_kwargs["session_id"] == "caller-supplied"
|
||||
@ -297,6 +297,437 @@ chat_input = ChatInput()
|
||||
assert "No 'graph' variable found" in str(exc_info.value)
|
||||
|
||||
|
||||
class TestRunFlowSessionId:
|
||||
"""Tests for run_flow session_id handling."""
|
||||
|
||||
@staticmethod
|
||||
def _mock_graph():
|
||||
graph = MagicMock()
|
||||
graph.context = {}
|
||||
graph.vertices = []
|
||||
graph.edges = []
|
||||
graph.prepare = MagicMock()
|
||||
|
||||
async def _async_start(_inputs, **_kwargs):
|
||||
yield
|
||||
|
||||
graph.async_start = _async_start
|
||||
return graph
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_id_autogenerated_when_not_provided(self, tmp_path):
|
||||
"""run_flow must assign a session_id so memory-store paths don't fail validation."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
mock_graph = self._mock_graph()
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = mock_graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path)
|
||||
|
||||
assert mock_graph.session_id, "session_id should be auto-generated when not provided"
|
||||
assert isinstance(mock_graph.session_id, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_caller_session_id_is_preserved(self, tmp_path):
|
||||
"""Caller-supplied session_id wins over auto-generation."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
mock_graph = self._mock_graph()
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = mock_graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path, session_id="my-fixed-session")
|
||||
|
||||
assert mock_graph.session_id == "my-fixed-session"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_autogenerated_session_ids_are_unique_across_runs(self, tmp_path):
|
||||
"""Each run without a session_id should produce a distinct value."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
graph_a = self._mock_graph()
|
||||
graph_b = self._mock_graph()
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
mock_load.return_value = graph_a
|
||||
await run_flow(script_path=script_path)
|
||||
mock_load.return_value = graph_b
|
||||
await run_flow(script_path=script_path)
|
||||
|
||||
assert graph_a.session_id != graph_b.session_id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_explicit_session_id_carries_across_consecutive_runs(self, tmp_path):
|
||||
"""The same caller-supplied session_id reaches both graphs — Memory continuity surface.
|
||||
|
||||
Counterpart to ``test_autogenerated_session_ids_are_unique_across_runs``: with an
|
||||
explicit session_id, two consecutive run_flow invocations must both stamp the
|
||||
graph with that exact value (otherwise --session-id would not actually achieve
|
||||
conversational continuity).
|
||||
"""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
graph_a = self._mock_graph()
|
||||
graph_b = self._mock_graph()
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
mock_load.return_value = graph_a
|
||||
await run_flow(script_path=script_path, session_id="continuity-session")
|
||||
mock_load.return_value = graph_b
|
||||
await run_flow(script_path=script_path, session_id="continuity-session")
|
||||
|
||||
assert graph_a.session_id == "continuity-session"
|
||||
assert graph_b.session_id == "continuity-session"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("bad_value", ["", " ", "\t\n"])
|
||||
async def test_empty_or_whitespace_session_id_is_rejected(self, tmp_path, bad_value):
|
||||
"""Empty/whitespace session_id surfaces as RunError instead of silently auto-generating.
|
||||
|
||||
The --session-id flag's purpose is Memory/MessageHistory continuity. If a shell
|
||||
quirk or env-var typo collapsed the value to empty, silently auto-generating
|
||||
a fresh UUID would mask the error: subsequent runs would not see prior state
|
||||
and the user would only notice via missing memory. Validate up-front instead.
|
||||
"""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
|
||||
with pytest.raises(RunError) as exc_info:
|
||||
await run_flow(script_path=script_path, session_id=bad_value)
|
||||
assert "session_id" in str(exc_info.value)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("bad_value", ["", " "])
|
||||
async def test_empty_or_whitespace_user_id_is_rejected(self, tmp_path, bad_value):
|
||||
"""Same validation as session_id, applied to user_id (variable-scoping in DB-backed setups)."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
|
||||
with pytest.raises(RunError) as exc_info:
|
||||
await run_flow(script_path=script_path, user_id=bad_value)
|
||||
assert "user_id" in str(exc_info.value)
|
||||
|
||||
|
||||
class TestRunFlowSessionIdPropagation:
|
||||
"""Session_id must reach Memory/MessageHistory components on the lfx run path.
|
||||
|
||||
lfx run uses ``graph.async_start`` (not ``graph.arun``), so it bypasses the
|
||||
``has_session_id_vertices`` propagation loop in ``Graph._run``. ``run_flow``
|
||||
must replicate that loop after assigning ``graph.session_id`` so components
|
||||
that read ``self.session_id`` from their input field (Memory.retrieve_messages
|
||||
etc.) actually see the configured value. Mirrors what
|
||||
``langflow/api/utils/flow_utils.build_graph_from_data`` does for the playground.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _mock_graph_with_vertices(vertex_specs):
|
||||
"""Build a mock graph whose has_session_id_vertices loop drives ``vertex_specs``.
|
||||
|
||||
Each spec is (vertex_id, raw_params_dict). Returns the graph plus the
|
||||
list of vertex mocks so tests can assert on update_raw_params calls.
|
||||
"""
|
||||
graph = MagicMock()
|
||||
graph.context = {}
|
||||
graph.vertices = []
|
||||
graph.edges = []
|
||||
graph.prepare = MagicMock()
|
||||
|
||||
async def _async_start(_inputs, **_kwargs):
|
||||
yield
|
||||
|
||||
graph.async_start = _async_start
|
||||
|
||||
vertex_mocks = {}
|
||||
for vertex_id, raw_params in vertex_specs:
|
||||
vertex = MagicMock()
|
||||
vertex.raw_params = dict(raw_params)
|
||||
vertex.update_raw_params = MagicMock()
|
||||
vertex_mocks[vertex_id] = vertex
|
||||
|
||||
graph.has_session_id_vertices = list(vertex_mocks.keys())
|
||||
graph.get_vertex = MagicMock(side_effect=lambda vid: vertex_mocks.get(vid))
|
||||
return graph, vertex_mocks
|
||||
|
||||
@staticmethod
|
||||
def _patches():
|
||||
return (
|
||||
patch("lfx.run.base.find_graph_variable"),
|
||||
patch("lfx.run.base.load_graph_from_script"),
|
||||
patch("lfx.run.base.validate_global_variables_for_env"),
|
||||
patch("lfx.run.base.extract_structured_result"),
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_id_propagates_to_vertex_with_empty_input(self, tmp_path):
|
||||
"""A Memory vertex with no hardcoded session_id receives the run's session_id."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
graph, vertices = self._mock_graph_with_vertices([("memory-1", {})])
|
||||
|
||||
find_p, load_p, validate_p, extract_p = self._patches()
|
||||
with find_p as mock_find, load_p as mock_load, validate_p as mock_validate, extract_p as mock_extract:
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path, session_id="from-cli")
|
||||
|
||||
vertices["memory-1"].update_raw_params.assert_called_once_with({"session_id": "from-cli"}, overwrite=True)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_id_does_not_overwrite_hardcoded_vertex_value(self, tmp_path):
|
||||
"""If the flow JSON pinned session_id on the Memory component, the CLI must not clobber it.
|
||||
|
||||
Matches Langflow's playground behavior: ``build_graph_from_data`` only writes
|
||||
when ``raw_params.get("session_id")`` is falsy.
|
||||
"""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
graph, vertices = self._mock_graph_with_vertices([("memory-pinned", {"session_id": "hardcoded-in-flow"})])
|
||||
|
||||
find_p, load_p, validate_p, extract_p = self._patches()
|
||||
with find_p as mock_find, load_p as mock_load, validate_p as mock_validate, extract_p as mock_extract:
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path, session_id="from-cli")
|
||||
|
||||
vertices["memory-pinned"].update_raw_params.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_id_propagation_handles_missing_vertex(self, tmp_path):
|
||||
"""A stale vertex_id in has_session_id_vertices (get_vertex returns None) is skipped, not raised."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
graph, vertices = self._mock_graph_with_vertices([("real-vertex", {})])
|
||||
graph.has_session_id_vertices = ["real-vertex", "ghost-vertex"]
|
||||
|
||||
find_p, load_p, validate_p, extract_p = self._patches()
|
||||
with find_p as mock_find, load_p as mock_load, validate_p as mock_validate, extract_p as mock_extract:
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path, session_id="from-cli")
|
||||
|
||||
vertices["real-vertex"].update_raw_params.assert_called_once()
|
||||
|
||||
|
||||
class TestRunFlowUserId:
|
||||
"""user_id auto-generation on the lfx run path.
|
||||
|
||||
AgentComponent (and any component that resolves variables) hits a precheck
|
||||
in ``custom_component.get_variable`` that requires a non-empty ``self.user_id``.
|
||||
lfx run has no notion of authenticated users, but the precheck still has to
|
||||
pass so the env-fallback variable service can answer. ``run_flow`` therefore
|
||||
auto-generates a ceremonial UUID when none is supplied; the value is unused
|
||||
for variable scoping in lfx (env vars are process-global) but exists to keep
|
||||
component initialization unblocked.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _mock_graph():
|
||||
graph = MagicMock()
|
||||
graph.context = {}
|
||||
graph.vertices = []
|
||||
graph.edges = []
|
||||
graph.prepare = MagicMock()
|
||||
|
||||
async def _async_start(_inputs, **_kwargs):
|
||||
yield
|
||||
|
||||
graph.async_start = _async_start
|
||||
return graph
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_id_autogenerated_when_not_provided(self, tmp_path):
|
||||
"""run_flow assigns a UUID user_id so the component precheck passes."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
mock_graph = self._mock_graph()
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = mock_graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path)
|
||||
|
||||
assert mock_graph.user_id, "user_id should be auto-generated when not provided"
|
||||
assert isinstance(mock_graph.user_id, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_caller_user_id_is_preserved(self, tmp_path):
|
||||
"""An explicit user_id wins over auto-generation."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
mock_graph = self._mock_graph()
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = mock_graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path, user_id="real-user-uuid")
|
||||
|
||||
assert mock_graph.user_id == "real-user-uuid"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_autogenerated_user_ids_are_unique_across_runs(self, tmp_path):
|
||||
"""Each run without a user_id should produce a distinct value (ceremonial UUIDs differ)."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
graph_a = self._mock_graph()
|
||||
graph_b = self._mock_graph()
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
mock_load.return_value = graph_a
|
||||
await run_flow(script_path=script_path)
|
||||
mock_load.return_value = graph_b
|
||||
await run_flow(script_path=script_path)
|
||||
|
||||
assert graph_a.user_id != graph_b.user_id
|
||||
|
||||
|
||||
class TestRunFlowFallbackToEnvVars:
|
||||
"""run_flow must plumb fallback_to_env_vars into ``graph.async_start``.
|
||||
|
||||
Without this, a langflow ``DatabaseVariableService`` registered alongside
|
||||
``database_service`` would raise ``variable not found`` for any
|
||||
``load_from_db=True`` field whose user_id has no Variable row (e.g., the
|
||||
ceremonial UUID lfx auto-generates). The flag tells
|
||||
``loading.update_params_with_load_from_db_fields`` to fall back to
|
||||
``os.environ`` when the DB lookup misses — same behavior as the langflow
|
||||
API path in ``processing.process.run_graph_internal``.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _mock_graph_capturing_kwargs(captured: dict):
|
||||
graph = MagicMock()
|
||||
graph.context = {}
|
||||
graph.vertices = []
|
||||
graph.edges = []
|
||||
graph.prepare = MagicMock()
|
||||
|
||||
async def _async_start(_inputs, **kwargs):
|
||||
captured.update(kwargs)
|
||||
yield
|
||||
|
||||
graph.async_start = _async_start
|
||||
return graph
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_passes_fallback_from_settings_default(self, tmp_path):
|
||||
"""Setting defaults True; run_flow forwards True into async_start."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
captured: dict = {}
|
||||
mock_graph = self._mock_graph_capturing_kwargs(captured)
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = mock_graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path)
|
||||
|
||||
assert captured.get("fallback_to_env_vars") is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_respects_settings_when_disabled(self, tmp_path):
|
||||
"""When LANGFLOW_FALLBACK_TO_ENV_VAR=false, the flag plumbs through as False."""
|
||||
script_path = tmp_path / "test.py"
|
||||
script_path.write_text("graph = None")
|
||||
captured: dict = {}
|
||||
mock_graph = self._mock_graph_capturing_kwargs(captured)
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.settings.fallback_to_env_var = False
|
||||
|
||||
with (
|
||||
patch("lfx.run.base.find_graph_variable") as mock_find,
|
||||
patch("lfx.run.base.load_graph_from_script") as mock_load,
|
||||
patch("lfx.run.base.validate_global_variables_for_env") as mock_validate,
|
||||
patch("lfx.run.base.extract_structured_result") as mock_extract,
|
||||
patch("lfx.run._defaults.get_settings_service", return_value=mock_settings),
|
||||
):
|
||||
mock_find.return_value = {"line_number": 1, "type": "Graph", "source_line": "graph = Graph(...)"}
|
||||
mock_load.return_value = mock_graph
|
||||
mock_validate.return_value = []
|
||||
mock_extract.return_value = {"success": True, "result": "test"}
|
||||
|
||||
await run_flow(script_path=script_path)
|
||||
|
||||
assert captured.get("fallback_to_env_vars") is False
|
||||
|
||||
|
||||
class TestRunFlowGlobalVariables:
|
||||
"""Tests for run_flow global variables injection."""
|
||||
|
||||
|
||||
@ -482,7 +482,7 @@ class TestEnvironmentVariableIntegration:
|
||||
|
||||
asyncio.run(manager.teardown())
|
||||
|
||||
def test_variable_service_uses_env(self, clean_manager):
|
||||
async def test_variable_service_uses_env(self, clean_manager):
|
||||
"""Test that variable service reads from environment."""
|
||||
from lfx.services.variable.service import VariableService
|
||||
|
||||
@ -491,12 +491,12 @@ class TestEnvironmentVariableIntegration:
|
||||
os.environ["TEST_API_KEY"] = "test_value_123" # pragma: allowlist secret
|
||||
try:
|
||||
variables = clean_manager.get(ServiceType.VARIABLE_SERVICE)
|
||||
value = variables.get_variable("TEST_API_KEY")
|
||||
value = await variables.get_variable("TEST_API_KEY")
|
||||
assert value == "test_value_123"
|
||||
finally:
|
||||
del os.environ["TEST_API_KEY"]
|
||||
|
||||
def test_variable_service_in_memory_overrides_env(self, clean_manager):
|
||||
async def test_variable_service_in_memory_overrides_env(self, clean_manager):
|
||||
"""Test that in-memory variables override environment."""
|
||||
from lfx.services.variable.service import VariableService
|
||||
|
||||
@ -506,7 +506,7 @@ class TestEnvironmentVariableIntegration:
|
||||
try:
|
||||
variables = clean_manager.get(ServiceType.VARIABLE_SERVICE)
|
||||
variables.set_variable("TEST_VAR", "memory_value")
|
||||
value = variables.get_variable("TEST_VAR")
|
||||
value = await variables.get_variable("TEST_VAR")
|
||||
assert value == "memory_value"
|
||||
finally:
|
||||
del os.environ["TEST_VAR"]
|
||||
|
||||
@ -206,31 +206,31 @@ class TestVariableService:
|
||||
assert variables.ready is True
|
||||
assert variables.name == "variable_service"
|
||||
|
||||
def test_set_and_get_variable(self, variables):
|
||||
async def test_set_and_get_variable(self, variables):
|
||||
"""Test setting and getting a variable."""
|
||||
variables.set_variable("test_key", "test_value")
|
||||
value = variables.get_variable("test_key")
|
||||
value = await variables.get_variable("test_key")
|
||||
assert value == "test_value"
|
||||
|
||||
def test_get_from_environment(self, variables):
|
||||
async def test_get_from_environment(self, variables):
|
||||
"""Test getting variable from environment."""
|
||||
os.environ["TEST_ENV_VAR"] = "env_value"
|
||||
try:
|
||||
value = variables.get_variable("TEST_ENV_VAR")
|
||||
value = await variables.get_variable("TEST_ENV_VAR")
|
||||
assert value == "env_value"
|
||||
finally:
|
||||
del os.environ["TEST_ENV_VAR"]
|
||||
|
||||
def test_get_nonexistent_variable(self, variables):
|
||||
async def test_get_nonexistent_variable(self, variables):
|
||||
"""Test getting a variable that doesn't exist."""
|
||||
value = variables.get_variable("nonexistent_key")
|
||||
value = await variables.get_variable("nonexistent_key")
|
||||
assert value is None
|
||||
|
||||
def test_delete_variable(self, variables):
|
||||
async def test_delete_variable(self, variables):
|
||||
"""Test deleting a variable."""
|
||||
variables.set_variable("test_key", "test_value")
|
||||
variables.delete_variable("test_key")
|
||||
value = variables.get_variable("test_key")
|
||||
value = await variables.get_variable("test_key")
|
||||
assert value is None
|
||||
|
||||
def test_list_variables(self, variables):
|
||||
@ -242,24 +242,55 @@ class TestVariableService:
|
||||
assert "key1" in vars_list
|
||||
assert "key2" in vars_list
|
||||
|
||||
def test_in_memory_overrides_env(self, variables):
|
||||
async def test_in_memory_overrides_env(self, variables):
|
||||
"""Test that in-memory variables override environment."""
|
||||
os.environ["TEST_VAR"] = "env_value"
|
||||
try:
|
||||
variables.set_variable("TEST_VAR", "memory_value")
|
||||
value = variables.get_variable("TEST_VAR")
|
||||
value = await variables.get_variable("TEST_VAR")
|
||||
assert value == "memory_value"
|
||||
finally:
|
||||
del os.environ["TEST_VAR"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_variable_is_coroutine(self, variables):
|
||||
"""get_variable must be async to match the langflow call site.
|
||||
|
||||
Callers use ``await variable_service.get_variable(...)`` (see
|
||||
``Component.get_variable`` in custom_component.py); awaiting a non-coroutine
|
||||
TypeErrors. lfx's implementation does no I/O — the async wrapper exists
|
||||
purely to match the interface langflow's DatabaseVariableService uses.
|
||||
"""
|
||||
import inspect
|
||||
|
||||
assert inspect.iscoroutinefunction(variables.get_variable)
|
||||
coro = variables.get_variable("anything")
|
||||
assert inspect.iscoroutine(coro)
|
||||
await coro # avoid 'coroutine never awaited' warning
|
||||
|
||||
async def test_get_variable_absorbs_extra_kwargs(self, variables):
|
||||
"""Extra kwargs from the langflow call site must not crash the lfx implementation.
|
||||
|
||||
Langflow calls ``await variable_service.get_variable(user_id=..., name=...,
|
||||
field=..., session=...)``. lfx's implementation must accept and ignore the
|
||||
extras so flows behave identically as long as the variable can be resolved
|
||||
by name.
|
||||
"""
|
||||
os.environ["LFX_KWARG_TEST"] = "value-from-env"
|
||||
try:
|
||||
value = await variables.get_variable(
|
||||
user_id="random-uuid", name="LFX_KWARG_TEST", field="value", session=None
|
||||
)
|
||||
assert value == "value-from-env"
|
||||
finally:
|
||||
del os.environ["LFX_KWARG_TEST"]
|
||||
|
||||
async def test_teardown(self, variables):
|
||||
"""Test service teardown clears variables."""
|
||||
variables.set_variable("test_key", "test_value")
|
||||
await variables.teardown()
|
||||
# Variables should be cleared (verify via public API)
|
||||
assert variables.list_variables() == []
|
||||
assert variables.get_variable("test_key") is None
|
||||
assert await variables.get_variable("test_key") is None
|
||||
|
||||
|
||||
class TestMinimalServicesIntegration:
|
||||
|
||||
@ -427,7 +427,7 @@ class TestRealWorldScenarios:
|
||||
await service_manager.teardown()
|
||||
assert ServiceType.STORAGE_SERVICE not in service_manager.services
|
||||
|
||||
def test_multiple_services_working_together(self, service_manager):
|
||||
async def test_multiple_services_working_together(self, service_manager):
|
||||
"""Test multiple services can coexist and work together."""
|
||||
# Register all minimal services
|
||||
service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService)
|
||||
@ -450,7 +450,7 @@ class TestRealWorldScenarios:
|
||||
# All should be usable
|
||||
tracing.add_log("test_trace", {"message": "test"})
|
||||
variables.set_variable("TEST_KEY", "test_value")
|
||||
assert variables.get_variable("TEST_KEY") == "test_value"
|
||||
assert await variables.get_variable("TEST_KEY") == "test_value"
|
||||
|
||||
def test_config_file_with_all_minimal_services(self, service_manager, temp_config_dir):
|
||||
"""Test loading all minimal services from config file."""
|
||||
|
||||
Reference in New Issue
Block a user