{"record":{"id":"daaae8c9ce170761","repo":"666ghj/MiroFish","slug":"len-self-failed-batches-zep-activity-batch-es","errorCode":null,"errorMessage":"{len(self._failed_batches)} Zep activity batch(es) failed; simulation graph ingestion is incomplete","messagePattern":"(.+?) Zep activity batch\\(es\\) failed; simulation graph ingestion is incomplete","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"backend/app/services/zep_graph_memory_updater.py","lineNumber":336,"sourceCode":"        # could enqueue after both the worker and final flush had exited.\n        with self._acceptance_lock:\n            self._running = False\n\n        if self._worker_thread and self._worker_thread.is_alive():\n            join_timeout = max(0.0, deadline - time.time())\n            self._worker_thread.join(timeout=join_timeout)\n            if self._worker_thread.is_alive():\n                raise TimeoutError(\n                    f\"Zep updater worker did not stop within {join_timeout:.0f}s\"\n                )\n\n        # The worker has drained the queue. Only now is it safe to flush\n        # buffers; doing this before join loses an item already dequeued by the\n        # worker but not yet buffered.\n        self._flush_remaining(deadline=deadline)\n\n        if self._failed_batches:\n            raise RuntimeError(\n                f\"{len(self._failed_batches)} Zep activity batch(es) failed; \"\n                \"simulation graph ingestion is incomplete\"\n            )\n\n        self._wait_for_pending_episodes(deadline=deadline)\n        \n        logger.info(f\"ZepGraphMemoryUpdater 已停止: graph_id={self.graph_id}, \"\n                   f\"total_activities={self._total_activities}, \"\n                   f\"batches_sent={self._total_sent}, \"\n                   f\"items_sent={self._total_items_sent}, \"\n                   f\"failed={self._failed_count}, \"\n                   f\"skipped={self._skipped_count}\")\n    \n    def add_activity(self, activity: AgentActivity):\n        \"\"\"\n        添加一个agent活动到队列\n        \n        所有有意义的行为都会被添加到队列，包括：","sourceCodeStart":318,"sourceCodeEnd":354,"githubUrl":"https://github.com/666ghj/MiroFish/blob/b5b53acc57189a4a42e44a23e149dc655c98fe82/backend/app/services/zep_graph_memory_updater.py#L318-L354","documentation":"After the worker drains and buffers flush, stop() checks self._failed_batches; if any activity batch could not be ingested into the Zep graph it raises RuntimeError('N Zep activity batch(es) failed; simulation graph ingestion is incomplete'). This is a deliberate fail-loud: the simulation's graph in Zep is missing data, so downstream graph analysis would be silently wrong.","triggerScenarios":"Any batch send to the Zep API failing past its retries during the run or the final flush (HTTP errors, auth expiry, rate limiting, payload size); then stop() (or the stop_simulation path that calls it) raises this instead of returning cleanly.","commonSituations":"Zep API key revoked mid-run; 429 rate limiting on large simulations; transient network drops whose retries also failed; oversized batches exceeding API limits.","solutions":["Check the logs for the per-batch failure reasons logged when each batch was first recorded as failed — that names the root cause.","Re-ingest the failed activities: replay actions.jsonl (or the recorded batch payloads) into the graph once the API issue (key/rate limit/network) is fixed.","Reduce batch size / increase retry budget for long simulations to avoid exhausting retries during transient outages."],"exampleFix":"# before\nupdater.stop()  # RuntimeError: 3 Zep activity batch(es) failed ...\n\n# after\ntry:\n    updater.stop()\nexcept RuntimeError as e:\n    if \"batch(es) failed\" not in str(e):\n        raise\n    failed = updater.export_failed_batches()  # persist payloads\n    quarantine_for_reingest(sim_id, failed)","handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":"try:\n    updater.stop()\nexcept RuntimeError as e:\n    if \"batch(es) failed\" not in str(e):\n        raise\n    persist_failed_batches(sim_id)      # fallback: keep payloads\n    enqueue_reingestion(sim_id)          # replay into Zep later","preventionTips":["Track failed-batch count as a live metric during the run.","Retry failed batches during the run instead of only at stop.","Cap batch size to stay under Zep API limits and reduce blast radius per failure."],"tags":["zep","updater","partial-write","data-loss","api-failure"],"backgroundTag":null,"analyzedSha":"b5b53acc57189a4a42e44a23e149dc655c98fe82","analyzedAt":"2026-08-14T22:29:33.146Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}