diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 7667c4555..a811c8f6f 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -347,7 +347,7 @@ metadata only. - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first. - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. -- Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. +- Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runtime, all other `/api/*` → Gateway REST APIs. diff --git a/backend/packages/harness/deerflow/runtime/goal.py b/backend/packages/harness/deerflow/runtime/goal.py index 3937ce3a8..e4f4e87e6 100644 --- a/backend/packages/harness/deerflow/runtime/goal.py +++ b/backend/packages/harness/deerflow/runtime/goal.py @@ -13,6 +13,7 @@ import hashlib import inspect import json import logging +import os import threading import weakref from collections.abc import AsyncIterator @@ -25,6 +26,7 @@ from langgraph.checkpoint.base import empty_checkpoint, uuid6 import deerflow.utils.llm_text as llm_text from deerflow.agents.goal_state import GoalBlocker, GoalEvaluation, GoalState from deerflow.models import create_chat_model +from deerflow.tracing import inject_langfuse_metadata from deerflow.utils.messages import message_to_text from deerflow.utils.time import now_iso @@ -242,15 +244,29 @@ def create_goal_evaluator_model( model_name: str | None = None, app_config: Any | None = None, ) -> Any: - """Create the non-thinking chat model used by the goal evaluator.""" + """Create the non-thinking chat model used by the goal evaluator. + + The evaluator runs from ``runtime/runs/worker.py`` after the main graph + run has already completed, so — unlike ``make_lead_agent``/ + ``DeerFlowClient.stream``, which attach ``build_tracing_callbacks()`` at + the graph root and correctly pass ``attach_tracing=False`` to avoid + double-attaching — there is no graph root here for the evaluator's model + call to inherit tracing from. It must attach its own model-level tracing + callbacks, same as the other standalone, non-graph callers + (``oneshot_llm.run_oneshot_llm``, ``MemoryUpdater``). + """ return create_chat_model( name=model_name, thinking_enabled=False, app_config=app_config, - attach_tracing=False, + attach_tracing=True, ) +def _resolve_environment() -> str | None: + return os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT") + + async def evaluate_goal_completion( goal: GoalState, messages: list[Any], @@ -258,8 +274,19 @@ async def evaluate_goal_completion( model: Any | None = None, model_name: str | None = None, app_config: Any | None = None, + thread_id: str | None = None, + user_id: str | None = None, + deerflow_trace_id: str | None = None, ) -> GoalEvaluation: - """Ask a small non-thinking model whether the active goal is satisfied.""" + """Ask a small non-thinking model whether the active goal is satisfied. + + ``thread_id``/``user_id``/``deerflow_trace_id`` are forwarded to Langfuse + trace metadata only (mirrors ``oneshot_llm.run_oneshot_llm``): this is a + standalone model call outside the main graph, so it must inject its own + Langfuse session/user attribution instead of relying on graph-root + callbacks to lift it — same fix as PR #2944 (main graph) and PR #3902 + (memory_agent/suggest_agent). + """ conversation = format_visible_conversation(messages) if not conversation or not has_visible_assistant_evidence(messages): return GoalEvaluation( @@ -283,9 +310,19 @@ async def evaluate_goal_completion( if model is None: model = create_goal_evaluator_model(model_name=model_name, app_config=app_config) + invoke_config: dict[str, Any] = {"run_name": "goal_evaluator"} + inject_langfuse_metadata( + invoke_config, + thread_id=thread_id, + user_id=user_id, + assistant_id="goal_evaluator", + model_name=model_name, + environment=_resolve_environment(), + deerflow_trace_id=deerflow_trace_id, + ) response = await model.ainvoke( [SystemMessage(content=system_instruction), HumanMessage(content=user_content)], - config={"run_name": "goal_evaluator"}, + config=invoke_config, ) return parse_goal_evaluation_response(_extract_response_text(response.content)) diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 71f3ebfb6..4f23065d4 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -528,6 +528,8 @@ async def run_agent( app_config=ctx.app_config, evaluator_model_factory=_get_goal_evaluator_model, abort_event=record.abort_event, + user_id=resolve_runtime_user_id(runtime), + deerflow_trace_id=deerflow_trace_id, ) if continuation_input is None or record.abort_event.is_set(): break @@ -842,6 +844,8 @@ async def _prepare_goal_continuation_input( app_config: AppConfig | None, evaluator_model_factory: Any | None = None, abort_event: asyncio.Event | None = None, + user_id: str | None = None, + deerflow_trace_id: str | None = None, ) -> dict[str, Any] | None: """Evaluate the active goal and return a hidden continuation input if needed. @@ -919,6 +923,9 @@ async def _prepare_goal_continuation_input( model=evaluator_model, model_name=model_name, app_config=app_config, + thread_id=thread_id, + user_id=user_id, + deerflow_trace_id=deerflow_trace_id, ) if abort_event is not None and abort_event.is_set(): return None diff --git a/backend/tests/test_goal_runtime.py b/backend/tests/test_goal_runtime.py index 61bd6b4f6..f539b7e59 100644 --- a/backend/tests/test_goal_runtime.py +++ b/backend/tests/test_goal_runtime.py @@ -116,11 +116,75 @@ def test_evaluate_goal_completion_uses_non_thinking_model(monkeypatch): assert result["satisfied"] is True assert result["blocker"] == "none" assert captured["thinking_enabled"] is False - assert captured["attach_tracing"] is False + # The goal evaluator runs from runtime/runs/worker.py after the main graph + # run has already finished, so there is no graph root for it to inherit + # tracing callbacks from (unlike make_lead_agent/DeerFlowClient.stream, + # which attach build_tracing_callbacks() at the graph root and correctly + # pass attach_tracing=False to avoid double-attaching). It must attach its + # own model-level tracing callbacks, same as the other standalone, + # non-graph callers (oneshot_llm.run_oneshot_llm, MemoryUpdater). + assert captured["attach_tracing"] is True fake_model.ainvoke.assert_awaited_once() + # No thread_id/user_id supplied here, and Langfuse is not enabled in the + # ambient test env, so inject_langfuse_metadata() is a no-op and the + # config is unchanged from the plain run_name — see + # test_evaluate_goal_completion_injects_langfuse_metadata below for the + # Langfuse-enabled case. assert fake_model.ainvoke.await_args.kwargs["config"] == {"run_name": "goal_evaluator"} +def test_evaluate_goal_completion_injects_langfuse_metadata(monkeypatch): + """Regression test for the goal evaluator's Langfuse tracing gap. + + Mirrors PR #2944 (main graph) and PR #3902 (memory_agent/suggest_agent): + a standalone, non-graph model call must inject Langfuse trace-attribute + metadata itself since there is no graph root to lift it from. + """ + from deerflow.config.tracing_config import reset_tracing_config + + for name in ("LANGFUSE_TRACING", "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_BASE_URL"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("LANGFUSE_TRACING", "true") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") + reset_tracing_config() + + fake_model = MagicMock() + fake_model.ainvoke = AsyncMock(return_value=SimpleNamespace(content='{"satisfied": true, "reason": "Done", "evidence_summary": "Done"}')) + state = goal.build_goal_state("Finish") + + try: + result = asyncio.run( + goal.evaluate_goal_completion( + state, + [ + HumanMessage(content="Please finish this."), + AIMessage(content="Done."), + ], + model=fake_model, + model_name="gpt-4o", + app_config=object(), + thread_id="thread-xyz", + user_id="alice", + deerflow_trace_id="gateway-trace-1", + ) + ) + finally: + reset_tracing_config() + + assert result["satisfied"] is True + fake_model.ainvoke.assert_awaited_once() + config = fake_model.ainvoke.await_args.kwargs["config"] + assert config["run_name"] == "goal_evaluator" + metadata = config.get("metadata") or {} + assert metadata.get("langfuse_session_id") == "thread-xyz", "goal evaluator trace must group under the thread's session" + assert metadata.get("langfuse_user_id") == "alice" + assert metadata.get("langfuse_trace_name") == "goal_evaluator" + assert metadata.get("deerflow_trace_id") == "gateway-trace-1" + tags = metadata.get("langfuse_tags") or [] + assert "model:gpt-4o" in tags + + def test_evaluate_goal_completion_uses_injected_model(monkeypatch): fake_model = MagicMock() fake_model.ainvoke = AsyncMock(return_value=SimpleNamespace(content='{"satisfied": true, "reason": "Done", "evidence_summary": "Done"}'))