diff --git a/README.md b/README.md index aa0e9ca8fb..35b87d2de1 100644 --- a/README.md +++ b/README.md @@ -453,6 +453,11 @@ For lightweight single-process event persistence, `run_events.backend: jsonl` keeps Unicode message content intact, including line and paragraph separators. Existing valid JSONL records remain readable without rewriting the files. +For per-call usage audits, an immediate LLM response replay with populated usage +updates both nested usage and top-level token counters, even if initial usage was absent or zero. +The original request, model, and status metadata stay intact; flushed events are +not rewritten. See the [run event contract](contracts/run_event_stream_contract.json). + The unified nginx endpoint is same-origin by default and does not emit browser CORS headers. If you run a split-origin or port-forwarded browser client, set `GATEWAY_CORS_ORIGINS` to comma-separated exact origins such as `http://localhost:3000`; the Gateway then applies the CORS allowlist and matching CSRF origin checks. When fine-grained authorization is enabled, Live Browser connections require `threads:write` as well as ownership of the thread, even when only viewing frames: the same connection can control the browser. Permission checks run when connecting. Restart Gateway after upgrading to disconnect sessions admitted by older code. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index a894bb04b8..cbcc524823 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -370,6 +370,7 @@ Automatic conversation summarization when approaching token limits: manual state updates). Its short-lived `checkpoint_write` thread operation shares the durable active-thread uniqueness constraint with run admission, preventing either worker-local or cross-worker checkpoint-write races. +- Each summarizer call emits a `middleware:summarize` journal event (`MIDDLEWARE_SUMMARIZE_TAG`; `action="summary_result"`, `changes.noop` true when the output is byte-identical to the prior `summary_text`) and bumps `summary_noop_count` / `summary_call_count` on the middleware. This is observation only; the summarizer call is never skipped. Per-call LLM telemetry lives on `RunJournal` (see the runtime guide). See [docs/summarization.md](docs/summarization.md) for details. diff --git a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py index e3ffa8251e..a89bdfdc88 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py @@ -27,6 +27,7 @@ from deerflow.config.task_continuity_config import TaskContinuityConfig from deerflow.extensions.notify import notify_context_compacted from deerflow.models import create_chat_model from deerflow.models.reasoning import resolve_reasoning_contract +from deerflow.runtime.events.catalog import MIDDLEWARE_SUMMARIZE_TAG logger = logging.getLogger(__name__) _SUMMARY_TRIGGER_MESSAGE_NAME = "summary" @@ -163,6 +164,8 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): extensions = get_agent_build_extensions() self._extensions = extensions + self.summary_call_count = 0 + self.summary_noop_count = 0 # Nostream generation models are either prebuilt by the factory or built # lazily by name. An eagerly seeded ``None`` pins the run-model fallback for # this middleware instance; unlike the legacy path, it does not allow one @@ -658,6 +661,38 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): ) notify_context_compacted(event, extensions=self._extensions) + def _record_summary_telemetry(self, runtime: Runtime, previous_summary: str | None, summary: str, summarized_count: int) -> None: + """P0 measurement only: count summarizer calls whose output equals the prior ``summary_text``. + + Never skips or alters the summarizer call; fails soft. + """ + try: + noop = previous_summary is not None and summary.encode("utf-8") == previous_summary.encode("utf-8") + self.summary_call_count = self.summary_call_count + 1 + if noop: + self.summary_noop_count = self.summary_noop_count + 1 + logger.info("Summarization no-op: summarizer output identical to existing summary_text (%d chars)", len(summary)) + context = getattr(runtime, "context", None) + journal = context.get("__run_journal") if isinstance(context, dict) else None + if journal is None: + return + journal.record_middleware( + MIDDLEWARE_SUMMARIZE_TAG, + name=type(self).__name__, + hook="before_model", + action="summary_result", + changes={ + "noop": noop, + "summary_chars": len(summary), + "previous_summary_chars": len(previous_summary) if previous_summary is not None else None, + "summarized_message_count": summarized_count, + "noop_count": self.summary_noop_count, + "call_count": self.summary_call_count, + }, + ) + except Exception: + logger.warning("Summarization telemetry failed", exc_info=True) + def compact_state( self, state: AgentState, @@ -684,6 +719,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): if raise_on_failure: raise SummaryGenerationError("summary generation failed") return None + self._record_summary_telemetry(runtime, previous_summary, summary, len(messages_to_summarize)) # Fire hooks only once a replacement summary exists — flushing pre-compaction # messages into durable memory for a summary that never materializes would # duplicate that work on the next attempt. Messages are still removed after @@ -733,6 +769,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): if raise_on_failure: raise SummaryGenerationError("summary generation failed") return None + self._record_summary_telemetry(runtime, previous_summary, summary, len(messages_to_summarize)) # Fire hooks only once a replacement summary exists (see compact_state). self._fire_hooks(messages_to_summarize, preserved_messages, runtime) self._record_compaction( diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 58d1dd69e3..e166e96022 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -33,6 +33,19 @@ the optional display summary. Image-only input has no text but must still stop the batch scan and later model calls from appending another human-input event. `tests/test_run_journal.py` covers callback and full/delta graph paths. +**Per-call LLM telemetry** (`runtime/journal.py`): `RunJournal` adds observation-only keys +to `llm.ai.response` metadata and to `llm.error` metadata (previously empty): `langchain_run_id`, +`langchain_parent_run_id`, `caller_category` (`lead_agent` / `middleware` / `subagent` / +`fallback` / `other`), `provider`, `model`, `stop_reason`, `status`, provider-reported token counts, +and the rendered request size measured at `on_chat_model_start` (`request_chars`, +`request_message_chars`, `request_tools_chars`, `request_message_count`, `request_started_at`). +Sizes count string-leaf characters without serializing the request (the callback runs inline on +the event loop); there is no pre-call token count, so `input_tokens` is the provider's figure. +Unavailable values are `None` or absent, never estimated. Every helper fails soft: a telemetry +error is logged and must not change the call, the staged events' existing fields, or token +accounting. The contract lists the optional keys in `contracts/run_event_stream_contract.json`. +Retries are not distinguishable from first attempts. + **LLM response callback coalescing** (`runtime/journal.py`): a provider may fire `on_llm_end` twice for one LangChain run id, first without usage (or with all token counts zero) and immediately again with usage populated. The first callback's generation @@ -43,7 +56,9 @@ It must not retain provider-owned message objects because a provider may mutate reuse the same response for the usage replay. Usage metadata is deep-snapshotted, including nested token-detail mappings, before it enters a staged or buffered event. An adjacent same-id positive-usage replay may enrich only each corresponding staged -event's metadata/content usage fields. Replay +event's metadata/content usage fields, including the `input_tokens`, `output_tokens`, +and `total_tokens` aliases derived from the accepted usage snapshot. Request size, +identity, caller, stop reason, and status retain their canonical values. Replay generation-count differences never add, remove, or replace canonical messages. The next unrelated event, an effective buffer size (committed plus pending events) reaching the flush threshold, or an explicit flush commits the staged unit and updates the message diff --git a/backend/packages/harness/deerflow/runtime/events/catalog.py b/backend/packages/harness/deerflow/runtime/events/catalog.py index a0e85aed58..74875a2271 100644 --- a/backend/packages/harness/deerflow/runtime/events/catalog.py +++ b/backend/packages/harness/deerflow/runtime/events/catalog.py @@ -83,6 +83,7 @@ MIDDLEWARE_SKILL_ACTIVATION_TAG = "skill_activation" MIDDLEWARE_SKILL_SECRETS_TAG = "skill_secrets" MIDDLEWARE_TOOL_PROMOTION_TAG = "tool_promotion" MIDDLEWARE_TOOL_PROGRESS_TAG = "tool_progress" +MIDDLEWARE_SUMMARIZE_TAG = "summarize" MIDDLEWARE_EVENT_TAGS = ( MIDDLEWARE_GUARDRAIL_TAG, MIDDLEWARE_LOOP_DETECTION_TAG, @@ -91,6 +92,7 @@ MIDDLEWARE_EVENT_TAGS = ( MIDDLEWARE_SKILL_SECRETS_TAG, MIDDLEWARE_TOOL_PROMOTION_TAG, MIDDLEWARE_TOOL_PROGRESS_TAG, + MIDDLEWARE_SUMMARIZE_TAG, ) JOURNAL_RUN_EVENT_DEFINITIONS = ( diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 29a6afb175..488189ac79 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -59,6 +59,77 @@ _PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES = frozenset({"ask_clarification", _MAX_RUN_SKILL_SNAPSHOTS = 64 +def _content_chars(content: Any) -> int: + """Characters in the string leaves and mapping keys of ``content``. + + Walks the structure without serializing it, so cost scales with the number of + nodes rather than the byte size (base64 image payloads are counted by ``len``). + """ + total = 0 + stack = [content] + while stack: + item = stack.pop() + if isinstance(item, str): + total += len(item) + elif isinstance(item, Mapping): + for key, value in item.items(): + total += len(key) if isinstance(key, str) else 0 + stack.append(value) + elif isinstance(item, (list, tuple)): + stack.extend(item) + return total + + +def _rendered_request_size(messages: Sequence[Sequence[BaseMessage]], invocation_params: Any) -> dict[str, Any]: + """Measure the request as rendered at ``on_chat_model_start`` (read-only). + + Sizes are characters of string content (see ``_content_chars``), not serialized + bytes. Token counts for the rendered request are not available without an extra + model/tokenizer call, so ``input_tokens`` from the provider's usage metadata + (recorded at ``on_llm_end``) is the token figure. ``request_tools_chars`` is + ``None`` when the callback carries no bound-tool schema. + """ + message_chars = 0 + message_count = 0 + for batch in messages: + for message in batch: + message_count += 1 + message_chars += _content_chars(getattr(message, "content", "")) + tool_calls = getattr(message, "tool_calls", None) + if tool_calls: + message_chars += _content_chars(tool_calls) + tools = invocation_params.get("tools") if isinstance(invocation_params, Mapping) else None + tools_chars = _content_chars(tools) if tools else None + return { + "request_chars": message_chars + (tools_chars or 0), + "request_message_chars": message_chars, + "request_tools_chars": tools_chars, + "request_message_count": message_count, + } + + +def _llm_caller_category(caller: str, *, error_fallback: bool) -> str: + if error_fallback: + return "fallback" + if caller.startswith("middleware:"): + return "middleware" + if caller.startswith("subagent:"): + return "subagent" + if caller == "lead_agent": + return "lead_agent" + return "other" + + +def _first_str(mapping: Any, *keys: str) -> str | None: + if not isinstance(mapping, Mapping): + return None + for key in keys: + value = mapping.get(key) + if isinstance(value, str) and value: + return value + return None + + @dataclass class _PendingLlmResponse: llm_run_id: str @@ -306,6 +377,9 @@ class RunJournal(BaseCallbackHandler): # Latency tracking self._llm_start_times: dict[str, float] = {} # langchain run_id -> start time + # P0 telemetry: per-call request measurement captured at start, merged into + # the llm.ai.response / llm.error metadata at end. Observation only. + self._llm_start_telemetry: dict[str, dict[str, Any]] = {} # LLM request/response tracking self._llm_call_index = 0 @@ -424,6 +498,7 @@ class RunJournal(BaseCallbackHandler): self._llm_start_times[rid] = time.monotonic() self._llm_call_index += 1 self._seen_llm_starts.add(rid) + self._capture_start_telemetry(rid, serialized, messages, kwargs) logger.debug( "on_chat_model_start %s: tags=%s num_batches=%d message_counts=%s", @@ -537,6 +612,7 @@ class RunJournal(BaseCallbackHandler): "usage": usage_dict, "latency_ms": latency_ms, "llm_call_index": call_index, + **self._end_telemetry(rid, message, usage_dict, caller, parent_run_id), }, ) ) @@ -572,6 +648,8 @@ class RunJournal(BaseCallbackHandler): should_schedule_progress = True + self._llm_start_telemetry.pop(rid, None) + if messages: self._queue_llm_response_events( str(run_id), @@ -584,13 +662,75 @@ class RunJournal(BaseCallbackHandler): self._schedule_progress_flush() def on_llm_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None: - self._llm_start_times.pop(str(run_id), None) + rid = str(run_id) + self._llm_start_times.pop(rid, None) + error_metadata = self._error_telemetry(rid, error, kwargs.get("tags"), kwargs.get("parent_run_id")) self._put( event_type=LLM_ERROR_EVENT.event_type, category=LLM_ERROR_EVENT.category, content=str(error), + metadata=error_metadata or None, ) + # -- P0 per-call telemetry (observation only; every helper fails soft) -- + + def _capture_start_telemetry(self, rid: str, serialized: Any, messages: list[list[BaseMessage]], kwargs: dict[str, Any]) -> None: + try: + start_metadata = kwargs.get("metadata") + telemetry = _rendered_request_size(messages, kwargs.get("invocation_params")) + telemetry["request_started_at"] = datetime.now(UTC).isoformat() + telemetry["provider"] = _first_str(start_metadata, "ls_provider") + telemetry["requested_model"] = _first_str(start_metadata, "ls_model_name") + if telemetry["provider"] is None and isinstance(serialized, Mapping): + serialized_id = serialized.get("id") + if isinstance(serialized_id, list) and serialized_id: + telemetry["provider"] = str(serialized_id[-1]) + self._llm_start_telemetry[rid] = telemetry + except Exception: + logger.warning("LLM start telemetry capture failed for %s", rid, exc_info=True) + + def _end_telemetry(self, rid: str, message: Any, usage: Mapping[str, Any], caller: str, parent_run_id: UUID | None) -> dict[str, Any]: + try: + start = self._llm_start_telemetry.get(rid) or {} + response_metadata = getattr(message, "response_metadata", None) or {} + additional_kwargs = getattr(message, "additional_kwargs", None) or {} + error_fallback = isinstance(additional_kwargs, Mapping) and bool(additional_kwargs.get("deerflow_error_fallback")) + return { + "langchain_run_id": rid, + "langchain_parent_run_id": str(parent_run_id) if parent_run_id else None, + "caller_category": _llm_caller_category(caller, error_fallback=error_fallback), + "provider": start.get("provider"), + "model": _first_str(response_metadata, "model_name", "model") or start.get("requested_model"), + "stop_reason": _first_str(response_metadata, "stop_reason", "finish_reason", "finishReason"), + "status": "fallback" if error_fallback else "ok", + "input_tokens": usage.get("input_tokens"), + "output_tokens": usage.get("output_tokens"), + "total_tokens": usage.get("total_tokens"), + **{k: v for k, v in start.items() if k.startswith("request_")}, + } + except Exception: + logger.warning("LLM end telemetry build failed for %s", rid, exc_info=True) + return {} + + def _error_telemetry(self, rid: str, error: BaseException, tags: list[str] | None, parent_run_id: UUID | None) -> dict[str, Any]: + try: + start = self._llm_start_telemetry.pop(rid, None) or {} + return { + "langchain_run_id": rid, + "langchain_parent_run_id": str(parent_run_id) if parent_run_id else None, + "caller_category": _llm_caller_category(self._identify_caller(tags), error_fallback=False), + "caller": self._identify_caller(tags), + "provider": start.get("provider"), + "model": start.get("requested_model"), + "status": "error", + "stop_reason": None, + "error_type": type(error).__name__, + **{k: v for k, v in start.items() if k.startswith("request_")}, + } + except Exception: + logger.warning("LLM error telemetry build failed for %s", rid, exc_info=True) + return {} + def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id=None, tags=None, metadata=None, inputs=None, **kwargs): """Cache the executing tool name for artifact attribution.""" tool_run_id = str(run_id) @@ -837,7 +977,10 @@ class RunJournal(BaseCallbackHandler): if isinstance(replay_metadata, Mapping): replay_usage = replay_metadata.get("usage") if isinstance(replay_usage, Mapping): - canonical["metadata"]["usage"] = deepcopy(dict(replay_usage)) + canonical_metadata = canonical["metadata"] + canonical_metadata["usage"] = deepcopy(dict(replay_usage)) + for key in ("input_tokens", "output_tokens", "total_tokens"): + canonical_metadata[key] = canonical_metadata["usage"].get(key) canonical_content = canonical.get("content") replay_content = replay.get("content") diff --git a/backend/tests/test_llm_call_telemetry.py b/backend/tests/test_llm_call_telemetry.py new file mode 100644 index 0000000000..02f73419a6 --- /dev/null +++ b/backend/tests/test_llm_call_telemetry.py @@ -0,0 +1,291 @@ +"""P0 per-LLM-call telemetry: observation only, fail-soft.""" + +from __future__ import annotations + +from copy import deepcopy +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage +from langchain_core.outputs import ChatGeneration, LLMResult + +from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.journal import RunJournal + + +def _response(*, usage=None, response_metadata=None, additional_kwargs=None) -> LLMResult: + message = AIMessage(content="hello", usage_metadata=usage, response_metadata=response_metadata or {}, additional_kwargs=additional_kwargs or {}) + return LLMResult(generations=[[ChatGeneration(message=message)]]) + + +async def _events(journal: RunJournal, store: MemoryRunEventStore, event_type: str) -> list[dict]: + await journal.flush() + return [e for e in await store.list_events("t1", "r1") if e["event_type"] == event_type] + + +@pytest.mark.anyio +async def test_normal_call_records_telemetry_and_rendered_request_size(): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + rid = uuid4() + messages = [[SystemMessage(content="sys"), HumanMessage(content="question")]] + journal.on_chat_model_start( + {"id": ["langchain", "ChatAnthropic"]}, + messages, + run_id=rid, + tags=["lead_agent"], + metadata={"ls_provider": "anthropic", "ls_model_name": "claude-x"}, + invocation_params={"tools": [{"name": "t", "description": "d"}]}, + ) + journal.on_llm_end( + _response(usage={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, response_metadata={"stop_reason": "end_turn", "model_name": "claude-x-1"}), + run_id=rid, + tags=["lead_agent"], + ) + + (event,) = await _events(journal, store, "llm.ai.response") + meta = event["metadata"] + assert meta["langchain_run_id"] == str(rid) + assert meta["caller"] == "lead_agent" + assert meta["caller_category"] == "lead_agent" + assert meta["provider"] == "anthropic" + assert meta["model"] == "claude-x-1" + assert meta["stop_reason"] == "end_turn" + assert (meta["input_tokens"], meta["output_tokens"], meta["total_tokens"]) == (10, 5, 15) + assert meta["request_message_chars"] == len("sys") + len("question") + assert meta["request_message_count"] == 2 + assert meta["request_tools_chars"] > 0 + assert meta["request_chars"] == meta["request_message_chars"] + meta["request_tools_chars"] + assert event["thread_id"] == "t1" and event["run_id"] == "r1" and event["created_at"] + + +@pytest.mark.anyio +async def test_unavailable_values_are_none_not_fabricated(): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + rid = uuid4() + journal.on_llm_end(_response(), run_id=rid, tags=None) # no on_chat_model_start + + (event,) = await _events(journal, store, "llm.ai.response") + meta = event["metadata"] + assert meta["stop_reason"] is None + assert meta["input_tokens"] is None + assert meta["provider"] is None + assert "request_chars" not in meta + + +@pytest.mark.anyio +async def test_middleware_and_fallback_caller_classification(): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + mid, fb = uuid4(), uuid4() + journal.on_llm_end(_response(), run_id=mid, tags=["middleware:summarize"]) + journal.on_llm_end(_response(additional_kwargs={"deerflow_error_fallback": True}), run_id=fb, tags=["lead_agent"]) + + events = await _events(journal, store, "llm.ai.response") + by_run = {e["metadata"]["langchain_run_id"]: e["metadata"] for e in events} + assert by_run[str(mid)]["caller_category"] == "middleware" + assert by_run[str(fb)]["caller_category"] == "fallback" + assert by_run[str(fb)]["status"] == "fallback" + + +@pytest.mark.anyio +async def test_llm_error_is_recorded_with_status(): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + rid = uuid4() + journal.on_chat_model_start({}, [[HumanMessage(content="abc")]], run_id=rid, tags=["middleware:title"]) + journal.on_llm_error(RuntimeError("boom"), run_id=rid, tags=["middleware:title"]) + + (event,) = await _events(journal, store, "llm.error") + assert event["metadata"]["status"] == "error" + assert event["metadata"]["caller_category"] == "middleware" + assert event["metadata"]["request_chars"] == 3 + assert event["content"] == "boom" + + +@pytest.mark.anyio +async def test_telemetry_failure_does_not_affect_call_or_event(monkeypatch): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + + def explode(*_a, **_k): + raise RuntimeError("telemetry broken") + + monkeypatch.setattr("deerflow.runtime.journal._rendered_request_size", explode) + monkeypatch.setattr("deerflow.runtime.journal._llm_caller_category", explode) + rid = uuid4() + journal.on_chat_model_start({}, [[HumanMessage(content="hi")]], run_id=rid, tags=["lead_agent"]) # must not raise + usage = {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5} + journal.on_llm_end(_response(usage=usage), run_id=rid, tags=["lead_agent"]) # must not raise + + (event,) = await _events(journal, store, "llm.ai.response") + assert event["metadata"]["usage"] == usage # pre-existing fields intact + assert event["metadata"]["caller"] == "lead_agent" + assert "langchain_run_id" not in event["metadata"] + assert journal._total_tokens == 5 + + +@pytest.mark.anyio +async def test_llm_error_without_prior_start_records_unavailable_request_size(): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + rid = uuid4() + journal.on_llm_error(ValueError("bad"), run_id=rid, tags=["subagent:general-purpose"]) # no on_chat_model_start + + (event,) = await _events(journal, store, "llm.error") + meta = event["metadata"] + assert meta["status"] == "error" + assert meta["caller"] == "subagent:general-purpose" + assert meta["caller_category"] == "subagent" + assert meta["error_type"] == "ValueError" + assert meta["provider"] is None and meta["model"] is None + assert "request_chars" not in meta # unavailable, not fabricated + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "initial_usage", + [None, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}], + ids=["missing-usage", "zero-usage"], +) +async def test_replayed_on_llm_end_keeps_canonical_request_metadata(initial_usage): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + rid, parent_id = uuid4(), uuid4() + journal.on_chat_model_start( + {}, + [[HumanMessage(content="question")]], + run_id=rid, + tags=["lead_agent"], + metadata={"ls_provider": "anthropic", "ls_model_name": "requested-model"}, + ) + response_metadata = {"model_name": "canonical-model", "finish_reason": "stop"} + # The first callback has missing or zero usage; the adjacent replay fills it in. + journal.on_llm_end(_response(usage=initial_usage, response_metadata=response_metadata), run_id=rid, parent_run_id=parent_id, tags=["lead_agent"]) + assert journal._pending_llm_response is not None + canonical_metadata = dict(journal._pending_llm_response.events[0]["metadata"]) + usage = {"input_tokens": 7, "output_tokens": 3, "total_tokens": 10} + journal.on_llm_end( + _response( + usage=usage, + response_metadata={"model_name": "replay-model", "finish_reason": "length"}, + additional_kwargs={"deerflow_error_fallback": True, "error_detail": "replay fallback"}, + ), + run_id=rid, + parent_run_id=uuid4(), + tags=["middleware:summarize"], + ) + + (event,) = await _events(journal, store, "llm.ai.response") + meta = event["metadata"] + assert meta == {**canonical_metadata, "usage": usage, **usage} + assert event["content"]["usage_metadata"] == usage + assert event["content"]["response_metadata"] == response_metadata + assert meta["request_message_chars"] == len("question") # canonical request size preserved + assert meta["request_message_count"] == 1 + assert meta["langchain_run_id"] == str(rid) + assert meta["langchain_parent_run_id"] == str(parent_id) + assert meta["caller_category"] == "lead_agent" and meta["status"] == "ok" + assert meta["model"] == "canonical-model" and meta["stop_reason"] == "stop" + assert not journal.had_llm_error_fallback + assert journal.get_completion_data()["total_tokens"] == 10 + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "initial_usage", + [None, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}], + ids=["missing-usage", "zero-usage"], +) +async def test_usage_replay_after_flush_keeps_committed_telemetry(initial_usage): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + rid = uuid4() + journal.on_llm_end(_response(usage=initial_usage), run_id=rid, tags=["lead_agent"]) + (canonical_event,) = await _events(journal, store, "llm.ai.response") + canonical_event = deepcopy(canonical_event) + + usage = {"input_tokens": 7, "output_tokens": 3, "total_tokens": 10} + journal.on_llm_end(_response(usage=usage), run_id=rid, tags=["lead_agent"]) + + (event,) = await _events(journal, store, "llm.ai.response") + assert event == canonical_event + assert journal.get_completion_data()["total_tokens"] == 10 + + +@pytest.mark.anyio +async def test_telemetry_metadata_round_trips_through_db_store(tmp_path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'test.db'}", sqlite_dir=str(tmp_path)) + try: + store = DbRunEventStore(get_session_factory()) + journal = RunJournal("r1", "t1", store, flush_threshold=100) + ok_id, err_id = uuid4(), uuid4() + journal.on_chat_model_start({}, [[HumanMessage(content="hello")]], run_id=ok_id, tags=["lead_agent"], metadata={"ls_provider": "anthropic"}) + journal.on_llm_end(_response(usage={"input_tokens": 2, "output_tokens": 1, "total_tokens": 3}, response_metadata={"stop_reason": "end_turn"}), run_id=ok_id, tags=["lead_agent"]) + journal.on_chat_model_start({}, [[HumanMessage(content="abc")]], run_id=err_id, tags=["middleware:title"]) + journal.on_llm_error(RuntimeError("boom"), run_id=err_id, tags=["middleware:title"]) + journal.record_middleware("summarize", name="DeerFlowSummarizationMiddleware", hook="before_model", action="summary_result", changes={"noop": False}) + await journal.flush() + + events = await store.list_events("t1", "r1") + by_type = {e["event_type"]: e for e in events} + ai = by_type["llm.ai.response"]["metadata"] + assert ai["langchain_run_id"] == str(ok_id) and ai["provider"] == "anthropic" and ai["stop_reason"] == "end_turn" + assert ai["input_tokens"] == 2 and ai["request_chars"] == len("hello") + err = by_type["llm.error"]["metadata"] + assert err["status"] == "error" and err["caller_category"] == "middleware" and err["request_chars"] == 3 + assert by_type["middleware:summarize"]["content"]["changes"] == {"noop": False} + finally: + await close_engine() + + +def _summarizer(text: str) -> DeerFlowSummarizationMiddleware: + model = MagicMock() + model.invoke.return_value = SimpleNamespace(text=text) + model.ainvoke = AsyncMock(return_value=SimpleNamespace(text=text)) + model.with_config.return_value = model + return DeerFlowSummarizationMiddleware(model=model, trigger=("messages", 4), keep=("messages", 2), token_counter=len) + + +def _state(summary: str | None) -> dict: + messages = [HumanMessage(content="u1"), AIMessage(content="a1"), HumanMessage(content="u2"), AIMessage(content="a2")] + return {"messages": messages, "summary_text": summary} + + +@pytest.mark.anyio +async def test_summarization_noop_detected_and_call_not_skipped(): + store = MemoryRunEventStore() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + runtime = SimpleNamespace(context={"thread_id": "t1", "__run_journal": journal}) + middleware = _summarizer("same summary") + + result = middleware.compact_state(_state("same summary"), runtime, force=True) + assert result is not None and result.summary_text == "same summary" # behavior unchanged + assert middleware.model.invoke.call_count == 1 # call NOT skipped + assert middleware.summary_noop_count == 1 + + middleware.compact_state(_state("different"), runtime, force=True) + middleware.compact_state(_state(None), runtime, force=True) + assert middleware.summary_noop_count == 1 + assert middleware.summary_call_count == 3 + + events = await _events(journal, store, "middleware:summarize") + assert [e["content"]["changes"]["noop"] for e in events] == [True, False, False] + + +def test_summarization_telemetry_failure_does_not_break_compaction(): + journal = MagicMock() + journal.record_middleware.side_effect = RuntimeError("journal down") + runtime = SimpleNamespace(context={"thread_id": "t1", "__run_journal": journal}) + middleware = _summarizer("s") + + result = middleware.compact_state(_state(None), runtime, force=True) + + assert result is not None and result.summary_text == "s" diff --git a/contracts/run_event_stream_contract.json b/contracts/run_event_stream_contract.json index 0d98532950..2a59d6bcc8 100644 --- a/contracts/run_event_stream_contract.json +++ b/contracts/run_event_stream_contract.json @@ -162,7 +162,22 @@ "caller": {"type": "string"}, "usage": {"type": "object"}, "latency_ms": {"type": ["integer", "null"], "minimum": 0}, - "llm_call_index": {"type": "integer", "minimum": 1} + "llm_call_index": {"type": "integer", "minimum": 1}, + "langchain_run_id": {"type": "string"}, + "langchain_parent_run_id": {"type": ["string", "null"]}, + "caller_category": {"enum": ["lead_agent", "middleware", "subagent", "fallback", "other"]}, + "provider": {"type": ["string", "null"]}, + "model": {"type": ["string", "null"]}, + "stop_reason": {"type": ["string", "null"]}, + "status": {"enum": ["ok", "fallback"]}, + "input_tokens": {"type": ["integer", "null"], "minimum": 0}, + "output_tokens": {"type": ["integer", "null"], "minimum": 0}, + "total_tokens": {"type": ["integer", "null"], "minimum": 0}, + "request_chars": {"type": "integer", "minimum": 0}, + "request_message_chars": {"type": "integer", "minimum": 0}, + "request_tools_chars": {"type": ["integer", "null"], "minimum": 0}, + "request_message_count": {"type": "integer", "minimum": 0}, + "request_started_at": {"type": "string"} }, "additionalProperties": true } @@ -188,7 +203,26 @@ "category": "trace", "producer": "RunJournal.on_llm_error()", "content_schema": {"type": "string"}, - "metadata_schema": {"type": "object", "additionalProperties": true} + "metadata_schema": { + "type": "object", + "properties": { + "langchain_run_id": {"type": "string"}, + "langchain_parent_run_id": {"type": ["string", "null"]}, + "caller": {"type": "string"}, + "caller_category": {"enum": ["lead_agent", "middleware", "subagent", "fallback", "other"]}, + "provider": {"type": ["string", "null"]}, + "model": {"type": ["string", "null"]}, + "status": {"const": "error"}, + "stop_reason": {"type": "null"}, + "error_type": {"type": "string"}, + "request_chars": {"type": "integer", "minimum": 0}, + "request_message_chars": {"type": "integer", "minimum": 0}, + "request_tools_chars": {"type": ["integer", "null"], "minimum": 0}, + "request_message_count": {"type": "integer", "minimum": 0}, + "request_started_at": {"type": "string"} + }, + "additionalProperties": true + } }, { "event_type": "context:memory", @@ -399,7 +433,7 @@ "pattern": "middleware:{tag}", "category": "middleware", "producer": "RunJournal.record_middleware(tag, ...)", - "known_tags": ["guardrail", "loop_detection", "safety_termination", "skill_activation", "skill_secrets", "tool_promotion", "tool_progress"], + "known_tags": ["guardrail", "loop_detection", "safety_termination", "skill_activation", "skill_secrets", "tool_promotion", "tool_progress", "summarize"], "event_type_schema": { "type": "string", "pattern": "^middleware:",