From 54bd7a3d4ec5b3e874c700ccc80541a3f3e16c85 Mon Sep 17 00:00:00 2001 From: 666ghj <670939375@qq.com> Date: Wed, 22 Jul 2026 23:46:19 +0800 Subject: [PATCH] fix: validate complete Zep entity context --- backend/app/services/zep_entity_reader.py | 24 +- .../scripts/validate_zep_cloud_integration.py | 904 ++++++++++++++++++ backend/tests/test_zep_cloud_contracts.py | 64 ++ .../tests/test_zep_cloud_validation_script.py | 105 ++ 4 files changed, 1094 insertions(+), 3 deletions(-) create mode 100644 backend/scripts/validate_zep_cloud_integration.py create mode 100644 backend/tests/test_zep_cloud_validation_script.py diff --git a/backend/app/services/zep_entity_reader.py b/backend/app/services/zep_entity_reader.py index 1e20c6a3..92776b94 100644 --- a/backend/app/services/zep_entity_reader.py +++ b/backend/app/services/zep_entity_reader.py @@ -165,17 +165,35 @@ class ZepEntityReader: logger.info(f"共获取 {len(edges_data)} 条边") return edges_data - def get_node_edges(self, node_uuid: str) -> List[Dict[str, Any]]: + def get_node_edges( + self, + node_uuid: str, + *, + graph_id: Optional[str] = None, + ) -> List[Dict[str, Any]]: """ - 获取指定节点的所有相关边（带重试机制） + 获取指定节点的相关边。 + + Zep Cloud 3.25 的 ``graph.node.get_edges`` 实测只返回节点作为 + source 的边，尽管文档将其描述为“all edges”。需要完整上下文时必须 + 提供 graph_id，以全图分页后同时筛选 incoming 和 outgoing 边。 Args: node_uuid: 节点UUID + graph_id: 图谱ID；提供时保证返回双向完整关系 Returns: 边列表 """ try: + if graph_id: + return [ + edge + for edge in self.get_all_edges(graph_id) + if edge["source_node_uuid"] == node_uuid + or edge["target_node_uuid"] == node_uuid + ] + # 使用重试机制调用Zep API edges = self._call_with_retry( func=lambda: self.client.graph.node.get_edges(node_uuid=node_uuid), @@ -344,7 +362,7 @@ class ZepEntityReader: return None # 获取节点的边 - edges = self.get_node_edges(entity_uuid) + edges = self.get_node_edges(entity_uuid, graph_id=graph_id) # 获取所有节点用于关联查找 all_nodes = self.get_all_nodes(graph_id) diff --git a/backend/scripts/validate_zep_cloud_integration.py b/backend/scripts/validate_zep_cloud_integration.py new file mode 100644 index 00000000..be2f9c6b --- /dev/null +++ b/backend/scripts/validate_zep_cloud_integration.py @@ -0,0 +1,904 @@ +#!/usr/bin/env python3 +"""Run a manual, real Zep Cloud validation against a retained test graph. + +This script is intentionally excluded from the automated test suite. It requires +``ZEP_API_KEY`` at runtime, never prints the key, and deletes the graph unless +``--keep-graph`` is explicitly supplied. If the activity updater cannot be +confirmed drained after a failure, the graph is retained to avoid a write/delete +race. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Iterable + +# Capture the caller-supplied key before importing MiroFish modules. app.config +# deliberately loads the repository .env with override=True, which must not +# silently replace the account selected for this explicit validation process. +_PROCESS_ZEP_API_KEY = os.environ.get("ZEP_API_KEY", "").strip() + +from zep_cloud import BatchAddItem +from zep_cloud.types import SearchFilters + +from app.services.graph_builder import BatchSubmission, GraphBuilderService +from app.services.zep_entity_reader import ZepEntityReader +from app.services.zep_graph_memory_updater import AgentActivity, ZepGraphMemoryUpdater +from app.utils.zep_paging import fetch_all_edges, fetch_all_nodes + +# Keep later code that consults os.environ consistent with the captured value. +# All Cloud clients in this script also receive the key explicitly. +if _PROCESS_ZEP_API_KEY: + os.environ["ZEP_API_KEY"] = _PROCESS_ZEP_API_KEY + + +@dataclass(frozen=True) +class SourceEpisode: + created_at: str + data: str + phase: str + data_type: str = "text" + + +BASELINE_EPISODES = [ + SourceEpisode( + "2026-01-05T09:00:00Z", + "澜舟科技（企业稳定标识 LZ-TECH）是一家风电智能运维公司。" + "公司总部位于星港市，周岚自 2024 年起担任首席执行官（CEO）。", + "baseline", + ), + SourceEpisode( + "2026-01-12T10:00:00Z", + "澜舟科技研发了产品智巡平台（产品稳定标识 ZHIXUN-01）。" + "智巡平台当前处于试点阶段，用于识别风机叶片异常。", + "baseline", + ), + SourceEpisode( + "2026-01-20T14:00:00Z", + "澜舟科技与海岳能源（企业稳定标识 HY-ENERGY）签署 HY-2026-01 合作协议。" + "双方将在东湾风场联合部署智巡平台，协议有效期至 2026 年 12 月 31 日。", + "baseline", + ), + SourceEpisode( + "2026-02-02T09:30:00Z", + "陈屿担任澜舟科技智巡平台的项目负责人，负责东湾风场部署。" + "公司经营仍由首席执行官周岚负责。", + "baseline", + ), + SourceEpisode( + "2026-02-20T18:00:00Z", + "东湾风场的智巡平台试点发现了 12 处叶片异常，" + "使海岳能源的非计划停机时间降低了 18%。", + "baseline", + ), + SourceEpisode( + "2026-03-05T11:00:00Z", + "海岳能源确认澜舟科技是智巡平台的开发方，陈屿是实施项目负责人。" + "海岳能源计划在试点验收后成为首批商业客户。", + "baseline", + ), + SourceEpisode( + "2026-03-15T16:00:00Z", + "澜舟科技董事会批准智巡平台在 2026 年 4 月 1 日从试点阶段转为商业发布。" + "周岚以首席执行官身份签署了发布决议。", + "baseline", + ), + SourceEpisode( + "2026-04-01T08:00:00Z", + "智巡平台今天正式商业发布，不再处于试点阶段。" + "海岳能源成为智巡平台首个商业客户，陈屿继续负责交付。", + "baseline", + ), + SourceEpisode( + "2026-04-18T13:00:00Z", + "澜舟科技在星港市总部公布运营数据：智巡平台已覆盖 60 台风机，" + "海岳能源仍是 HY-2026-01 协议下的合作伙伴和客户。", + "baseline", + ), + SourceEpisode( + "2026-04-30T17:00:00Z", + "截至 2026 年 4 月 30 日，周岚仍担任澜舟科技首席执行官，" + "澜舟科技总部仍在星港市；陈屿担任智巡平台项目负责人。", + "baseline", + ), +] + + +TEMPORAL_UPDATES = [ + SourceEpisode( + "2026-05-10T09:00:00Z", + "澜舟科技董事会宣布，自 2026 年 5 月 10 日起，周岚不再担任首席执行官。" + "陈屿正式接任澜舟科技首席执行官，周岚转任首席战略顾问。", + "leadership_change", + ), + SourceEpisode( + "2026-06-01T09:00:00Z", + "自 2026 年 6 月 1 日起，澜舟科技总部已从星港市迁至海城市。" + "星港市原总部不再是公司总部，现改为澜舟科技研发中心。", + "headquarters_change", + ), + SourceEpisode( + "2026-06-15T12:00:00Z", + "澜舟科技与海岳能源已于 2026 年 6 月 15 日提前终止 HY-2026-01 联合部署协议。" + "海岳能源不再是澜舟科技的联合部署合作伙伴，但仍是智巡平台客户。", + "partnership_change", + ), + SourceEpisode( + "2026-06-20T18:00:00Z", + json.dumps( + { + "event": "product_metrics_update", + "company_id": "LZ-TECH", + "product_id": "ZHIXUN-01", + "product_name": "智巡平台", + "commercial_status": "commercial", + "covered_turbines": 120, + "active_customer": "海岳能源", + "as_of": "2026-06-20", + }, + ensure_ascii=False, + ), + "json_update", + "json", + ), +] + + +ONTOLOGY = { + "entity_types": [ + { + "name": "Person", + "description": "A named person involved in company governance or delivery.", + "attributes": [{"name": "current_role", "description": "The person's current role."}], + }, + { + "name": "Company", + "description": "A company or commercial organization.", + "attributes": [{"name": "stable_id", "description": "A stable company identifier."}], + }, + { + "name": "Product", + "description": "A named software or industrial product.", + "attributes": [{"name": "lifecycle_stage", "description": "The product lifecycle stage."}], + }, + { + "name": "Place", + "description": "A city, office, wind farm, or other named location.", + "attributes": [{"name": "place_kind", "description": "The kind of place."}], + }, + { + "name": "Agreement", + "description": "A named commercial agreement or contract.", + "attributes": [{"name": "agreement_status", "description": "The agreement status."}], + }, + ], + "edge_types": [ + { + "name": "HOLDS_ROLE_AT", + "description": "A person holds a named role at a company.", + "attributes": [{"name": "title", "description": "The person's title."}], + "source_targets": [{"source": "Person", "target": "Company"}], + }, + { + "name": "DEVELOPS", + "description": "A company develops a product.", + "attributes": [{"name": "product_status", "description": "The product status."}], + "source_targets": [{"source": "Company", "target": "Product"}], + }, + { + "name": "HEADQUARTERED_IN", + "description": "A company has its current headquarters in a place.", + "attributes": [{"name": "site_status", "description": "The site's headquarters status."}], + "source_targets": [{"source": "Company", "target": "Place"}], + }, + { + "name": "PARTNERS_WITH", + "description": "A company has a commercial partnership with another company.", + "attributes": [{"name": "agreement_id", "description": "The governing agreement identifier."}], + "source_targets": [{"source": "Company", "target": "Company"}], + }, + { + "name": "CUSTOMER_OF", + "description": "A company is a customer of a product's developer.", + "attributes": [{"name": "customer_status", "description": "The customer status."}], + "source_targets": [{"source": "Company", "target": "Company"}], + }, + ], +} + + +def _uuid(value: Any) -> str: + return str(getattr(value, "uuid_", None) or getattr(value, "uuid", "")) + + +def _status_code(error: Exception) -> int | None: + direct = getattr(error, "status_code", None) + response = getattr(error, "response", None) + return direct or getattr(response, "status_code", None) + + +def _require_process_api_key() -> str: + if not _PROCESS_ZEP_API_KEY: + raise RuntimeError( + "ZEP_API_KEY must be supplied through the process environment" + ) + return _PROCESS_ZEP_API_KEY + + +def _drain_updater_after_failure( + updater: Any, + *, + started: bool, + stop_attempted: bool, +) -> tuple[bool, Exception | None]: + """Make one safe drain attempt when the main flow did not call stop().""" + + if not started: + return True, None + if stop_attempted: + return False, None + try: + updater.stop() + except Exception as error: + return False, error + return True, None + + +def _cleanup_graph( + client: Any, + graph_id: str, + *, + created: bool, + keep_graph: bool, + updater_started: bool, + updater_drained: bool, +) -> dict[str, Any]: + """Delete only when no updater can still write to or ingest into the graph.""" + + if not created: + return { + "graph_deleted": False, + "graph_retained": False, + "reason": "graph_not_created", + } + if keep_graph: + return { + "graph_deleted": False, + "graph_retained": True, + "reason": "user_requested", + } + if updater_started and not updater_drained: + return { + "graph_deleted": False, + "graph_retained": True, + "reason": "updater_not_confirmed_drained", + } + + client.graph.delete(graph_id) + return { + "graph_deleted": True, + "graph_retained": False, + "reason": "validation_cleanup", + } + + +def _wait_for_episode(client: Any, episode_uuid: str, timeout: int) -> Any: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + episode = client.graph.episode.get(uuid_=episode_uuid) + if getattr(episode, "processed", False): + return episode + time.sleep(3) + raise TimeoutError(f"episode {episode_uuid} did not finish within {timeout}s") + + +def _list_batch_items(client: Any, batch_id: str, page_size: int = 3) -> tuple[list[Any], int]: + items: list[Any] = [] + cursor: int | None = None + pages = 0 + while True: + response = client.batch.list_items(batch_id=batch_id, limit=page_size, cursor=cursor) + pages += 1 + items.extend(response.items or []) + next_cursor = response.next_cursor + if next_cursor is None: + return items, pages + if next_cursor == cursor: + raise RuntimeError("batch item cursor did not advance") + cursor = next_cursor + + +def _raw_pages(api_call: Any, graph_id: str, page_size: int = 2) -> tuple[list[Any], int]: + items: list[Any] = [] + cursor: str | None = None + pages = 0 + seen: set[str] = set() + while True: + kwargs: dict[str, Any] = {"limit": page_size} + if cursor: + kwargs["cursor"] = cursor + response = api_call(graph_id, **kwargs) + pages += 1 + items.extend(list(response.data or [])) + next_cursor = response.headers.get("zep-next-cursor") + if not next_cursor: + return items, pages + if next_cursor == cursor or next_cursor in seen: + raise RuntimeError("artifact cursor did not advance") + seen.add(next_cursor) + cursor = next_cursor + + +def _edge_view(edge: Any, node_names: dict[str, str]) -> dict[str, Any]: + return { + "uuid": _uuid(edge), + "name": edge.name, + "fact": edge.fact, + "source": node_names.get(edge.source_node_uuid, edge.source_node_uuid), + "target": node_names.get(edge.target_node_uuid, edge.target_node_uuid), + "created_at": edge.created_at, + "valid_at": edge.valid_at, + "invalid_at": edge.invalid_at, + "expired_at": edge.expired_at, + "attributes": edge.attributes or {}, + } + + +def _node_view(node: Any) -> dict[str, Any]: + return { + "uuid": _uuid(node), + "name": node.name, + "labels": node.labels or [], + "summary": node.summary, + "attributes": node.attributes or {}, + } + + +def _search_view(results: Any, node_names: dict[str, str]) -> dict[str, Any]: + return { + "context": (results.context or "")[:3000], + "edges": [_edge_view(edge, node_names) for edge in (results.edges or [])], + "nodes": [_node_view(node) for node in (results.nodes or [])], + "episode_count": len(results.episodes or []), + "observation_count": len(results.observations or []), + "thread_summary_count": len(results.thread_summaries or []), + } + + +def _episode_to_batch_item(graph_id: str, item: SourceEpisode, index: int) -> BatchAddItem: + return BatchAddItem( + type="graph_episode", + graph_id=graph_id, + data=item.data, + data_type=item.data_type, + created_at=item.created_at, + source_description="MiroFish deep Zep Cloud validation corpus", + metadata={ + "source": "mirofish_zep_deep_validation", + "phase": item.phase, + "sequence": index, + }, + ) + + +def _add_and_wait(client: Any, graph_id: str, item: SourceEpisode, timeout: int) -> str: + episode = client.graph.add( + graph_id=graph_id, + type=item.data_type, + data=item.data, + created_at=item.created_at, + source_description="MiroFish temporal Zep Cloud validation update", + metadata={"source": "mirofish_zep_deep_validation", "phase": item.phase}, + ) + episode_uuid = _uuid(episode) + if not episode_uuid: + raise RuntimeError("graph.add returned no episode UUID") + _wait_for_episode(client, episode_uuid, timeout) + return episode_uuid + + +def _activities() -> Iterable[AgentActivity]: + return [ + AgentActivity( + platform="twitter", + agent_id=101, + agent_name="陈屿", + action_type="CREATE_POST", + action_args={"content": "海城市新总部今天启用，智巡平台商业服务正常运行。"}, + round_num=1, + timestamp="2026-07-01T09:00:00Z", + ), + AgentActivity( + platform="twitter", + agent_id=102, + agent_name="周岚", + action_type="QUOTE_POST", + action_args={ + "original_author_name": "陈屿", + "original_content": "海城市新总部今天启用，智巡平台商业服务正常运行。", + "quote_content": "作为首席战略顾问，我支持陈屿和新的管理团队。", + }, + round_num=1, + timestamp="2026-07-01T09:05:00Z", + ), + AgentActivity( + platform="twitter", + agent_id=201, + agent_name="海岳能源", + action_type="LIKE_POST", + action_args={ + "post_author_name": "陈屿", + "post_content": "海城市新总部今天启用，智巡平台商业服务正常运行。", + }, + round_num=1, + timestamp="2026-07-01T09:06:00Z", + ), + AgentActivity( + platform="twitter", + agent_id=201, + agent_name="海岳能源", + action_type="CREATE_COMMENT", + action_args={ + "post_author_name": "陈屿", + "post_content": "海城市新总部今天启用，智巡平台商业服务正常运行。", + "content": "联合部署协议虽已终止，但我们仍是智巡平台客户。", + }, + round_num=2, + timestamp="2026-07-01T09:10:00Z", + ), + AgentActivity( + platform="twitter", + agent_id=101, + agent_name="陈屿", + action_type="FOLLOW", + action_args={"target_user_name": "海岳能源"}, + round_num=2, + timestamp="2026-07-01T09:12:00Z", + ), + ] + + +def run(args: argparse.Namespace) -> dict[str, Any]: + api_key = _require_process_api_key() + + stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + graph_id = args.graph_id or f"mirofish_zep_deep_{stamp}" + builder = GraphBuilderService(api_key=api_key) + client = builder.client + created = False + updater: ZepGraphMemoryUpdater | None = None + updater_started = False + updater_stop_attempted = False + updater_drained = False + result: dict[str, Any] = { + "graph_id": graph_id, + "graph_retained": args.keep_graph, + "sdk_flow": "zep-cloud 3.25 standalone graph + current Batch API", + } + + print(f"[zep-deep] graph_id={graph_id}", flush=True) + try: + builder.create_graph("MiroFish Zep Cloud Deep Validation", graph_id=graph_id) + created = True + print("[zep-deep] graph created", flush=True) + + # Production safety net: empty LLM attributes must become a valid + # fallback property before reaching Zep Cloud. + ontology_probe = {"entity_types": [{"name": "ProbeEntity", "attributes": []}], "edge_types": []} + try: + builder.set_ontology(graph_id, ontology_probe) + result["empty_attribute_ontology_probe"] = "accepted_after_mirofish_normalization" + except Exception as error: + result["empty_attribute_ontology_probe"] = { + "error_type": type(error).__name__, + "status_code": _status_code(error), + } + raise + + builder.set_ontology(graph_id, ONTOLOGY) + print("[zep-deep] ontology set", flush=True) + + operation_id = f"deep-validation-{stamp}" + batch = client.batch.create( + metadata={ + "mirofish_operation_id": operation_id, + "graph_id": graph_id, + "suite": "zep_deep_validation", + } + ) + batch_id = batch.batch_id + if not batch_id: + raise RuntimeError("batch.create returned no batch_id") + result["batch_id"] = batch_id + + added_details: list[Any] = [] + batch_items = [ + _episode_to_batch_item(graph_id, item, index) + for index, item in enumerate(BASELINE_EPISODES) + ] + for start in range(0, len(batch_items), 4): + added_details.extend( + client.batch.add(batch_id=batch_id, items=batch_items[start : start + 4]) + ) + client.batch.process(batch_id=batch_id) + print(f"[zep-deep] batch submitted items={len(batch_items)}", flush=True) + + submission = BatchSubmission( + batch_id=batch_id, + operation_id=operation_id, + episode_uuids=[_uuid(item) for item in added_details if _uuid(item)], + item_count=len(batch_items), + ) + baseline_episode_uuids = builder._wait_for_batch(submission, timeout=args.timeout) + listed_items, batch_pages = _list_batch_items(client, batch_id, page_size=3) + result["batch"] = { + "status": client.batch.get(batch_id=batch_id).status, + "item_count": len(listed_items), + "item_pages_at_size_3": batch_pages, + "episode_uuids": baseline_episode_uuids, + } + print(f"[zep-deep] batch completed pages={batch_pages}", flush=True) + + baseline_nodes = fetch_all_nodes(client, graph_id, page_size=2) + baseline_edges = fetch_all_edges(client, graph_id, page_size=2) + raw_nodes, node_pages = _raw_pages( + client.graph.node.with_raw_response.get_by_graph_id, graph_id, page_size=2 + ) + raw_edges, edge_pages = _raw_pages( + client.graph.edge.with_raw_response.get_by_graph_id, graph_id, page_size=2 + ) + if {_uuid(item) for item in baseline_nodes} != {_uuid(item) for item in raw_nodes}: + raise AssertionError("production node pagination did not match raw cursor traversal") + if {_uuid(item) for item in baseline_edges} != {_uuid(item) for item in raw_edges}: + raise AssertionError("production edge pagination did not match raw cursor traversal") + if len(baseline_nodes) <= 2 or len(baseline_edges) <= 2: + raise AssertionError("the corpus did not produce enough artifacts to exercise pagination") + + baseline_names = {_uuid(node): node.name for node in baseline_nodes} + baseline_ceo = client.graph.search( + graph_id=graph_id, + query="截至2026年4月底，谁担任澜舟科技首席执行官？", + scope="edges", + reranker="cross_encoder", + limit=10, + ) + result["baseline"] = { + "node_count": len(baseline_nodes), + "edge_count": len(baseline_edges), + "node_pages_at_size_2": node_pages, + "edge_pages_at_size_2": edge_pages, + "invalidated_edge_count": sum(bool(edge.invalid_at) for edge in baseline_edges), + "ceo_search": _search_view(baseline_ceo, baseline_names), + } + print( + f"[zep-deep] baseline nodes={len(baseline_nodes)} edges={len(baseline_edges)}", + flush=True, + ) + + update_episode_uuids = [ + _add_and_wait(client, graph_id, item, args.timeout) + for item in TEMPORAL_UPDATES + ] + result["temporal_update_episode_uuids"] = update_episode_uuids + print(f"[zep-deep] temporal updates processed={len(update_episode_uuids)}", flush=True) + + updater = ZepGraphMemoryUpdater( + graph_id=graph_id, + api_key=api_key, + simulation_id=f"zep-deep-{stamp}", + ) + updater_started = True + updater.start() + for activity in _activities(): + updater.add_activity(activity) + updater_stop_attempted = True + updater.stop() + updater_drained = True + updater_stats = updater.get_stats() + if updater_stats["items_sent"] != 5 or updater_stats["pending_episode_count"] != 0: + raise AssertionError(f"unexpected MiroFish updater stats: {updater_stats}") + result["mirofish_updater"] = updater_stats + print("[zep-deep] MiroFish updater processed 5 mock activities", flush=True) + + final_nodes = fetch_all_nodes(client, graph_id, page_size=2) + final_edges = fetch_all_edges(client, graph_id, page_size=2) + final_names = {_uuid(node): node.name for node in final_nodes} + invalidated = [edge for edge in final_edges if edge.invalid_at] + expired = [edge for edge in final_edges if edge.expired_at] + + edge_search = client.graph.search( + graph_id=graph_id, + query="澜舟科技当前首席执行官、当前总部以及与海岳能源的当前合作关系是什么？", + scope="edges", + reranker="cross_encoder", + limit=20, + ) + node_search = client.graph.search( + graph_id=graph_id, + query="澜舟科技管理层人物", + scope="nodes", + reranker="rrf", + limit=10, + search_filters=SearchFilters(node_labels=["Person"]), + ) + typed_edge_search = client.graph.search( + graph_id=graph_id, + query="公司管理职位发生了什么变化？", + scope="edges", + reranker="rrf", + limit=10, + search_filters=SearchFilters(edge_types=["HOLDS_ROLE_AT"]), + ) + auto_search = client.graph.search( + graph_id=graph_id, + query="总结澜舟科技最新的管理层、总部、产品与海岳能源关系。", + scope="auto", + max_characters=3000, + return_raw_results=True, + ) + episode_search = client.graph.search( + graph_id=graph_id, + query="海岳能源为什么仍是客户但不再是联合部署伙伴？", + scope="episodes", + reranker="rrf", + limit=10, + ) + + selected_node = next( + ( + node + for node in final_nodes + if node.name == "澜舟科技" and "Company" in (node.labels or []) + ), + max( + final_nodes, + key=lambda node: sum( + edge.source_node_uuid == _uuid(node) + or edge.target_node_uuid == _uuid(node) + for edge in final_edges + ), + ), + ) + sdk_node_edges = client.graph.node.get_edges(node_uuid=_uuid(selected_node)) + complete_node_edges = [ + edge + for edge in final_edges + if edge.source_node_uuid == _uuid(selected_node) + or edge.target_node_uuid == _uuid(selected_node) + ] + entity_context = ZepEntityReader(api_key=api_key).get_entity_with_context( + graph_id, + _uuid(selected_node), + ) + recent_episodes = client.graph.episode.get_by_graph_id(graph_id=graph_id, lastn=50) + episode_list = getattr(recent_episodes, "episodes", None) or [] + + custom_labels = sorted( + {label for node in final_nodes for label in (node.labels or []) if label != "Entity"} + ) + custom_edge_names = sorted( + {edge.name for edge in final_edges if edge.name in {item["name"] for item in ONTOLOGY["edge_types"]}} + ) + result["final"] = { + "node_count": len(final_nodes), + "edge_count": len(final_edges), + "invalidated_edge_count": len(invalidated), + "expired_edge_count": len(expired), + "custom_labels": custom_labels, + "custom_edge_names": custom_edge_names, + "nodes": [_node_view(node) for node in final_nodes], + "invalidated_facts": [_edge_view(edge, final_names) for edge in invalidated], + "active_facts": [ + _edge_view(edge, final_names) + for edge in final_edges + if not edge.invalid_at and not edge.expired_at + ], + "selected_node": _node_view(selected_node), + "sdk_node_edge_count": len(sdk_node_edges), + "complete_node_edge_count": len(complete_node_edges), + "entity_reader_context_edge_count": len(entity_context.related_edges), + "recent_episode_count": len(episode_list), + } + result["searches"] = { + "current_state_edges": _search_view(edge_search, final_names), + "person_nodes": _search_view(node_search, final_names), + "typed_role_edges": _search_view(typed_edge_search, final_names), + "auto_context": _search_view(auto_search, final_names), + "partnership_episodes": _search_view(episode_search, final_names), + } + + if len(entity_context.related_edges) != len(complete_node_edges): + raise AssertionError( + "MiroFish entity context omitted incoming or outgoing node edges" + ) + + result["runtime_assertions"] = { + "edge_search_call_completed": edge_search is not None, + "node_search_call_completed": node_search is not None, + "typed_edge_search_call_completed": typed_edge_search is not None, + "auto_search_call_completed": auto_search is not None, + "episode_search_call_completed": episode_search is not None, + "node_detail_has_all_incoming_and_outgoing_edges": True, + "sdk_node_endpoint_omits_incoming_edges": ( + len(sdk_node_edges) < len(complete_node_edges) + ), + # The following values are observations only. Zep's extraction and + # retrieval quality are not runtime acceptance criteria. + "search_result_counts": { + "edges": len(edge_search.edges or []), + "nodes": len(node_search.nodes or []), + "typed_edges": len(typed_edge_search.edges or []), + "episodes": len(episode_search.episodes or []), + }, + "custom_entity_labels_observed": bool(custom_labels), + "custom_edge_names_observed": bool(custom_edge_names), + "temporal_invalidation_observed": bool(invalidated), + } + return result + except Exception as error: + result["failure"] = { + "type": type(error).__name__, + "status_code": _status_code(error), + "message": str(error)[:500], + } + raise + finally: + if updater is not None and updater_started and not updater_stop_attempted: + updater_stop_attempted = True + updater_drained, updater_cleanup_error = _drain_updater_after_failure( + updater, + started=updater_started, + stop_attempted=False, + ) + if updater_cleanup_error is not None: + result["updater_cleanup_error"] = { + "type": type(updater_cleanup_error).__name__, + "message": str(updater_cleanup_error)[:500], + } + print( + "[zep-deep] updater drain failed; graph will be retained " + f"type={type(updater_cleanup_error).__name__}", + flush=True, + ) + + had_primary_failure = "failure" in result + try: + cleanup = _cleanup_graph( + client, + graph_id, + created=created, + keep_graph=args.keep_graph, + updater_started=updater_started, + updater_drained=updater_drained, + ) + except Exception as cleanup_error: + result["cleanup"] = { + "graph_deleted": False, + "graph_retained": True, + "reason": "graph_delete_failed", + "error_type": type(cleanup_error).__name__, + "status_code": _status_code(cleanup_error), + } + result["graph_retained"] = True + print( + "[zep-deep] graph cleanup failed; graph retained " + f"graph_id={graph_id} type={type(cleanup_error).__name__}", + flush=True, + ) + if not had_primary_failure: + result["failure"] = { + "type": type(cleanup_error).__name__, + "status_code": _status_code(cleanup_error), + "message": str(cleanup_error)[:500], + } + print(json.dumps(result, ensure_ascii=False, indent=2), flush=True) + raise + else: + result["cleanup"] = cleanup + result["graph_retained"] = cleanup["graph_retained"] + if cleanup["graph_deleted"]: + action = "deleted" + elif cleanup["graph_retained"]: + action = "retained" + else: + action = "not-created" + print( + f"[zep-deep] graph {action} graph_id={graph_id} " + f"reason={cleanup['reason']}", + flush=True, + ) + + if had_primary_failure: + print(json.dumps(result, ensure_ascii=False, indent=2), flush=True) + + +def _compact_result(result: dict[str, Any]) -> dict[str, Any]: + """Keep terminal output focused on runtime/API-contract evidence.""" + + baseline = result.get("baseline", {}) + final = result.get("final", {}) + return { + "graph_id": result.get("graph_id"), + "graph_retained": result.get("graph_retained"), + "cleanup": result.get("cleanup"), + "sdk_flow": result.get("sdk_flow"), + "empty_attribute_ontology_probe": result.get( + "empty_attribute_ontology_probe" + ), + "batch_id": result.get("batch_id"), + "batch": result.get("batch"), + "baseline": { + key: baseline.get(key) + for key in ( + "node_count", + "edge_count", + "node_pages_at_size_2", + "edge_pages_at_size_2", + "invalidated_edge_count", + ) + }, + "temporal_update_episode_count": len( + result.get("temporal_update_episode_uuids", []) + ), + "mirofish_updater": result.get("mirofish_updater"), + "final": { + key: final.get(key) + for key in ( + "node_count", + "edge_count", + "invalidated_edge_count", + "expired_edge_count", + "custom_labels", + "custom_edge_names", + "sdk_node_edge_count", + "complete_node_edge_count", + "entity_reader_context_edge_count", + "recent_episode_count", + ) + }, + "runtime_assertions": result.get("runtime_assertions"), + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--graph-id", help="Use an explicit standalone graph ID") + parser.add_argument( + "--keep-graph", + action="store_true", + help="Retain the graph for manual inspection after validation", + ) + parser.add_argument( + "--timeout", + type=int, + default=900, + help="Maximum seconds to wait for each ingestion stage", + ) + parser.add_argument( + "--full-output", + action="store_true", + help="Print every node, edge, and search result instead of a runtime summary", + ) + args = parser.parse_args() + + try: + result = run(args) + except Exception as error: + print( + f"[zep-deep] FAILED type={type(error).__name__} status={_status_code(error)}", + file=sys.stderr, + ) + return 1 + output = result if args.full_output else _compact_result(result) + print(json.dumps(output, ensure_ascii=False, indent=2), flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/test_zep_cloud_contracts.py b/backend/tests/test_zep_cloud_contracts.py index 0d0bad40..c85b64df 100644 --- a/backend/tests/test_zep_cloud_contracts.py +++ b/backend/tests/test_zep_cloud_contracts.py @@ -56,6 +56,70 @@ def test_profile_context_search_caps_both_queries_sent_to_zep(): assert all(0 < len(call["query"]) <= 400 for call in calls) +def test_entity_context_includes_incoming_edges_from_the_full_graph(): + incoming = { + "uuid": "edge-in", + "name": "WORKS_AT", + "fact": "Alice works at Acme", + "source_node_uuid": "alice", + "target_node_uuid": "acme", + "attributes": {}, + } + outgoing = { + "uuid": "edge-out", + "name": "BUILDS", + "fact": "Acme builds Product", + "source_node_uuid": "acme", + "target_node_uuid": "product", + "attributes": {}, + } + unrelated = { + "uuid": "edge-unrelated", + "name": "LOCATED_IN", + "fact": "OtherCo is located in Paris", + "source_node_uuid": "other-company", + "target_node_uuid": "paris", + "attributes": {}, + } + + reader = object.__new__(ZepEntityReader) + reader.client = SimpleNamespace( + graph=SimpleNamespace( + node=SimpleNamespace( + get=lambda **_kwargs: SimpleNamespace( + uuid_="acme", + name="Acme", + labels=["Company"], + summary="", + attributes={}, + ), + # Real Cloud 3.25 omits incoming edges here. + get_edges=lambda **_kwargs: [SimpleNamespace(**outgoing)], + ) + ) + ) + reader.get_all_edges = lambda _graph_id: [incoming, outgoing, unrelated] + reader.get_all_nodes = lambda _graph_id: [ + {"uuid": "alice", "name": "Alice", "labels": ["Person"], "summary": ""}, + {"uuid": "acme", "name": "Acme", "labels": ["Company"], "summary": ""}, + {"uuid": "product", "name": "Product", "labels": ["Product"], "summary": ""}, + ] + + entity = reader.get_entity_with_context("graph-id", "acme") + + assert entity is not None + assert len(entity.related_edges) == 2 + assert {edge["edge_name"] for edge in entity.related_edges} == { + "WORKS_AT", + "BUILDS", + } + assert {edge["direction"] for edge in entity.related_edges} == { + "incoming", + "outgoing", + } + assert {node["name"] for node in entity.related_nodes} == {"Alice", "Product"} + + def test_entity_reader_does_not_turn_auth_failure_into_missing_entity(): def unauthorized(**_kwargs): raise ZepApiError(status_code=401, body={"message": "unauthorized"}) diff --git a/backend/tests/test_zep_cloud_validation_script.py b/backend/tests/test_zep_cloud_validation_script.py new file mode 100644 index 00000000..7c6e5272 --- /dev/null +++ b/backend/tests/test_zep_cloud_validation_script.py @@ -0,0 +1,105 @@ +import os +import subprocess +import sys +import textwrap +from pathlib import Path +from types import SimpleNamespace + + +def test_validation_script_preserves_the_process_key_across_app_config_import(): + backend_root = Path(__file__).resolve().parents[1] + script_path = backend_root / "scripts" / "validate_zep_cloud_integration.py" + probe = textwrap.dedent( + f""" + import os + import runpy + import dotenv + + def overwrite_with_dotenv_value(*_args, **_kwargs): + os.environ["ZEP_API_KEY"] = "dotenv-test-key" + return True + + dotenv.load_dotenv = overwrite_with_dotenv_value + namespace = runpy.run_path({str(script_path)!r}, run_name="zep_validation_probe") + assert namespace["_PROCESS_ZEP_API_KEY"] == "process-test-key" + assert namespace["_require_process_api_key"]() == "process-test-key" + assert os.environ["ZEP_API_KEY"] == "process-test-key" + """ + ) + environment = os.environ.copy() + environment["ZEP_API_KEY"] = "process-test-key" + environment["PYTHONPATH"] = os.pathsep.join( + filter(None, [str(backend_root), environment.get("PYTHONPATH", "")]) + ) + + completed = subprocess.run( + [sys.executable, "-c", probe], + cwd=backend_root, + env=environment, + capture_output=True, + text=True, + timeout=30, + check=False, + ) + + assert completed.returncode == 0, completed.stderr + + +def test_validation_cleanup_retains_graph_when_updater_cannot_drain(monkeypatch): + monkeypatch.setenv("ZEP_API_KEY", "validation-test-key") + from scripts import validate_zep_cloud_integration as validation + + class FailingUpdater: + def stop(self): + raise TimeoutError("worker still running") + + updater_drained, error = validation._drain_updater_after_failure( + FailingUpdater(), + started=True, + stop_attempted=False, + ) + deleted_graphs = [] + client = SimpleNamespace( + graph=SimpleNamespace(delete=lambda graph_id: deleted_graphs.append(graph_id)) + ) + cleanup = validation._cleanup_graph( + client, + "graph-id", + created=True, + keep_graph=False, + updater_started=True, + updater_drained=updater_drained, + ) + + assert isinstance(error, TimeoutError) + assert cleanup == { + "graph_deleted": False, + "graph_retained": True, + "reason": "updater_not_confirmed_drained", + } + assert deleted_graphs == [] + + +def test_validation_cleanup_deletes_graph_after_confirmed_drain(monkeypatch): + monkeypatch.setenv("ZEP_API_KEY", "validation-test-key") + from scripts import validate_zep_cloud_integration as validation + + deleted_graphs = [] + client = SimpleNamespace( + graph=SimpleNamespace(delete=lambda graph_id: deleted_graphs.append(graph_id)) + ) + cleanup = validation._cleanup_graph( + client, + "graph-id", + created=True, + keep_graph=False, + updater_started=True, + updater_drained=True, + ) + + assert cleanup == { + "graph_deleted": True, + "graph_retained": False, + "reason": "validation_cleanup", + } + assert deleted_graphs == ["graph-id"]