fix(sdk): coding reconcile model failures fail the write, nothing half-applied

Experience reconcile no longer falls back to ADD-all, and fact reconcile no
longer inserts without superseding, when the model call fails, returns empty
content, or returns a reply that is not a JSON object with an ops list. The
write returns success=false, agent_error_code=RECONCILE_FAILED and the
cause in reconcile_error_code (LLM_UNAVAILABLE / EMPTY_RESPONSE /
JSON_PARSE_FAILED), the same as chat.

Each reconciler is split into plan() (retrieve + model decision, no store)
and apply() (store). CodingWriter plans both layers before applying either,
so a decision failure in either layer stores no experience or fact and
supersedes no older fact. reconcile() still runs plan + apply.

An experience reply of {"ops": []} is a valid "no changes" decision and
stores nothing. A non-empty ops list with no recognizable op is a parse
failure. A fact reply of {"ops": []} inserts without superseding.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -31,6 +31,53 @@ Hy Memory and MemHub are versioned and released independently.
     id need not equal the Hermes profile name (e.g. `--user-id wife-work
     --root ~/.hermes/profiles/wife:work`).
 
+- Hy Memory: a model failure during a write now reports failure and stores
+  nothing. **Behaviour change for direct SDK users:** writes that used to
+  return `success: true` in a degraded state now return `success: false`;
+  retry them with a new `request_id` once the model is back.
+  - Reconcile (pro / ultra `add` and `add_extracted`): when the reconcile model
+    call fails, returns HTTP 200 with empty content, or returns a reply that
+    cannot be parsed, or reconcile raises before anything is stored, the write
+    used to fall back to adding every extracted memory as-is (no dedup or
+    evolution) and report success. That fallback is removed, with its
+    `degraded` / `degraded_reason` result fields. The write now returns
+    `success: false`, `error_code: 502`,
+    `agent_error_code: "RECONCILE_FAILED"`, and the cause in the new
+    `reconcile_error_code` (`LLM_UNAVAILABLE`, `EMPTY_RESPONSE`,
+    `JSON_PARSE_FAILED`, `RECONCILE_EXCEPTION`). None of the new memories are
+    stored, the episode counter does not advance, and `add` rolls back its
+    raw turn the same way as an extraction failure
+    (`l1_rollback.retry_safe: true`). An empty 200 reply was previously
+    counted as "no changes" and reported success with 0 memories; only a
+    literal `[]` reply still means that. `RECONCILE_ENABLED=false` is a
+    configuration choice and still writes directly.
+  - Working / coding path: an extraction model failure used to return
+    success with 0 operations. It now returns `success: false` with
+    `agent_error_code` `LLM_ERROR`, `EMPTY_RESPONSE` or `JSON_PARSE_FAILED`.
+    A model that answers with nothing worth remembering is still a success.
+    An embedding failure used to store the experience or fact without a
+    vector, where search never finds it, and report success; it now returns
+    `EMBEDDING_FAILED` and the node is not stored. A fact embedding is checked
+    before any older fact is superseded. A failed write no longer appends its
+    turns to the recent trace and records no `chain_rounds`.
+    With `MEMORY_CODING_VERBATIM_FALLBACK_ENABLED` turned on, extraction
+    failures keep the verbatim fallback.
+  - Working / coding reconcile: when the experience reconcile or the fact
+    reconcile model call fails, returns empty content, or returns a reply that
+    is not a JSON object with an `ops` list, the write returns
+    `success: false`, `agent_error_code: "RECONCILE_FAILED"` and the cause in
+    `reconcile_error_code` (`LLM_UNAVAILABLE`, `EMPTY_RESPONSE`,
+    `JSON_PARSE_FAILED`), the same as chat. Before, experience reconcile fell
+    back to adding every draft, and fact reconcile inserted the new fact
+    without superseding older ones. Both model decisions now run before
+    anything is stored, so a failure in either leaves no experience or fact
+    from the write and supersedes no older fact. An experience reply of
+    `{"ops": []}` is a valid "no changes" decision: the write succeeds and no
+    experience is stored. A non-empty `ops` list with no recognizable op is a
+    parse failure. A fact reply of `{"ops": []}` inserts the new fact without
+    superseding anything. Fact reconcile still makes no model call when there
+    are no candidates.
+
 - MemHub's DeepSeek Harness app page can grant recall again (#230). The
   "一键插入应用" switch used to be greyed out with `unsupported_host`, and a
   profile installed from the CLI stayed `install-only` with no way to approve

docs/multi_speaker_design.md

@@ -107,5 +107,5 @@ L1_RAW 保存的内容使用上述渲染格式。历史回读时：
 ```bash
 python -m pytest -q \
   tests/test_multi_speaker_format.py \
-  tests/test_reconcile_fallback.py
+  tests/test_reconcile_failure.py
 ```

docs/scene_coding_usage.md

@@ -150,7 +150,23 @@ working/coding 写入返回双层结果：
 }
 ```
 
-`scene="coding"` 还会返回 `chain_rounds`。出于历史兼容，该字段当前统计本次识别并处理
+抽取 LLM 调用失败、返回空 content 或回复解析不出 JSON，以及经验 / 事实向量化失败时，
+写入返回 `success=False`、`error_code=502`，`agent_error_code` 依次为 `LLM_ERROR`、
+`EMPTY_RESPONSE`、`JSON_PARSE_FAILED`、`EMBEDDING_FAILED`，`error_message` 以
+`[CODE]` 开头；没有向量的经验 / 事实不会落库，失败的写入也不追加对话 trace，换一个
+`request_id` 重跑即可。模型正常作答但没有值得记的内容仍是 `success=True`、0 条。
+显式开启 `MEMORY_CODING_VERBATIM_FALLBACK_ENABLED` 时抽取失败沿用原文兜底。
+
+经验层 / 事实层 reconcile 的模型决策同样不再兜底：LLM 调用失败、返回空 content 或回复
+不是带 `ops` 列表的 JSON 对象时，写入返回 `success=False`、
+`agent_error_code="RECONCILE_FAILED"`，原因在 `reconcile_error_code`（`LLM_UNAVAILABLE` /
+`EMPTY_RESPONSE` / `JSON_PARSE_FAILED`，与 chat 链路一致）。两层的模型决策都在落库之前
+完成，任何一层失败时本次一条经验 / 事实都不存，旧事实也不会被覆盖。经验层回复
+`{"ops": []}` 表示「无需任何操作」，写入成功、经验层不存；`ops` 非空却没有一条可识别的
+op 视为解析失败。事实层回复 `{"ops": []}` 表示都不覆盖、照常插入新事实；没有候选时不
+调用 LLM。
+
+`scene="coding"` 还会返回 `chain_rounds`（写入失败时不记录演进链，也不返回该字段）。出于历史兼容，该字段当前统计本次识别并处理
 的 diff hunk 数，而不保证等于最终合并后的唯一演进 round 数。`0` 表示没有识别到可记录
 的代码修改，不代表双层 memory 写入失败。
 

docs/usage.md

@@ -128,6 +128,14 @@ result = client.add(
 双层 reconcile 操作、`experience_ids`、`fact_ids` 等字段，完整结构见
 [scene_coding_usage.md](scene_coding_usage.md#4-返回结构)。
 
+写入失败时 `success=False`，`error_code` 为 HTTP 风格的整数，`agent_error_code` 给出稳定
+失败码。pro / ultra 的 reconcile（新事实与已有记忆判重、合并）模型调用失败、返回空
+content 或回复解析不出时，`add` 与 `add_extracted` 都报 `agent_error_code="RECONCILE_FAILED"`，
+具体原因在 `reconcile_error_code`（`LLM_UNAVAILABLE` / `EMPTY_RESPONSE` /
+`JSON_PARSE_FAILED` / `RECONCILE_EXCEPTION`）。此时本次一条新记忆都不落库，`add` 已写入
+的原始对话按抽取失败同样回滚（`l1_rollback.retry_safe=True`）；模型恢复后换一个
+`request_id` 重跑即可。`RECONCILE_ENABLED=false` 时不调用 reconcile，直接写入。
+
 `mode="lite"` 不执行 LLM 抽取，写入内容直接以 `L1_RAW` 形式保存。lite 的 Search 是
 严格的 raw-only 向量检索：无论调用方传入什么 `layers`、`scene` 或 `search_mode`，都只
 召回 `L1_RAW`，不会进入 working/coding、Schema 或 Intention 链路，也不会调用场景判断

hy_memory/agent/reconciler.py

@@ -136,10 +136,10 @@ class ReconcileResult:
     ops: List[ReconcileOp] = field(default_factory=list)
     success: bool = True
     error: Optional[str] = None
-    # True = reconcile 失败后走了「全部 ADD」兜底（见 _fallback_add_ops）。
-    # 结果仍然可用（success=True），但没有做过去重/演化判定，可能引入重复。
-    degraded: bool = False
-    degraded_reason: Optional[str] = None
+    # success=False 时的稳定失败码：LLM_UNAVAILABLE / EMPTY_RESPONSE /
+    # JSON_PARSE_FAILED / RECONCILE_EXCEPTION。失败时 ops 恒为空：writer 什么都
+    # 不落库，整次写入报失败，调用方换个 request_id 重跑即可（reconcile 会判重）。
+    error_code: Optional[str] = None
     # reconcile 链路 LLM token 消耗（search-query + reconcile 两次 LLM 调用之和）
     prompt_tokens: int = 0
     completion_tokens: int = 0
@@ -151,7 +151,7 @@ class ReconcileResult:
     # 用 content 作 key 而不是 node_id：新记忆此刻还没入库、没有 id。
     candidate_scores: Dict[str, List[Tuple[str, float]]] = field(default_factory=dict)
     # reconcile 候选检索已经为每条新 memory 算过 embedding。把原始文本对应向量交回
-    # writer，ADD-all / degraded ADD 可直接落库，避免同一事实再次请求 embedding。
+    # writer，原文不变的 ADD 可直接落库，避免同一事实再次请求 embedding。
     # 仅按完全相同 content 复用；LLM 改写后的 UPDATE/SUPERSEDE 会重新计算，防止错配。
     new_memory_embeddings: Dict[str, List[float]] = field(default_factory=dict)
 
@@ -394,49 +394,6 @@ class MemoryReconciler:
 
         return [], None
 
-    @staticmethod
-    def _fallback_add_ops(
-        new_memories: List[str],
-        new_memories_with_meta: Optional[List[Dict]] = None,
-    ) -> List[ReconcileOp]:
-        """reconcile 彻底失败时的兜底：把抽取结果**原样全部 ADD**。
-
-        为什么必须有这个兜底：reconcile 只是决定「新事实和旧记忆怎么合」。它挂了（网关 500、
-        JSON 解析不出）不代表抽取结果没价值 —— 此前的行为是把一整批完好的事实**全部丢弃**，
-        线上实测因此丢掉过整块内容（含评测题的正确答案）。宁可留重复也不要丢内容：
-        重复还能被后续 dedup / reconcile 收拾，丢了就永远找不回来。
-        """
-        metas = new_memories_with_meta or []
-        by_content = {
-            (m.get("content") or ""): m for m in metas if isinstance(m, dict)
-        }
-        ops: List[ReconcileOp] = []
-        for text in new_memories:
-            if not text:
-                continue
-            meta = by_content.get(text) or {}
-            ops.append(ReconcileOp(
-                op="ADD",
-                content=text,
-                layer=meta.get("layer"),
-                tags=meta.get("tags") or [],
-                owner=meta.get("owner"),
-                topic_paths=meta.get("topic_paths") or [],
-                event_facet=meta.get("event_facet") or {},
-                topic_assignments=meta.get("topic_assignments") or [],
-                topic_paths_provided=bool(
-                    meta.get("topic_paths_provided", "topic_paths" in meta)
-                ),
-                event_facet_provided=bool(
-                    meta.get("event_facet_provided", "event_facet" in meta)
-                ),
-                topic_assignments_provided=bool(meta.get(
-                    "topic_assignments_provided", "topic_assignments" in meta,
-                )),
-                reason="reconcile unavailable → stored as-is (degraded)",
-            ))
-        return ops
-
     async def reconcile(
         self,
         new_memories: List[str],
@@ -1088,8 +1045,8 @@ class MemoryReconciler:
                     )
                 except Exception as e:
                     # LLM 彻底不可用（provider 已按 max_retries 重试过，网关 500 /
-                    # 连接错误等）。**不能让它冒泡** —— 冒泡会让整批抽取结果连带丢弃。
-                    # 交给下面的 ADD-all 兜底。
+                    # 连接错误等）。不冒泡：下面统一记 RECONCILE 失败日志，再返回
+                    # success=False —— writer 不落任何新记忆，整次写入报失败。
                     _llm_error = f"{type(e).__name__}: {e}"
                     logger.warning(f"[reconciler] RECONCILE LLM unavailable: {_llm_error}")
                     break
@@ -1097,7 +1054,7 @@ class MemoryReconciler:
                 _tok_pt += response.prompt_tokens
                 _tok_ct += response.completion_tokens
                 _tok_total += response.tokens_used
-                raw_response = response.content
+                raw_response = response.content or ""
                 ops = self._parse_ops(raw_response)
                 ops = enforce_time_order(ops, current_time, candidate_map)
                 ops = self._inherit_exact_input_structure(
@@ -1106,7 +1063,9 @@ class MemoryReconciler:
                 if _structure_metadata_on:
                     ops = self._resolve_structure_references(ops, candidate_map)
 
-                _is_empty_array = self._strip_code_fence(raw_response) in ("[]", "")
+                # 只有字面量 "[]" 才是「模型判定无需任何操作」。HTTP 200 但 content
+                # 为空不是决策，是失败：按 EMPTY_RESPONSE 处理，不能当成 0 条成功。
+                _is_empty_array = self._strip_code_fence(raw_response) == "[]"
                 if ops or _is_empty_array:
                     _parse_ok = True
                     break
@@ -1136,7 +1095,7 @@ class MemoryReconciler:
                     error_msg = (
                         f"RECONCILE LLM returned empty response after {_n_attempts} attempts"
                     )
-                    error_code = "EMPTY_RESPONSE" 
+                    error_code = "EMPTY_RESPONSE"
                 logger.warning(f"[reconciler] {error_msg}")
 
                 if cache and request_id:
@@ -1153,14 +1112,13 @@ class MemoryReconciler:
                                 {
                                     "error": error_code,
                                     "message": error_msg,
-                                    "fallback": "ADD_ALL",
-                                    "fallback_count": len(new_memories or []),
+                                    "stored": 0,
                                 },
                                 ensure_ascii=False,
                             ),
                             elapsed_ms=recon_elapsed,
-                            # LLM 彻底不可用时 response 是 None —— 这行本身会抛
-                            # AttributeError，而它正好在兜底路径上，会把兜底也一起废掉
+                            # LLM 彻底不可用时 response 是 None —— 直接取属性会抛
+                            # AttributeError，把失败日志本身也废掉
                             prompt_tokens=getattr(response, "prompt_tokens", 0) or 0,
                             completion_tokens=getattr(response, "completion_tokens", 0) or 0,
                             total_tokens=getattr(response, "tokens_used", 0) or 0,
@@ -1170,22 +1128,16 @@ class MemoryReconciler:
                             f"[reconciler] store failed RECONCILE log failed: {e}"
                         )
 
-                # 兜底：JSON 解析不出来不等于抽取结果没价值 —— 全部 ADD，别丢内容。
-                _fb = self._fallback_add_ops(new_memories, new_memories_with_meta)
-                logger.warning(
-                    f"[reconciler] falling back to ADD-all: {len(_fb)} memories "
-                    f"({error_code})"
-                )
+                # 不做 ADD-all 兜底：没经过判重 / 演化的整批 ADD 会在重跑时产生重复，
+                # 而且调用方看到 success 就不会重试。这里报失败、不落库，由调用方重跑。
                 return ReconcileResult(
-                    ops=_fb,
-                    success=bool(_fb),
-                    error=None if _fb else error_msg,
-                    degraded=bool(_fb),
-                    degraded_reason=error_code,
+                    ops=[],
+                    success=False,
+                    error=error_msg,
+                    error_code=error_code,
                     prompt_tokens=_tok_pt,
                     completion_tokens=_tok_ct,
                     total_tokens=_tok_total,
-                    new_memory_embeddings=new_memory_embeddings,
                 )
 
             # 写 RECONCILE pipeline log
@@ -1250,20 +1202,14 @@ class MemoryReconciler:
 
         except Exception as e:
             logger.error(f"[reconciler] failed: {e}", exc_info=True)
-            # 兜底同样适用于「reconcile 链路任意环节炸了」（候选检索、embed、
-            # prompt 组装…）。此前这里返回 ops=[] → writer 直接丢弃整批抽取结果。
-            _fb = self._fallback_add_ops(new_memories, new_memories_with_meta)
-            if _fb:
-                logger.warning(
-                    f"[reconciler] falling back to ADD-all after unexpected error: "
-                    f"{len(_fb)} memories"
-                )
-                return ReconcileResult(
-                    ops=_fb, success=True,
-                    degraded=True, degraded_reason=f"EXCEPTION:{type(e).__name__}",
-                    new_memory_embeddings=dict(new_memory_embeddings or {}),
-                )
-            return ReconcileResult(ops=[], success=False, error=str(e))
+            # reconcile 链路任意环节炸了（候选检索、embed、prompt 组装…）：此时还
+            # 没有任何新记忆落库，报失败让整次写入可以干净地重跑。
+            return ReconcileResult(
+                ops=[],
+                success=False,
+                error=f"{type(e).__name__}: {e}",
+                error_code="RECONCILE_EXCEPTION",
+            )
 
     @staticmethod
     def _inherit_exact_input_structure(

hy_memory/client.py

@@ -134,6 +134,7 @@ _WRITE_DIAGNOSTIC_KEYS = (
     "agent_status",
     "agent_error_code",
     "agent_error",
+    "reconcile_error_code",
     "l1_rollback",
     "l1_rollback_failed",
     "partial_agent_persist",
@@ -3032,7 +3033,9 @@ class HyMemoryClient:
                 # 版本演进链层（scene="coding" 独有）。放在 writer.write 之后、返回
                 # 之前，因为它只依赖对话原文，与向量层写入结果无关。
                 # best-effort：这层挂不上不影响本次 add 的向量层产出。
-                if scene_variant == "coding":
+                # 向量层写入失败（success=False）时跳过：调用方会重跑整段，失败的
+                # 那次不该先把改动记进演进链。
+                if scene_variant == "coding" and resp.get("success"):
                     resp["chain_rounds"] = self._observe_chain(
                         chat_messages, user_id=uid,
                         workspace_id=workspace_id, branch=branch,

hy_memory/coding/__init__.py

@@ -12,6 +12,7 @@ HY Memory - Coding Memory 模块
 
 from .types import (
     CodingMemory,
+    CodingWriteError,
     CodingMemoryDraft,
     ReconcileOp,
     WorkingFact,
@@ -26,6 +27,7 @@ from .types import (
 
 __all__ = [
     "CodingMemory",
+    "CodingWriteError",
     "CodingMemoryDraft",
     "ReconcileOp",
     "WorkingFact",

hy_memory/coding/extractor.py

@@ -33,7 +33,7 @@ from .preproc import (
     prepare_extractor_messages,
 )
 from .judge import _store_coding_log
-from .types import CodingMemoryDraft, WorkingFactDraft
+from .types import CodingMemoryDraft, CodingWriteError, WorkingFactDraft
 
 
 
@@ -217,6 +217,10 @@ class CodingMemoryExtractor:
         Returns:
             (经验草稿, 事实草稿)。两层都可能为空。只有显式启用
             verbatim_fallback_enabled 时，模型没产出 fact 才调用 _verbatim_fact。
+
+        Raises:
+            CodingWriteError: 未开启原文兜底时，LLM 调用失败（LLM_ERROR）、返回空
+                content（EMPTY_RESPONSE）或回复解析不出 JSON 对象（JSON_PARSE_FAILED）。
         """
         if not messages:
             return [], []
@@ -268,8 +272,9 @@ class CodingMemoryExtractor:
             )
         except Exception as e:
             logger.warning(f"[coding-extract] LLM call failed: {e}")
-            # 经验层不兜底。事实层原文兜底也默认关闭，避免 assistant/tool 轨迹被
-            # 当作 statement 直接展示；仅在配置显式开启时保留旧行为。
+            # 经验层不兜底。事实层原文兜底默认关闭，避免 assistant/tool 轨迹被
+            # 当作 statement 直接展示；关闭时 LLM 失败抛 CodingWriteError(LLM_ERROR)，
+            # 整次写入报失败，而不是「成功、0 条」。仅在配置显式开启时保留旧行为。
             verbatim = (
                 self._verbatim_fact(messages)
                 if self.verbatim_fallback_enabled else None
@@ -296,9 +301,33 @@ class CodingMemoryExtractor:
                 ),
                 elapsed_ms=round((time.perf_counter() - _t0) * 1000, 2),
             )
+            if not self.verbatim_fallback_enabled:
+                raise CodingWriteError(
+                    "LLM_ERROR",
+                    f"coding extract LLM call failed: {type(e).__name__}: {e}",
+                ) from e
             return [], ([verbatim] if verbatim else [])
         _elapsed = round((time.perf_counter() - _t0) * 1000, 2)
 
+        # 模型答了且没有可记的内容（如 {"experiences": [], "fact": null}）是正常的空
+        # 抽取；200 但 content 为空、或回复解析不出 JSON 对象，是模型失败，不能当成
+        # 「成功、0 条」。原文兜底显式开启时沿用旧行为（交给下面的 verbatim 兜底）。
+        failure = self._output_failure(resp.content)
+        if failure is not None and not self.verbatim_fallback_enabled:
+            code, message = failure
+            await _store_coding_log(
+                self.cache, request_id=request_id, user_id=user_id, agent_id=agent_id,
+                session_id=session_id or "", step="CODING_EXTRACT", prompt=prompt,
+                response=resp.content or "",
+                parsed=json.dumps(
+                    {"drafts": 0, "error": code, "message": message},
+                    ensure_ascii=False,
+                ),
+                elapsed_ms=_elapsed, prompt_tokens=resp.prompt_tokens,
+                completion_tokens=resp.completion_tokens, total_tokens=resp.tokens_used,
+            )
+            raise CodingWriteError(code, message)
+
         raw_experiences, raw_facts = self._parse_output(resp.content)
 
         # 经验层的 files 走规则抽取（整段轨迹碰过的文件）；事实层的 files 由 LLM
@@ -478,12 +507,22 @@ class CodingMemoryExtractor:
         return "\n".join(rows)
 
     @classmethod
-    def _parse_output(
-        cls, text: str
-    ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
-        """宽松解析 LLM 输出 JSON，返回 (experiences, facts) 两个原始列表。"""
+    def _output_failure(cls, text: Optional[str]) -> Optional[Tuple[str, str]]:
+        """回复本身是失败时返回 (error_code, message)；正常（含空抽取）返回 None。"""
+        if not (text or "").strip():
+            return "EMPTY_RESPONSE", "coding extract LLM returned empty response"
+        if cls._load_output_object(text) is None:
+            return (
+                "JSON_PARSE_FAILED",
+                f"coding extract output is not a JSON object ({len(text)} chars)",
+            )
+        return None
+
+    @staticmethod
+    def _load_output_object(text: Optional[str]) -> Optional[Dict[str, Any]]:
+        """宽松解析 LLM 输出，返回顶层 JSON 对象；解析不出返回 None。"""
         if not text:
-            return [], []
+            return None
         s = text.strip()
         # 去 markdown fence
         if s.startswith("```"):
@@ -509,8 +548,17 @@ class CodingMemoryExtractor:
                     obj = json.loads(s[start: end + 1])
                 except Exception:
                     obj = None
+        return obj if isinstance(obj, dict) else None
 
-        if not isinstance(obj, dict):
+    @classmethod
+    def _parse_output(
+        cls, text: str
+    ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+        """宽松解析 LLM 输出 JSON，返回 (experiences, facts) 两个原始列表。"""
+        if not text:
+            return [], []
+        obj = cls._load_output_object(text)
+        if obj is None:
             logger.warning(
                 f"[coding-extract] failed to parse output ({len(text or '')} chars)"
             )

hy_memory/coding/fact_reconciler.py

@@ -18,6 +18,7 @@ import json
 import logging
 import os
 import uuid
+from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -34,6 +35,7 @@ from .types import (
 
 
     MAX_FACT_CHAIN_ENTRIES,
+    CodingWriteError,
     FactReconcileOp,
     WorkingFact,
     WorkingFactDraft,
@@ -259,6 +261,16 @@ def build_chain_summary_prompt(
     return with_output_language(prompt, code, explicit=explicit is not None), code
 
 
+@dataclass
+class FactReconcilePlan:
+    """plan() 的产物：覆盖判定已完成、尚未落库。apply() 才执行 store 操作。"""
+
+    drafts: List[WorkingFactDraft]
+    candidates: List[List[WorkingFact]] = field(default_factory=list)
+    supersedes: Dict[int, List[str]] = field(default_factory=dict)
+    context: Dict[str, Any] = field(default_factory=dict)
+
+
 # ================================================================
 # WorkingFactReconciler
 # ================================================================
@@ -327,15 +339,63 @@ class WorkingFactReconciler:
         language: Optional[str] = None,
     ) -> Tuple[List[WorkingFact], List[FactReconcileOp]]:
         """
-        执行 reconcile + 持久化。memory_at 是这段对话的时间，记到新事实上。
-        language 为显式 prompt 语言码（None → zh/en 自动检测），同时用于演变总结。
+        执行 reconcile + 持久化（plan() 后紧接 apply()）。memory_at 是这段对话的
+        时间，记到新事实上。language 为显式 prompt 语言码（None → zh/en 自动检测），
+        同时用于演变总结。
 
         Returns:
-            (真正落库的新事实, 执行过的 ops)。LLM 挂掉时退化为全 INSERT 不覆盖，
-            宁可多留一条过期事实，也不因为一次调用失败丢结论。
+            (真正落库的新事实, 执行过的 ops)。
+
+        Raises:
+            CodingWriteError(RECONCILE_FAILED)：覆盖判定的 LLM 失败 / 空回复 /
+                解析不出。此时既不插入新事实，也不覆盖任何旧事实。
         """
+        plan = await self.plan(
+            drafts,
+            user_id=user_id,
+            agent_id=agent_id,
+            session_id=session_id,
+            workspace_id=workspace_id,
+            branch=branch,
+            source=source,
+            memory_at=memory_at,
+            language=language,
+        )
+        return await self.apply(plan)
+
+    async def plan(
+        self,
+        drafts: List[WorkingFactDraft],
+        *,
+        user_id: str,
+        agent_id: str = "default_agent",
+        session_id: Optional[str] = None,
+        workspace_id: Optional[str] = None,
+        branch: Optional[str] = None,
+        source: str = "default",
+        memory_at: Optional[datetime] = None,
+        language: Optional[str] = None,
+    ) -> FactReconcilePlan:
+        """检索 candidates + 覆盖判定，不落库。
+
+        没有任何候选时不调 LLM（全部 INSERT 是确定的结论，不是兜底）。调了 LLM 时，
+        有效回复 = JSON 对象且 "ops" 是列表（空列表 = 都不覆盖、只插入）。LLM 调用
+        失败（LLM_UNAVAILABLE）、200 但 content 为空（EMPTY_RESPONSE）、回复不是
+        上述形状（JSON_PARSE_FAILED）时抛 CodingWriteError(RECONCILE_FAILED)。
+        此前 LLM 失败会退化为全 INSERT 不覆盖，留下本该被接续的旧事实。
+        """
+        context = dict(
+            user_id=user_id,
+            agent_id=agent_id,
+            session_id=session_id,
+            workspace_id=workspace_id,
+            branch=branch,
+            source=source,
+            memory_at=memory_at,
+            language=language,
+        )
         if not drafts:
-            return [], []
+            return FactReconcilePlan(drafts=[], context=context)
 
         candidates = await self._retrieve_candidates(
             drafts,
@@ -349,27 +409,37 @@ class WorkingFactReconciler:
 
         try:
             supersedes = await self._llm_decide(drafts, candidates, language=language)
+        except CodingWriteError:
+            raise
         except Exception as e:
-            logger.warning(
-                f"[fact-reconcile] LLM decide failed, insert without supersede: {e}"
-            )
-            supersedes = {}
+            logger.warning(f"[fact-reconcile] LLM decide failed: {e}")
+            raise CodingWriteError(
+                "RECONCILE_FAILED",
+                f"working fact reconcile LLM call failed: {type(e).__name__}: {e}",
+                cause="LLM_UNAVAILABLE",
+            ) from e
 
+        return FactReconcilePlan(
+            drafts=list(drafts),
+            candidates=candidates,
+            supersedes=supersedes,
+            context=context,
+        )
+
+    async def apply(
+        self, plan: FactReconcilePlan,
+    ) -> Tuple[List[WorkingFact], List[FactReconcileOp]]:
+        """执行 plan() 的决策（覆盖旧事实 + 插入新事实）。"""
+        if not plan.drafts:
+            return [], []
         facts, ops = await self._execute(
-            drafts,
-            candidates,
-            supersedes,
-            user_id=user_id,
-            agent_id=agent_id,
-            session_id=session_id,
-            workspace_id=workspace_id,
-            branch=branch,
-            source=source,
-            memory_at=memory_at,
-            language=language,
+            plan.drafts,
+            plan.candidates,
+            plan.supersedes,
+            **plan.context,
         )
         logger.info(
-            f"[fact-reconcile] drafts={len(drafts)} inserted={len(facts)} "
+            f"[fact-reconcile] drafts={len(plan.drafts)} inserted={len(facts)} "
             f"superseded={sum(1 for o in ops if o.action == 'SUPERSEDE')}"
         )
         return facts, ops
@@ -473,6 +543,11 @@ class WorkingFactReconciler:
             max_tokens=self.max_tokens,
             audit_stage="coding_fact_reconcile",
         )
+        failure = self._reply_failure(resp.content)
+        if failure is not None:
+            code, message = failure
+            logger.warning(f"[fact-reconcile] {code}: {message}")
+            raise CodingWriteError("RECONCILE_FAILED", message, cause=code)
         return self._parse_ops(resp.content)
 
     # ------------------------------------------------------------
@@ -599,11 +674,24 @@ class WorkingFactReconciler:
             rows.append(f"for draft[{i}]:\n" + "\n".join(cand_rows))
         return "\n\n".join(rows) if rows else "(none)"
 
+    @classmethod
+    def _reply_failure(cls, text: Optional[str]) -> Optional[Tuple[str, str]]:
+        """回复本身是失败时返回 (cause, message)；有效回复（含 {"ops": []}）返回 None。"""
+        if not (text or "").strip():
+            return "EMPTY_RESPONSE", "working fact reconcile LLM returned empty response"
+        obj = cls._load_object(text)
+        if not isinstance(obj, dict) or not isinstance(obj.get("ops"), list):
+            return (
+                "JSON_PARSE_FAILED",
+                f"working fact reconcile output has no ops list ({len(text)} chars)",
+            )
+        return None
+
     @staticmethod
-    def _parse_ops(text: str) -> Dict[int, List[str]]:
-        """宽松解析 ops，返回 {draft_idx: [fact_id...]}。"""
+    def _load_object(text: Optional[str]) -> Optional[Dict[str, Any]]:
+        """宽松解析出顶层 JSON 对象；解析不出返回 None。"""
         if not text:
-            return {}
+            return None
         s = text.strip()
         # 去 markdown fence
         if s.startswith("```"):
@@ -621,7 +709,14 @@ class WorkingFactReconciler:
                     obj = json.loads(s[start: end + 1])
                 except Exception:
                     obj = None
+        return obj if isinstance(obj, dict) else None
 
+    @classmethod
+    def _parse_ops(cls, text: str) -> Dict[int, List[str]]:
+        """宽松解析 ops，返回 {draft_idx: [fact_id...]}。"""
+        if not text:
+            return {}
+        obj = cls._load_object(text)
         if not isinstance(obj, dict):
             logger.warning(
                 f"[fact-reconcile] failed to parse ops output ({len(text or '')} chars)"
@@ -689,6 +784,11 @@ class WorkingFactReconciler:
                 memory_at=memory_at,
             )
 
+            # 先确认新事实能向量化，再去覆盖旧事实：否则 embedding 挂掉时旧事实
+            # 已被软删、新事实又落不了库，这条结论就凭空没了。失败抛
+            # CodingWriteError(EMBEDDING_FAILED)，整次写入报失败。
+            new_embedding = await self.store.embed_fact(new_fact)
+
             allowed = {c.fact_id for c in (candidates[idx] if idx < len(candidates) else [])}
             targets = [
                 fid for fid in supersedes.get(idx, [])
@@ -769,7 +869,9 @@ class WorkingFactReconciler:
                         )
 
             try:
-                await self.store.insert_fact(new_fact)
+                await self.store.insert_fact(new_fact, embedding=new_embedding)
+            except CodingWriteError:
+                raise
             except Exception as e:
                 logger.warning(f"[fact-reconcile] INSERT failed: {e}")
                 continue

hy_memory/coding/reconciler.py

@@ -7,6 +7,9 @@ Coding Memory Reconciler
         （DELETE 由 LLM 在 draft 与 candidate 关系中观察到"明确否定/作废"时附带产出）
 
 执行端：reconcile() 调用 store 完成实际 ADD / UPDATE / DELETE 持久化操作。
+CodingWriter 分两步用：先 plan()（检索 + LLM 决策，不落库），等经验层与事实层的
+模型决策都成功后再 apply()（落库）。LLM 失败 / 空回复 / 解析不出时 plan() 抛
+CodingWriteError(RECONCILE_FAILED)，本次写入什么都不存。
 
 复用 LLMConfig.model（与 chat reconciler 共用）。
 
@@ -17,6 +20,7 @@ import json
 import logging
 import time
 import uuid
+from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional, Tuple
 
 from ..agent.llm_provider import LLMProvider
@@ -29,7 +33,7 @@ from ..prompts import (
 )
 from .judge import _store_coding_log
 from .store import CodingMemoryStore
-from .types import CodingMemory, CodingMemoryDraft, ReconcileOp
+from .types import CodingMemory, CodingMemoryDraft, CodingWriteError, ReconcileOp
 
 
 
@@ -93,6 +97,15 @@ def build_experience_reconcile_prompt(
     return with_output_language(prompt, code, explicit=explicit is not None), code
 
 
+@dataclass
+class ExperienceReconcilePlan:
+    """plan() 的产物：模型已决策、尚未落库。apply() 才执行 store 操作。"""
+
+    drafts: List[CodingMemoryDraft]
+    candidates_per_draft: List[List[CodingMemory]] = field(default_factory=list)
+    ops: List[ReconcileOp] = field(default_factory=list)
+
+
 # ================================================================
 # CodingMemoryReconciler
 # ================================================================
@@ -143,21 +156,52 @@ class CodingMemoryReconciler:
         language: Optional[str] = None,
     ) -> List[ReconcileOp]:
         """
-        执行 reconcile + 持久化。返回执行的 ReconcileOp[]（含成功执行的 ADD/UPDATE/DELETE/SKIP）。
+        执行 reconcile + 持久化（plan() 后紧接 apply()）。返回执行的 ReconcileOp[]
+        （含成功执行的 ADD/UPDATE/DELETE/SKIP）。
 
-        执行错误（如 LLM 失败）会 fail-safe 回退到对每个 draft 直接 ADD。
+        LLM 失败、空回复或解析不出时抛 CodingWriteError(RECONCILE_FAILED)，什么都
+        不落库——不再回退成「全部 ADD」。
+        """
+        plan = await self.plan(
+            drafts,
+            request_id=request_id,
+            user_id=user_id,
+            agent_id=agent_id,
+            session_id=session_id,
+            language=language,
+        )
+        return await self.apply(plan)
 
-        cache 非空时落 CODING_RECONCILE pipeline log——在 LLM 决策后、执行入库前打点，
-        这样即使 store 入库崩溃，reconcile 决策仍已可见（crash 不丢）。
+    async def plan(
+        self,
+        drafts: List[CodingMemoryDraft],
+        *,
+        request_id: str = "",
+        user_id: str = "",
+        agent_id: str = "default_agent",
+        session_id: str = "",
+        language: Optional[str] = None,
+    ) -> ExperienceReconcilePlan:
+        """检索 candidates + LLM 决策，不落库。
+
+        有效回复 = 一个 JSON 对象，其 "ops" 是列表：
+          - {"ops": []}：模型判定「无需任何操作」（与 chat reconcile 的 [] 同义），
+            成功、本层什么都不存；
+          - 非空列表：至少要有一条可识别的 op（ADD/UPDATE/DELETE/SKIP），否则视为
+            解析失败。漏掉的 draft 仍按原规则补 ADD（那是对有效回复的校验兜底，
+            不是模型失败兜底）。
+        LLM 调用失败（LLM_UNAVAILABLE）、200 但 content 为空（EMPTY_RESPONSE）、
+        回复不是上述形状（JSON_PARSE_FAILED）时抛 CodingWriteError(RECONCILE_FAILED)。
+
+        cache 非空时落 CODING_RECONCILE pipeline log（成功、失败都落）。
         """
         if not drafts:
-            return []
+            return ExperienceReconcilePlan(drafts=[])
 
         # 从 draft 回填 trace 身份（writer 通常显式传，这里兜底）
-        if drafts:
-            user_id = user_id or drafts[0].user_id
-            agent_id = agent_id or drafts[0].agent_id
-            session_id = session_id or (drafts[0].session_id or "")
+        user_id = user_id or drafts[0].user_id
+        agent_id = agent_id or drafts[0].agent_id
+        session_id = session_id or (drafts[0].session_id or "")
 
         # 1. 检索 candidates
         candidates_per_draft: List[List[CodingMemory]] = await self._retrieve_candidates(drafts)
@@ -167,19 +211,20 @@ class CodingMemoryReconciler:
         _decide_prompt = ""
         _decide_raw = ""
         _decide_resp = None
+        failure: Optional[Tuple[str, str]] = None
+        ops: List[ReconcileOp] = []
         try:
             ops, _decide_prompt, _decide_raw, _decide_resp = await self._llm_decide(
                 drafts, candidates_per_draft, language=language,
             )
+            failure = self._reply_failure(_decide_raw)
         except Exception as e:
-            logger.warning(f"[coding-reconcile] LLM decide failed, fallback to ADD-all: {e}")
-            ops = [ReconcileOp(action="ADD", draft_idx=i) for i in range(len(drafts))]
+            logger.warning(f"[coding-reconcile] LLM decide failed: {e}")
             _decide_raw = f"[LLM error] {type(e).__name__}: {str(e)[:300]}"
-
-        if not ops:
-            # LLM 输出无 ops，但有 drafts → fallback ADD-all
-            logger.info("[coding-reconcile] LLM returned no ops, fallback to ADD-all")
-            ops = [ReconcileOp(action="ADD", draft_idx=i) for i in range(len(drafts))]
+            failure = (
+                "LLM_UNAVAILABLE",
+                f"coding reconcile LLM call failed: {type(e).__name__}: {e}",
+            )
 
         _decide_elapsed = round((time.perf_counter() - _t0) * 1000, 2)
         # CODING_RECONCILE：LLM 决策完成即打点（执行入库之前，crash 不丢）
@@ -187,17 +232,42 @@ class CodingMemoryReconciler:
             self.cache, request_id=request_id, user_id=user_id, agent_id=agent_id,
             session_id=session_id, step="CODING_RECONCILE",
             prompt=_decide_prompt, response=_decide_raw,
-            parsed=json.dumps({"ops": [o.to_dict() for o in ops]}, ensure_ascii=False, default=str),
+            parsed=json.dumps(
+                {"error": failure[0], "message": failure[1], "stored": 0}
+                if failure else {"ops": [o.to_dict() for o in ops]},
+                ensure_ascii=False, default=str,
+            ),
             elapsed_ms=_decide_elapsed,
             prompt_tokens=getattr(_decide_resp, "prompt_tokens", 0) if _decide_resp else 0,
             completion_tokens=getattr(_decide_resp, "completion_tokens", 0) if _decide_resp else 0,
             total_tokens=getattr(_decide_resp, "tokens_used", 0) if _decide_resp else 0,
         )
+        if failure is not None:
+            code, message = failure
+            logger.warning(f"[coding-reconcile] {code}: {message}")
+            raise CodingWriteError("RECONCILE_FAILED", message, cause=code)
 
-        # 3. 校验 + 执行
-        executed = await self._execute(ops, drafts, candidates_per_draft)
+        return ExperienceReconcilePlan(
+            drafts=list(drafts),
+            candidates_per_draft=candidates_per_draft,
+            ops=ops,
+        )
+
+    async def apply(self, plan: ExperienceReconcilePlan) -> List[ReconcileOp]:
+        """执行 plan() 的决策（store ADD / UPDATE / DELETE）。"""
+        if not plan.drafts:
+            return []
+        if not plan.ops:
+            # 有效的 {"ops": []}：模型判定无需任何操作，本层什么都不存。
+            logger.info(
+                f"[coding-reconcile] drafts={len(plan.drafts)} LLM decided no ops"
+            )
+            return []
+        executed = await self._execute(
+            plan.ops, plan.drafts, plan.candidates_per_draft,
+        )
         logger.info(
-            f"[coding-reconcile] drafts={len(drafts)} ops={len(executed)} "
+            f"[coding-reconcile] drafts={len(plan.drafts)} ops={len(executed)} "
             + ", ".join(f"{a}={sum(1 for o in executed if o.action == a)}"
                         for a in ("ADD", "UPDATE", "DELETE", "SKIP"))
         )
@@ -316,11 +386,30 @@ class CodingMemoryReconciler:
             rows.append(f"for draft[{i}]:\n" + "\n".join(cand_rows))
         return "\n\n".join(rows) if rows else "(none)"
 
+    @classmethod
+    def _reply_failure(cls, text: Optional[str]) -> Optional[Tuple[str, str]]:
+        """回复本身是失败时返回 (cause, message)；有效回复（含 {"ops": []}）返回 None。"""
+        if not (text or "").strip():
+            return "EMPTY_RESPONSE", "coding reconcile LLM returned empty response"
+        obj = cls._load_object(text)
+        ops_raw = obj.get("ops") if isinstance(obj, dict) else None
+        if not isinstance(ops_raw, list):
+            return (
+                "JSON_PARSE_FAILED",
+                f"coding reconcile output has no ops list ({len(text)} chars)",
+            )
+        if ops_raw and not cls._parse_ops(text):
+            return (
+                "JSON_PARSE_FAILED",
+                f"coding reconcile output has no recognizable op ({len(text)} chars)",
+            )
+        return None
+
     @staticmethod
-    def _parse_ops(text: str) -> List[ReconcileOp]:
-        """宽松解析 ops 列表。"""
+    def _load_object(text: Optional[str]) -> Optional[Dict[str, Any]]:
+        """宽松解析出顶层 JSON 对象；解析不出返回 None。"""
         if not text:
-            return []
+            return None
         s = text.strip()
         # 去 markdown fence
         if s.startswith("```"):
@@ -344,8 +433,15 @@ class CodingMemoryReconciler:
                     obj = json.loads(s[start: end + 1])
                 except Exception:
                     obj = None
+        return obj if isinstance(obj, dict) else None
 
-        if not obj or not isinstance(obj, dict):
+    @classmethod
+    def _parse_ops(cls, text: str) -> List[ReconcileOp]:
+        """宽松解析 ops 列表。"""
+        if not text:
+            return []
+        obj = cls._load_object(text)
+        if not obj:
             logger.warning(
                 f"[coding-reconcile] failed to parse ops output ({len(text or '')} chars)"
             )
@@ -394,6 +490,9 @@ class CodingMemoryReconciler:
         - UPDATE  需 draft_idx 合法 + target_experience_id 非空
         - DELETE  需 target_experience_id 非空
         - SKIP    不执行任何 store 操作
+
+        向量化失败（CodingWriteError(EMBEDDING_FAILED)）不吞成「少执行一条」，
+        原样抛给 CodingWriter，整次写入报失败。
         """
         executed: List[ReconcileOp] = []
         # 已被 UPDATE 占用的 draft_idx → 防止同 draft 被重复 ADD/UPDATE
@@ -414,6 +513,8 @@ class CodingMemoryReconciler:
                     op.target_experience_id = memory.experience_id  # 回写 id 供 caller 读
                     executed.append(op)
                     consumed.add(op.draft_idx)
+                except CodingWriteError:
+                    raise
                 except Exception as e:
                     logger.warning(f"[coding-reconcile] ADD failed: {e}")
 
@@ -444,6 +545,8 @@ class CodingMemoryReconciler:
                     await self.store.update(memory)
                     executed.append(op)
                     consumed.add(op.draft_idx)
+                except CodingWriteError:
+                    raise
                 except Exception as e:
                     logger.warning(f"[coding-reconcile] UPDATE failed: {e}")
 
@@ -487,6 +590,8 @@ class CodingMemoryReconciler:
                     target_experience_id=memory.experience_id,
                     reason="fallback ADD (no LLM op)",
                 ))
+            except CodingWriteError:
+                raise
             except Exception as e:
                 logger.warning(f"[coding-reconcile] fallback ADD failed: {e}")
 

hy_memory/coding/store.py

@@ -44,7 +44,7 @@ from ..data.vector_store import create_vector_store
 from ..data.vector_store_base import VectorStoreBase
 from ..models.memory import MemoryLayer, MemoryNode, MemoryStatus
 from ..utils.async_executor import run_in_executor_guarded
-from .types import CodingMemory, WorkingFact
+from .types import CodingMemory, CodingWriteError, WorkingFact
 
 logger = logging.getLogger(__name__)
 
@@ -599,10 +599,15 @@ class CodingMemoryStore:
     # ------------------------------------------------------------
 
     async def insert(self, memory: CodingMemory) -> None:
-        """新增一条 coding memory（SQLite + VDB 全套）"""
+        """新增一条 coding memory（SQLite + VDB 全套）。
+
+        先向量化再落 SQLite：向量化失败抛 CodingWriteError(EMBEDDING_FAILED)，
+        什么都不写，不留一条没有向量、永远检索不到的经验。
+        """
         self._fill_timestamps(memory)
+        embeddings = await self._embed_vec_keys(memory)
         await _run_in_pool(self._upsert_sql_sync, memory)
-        await self._upsert_vec_keys(memory)
+        await self._upsert_vec_keys(memory, embeddings=embeddings)
         logger.info(
             f"[coding-store] inserted experience_id={memory.experience_id} "
             f"keys={1 + len(memory.search_keys)}"
@@ -627,6 +632,8 @@ class CodingMemoryStore:
         ):
             raise ValueError("coding_memory_update_authority_mismatch")
         prev_key_count = 1 + len(prev.search_keys)
+        # 先向量化：失败时旧经验（SQLite 与向量行）原样保留，不改一个字。
+        embeddings = await self._embed_vec_keys(memory)
 
         memory.updated_at = datetime.utcnow()
         if memory.created_at is None:
@@ -637,7 +644,7 @@ class CodingMemoryStore:
         # 删旧 VDB
         await self._delete_vec_keys_by_count(memory.experience_id, prev_key_count)
         # 写新 VDB
-        await self._upsert_vec_keys(memory)
+        await self._upsert_vec_keys(memory, embeddings=embeddings)
         logger.info(
             f"[coding-store] updated experience_id={memory.experience_id} "
             f"old_keys={prev_key_count} new_keys={1 + len(memory.search_keys)}"
@@ -1000,12 +1007,22 @@ class CodingMemoryStore:
     # 事实层
     # ------------------------------------------------------------
 
-    async def insert_fact(self, fact: WorkingFact) -> None:
-        """新增一条事实（SQLite + 事实 collection 各一行）。"""
+    async def insert_fact(
+        self, fact: WorkingFact, embedding: Optional[List[float]] = None,
+    ) -> None:
+        """新增一条事实（SQLite + 事实 collection 各一行）。
+
+        embedding 可由调用方预先用 embed_fact() 算好（fact reconcile 要在覆盖旧
+        事实之前确认新事实能向量化）；不传时这里算。向量化失败抛
+        CodingWriteError(EMBEDDING_FAILED)，什么都不写。
+        """
         if fact.created_at is None:
             fact.created_at = datetime.utcnow()
+        # 先向量化再落 SQLite，理由同 insert()：不落没有向量的事实。
+        if embedding is None:
+            embedding = await self.embed_fact(fact)
         await _run_in_pool(self._upsert_fact_sql_sync, fact)
-        await self._upsert_fact_vec(fact)
+        await self._upsert_fact_vec(fact, embedding=embedding)
         logger.info(f"[coding-store] inserted fact_id={fact.fact_id}")
         logger.debug(
             f"[coding-store] inserted fact_id={fact.fact_id} "
@@ -1574,25 +1591,64 @@ class CodingMemoryStore:
     # VDB 实现
     # ------------------------------------------------------------
 
-    async def _upsert_vec_keys(self, memory: CodingMemory) -> None:
-        """把 task + search_keys 写入 VDB（每行一个 MemoryNode）。"""
-        if self._vec is None:
-            return
-
+    @staticmethod
+    def _vec_key_texts(memory: CodingMemory) -> List[Tuple[str, str]]:
         all_keys: List[Tuple[str, str]] = [("task", memory.task)]
         for k in memory.search_keys:
             if k:
                 all_keys.append(("aspect", k))
+        return all_keys
+
+    async def _embed_vec_keys(
+        self, memory: CodingMemory
+    ) -> Optional[List[List[float]]]:
+        """写入链路用：向量化 task + search_keys，失败抛 CodingWriteError。
+
+        无 VDB 时返回 None（不需要向量）。
+        """
+        if self._vec is None:
+            return None
+        texts = [t for _, t in self._vec_key_texts(memory)]
+        try:
+            embeddings = await self.embed_service.embed_batch(texts)
+        except Exception as e:
+            logger.warning(f"[coding-store] embed_batch failed: {e}")
+            raise CodingWriteError(
+                "EMBEDDING_FAILED",
+                f"coding experience embedding failed: {type(e).__name__}: {e}",
+            ) from e
+        if len(embeddings or []) != len(texts) or not all(embeddings):
+            raise CodingWriteError(
+                "EMBEDDING_FAILED",
+                "coding experience embedding returned no vector",
+            )
+        return embeddings
+
+    async def _upsert_vec_keys(
+        self,
+        memory: CodingMemory,
+        embeddings: Optional[List[List[float]]] = None,
+    ) -> None:
+        """把 task + search_keys 写入 VDB（每行一个 MemoryNode）。
+
+        embeddings 由写入链路预先算好（见 _embed_vec_keys）；不传时现算，
+        失败只记 warning（edit_experience 的手动编辑路径沿用这一行为）。
+        """
+        if self._vec is None:
+            return
+
+        all_keys = self._vec_key_texts(memory)
         if not all_keys:
             return
 
         # batch embed
         texts = [t for _, t in all_keys]
-        try:
-            embeddings = await self.embed_service.embed_batch(texts)
-        except Exception as e:
-            logger.warning(f"[coding-store] embed_batch failed: {e}")
-            return
+        if embeddings is None:
+            try:
+                embeddings = await self.embed_service.embed_batch(texts)
+            except Exception as e:
+                logger.warning(f"[coding-store] embed_batch failed: {e}")
+                return
 
         nodes: List[MemoryNode] = []
         for i, ((kind, text), emb) in enumerate(zip(all_keys, embeddings)):
@@ -1623,7 +1679,34 @@ class CodingMemoryStore:
         except Exception as e:
             logger.warning(f"[coding-store] vdb upsert_batch failed: {e}")
 
-    async def _upsert_fact_vec(self, fact: WorkingFact) -> None:
+    @staticmethod
+    def _fact_vector_text(fact: WorkingFact) -> str:
+        return f"{fact.subject}: {fact.statement}" if fact.subject else fact.statement
+
+    async def embed_fact(self, fact: WorkingFact) -> Optional[List[float]]:
+        """写入链路用：向量化事实检索文本，失败抛 CodingWriteError。
+
+        无事实 VDB 或 statement 为空时返回 None（不需要向量）。
+        """
+        if self._fact_vec is None or not fact.statement:
+            return None
+        try:
+            embedding = await self.embed_service.embed(self._fact_vector_text(fact))
+        except Exception as e:
+            logger.warning(f"[coding-store] fact embed failed: {e}")
+            raise CodingWriteError(
+                "EMBEDDING_FAILED",
+                f"working fact embedding failed: {type(e).__name__}: {e}",
+            ) from e
+        if not embedding:
+            raise CodingWriteError(
+                "EMBEDDING_FAILED", "working fact embedding returned no vector",
+            )
+        return embedding
+
+    async def _upsert_fact_vec(
+        self, fact: WorkingFact, embedding: Optional[List[float]] = None,
+    ) -> None:
         """事实入向量库。一条事实一行，node_id 就是 fact_id。
 
         检索文本是 statement，前面缀上 subject（有的话）。subject 通常已经在
@@ -1634,14 +1717,14 @@ class CodingMemoryStore:
         """
         if self._fact_vec is None or not fact.statement:
             return
-        vector_text = (
-            f"{fact.subject}: {fact.statement}" if fact.subject else fact.statement
-        )
-        try:
-            embedding = await self.embed_service.embed(vector_text)
-        except Exception as e:
-            logger.warning(f"[coding-store] fact embed failed: {e}")
-            return
+        vector_text = self._fact_vector_text(fact)
+        if embedding is None:
+            # edit_fact 的手动编辑路径：现算，失败只记 warning。
+            try:
+                embedding = await self.embed_service.embed(vector_text)
+            except Exception as e:
+                logger.warning(f"[coding-store] fact embed failed: {e}")
+                return
 
         node = MemoryNode(
             node_id=fact.fact_id,

hy_memory/coding/types.py

@@ -25,6 +25,25 @@ FactStatus = Literal["active", "superseded"]
 MAX_FACT_CHAIN_ENTRIES = 8
 
 
+class CodingWriteError(RuntimeError):
+    """working / coding 写入链路里让整次写入失败的模型错误。
+
+    error_code 是稳定失败码，CodingWriter 原样透传为 agent_error_code：
+    - LLM_ERROR          抽取 LLM 调用失败
+    - EMPTY_RESPONSE     抽取 LLM 返回 200 但 content 为空
+    - JSON_PARSE_FAILED  抽取 LLM 的回复解析不出 JSON 对象
+    - EMBEDDING_FAILED   经验 / 事实向量化失败（不落没有向量的节点）
+    - RECONCILE_FAILED   经验 / 事实 reconcile 的模型决策失败；具体原因在 cause
+                         （LLM_UNAVAILABLE / EMPTY_RESPONSE / JSON_PARSE_FAILED），
+                         透传为 reconcile_error_code，与 chat 链路一致
+    """
+
+    def __init__(self, error_code: str, message: str, cause: Optional[str] = None):
+        super().__init__(message)
+        self.error_code = error_code
+        self.cause = cause
+
+
 def _epoch(value: Optional[datetime]) -> Optional[int]:
     """memory_at 对外是秒级 epoch，与 chat 一致。"""
     return int(value.timestamp()) if isinstance(value, datetime) else None

hy_memory/coding/writer.py

@@ -10,8 +10,9 @@ CodingWriter - 写入端协调器
              ├── recent history: 最近 K 个 user-turn（只作上下文）
              ├── new messages: 本次全部 turn（只做单条限长）
              ├── extractor.extract(...)            ← 一次 LLM 产出双层草稿
-             ├── reconciler.reconcile(经验草稿)     ← ADD/UPDATE/DELETE/SKIP
-             ├── fact_reconciler.reconcile(事实草稿) ← INSERT/SUPERSEDE
+             ├── reconciler.plan(经验草稿)          ← ADD/UPDATE/DELETE/SKIP 决策
+             ├── fact_reconciler.plan(事实草稿)      ← INSERT/SUPERSEDE 决策
+             ├── reconciler.apply / fact_reconciler.apply ← 两层决策都成功后才落库
              └── _crosslink(...)                   ← 同批全互链回填
 """
 
@@ -30,7 +31,7 @@ from .judge import classify_messages_is_coding
 from .preproc import has_any_tool_message, prepare_extractor_messages
 from .reconciler import CodingMemoryReconciler
 from .store import CodingMemoryStore
-from .types import FactReconcileOp, ReconcileOp, WorkingFact
+from .types import CodingWriteError, FactReconcileOp, ReconcileOp, WorkingFact
 
 logger = logging.getLogger(__name__)
 
@@ -104,6 +105,13 @@ class CodingWriter:
                 ],
                 "fact_ids": ["..."],         # 这趟真正落库的 fact_id
             }
+
+        模型失败（抽取 LLM 调用失败 / 空回复 / 解析不出）或向量化失败时返回
+        success=False，agent_error_code 为 LLM_ERROR / EMPTY_RESPONSE /
+        JSON_PARSE_FAILED / EMBEDDING_FAILED，error_code=502。经验层 / 事实层
+        reconcile 的模型决策失败时 agent_error_code=RECONCILE_FAILED，原因在
+        reconcile_error_code（LLM_UNAVAILABLE / EMPTY_RESPONSE / JSON_PARSE_FAILED），
+        本次什么都不落库。模型答了但没有值得记的内容仍是 success=True、0 条。
         """
         request_id = request_id or str(uuid.uuid4())
         t0 = time.perf_counter()
@@ -143,6 +151,10 @@ class CodingWriter:
         # 跟库里已有的什么关系是 reconcile 的事。把已有 task 摆到它面前，它会把
         # 一个其实不同的任务往清单里某个措辞上靠，reconciler 随后判成 UPDATE 合并
         # 掉，那个区别就没了——措辞对不上顶多多存一条，措辞被带偏是直接写错。
+        #
+        # 模型失败（LLM 调用失败 / 空回复 / 解析不出）抛 CodingWriteError：整次写入
+        # 报失败、什么都不存，连 trace 也不追加——调用方换个 request_id 重跑时，
+        # 历史窗口里不会多出这段本该是「新消息」的对话。
         try:
             drafts, fact_drafts = await self.extractor.extract(
                 messages,
@@ -156,26 +168,26 @@ class CodingWriter:
                 request_id=request_id,
                 language=language,
             )
-        finally:
-            # 历史服务于下一次 write；即使本次没跨过价值门槛，也不能丢掉上下文。
-            try:
-                await self.store.append_trace(
-                    current_messages,
-                    request_id=request_id,
-                    user_id=user_id,
-                    agent_id=agent_id or "default_agent",
-                    session_id=session_id,
-                    workspace_id=workspace_id,
-                    branch=branch,
-                    source=source,
-                )
-            except Exception as e:
-                logger.warning(f"[coding-write] append trace failed: {e}")
+        except CodingWriteError as e:
+            return self._failure_response(request_id, t0, e)
+
+        trace_kwargs = dict(
+            request_id=request_id,
+            user_id=user_id,
+            agent_id=agent_id or "default_agent",
+            session_id=session_id,
+            workspace_id=workspace_id,
+            branch=branch,
+            source=source,
+        )
 
         if not drafts and not fact_drafts:
+            # 模型答了、只是没有值得记的：正常的空抽取，仍算成功。历史服务于下一次
+            # write；即使本次没跨过价值门槛，也不能丢掉上下文。
             logger.info(
                 f"[coding-write] no drafts produced (value bar); request_id={request_id}"
             )
+            await self._append_trace(current_messages, **trace_kwargs)
             return self._empty_response(request_id, t0)
 
         # extractor 只负责内容；调用来源由 SDK 入口注入，不能交给 LLM 生成。
@@ -183,34 +195,51 @@ class CodingWriter:
             draft.source = source
             draft.memory_at = memory_at
 
-        # 3. 经验层 reconcile + 持久化
-        ops: List[ReconcileOp] = (
-            await self.reconciler.reconcile(
-                drafts,
-                request_id=request_id,
-                user_id=user_id,
-                agent_id=agent_id,
-                session_id=session_id or "",
-                language=language,
-            )
-            if drafts else []
-        )
-
-        # 4. 事实层 reconcile + 持久化（独立链路，不受经验层结果影响）
+        # 3. 两层 reconcile 的模型决策（只检索 + 调 LLM，不落库）
+        # 4. 两层决策都成功后才落库：经验层 apply → 事实层 apply
+        #
+        # 先把所有模型决策做完再写库，任何一层的决策失败（LLM 失败 / 空回复 /
+        # 解析不出 → CodingWriteError(RECONCILE_FAILED)）都发生在落库之前，本次
+        # 写入一条经验 / 事实都没存、也没覆盖旧事实，换个 request_id 重跑即干净。
+        # 两层决策互不依赖（各自检索各自的候选），提前做不改变结果。
+        #
+        # 向量化失败抛 CodingWriteError(EMBEDDING_FAILED)：不落没有向量的节点，
+        # 整次写入报失败。
         facts: List[WorkingFact] = []
         fact_ops: List[FactReconcileOp] = []
-        if fact_drafts:
-            facts, fact_ops = await self.fact_reconciler.reconcile(
-                fact_drafts,
-                user_id=user_id,
-                agent_id=agent_id,
-                session_id=session_id,
-                workspace_id=workspace_id,
-                branch=branch,
-                source=source,
-                memory_at=memory_at,
-                language=language,
+        ops: List[ReconcileOp] = []
+        try:
+            experience_plan = (
+                await self.reconciler.plan(
+                    drafts,
+                    request_id=request_id,
+                    user_id=user_id,
+                    agent_id=agent_id,
+                    session_id=session_id or "",
+                    language=language,
+                )
+                if drafts else None
             )
+            fact_plan = (
+                await self.fact_reconciler.plan(
+                    fact_drafts,
+                    user_id=user_id,
+                    agent_id=agent_id,
+                    session_id=session_id,
+                    workspace_id=workspace_id,
+                    branch=branch,
+                    source=source,
+                    memory_at=memory_at,
+                    language=language,
+                )
+                if fact_drafts else None
+            )
+            if experience_plan is not None:
+                ops = await self.reconciler.apply(experience_plan)
+            if fact_plan is not None:
+                facts, fact_ops = await self.fact_reconciler.apply(fact_plan)
+        except CodingWriteError as e:
+            return self._failure_response(request_id, t0, e)
 
         experience_ids = [op.target_experience_id for op in ops
                           if op.action in ("ADD", "UPDATE") and op.target_experience_id]
@@ -218,6 +247,7 @@ class CodingWriter:
 
         # 5. 同批全互链
         await self._crosslink(experience_ids, facts, user_id=user_id)
+        await self._append_trace(current_messages, **trace_kwargs)
 
         add_cnt = sum(1 for op in ops if op.action == "ADD")
         update_cnt = sum(1 for op in ops if op.action == "UPDATE")
@@ -298,6 +328,41 @@ class CodingWriter:
                         f"[coding-write] crosslink fact {f.fact_id} failed: {e}"
                     )
 
+    async def _append_trace(self, current_messages: List[ChatMessage], **kwargs) -> None:
+        """把本次对话追加进 recent trace（下一次 write 的历史上下文），best-effort。"""
+        try:
+            await self.store.append_trace(current_messages, **kwargs)
+        except Exception as e:
+            logger.warning(f"[coding-write] append trace failed: {e}")
+
+    def _failure_response(
+        self, request_id: str, t0: float, error: CodingWriteError,
+    ) -> Dict[str, Any]:
+        """模型 / 向量化失败：success=False，稳定失败码在 agent_error_code。
+
+        形状与 chat 链路的失败回执一致（error_code=502，error_message 带
+        「[CODE] 」前缀），调用方据 success 判定并重跑。
+        """
+        code = error.error_code
+        cause = getattr(error, "cause", None)
+        message = f"{cause}: {error}" if cause else str(error)
+        logger.warning(
+            f"[coding-write] failed: code={code} request_id={request_id}: {message}"
+        )
+        resp = self._empty_response(request_id, t0)
+        resp.update({
+            "success": False,
+            "error_code": 502,
+            "error_message": f"[{code}] {message}",
+            "agent_status": "failed",
+            "agent_error_code": code,
+            "agent_error": message,
+        })
+        if cause:
+            # 与 chat 链路一致：reconcile 失败的具体原因
+            resp["reconcile_error_code"] = cause
+        return resp
+
     def _empty_response(self, request_id: str, t0: float) -> Dict[str, Any]:
         return {
             "success": True,

hy_memory/pipelines/writer.py

@@ -73,6 +73,19 @@ def _write_apply_verify_enabled() -> bool:
 _RECONCILE_LOOP_USER_LOCKS = weakref.WeakKeyDictionary()
 
 
+class ReconcileFailure(str):
+    """reconcile 失败时 `_reconcile_and_store` 返回的 error：仍是错误信息字符串，
+    另带稳定的失败码 `code`（LLM_UNAVAILABLE / EMPTY_RESPONSE / JSON_PARSE_FAILED /
+    RECONCILE_EXCEPTION），供 writer 透传给调用方。"""
+
+    code: str = "RECONCILE_FAILED"
+
+    def __new__(cls, message: str, code: str = "RECONCILE_FAILED"):
+        obj = super().__new__(cls, message)
+        obj.code = code
+        return obj
+
+
 def _reconcile_user_lock(user_id: str) -> asyncio.Lock:
     loop = asyncio.get_running_loop()
     locks = _RECONCILE_LOOP_USER_LOCKS.get(loop)
@@ -1262,15 +1275,6 @@ class MemoryWriter(WritePipeline):
             recon_tokens 含 prompt/completion/total（reconcile 链路 LLM 消耗）
         """
         stored_ids: List[str] = []
-        # Episode：本次写入归属的内部会话 id（跨 write 把"一件事"归到一起）。
-        # 只数 user 侧记忆（与 rolling summary 只数 pending_user_count 一致）——
-        # agent 侧记忆若参与计数会把 episode 提前撑到上限、切得过碎，白付摘要的 token。
-        episode_id = await self._resolve_episode_id(
-            request, _count_user_owned(new_memories_meta)
-        )
-        session_episode_id = self._session_episode_id(request)
-        if session_episode_id:
-            episode_id = session_episode_id
         # 关系图用：收集本次落库的 (node, op)，供 graph 投影建边。
         # 收 op 是为了拿 op.related（LLM 标注的 typed 关系）。
         graph_payload: List[Tuple[Any, Any]] = []
@@ -1297,7 +1301,32 @@ class MemoryWriter(WritePipeline):
         )
 
         if not recon_result.success:
-            return [], recon_result.error, {"add": 0, "supersede": 0, "update": 0, "total": 0}, {"prompt": 0, "completion": 0, "total": 0}
+            # reconcile 失败：本次一条新记忆都不落库（不做 ADD-all 兜底），由调用方重跑。
+            # episode 状态也还没推进，重跑不会把同一批记忆重复计数。
+            error = ReconcileFailure(
+                recon_result.error or "reconcile failed",
+                code=recon_result.error_code or "RECONCILE_FAILED",
+            )
+            logger.warning(
+                f"TRACE_PERF [{req_id}] S1_RECONCILE_FAILED "
+                f"code={error.code} candidates={len(new_memory_texts)} stored=0"
+            )
+            return [], error, {"add": 0, "supersede": 0, "update": 0, "total": 0}, {
+                "prompt": recon_result.prompt_tokens,
+                "completion": recon_result.completion_tokens,
+                "total": recon_result.total_tokens,
+            }
+
+        # Episode：本次写入归属的内部会话 id（跨 write 把"一件事"归到一起）。
+        # 只数 user 侧记忆（与 rolling summary 只数 pending_user_count 一致）——
+        # agent 侧记忆若参与计数会把 episode 提前撑到上限、切得过碎，白付摘要的 token。
+        # 放在 reconcile 成功之后：失败的写入不能推进 episode 计数。
+        episode_id = await self._resolve_episode_id(
+            request, _count_user_owned(new_memories_meta)
+        )
+        session_episode_id = self._session_episode_id(request)
+        if session_episode_id:
+            episode_id = session_episode_id
 
         # 分类统计
         _op_counts = {}
@@ -1308,16 +1337,7 @@ class MemoryWriter(WritePipeline):
         logger.info(
             f"TRACE_PERF [{req_id}] S1_RECONCILE_DONE "
             f"ops={len(recon_result.ops)} candidates={len(new_memory_texts)} | {_op_counts_str}"
-            + (f" | DEGRADED({recon_result.degraded_reason})"
-               if getattr(recon_result, "degraded", False) else "")
         )
-        if getattr(recon_result, "degraded", False):
-            # 走了 ADD-all 兜底：内容保住了，但没做去重/演化判定。必须可观测，
-            # 否则线上只会看到"记忆变多了"而不知道为什么。
-            logger.warning(
-                f"[write] reconcile degraded → ADD-all "
-                f"({recon_result.degraded_reason}), {len(recon_result.ops)} memories stored as-is"
-            )
         # 逐条 op 详情由 reconciler 的 "[reconciler] ops detail" 单条 JSON list 输出，
         # 这里不再逐条打印（避免重复刷屏）。
 
@@ -2507,12 +2527,10 @@ class MemoryWriter(WritePipeline):
             response.extra["extract_tokens"] = {"prompt": 0, "completion": 0, "total": 0}
 
             if recon_error is not None:
-                response.success = False
-                response.error_code = 502
-                response.error_message = f"[RECONCILE_FAILED] {recon_error}"
-                response.extra["agent_status"] = "failed"
-                response.extra["agent_error"] = recon_error
-                response.extra["agent_nodes"] = 0
+                # 与 write() 一致：reconcile 失败不落任何记忆，报失败供调用方重跑。
+                self._mark_reconcile_failure(response, recon_error)
+                response.extra["agent_mode"] = "extracted"
+                logger.warning(f"[write_extracted] reconcile failed: {recon_error}")
                 return response
 
             response.success = True
@@ -3531,12 +3549,11 @@ class MemoryWriter(WritePipeline):
                 response.extra["reconcile_ops"] = ops_counts
                 response.extra["reconcile_tokens"] = recon_tokens
                 if recon_error is not None:
+                    # stored_ids 为空：reconcile 失败时一条新记忆都不落库，write()
+                    # 随后回滚 L1_RAW（retry_safe=True），换个 request_id 重跑即可。
                     response.extra["agent_stored_ids"] = stored_ids
-                    self._mark_agent_failure(
-                        response,
-                        error_code="RECONCILE_FAILED",
-                        error_message=recon_error,
-                        agent_nodes=len(stored_ids),
+                    self._mark_reconcile_failure(
+                        response, recon_error, agent_nodes=len(stored_ids),
                     )
                     logger.warning(f"[write] reconcile failed: {recon_error}")
                     return
@@ -3659,6 +3676,29 @@ class MemoryWriter(WritePipeline):
         response.error_code = 502
         response.error_message = f"[{code}] {message}"
 
+    @classmethod
+    def _mark_reconcile_failure(
+        cls,
+        response: WriteResponse,
+        recon_error: str,
+        *,
+        agent_nodes: int = 0,
+    ) -> None:
+        """reconcile 失败：agent_error_code 恒为 RECONCILE_FAILED，具体原因
+        （LLM_UNAVAILABLE / EMPTY_RESPONSE / JSON_PARSE_FAILED / RECONCILE_EXCEPTION）
+        放在 reconcile_error_code。"""
+        reason = getattr(recon_error, "code", "") or "RECONCILE_FAILED"
+        response.extra["reconcile_error_code"] = reason
+        cls._mark_agent_failure(
+            response,
+            error_code="RECONCILE_FAILED",
+            error_message=(
+                f"{reason}: {recon_error}"
+                if reason != "RECONCILE_FAILED" else str(recon_error)
+            ),
+            agent_nodes=agent_nodes,
+        )
+
     async def _handle_agent_failure(self, request, response, agent_result, tracer_span, extract_timing: Optional[Dict[str, Any]] = None) -> None:
         """处理 agent extract 失败的情况。"""
         if agent_result is None:

tests/coding/test_extractor.py

@@ -6,6 +6,7 @@ import json
 import pytest
 
 from hy_memory.coding.extractor import CodingMemoryExtractor, EXTRACT_PROMPT
+from hy_memory.coding.types import CodingWriteError
 from hy_memory.pipelines.base import ChatMessage, ToolCall
 
 from tests.coding._helpers import MockLLMProvider, FailingLLMProvider
@@ -267,11 +268,31 @@ class TestExtractor:
         assert drafts == []
         _assert_verbatim_fallback(facts)
 
-    async def test_failsafe_on_llm_error(self):
+    async def test_llm_error_raises_instead_of_empty_success(self):
+        """LLM 失败不再返回 ([], [])（等于「成功、0 条」），而是抛 LLM_ERROR。"""
         ext = CodingMemoryExtractor(FailingLLMProvider())
-        drafts, facts = await ext.extract(_msgs(), user_id="u1")
-        assert drafts == []
-        assert facts == []
+        with pytest.raises(CodingWriteError) as exc:
+            await ext.extract(_msgs(), user_id="u1")
+        assert exc.value.error_code == "LLM_ERROR"
+
+    async def test_empty_200_reply_raises(self):
+        ext = CodingMemoryExtractor(MockLLMProvider(""))
+        with pytest.raises(CodingWriteError) as exc:
+            await ext.extract(_msgs(), user_id="u1")
+        assert exc.value.error_code == "EMPTY_RESPONSE"
+
+    async def test_unparseable_reply_raises(self):
+        ext = CodingMemoryExtractor(MockLLMProvider("sorry, I can't do that"))
+        with pytest.raises(CodingWriteError) as exc:
+            await ext.extract(_msgs(), user_id="u1")
+        assert exc.value.error_code == "JSON_PARSE_FAILED"
+
+    async def test_model_answered_nothing_to_remember_is_not_an_error(self):
+        """模型答了 {} / 空列表：正常的空抽取，不是失败。"""
+        for reply in ("{}", _llm_resp([]), _llm_resp([], None)):
+            ext = CodingMemoryExtractor(MockLLMProvider(reply))
+            drafts, facts = await ext.extract(_msgs(), user_id="u1")
+            assert drafts == [] and facts == []
 
     async def test_failsafe_on_llm_error_fallback_can_be_enabled(self):
         ext = CodingMemoryExtractor(

tests/coding/test_fact_reconciler.py

@@ -12,7 +12,12 @@ from hy_memory.coding.fact_reconciler import (
     FACT_RECONCILE_PROMPT,
     WorkingFactReconciler,
 )
-from hy_memory.coding.types import CodingMemory, WorkingFact, WorkingFactDraft
+from hy_memory.coding.types import (
+    CodingMemory,
+    CodingWriteError,
+    WorkingFact,
+    WorkingFactDraft,
+)
 
 from tests.coding._helpers import FailingLLMProvider, MockLLMProvider
 
@@ -602,39 +607,79 @@ class TestCleanSummary:
         assert s == '现在读 "APP_CONFIG" 这个变量。'
 
 
+async def _seed_old_fact(store):
+    await store.insert_fact(WorkingFact(
+        fact_id="f_old", user_id="u1", session_id="test-session",
+        statement="config 位于 /opt/app/config.yaml",
+    ))
+
+
+async def _active_fact_ids(store):
+    with store._lock:
+        rows = store._conn.execute(
+            "SELECT fact_id FROM working_fact WHERE user_id=? AND status='active'",
+            ("u1",),
+        ).fetchall()
+    return sorted(r[0] for r in rows)
+
+
 @pytest.mark.asyncio
-class TestFailSafe:
-    async def test_llm_failure_still_inserts(self, coding_store, mock_embed):
-        """LLM 挂了宁可多留一条过期事实，也不能因为一次调用失败丢结论。"""
-        await coding_store.insert_fact(WorkingFact(
-            fact_id="f_old", user_id="u1", session_id="test-session",
-            statement="config 位于 /opt/app/config.yaml",
-        ))
-        rec = WorkingFactReconciler(coding_store, mock_embed, FailingLLMProvider())
+class TestDecisionFailure:
+    """覆盖判定的 LLM 失败 / 空回复 / 解析不出：抛 RECONCILE_FAILED，既不插入新事实，
+    也不覆盖旧事实（此前退化为全 INSERT 不覆盖）。"""
+
+    @pytest.mark.parametrize("llm,cause", [
+        (FailingLLMProvider(), "LLM_UNAVAILABLE"),
+        (MockLLMProvider(""), "EMPTY_RESPONSE"),
+        (MockLLMProvider("不是 JSON 的一段话"), "JSON_PARSE_FAILED"),
+        (MockLLMProvider('{"result": "keep"}'), "JSON_PARSE_FAILED"),
+    ])
+    async def test_failure_stores_nothing(self, coding_store, mock_embed, llm, cause):
+        await _seed_old_fact(coding_store)
+        rec = WorkingFactReconciler(coding_store, mock_embed, llm)
+
+        with pytest.raises(CodingWriteError) as exc:
+            await rec.reconcile(
+                [_draft("config 位于 /etc/app/config.yaml")],
+                user_id="u1", session_id="test-session",
+            )
+        assert exc.value.error_code == "RECONCILE_FAILED"
+        assert exc.value.cause == cause
+        assert await _active_fact_ids(coding_store) == ["f_old"]
+        old = await coding_store.get_fact("f_old", user_id="u1")
+        assert old.status == "active"
+
+    async def test_valid_empty_ops_inserts_without_supersede(
+        self, coding_store, mock_embed,
+    ):
+        """{"ops": []} 是有效决策：都不覆盖、只插入。"""
+        await _seed_old_fact(coding_store)
+        rec = WorkingFactReconciler(
+            coding_store, mock_embed, MockLLMProvider(_ops_resp([])),
+        )
 
         facts, ops = await rec.reconcile(
-            [_draft("config 位于 /etc/app/config.yaml")], user_id="u1",
+            [_draft("config 位于 /etc/app/config.yaml")],
+            user_id="u1", session_id="test-session",
         )
         assert len(facts) == 1
         assert [o.action for o in ops] == ["INSERT"]
         old = await coding_store.get_fact("f_old", user_id="u1")
         assert old.status == "active"
 
-    async def test_unparsable_llm_output(self, coding_store, mock_embed):
-        await coding_store.insert_fact(WorkingFact(
-            fact_id="f_old", user_id="u1", session_id="test-session",
-            statement="config 位于 /opt/app/config.yaml",
-        ))
-        rec = WorkingFactReconciler(
-            coding_store, mock_embed, MockLLMProvider("不是 JSON 的一段话")
+    async def test_no_candidates_needs_no_llm(self, coding_store, mock_embed):
+        """没有候选时不调 LLM：全部 INSERT 是确定结论，LLM 挂着也不影响。"""
+        llm = FailingLLMProvider()
+        rec = WorkingFactReconciler(coding_store, mock_embed, llm)
+        facts, _ = await rec.reconcile(
+            [_draft("config 位于 /etc/app/config.yaml")],
+            user_id="u1", session_id="test-session",
         )
+        assert len(facts) == 1 and llm.calls == 0
 
-        facts, ops = await rec.reconcile(
-            [_draft("config 位于 /etc/app/config.yaml")], user_id="u1",
-        )
-        assert len(facts) == 1
-        assert [o.action for o in ops] == ["INSERT"]
 
+@pytest.mark.asyncio
+class TestReplyParsing:
     async def test_markdown_fence_parsed(self, coding_store, mock_embed):
         await coding_store.insert_fact(WorkingFact(
             fact_id="f_old", user_id="u1", session_id="test-session",

tests/coding/test_reconciler.py

@@ -6,7 +6,7 @@ import json
 import pytest
 
 from hy_memory.coding.reconciler import CodingMemoryReconciler
-from hy_memory.coding.types import CodingMemory, CodingMemoryDraft
+from hy_memory.coding.types import CodingMemory, CodingMemoryDraft, CodingWriteError
 
 from tests.coding._helpers import MockLLMProvider, FailingLLMProvider
 
@@ -217,12 +217,38 @@ class TestReconcilerActions:
 
 @pytest.mark.asyncio
 class TestReconcilerFallback:
-    async def test_llm_failure_fallback_add_all(self, coding_store, mock_embed):
-        rec = CodingMemoryReconciler(coding_store, mock_embed, FailingLLMProvider())
+    @pytest.mark.parametrize("llm,cause", [
+        (FailingLLMProvider(), "LLM_UNAVAILABLE"),
+        (MockLLMProvider(""), "EMPTY_RESPONSE"),
+        (MockLLMProvider("not json"), "JSON_PARSE_FAILED"),
+        (MockLLMProvider("{}"), "JSON_PARSE_FAILED"),
+        (MockLLMProvider('{"ops": [{"draft_idx": 0, "action": "MERGE"}]}'),
+         "JSON_PARSE_FAILED"),
+    ])
+    async def test_decision_failure_raises_and_stores_nothing(
+        self, coding_store, mock_embed, llm, cause,
+    ):
+        """LLM 失败 / 空回复 / 解析不出不再回退成 ADD-all：抛 RECONCILE_FAILED。"""
+        rec = CodingMemoryReconciler(coding_store, mock_embed, llm)
         d1 = CodingMemoryDraft(task="T1", solution="S", user_id="u1")
         d2 = CodingMemoryDraft(task="T2", solution="S", user_id="u1")
-        ops = await rec.reconcile([d1, d2])
-        assert len(ops) == 2 and all(op.action == "ADD" for op in ops)
+        with pytest.raises(CodingWriteError) as exc:
+            await rec.reconcile([d1, d2])
+        assert exc.value.error_code == "RECONCILE_FAILED"
+        assert exc.value.cause == cause
+        assert await coding_store.list_user_tasks("u1") == []
+
+    async def test_valid_empty_ops_is_a_no_change_decision(
+        self, coding_store, mock_embed,
+    ):
+        """{"ops": []} 与 chat 的 [] 同义：成功、什么都不存。"""
+        rec = CodingMemoryReconciler(
+            coding_store, mock_embed, MockLLMProvider(_llm_resp([])),
+        )
+        d = CodingMemoryDraft(task="T", solution="S", user_id="u1")
+        ops = await rec.reconcile([d])
+        assert ops == []
+        assert await coding_store.list_user_tasks("u1") == []
 
     async def test_missing_draft_idx_fallback_add(self, coding_store, mock_embed):
         # LLM 只处理了 draft 0，draft 1 漏掉
@@ -251,9 +277,13 @@ class TestReconcilerFallback:
         ops = await rec.reconcile([])
         assert ops == []
 
-    async def test_garbage_llm_output_fallback(self, coding_store, mock_embed):
-        rec = CodingMemoryReconciler(coding_store, mock_embed, MockLLMProvider("not json"))
+    async def test_plan_does_not_store(self, coding_store, mock_embed):
+        """plan() 只做检索 + 决策；apply() 之前库里什么都没有。"""
+        llm = MockLLMProvider(_llm_resp([{"draft_idx": 0, "action": "ADD"}]))
+        rec = CodingMemoryReconciler(coding_store, mock_embed, llm)
         d = CodingMemoryDraft(task="T", solution="S", user_id="u1")
-        ops = await rec.reconcile([d])
-        # 解析失败 → fallback ADD
-        assert any(op.action == "ADD" for op in ops)
+        plan = await rec.plan([d])
+        assert [op.action for op in plan.ops] == ["ADD"]
+        assert await coding_store.list_user_tasks("u1") == []
+        await rec.apply(plan)
+        assert await coding_store.list_user_tasks("u1") == ["T"]

tests/coding/test_writer_e2e.py

@@ -36,6 +36,16 @@ def _reconcile_resp(ops):
     return json.dumps({"ops": ops})
 
 
+class _FailingEmbed:
+    """向量化服务整体不可用。"""
+
+    async def embed(self, text):
+        raise RuntimeError("embedding service down")
+
+    async def embed_batch(self, texts):
+        raise RuntimeError("embedding service down")
+
+
 def _coding_messages():
     return [
         ChatMessage(role="user", content="帮我打包发布到私有包索引"),
@@ -220,16 +230,117 @@ class TestCodingWriterE2E:
         assert sorted(await coding_store.list_user_tasks("u1")) == ["T1", "T2"]
 
     async def test_extractor_failure(self, coding_store, mock_embed):
-        """extractor LLM 失败 → 空 drafts → 空 ops"""
+        """extractor LLM 失败 → success=False / LLM_ERROR，什么都不存，trace 也不追加"""
         ext = CodingMemoryExtractor(FailingLLMProvider())
         rec = CodingMemoryReconciler(coding_store, mock_embed, MockLLMProvider("{}"))
         writer = CodingWriter(
             store=coding_store, extractor=ext, reconciler=rec,
             llm_provider=FailingLLMProvider(), embed_service=mock_embed,
         )
-        resp = await writer.write(_coding_messages(), user_id="u1", workspace_id="ws")
-        assert resp["success"] is True
-        assert resp["ops"] == []
+        resp = await writer.write(
+            _coding_messages(), user_id="u1", workspace_id="ws", session_id="s1",
+        )
+        assert resp["success"] is False
+        assert resp["error_code"] == 502
+        assert resp["agent_error_code"] == "LLM_ERROR"
+        assert resp["error_message"].startswith("[LLM_ERROR]")
+        assert resp["ops"] == [] and resp["experience_ids"] == []
+        assert resp["fact_ids"] == []
+        assert await coding_store.list_user_tasks("u1") == []
+        assert await coding_store.get_recent_trace_messages(
+            user_id="u1", agent_id="default_agent", max_user_turns=5,
+        ) == [], "失败的写入不追加 trace：重跑时历史里不会多出这段"
+
+    async def test_extractor_failure_then_rerun_stores(self, coding_store, mock_embed):
+        """LLM 恢复后换 request_id 重跑，正常落库。"""
+        failing = CodingMemoryExtractor(FailingLLMProvider())
+        rec = CodingMemoryReconciler(coding_store, mock_embed, MockLLMProvider("{}"))
+        writer = CodingWriter(
+            store=coding_store, extractor=failing, reconciler=rec,
+            llm_provider=FailingLLMProvider(), embed_service=mock_embed,
+        )
+        first = await writer.write(
+            _coding_messages(), user_id="u1", session_id="s1", request_id="r1",
+        )
+        assert first["success"] is False
+
+        llm = MockLLMProvider([
+            _extract_resp([{"task": "T1", "solution": "S", "confidence": 0.9}]),
+            _reconcile_resp([{"draft_idx": 0, "action": "ADD"}]),
+        ])
+        writer.extractor = CodingMemoryExtractor(llm)
+        writer.reconciler = CodingMemoryReconciler(coding_store, mock_embed, llm)
+        second = await writer.write(
+            _coding_messages(), user_id="u1", session_id="s1", request_id="r2",
+        )
+        assert second["success"] is True
+        assert await coding_store.list_user_tasks("u1") == ["T1"]
+
+    async def test_extractor_empty_reply_is_failure(self, coding_store, mock_embed):
+        llm = MockLLMProvider("")
+        writer = CodingWriter(
+            store=coding_store, extractor=CodingMemoryExtractor(llm),
+            reconciler=CodingMemoryReconciler(coding_store, mock_embed, llm),
+            llm_provider=llm, embed_service=mock_embed,
+        )
+        resp = await writer.write(_coding_messages(), user_id="u1", session_id="s1")
+        assert resp["success"] is False
+        assert resp["agent_error_code"] == "EMPTY_RESPONSE"
+
+    async def test_experience_embedding_failure_is_failure(
+        self, coding_config, mock_embed,
+    ):
+        """经验向量化失败 → EMBEDDING_FAILED，SQLite 里也不留没有向量的经验。"""
+        from hy_memory.coding.store import CodingMemoryStore
+
+        store = CodingMemoryStore(coding_config, embed_service=_FailingEmbed())
+        await store.initialize()
+        try:
+            llm = MockLLMProvider([
+                _extract_resp([{"task": "T1", "solution": "S", "confidence": 0.9}]),
+                _reconcile_resp([{"draft_idx": 0, "action": "ADD"}]),
+            ])
+            writer = CodingWriter(
+                store=store, extractor=CodingMemoryExtractor(llm),
+                reconciler=CodingMemoryReconciler(store, _FailingEmbed(), llm),
+                llm_provider=llm, embed_service=_FailingEmbed(),
+            )
+            resp = await writer.write(
+                _coding_messages(), user_id="u1", session_id="s1",
+            )
+            assert resp["success"] is False
+            assert resp["agent_error_code"] == "EMBEDDING_FAILED"
+            assert resp["experience_ids"] == []
+            assert await store.list_user_tasks("u1") == []
+        finally:
+            await store.close()
+
+    async def test_fact_embedding_failure_is_failure(self, coding_config, mock_embed):
+        """事实向量化失败 → EMBEDDING_FAILED，事实不落库。"""
+        from hy_memory.coding.store import CodingMemoryStore
+
+        store = CodingMemoryStore(coding_config, embed_service=_FailingEmbed())
+        await store.initialize()
+        try:
+            llm = MockLLMProvider(_extract_resp([], {"statement": "构建产物在 dist/"}))
+            writer = CodingWriter(
+                store=store, extractor=CodingMemoryExtractor(llm),
+                reconciler=CodingMemoryReconciler(store, _FailingEmbed(), llm),
+                llm_provider=llm, embed_service=_FailingEmbed(),
+            )
+            resp = await writer.write(
+                _coding_messages(), user_id="u1", session_id="s1",
+            )
+            assert resp["success"] is False
+            assert resp["agent_error_code"] == "EMBEDDING_FAILED"
+            assert resp["fact_ids"] == []
+            with store._lock:
+                n = store._conn.execute(
+                    "SELECT COUNT(*) FROM working_fact WHERE user_id=?", ("u1",),
+                ).fetchone()[0]
+            assert n == 0
+        finally:
+            await store.close()
 
     async def test_existing_tasks_not_shown_to_extractor(self, coding_store, mock_embed):
         """库里已有的 task 不得进抽取 prompt——抽取只读这一段，去重是 reconcile 的事"""
@@ -249,6 +360,142 @@ class TestCodingWriterE2E:
         assert "老 task" not in llm.calls[0]["prompt"]
 
 
+def _count_rows(store, sql, *args):
+    with store._lock:
+        return store._conn.execute(sql, args).fetchone()[0]
+
+
+def _writer_with(store, embed, *, extract_llm, exp_llm, fact_llm):
+    from hy_memory.coding.fact_reconciler import WorkingFactReconciler
+
+    return CodingWriter(
+        store=store,
+        extractor=CodingMemoryExtractor(extract_llm),
+        reconciler=CodingMemoryReconciler(store, embed, exp_llm),
+        fact_reconciler=WorkingFactReconciler(store, embed, fact_llm),
+        llm_provider=extract_llm,
+        embed_service=embed,
+    )
+
+
+_TWO_LAYER_EXTRACT = _extract_resp(
+    [{"task": "T1", "solution": "S", "confidence": 0.9}],
+    {"statement": "config 位于 /etc/app/config.yaml"},
+)
+
+
+@pytest.mark.asyncio
+class TestReconcileDecisionFailure:
+    """经验层 / 事实层 reconcile 的模型决策失败：整次写入 success=False、
+    agent_error_code=RECONCILE_FAILED，本次一条经验 / 事实都不落库，也不覆盖旧事实。"""
+
+    @pytest.mark.parametrize("exp_llm,cause", [
+        (FailingLLMProvider(), "LLM_UNAVAILABLE"),
+        (MockLLMProvider(""), "EMPTY_RESPONSE"),
+        (MockLLMProvider("not json at all"), "JSON_PARSE_FAILED"),
+    ])
+    async def test_experience_reconcile_failure_stores_nothing(
+        self, coding_store, mock_embed, exp_llm, cause,
+    ):
+        writer = _writer_with(
+            coding_store, mock_embed,
+            extract_llm=MockLLMProvider(_TWO_LAYER_EXTRACT),
+            exp_llm=exp_llm,
+            fact_llm=MockLLMProvider(_reconcile_resp([])),
+        )
+        resp = await writer.write(_coding_messages(), user_id="u1", session_id="s1")
+
+        assert resp["success"] is False
+        assert resp["error_code"] == 502
+        assert resp["agent_error_code"] == "RECONCILE_FAILED"
+        assert resp["reconcile_error_code"] == cause
+        assert resp["error_message"].startswith(f"[RECONCILE_FAILED] {cause}")
+        assert resp["experience_ids"] == [] and resp["fact_ids"] == []
+        assert await coding_store.list_user_tasks("u1") == []
+        assert _count_rows(
+            coding_store, "SELECT COUNT(*) FROM working_fact WHERE user_id=?", "u1",
+        ) == 0, "事实层的决策虽然没问题，也不能先落库"
+        assert await coding_store.get_recent_trace_messages(
+            user_id="u1", agent_id="default_agent", max_user_turns=5,
+        ) == []
+
+    async def test_valid_no_ops_reply_is_success(self, coding_store, mock_embed):
+        """经验层 {"ops": []}：有效的「无需任何操作」，写入成功，经验层不存。"""
+        writer = _writer_with(
+            coding_store, mock_embed,
+            extract_llm=MockLLMProvider(_TWO_LAYER_EXTRACT),
+            exp_llm=MockLLMProvider(_reconcile_resp([])),
+            fact_llm=MockLLMProvider(_reconcile_resp([])),
+        )
+        resp = await writer.write(_coding_messages(), user_id="u1", session_id="s1")
+
+        assert resp["success"] is True
+        assert resp["ops"] == [] and resp["experience_ids"] == []
+        assert await coding_store.list_user_tasks("u1") == []
+        assert len(resp["fact_ids"]) == 1, "事实层照常插入"
+
+    @pytest.mark.parametrize("fact_llm,cause", [
+        (FailingLLMProvider(), "LLM_UNAVAILABLE"),
+        (MockLLMProvider(""), "EMPTY_RESPONSE"),
+        (MockLLMProvider("不是 JSON"), "JSON_PARSE_FAILED"),
+    ])
+    async def test_fact_reconcile_failure_stores_nothing(
+        self, coding_store, mock_embed, fact_llm, cause,
+    ):
+        """事实层覆盖判定失败：新事实不插入、旧事实不被覆盖，经验层也不先落库。"""
+        from hy_memory.coding.types import WorkingFact
+
+        await coding_store.insert_fact(WorkingFact(
+            fact_id="f_old", user_id="u1", agent_id="default_agent",
+            session_id="s1", statement="config 位于 /opt/app/config.yaml",
+        ))
+        writer = _writer_with(
+            coding_store, mock_embed,
+            extract_llm=MockLLMProvider(_TWO_LAYER_EXTRACT),
+            exp_llm=MockLLMProvider(_reconcile_resp([{"draft_idx": 0, "action": "ADD"}])),
+            fact_llm=fact_llm,
+        )
+        resp = await writer.write(_coding_messages(), user_id="u1", session_id="s1")
+
+        assert resp["success"] is False
+        assert resp["agent_error_code"] == "RECONCILE_FAILED"
+        assert resp["reconcile_error_code"] == cause
+        assert await coding_store.list_user_tasks("u1") == [], "经验层不得先落库"
+        assert _count_rows(
+            coding_store, "SELECT COUNT(*) FROM working_fact WHERE user_id=?", "u1",
+        ) == 1, "只剩旧事实，新事实没插入"
+        old = await coding_store.get_fact("f_old", user_id="u1")
+        assert old.status == "active", "旧事实不得被覆盖"
+
+    async def test_rerun_with_new_request_id_after_recovery(
+        self, coding_store, mock_embed,
+    ):
+        writer = _writer_with(
+            coding_store, mock_embed,
+            extract_llm=MockLLMProvider(_TWO_LAYER_EXTRACT),
+            exp_llm=FailingLLMProvider(),
+            fact_llm=MockLLMProvider(_reconcile_resp([])),
+        )
+        first = await writer.write(
+            _coding_messages(), user_id="u1", session_id="s1", request_id="r1",
+        )
+        assert first["success"] is False
+
+        writer.reconciler = CodingMemoryReconciler(
+            coding_store, mock_embed,
+            MockLLMProvider(_reconcile_resp([{"draft_idx": 0, "action": "ADD"}])),
+        )
+        second = await writer.write(
+            _coding_messages(), user_id="u1", session_id="s1", request_id="r2",
+        )
+        assert second["success"] is True
+        assert await coding_store.list_user_tasks("u1") == ["T1"]
+        assert len(second["fact_ids"]) == 1
+        assert _count_rows(
+            coding_store, "SELECT COUNT(*) FROM working_fact WHERE user_id=?", "u1",
+        ) == 1
+
+
 @pytest.mark.asyncio
 class TestDualLayerWrite:
     """双层写入 + 同批全互链（§4）"""

tests/test_prompt_language_selection.py

@@ -342,10 +342,13 @@ async def test_coding_writer_threads_language_to_every_step():
     fact = WorkingFactDraft(subject="x", statement="y")
     writer.extractor = type("X", (), {})()
     writer.extractor.extract = AsyncMock(return_value=([draft], [fact]))
+    # writer 先做两层的模型决策（plan），都成功后才落库（apply）
     writer.reconciler = type("R", (), {})()
-    writer.reconciler.reconcile = AsyncMock(return_value=[])
+    writer.reconciler.plan = AsyncMock(return_value=object())
+    writer.reconciler.apply = AsyncMock(return_value=[])
     writer.fact_reconciler = type("F", (), {})()
-    writer.fact_reconciler.reconcile = AsyncMock(return_value=([], []))
+    writer.fact_reconciler.plan = AsyncMock(return_value=object())
+    writer.fact_reconciler.apply = AsyncMock(return_value=([], []))
     writer._crosslink = AsyncMock()
     writer.cache = None
     writer.variant = "working"
@@ -353,8 +356,8 @@ async def test_coding_writer_threads_language_to_every_step():
     await writer.write([ChatMessage(role="user", content="hi")], user_id="u1",
                        language="de")
     assert writer.extractor.extract.await_args.kwargs["language"] == "de"
-    assert writer.reconciler.reconcile.await_args.kwargs["language"] == "de"
-    assert writer.fact_reconciler.reconcile.await_args.kwargs["language"] == "de"
+    assert writer.reconciler.plan.await_args.kwargs["language"] == "de"
+    assert writer.fact_reconciler.plan.await_args.kwargs["language"] == "de"
 
 
 # ── summaries ──

tests/test_reconcile_failure.py

@@ -0,0 +1,377 @@
+"""reconcile 的模型调用失败时，整次写入报失败、一条新记忆都不落库。
+
+此前 reconcile 失败（LLM 不可用 / 返回空 / 解析不出 / 链路异常）会走「全部 ADD」
+兜底并报 success=True：调用方不会重试，而没经过判重 / 演化的整批 ADD 又会在之后
+的写入里制造重复。现在：
+
+- ReconcileResult(success=False, ops=[], error_code=...)，不再有 ADD-all 兜底；
+- add（write）：agent_error_code=RECONCILE_FAILED，L1_RAW 按抽取失败同样回滚
+  （l1_rollback.retry_safe=True），换个 request_id 重跑即可落库；
+- add_extracted（write_extracted）：同样报失败、什么都不存；
+- RECONCILE_ENABLED=false 是配置选择，不是失败，仍直写。
+"""
+
+import asyncio
+import json
+import uuid
+from typing import Any, Dict, List
+
+import pytest
+
+from hy_memory.agent.llm_provider import LLMResponse
+from hy_memory.agent.reconciler import MemoryReconciler, ReconcileResult
+from hy_memory.config import MemoryConfig
+from hy_memory.models.memory import MemoryLayer, MemoryNode, MemoryStatus
+from hy_memory.pipelines.base import WriteRequest
+
+
+FACTS = [
+    "The user took a road trip in July 2023 to Banff, where he went skiing.",
+    "The user is in a new romantic relationship as of August 2023.",
+]
+EXISTING = "The user enjoys skiing."
+DIMS = 16
+
+
+def run(coro):
+    loop = asyncio.new_event_loop()
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+def _vec(text: str) -> List[float]:
+    seed = sum(ord(c) for c in text) or 1
+    v = [((seed * (i + 7)) % 97) / 97.0 + 0.01 for i in range(DIMS)]
+    n = sum(x * x for x in v) ** 0.5
+    return [x / n for x in v]
+
+
+class _FakeEmbed:
+    async def embed(self, text, **kwargs):
+        return _vec(text)
+
+    async def embed_batch(self, texts, **kwargs):
+        return [_vec(t) for t in texts]
+
+    async def embed_queued(self, text, use_cache=True):
+        return _vec(text)
+
+    async def embed_queued_with_meta(self, text, use_cache=True):
+        return _vec(text), {"mock": True}
+
+
+class _ReconcileLLM:
+    """reconcile 用的假 LLM：mode 决定这一次怎么失败 / 恢复。"""
+
+    def __init__(self, mode: str = "fail"):
+        self.mode = mode
+        self.calls = 0
+
+    async def complete(self, **kwargs):
+        self.calls += 1
+        if self.mode == "fail":
+            raise RuntimeError("gateway 500")
+        if self.mode == "empty":
+            return LLMResponse(content="", tokens_used=3, prompt_tokens=3)
+        if self.mode == "garbage":
+            return LLMResponse(content="I cannot help with that.", tokens_used=3)
+        if self.mode == "noop":
+            return LLMResponse(content="[]", tokens_used=3)
+        ops = [
+            {"op": "ADD", "content": f, "layer": "L2_FACT", "owner": "user"}
+            for f in FACTS
+        ]
+        return LLMResponse(content=json.dumps(ops), tokens_used=5)
+
+
+class _AgentResult:
+    success = True
+    suggested_layer = None
+    summary = ""
+    conflicts: List[Dict[str, Any]] = []
+    should_merge = False
+    merge_target_id = None
+    tokens_used = 0
+    extract_elapsed_ms = 1.0
+    summary_elapsed_ms = 0.0
+    error = None
+    error_code = None
+
+    def __init__(self):
+        self.extracted_info = {
+            "memory": [
+                {"content": f, "layer": "L2_FACT", "owner": "user", "tags": []}
+                for f in FACTS
+            ],
+        }
+
+
+class _FakeMemAgent:
+    async def process_add(self, **kwargs):
+        return _AgentResult()
+
+
+@pytest.fixture
+def store(tmp_path):
+    from hy_memory.data.vector_store_chroma import ChromaVectorStore
+
+    config = MemoryConfig.from_dict({"vector_store": {
+        "provider": "chroma",
+        "collection_name": f"test_recon_fail_{uuid.uuid4().hex[:8]}",
+        "persist_directory": str(tmp_path / "chroma"),
+        "embedding_dims": DIMS,
+    }})
+    vs = ChromaVectorStore(config)
+    run(vs.initialize())
+    # 一条已有记忆，与第一条新事实同向量：保证 reconcile 有候选、真的去调 LLM
+    # （没有任何候选时 reconcile 不调 LLM，直接全部 ADD，那不是模型失败的场景）。
+    run(vs.upsert(MemoryNode(
+        node_id="existing-1",
+        user_id="u1",
+        agent_id="default_agent",
+        session_id="s0",
+        content=EXISTING,
+        layer=MemoryLayer.L2_FACT,
+        status=MemoryStatus.ACTIVE,
+        is_latest=True,
+        embedding=_vec(FACTS[0]),
+    )))
+    yield config, vs
+    run(vs.close())
+
+
+def _writer(config, vs, llm):
+    from hy_memory.pipelines.writer import MemoryWriter
+
+    writer = MemoryWriter(config, embed_service=_FakeEmbed(), vector_store=vs, cache=None)
+    run(writer.initialize())
+    writer._embed_service = _FakeEmbed()
+    writer._mem_agent = _FakeMemAgent()
+    writer.reconciler._llm = llm
+    writer.reconciler._enable_search_query = False
+    return writer
+
+
+def _add_request(request_id: str) -> WriteRequest:
+    return WriteRequest(
+        content="user: I went skiing in Banff last July.",
+        user_id="u1",
+        agent_id="default_agent",
+        session_id="s1",
+        extra={"agent_mode": "full"},
+        request_id=request_id,
+    )
+
+
+def _nodes(vs, layer: MemoryLayer):
+    nodes = run(vs.list_by_user(
+        user_id="u1", layers=[layer], limit=100, allow_raw_layer=True,
+    ))
+    return [n for n in nodes if n.layer == layer]
+
+
+def _reconcile(vs, llm, embed=None) -> ReconcileResult:
+    rec = MemoryReconciler(MemoryConfig(), llm_provider=llm)
+    rec._enable_search_query = False
+    return run(rec.reconcile(
+        new_memories=list(FACTS),
+        user_id="u1",
+        agent_id="default_agent",
+        vector_store=vs,
+        embed_service=embed or _FakeEmbed(),
+    ))
+
+
+def _new_facts(vs) -> List[str]:
+    return sorted(
+        n.content for n in _nodes(vs, MemoryLayer.L2_FACT) if n.node_id != "existing-1"
+    )
+
+
+class TestReconcilerReportsFailure:
+    def test_llm_unavailable_is_failure_without_ops(self, store):
+        llm = _ReconcileLLM("fail")
+        r = _reconcile(store[1], llm)
+        assert llm.calls >= 1
+        assert r.success is False
+        assert r.ops == []
+        assert r.error_code == "LLM_UNAVAILABLE"
+
+    def test_empty_200_reply_is_failure_not_zero_ops(self, store):
+        llm = _ReconcileLLM("empty")
+        r = _reconcile(store[1], llm)
+        assert r.success is False
+        assert r.ops == []
+        assert r.error_code == "EMPTY_RESPONSE"
+        assert llm.calls == 2, "空回复按解析失败重试一次"
+
+    def test_unparseable_reply_is_failure(self, store):
+        r = _reconcile(store[1], _ReconcileLLM("garbage"))
+        assert r.success is False
+        assert r.ops == []
+        assert r.error_code == "JSON_PARSE_FAILED"
+
+    def test_explicit_empty_array_is_a_decision(self, store):
+        """字面量 [] 是模型判定「无需任何操作」，仍是成功。"""
+        r = _reconcile(store[1], _ReconcileLLM("noop"))
+        assert r.success is True
+        assert r.ops == []
+
+    def test_unexpected_exception_is_failure(self, store):
+        class _BrokenEmbed(_FakeEmbed):
+            async def embed_batch(self, texts, **kwargs):
+                raise RuntimeError("embedding down")
+
+        r = _reconcile(store[1], _ReconcileLLM("add"), embed=_BrokenEmbed())
+        assert r.success is False
+        assert r.ops == []
+        assert r.error_code == "RECONCILE_EXCEPTION"
+
+    def test_no_add_all_fallback_left(self):
+        assert not hasattr(MemoryReconciler, "_fallback_add_ops")
+        assert not hasattr(ReconcileResult(), "degraded")
+
+
+class TestAddReconcileFailure:
+    @pytest.mark.parametrize(
+        "mode,reason",
+        [("fail", "LLM_UNAVAILABLE"), ("empty", "EMPTY_RESPONSE"),
+         ("garbage", "JSON_PARSE_FAILED")],
+    )
+    def test_failure_stores_nothing_and_rolls_back_raw(self, store, mode, reason):
+        config, vs = store
+        writer = _writer(config, vs, _ReconcileLLM(mode))
+
+        resp = run(writer.write(_add_request("req-fail")))
+
+        assert resp.success is False
+        assert resp.error_code == 502
+        assert resp.extra["agent_status"] == "failed"
+        assert resp.extra["agent_error_code"] == "RECONCILE_FAILED"
+        assert resp.extra["reconcile_error_code"] == reason
+        assert resp.error_message.startswith("[RECONCILE_FAILED]")
+        assert resp.extra.get("agent_stored_ids") == []
+        assert _new_facts(vs) == [], "reconcile 失败不得落任何新 L2"
+        rollback = resp.extra["l1_rollback"]
+        assert rollback["deleted"] is True
+        assert rollback["retry_safe"] is True
+        assert _nodes(vs, MemoryLayer.L1_RAW) == [], "L1_RAW 应已回滚"
+
+    def test_rerun_with_new_request_id_after_recovery_stores(self, store):
+        config, vs = store
+        llm = _ReconcileLLM("fail")
+        writer = _writer(config, vs, llm)
+
+        failed = run(writer.write(_add_request("req-1")))
+        assert failed.success is False
+
+        llm.mode = "add"
+        ok = run(writer.write(_add_request("req-2")))
+
+        assert ok.success is True
+        assert ok.extra["agent_status"] == "success"
+        assert _new_facts(vs) == sorted(FACTS)
+        raws = _nodes(vs, MemoryLayer.L1_RAW)
+        assert len(raws) == 1, "失败那次的 raw 已回滚，只剩重跑这次的"
+        assert raws[0].status == MemoryStatus.SHADOW
+
+
+class TestAddExtractedReconcileFailure:
+    def _request(self, request_id: str) -> WriteRequest:
+        return WriteRequest(
+            content=FACTS[0],
+            user_id="u1",
+            agent_id="default_agent",
+            session_id="s1",
+            extra={"extracted_contents": list(FACTS), "agent_mode": "extracted"},
+            request_id=request_id,
+        )
+
+    def test_failure_stores_nothing(self, store):
+        config, vs = store
+        writer = _writer(config, vs, _ReconcileLLM("empty"))
+
+        resp = run(writer.write_extracted(self._request("req-x1")))
+
+        assert resp.success is False
+        assert resp.error_code == 502
+        assert resp.extra["agent_error_code"] == "RECONCILE_FAILED"
+        assert resp.extra["reconcile_error_code"] == "EMPTY_RESPONSE"
+        assert _new_facts(vs) == []
+
+    def test_rerun_after_recovery_stores(self, store):
+        config, vs = store
+        llm = _ReconcileLLM("fail")
+        writer = _writer(config, vs, llm)
+        assert run(writer.write_extracted(self._request("req-x1"))).success is False
+
+        llm.mode = "add"
+        resp = run(writer.write_extracted(self._request("req-x2")))
+
+        assert resp.success is True
+        assert _new_facts(vs) == sorted(FACTS)
+
+    def test_reconcile_disabled_still_writes_directly(self, store, monkeypatch):
+        """RECONCILE_ENABLED=false 是配置选择：不调 reconcile LLM，直写。"""
+        import hy_memory.pipelines.writer as writer_mod
+
+        monkeypatch.setattr(writer_mod, "_RECONCILE_ENABLED", False)
+        config, vs = store
+        llm = _ReconcileLLM("fail")
+        writer = _writer(config, vs, llm)
+
+        resp = run(writer.write_extracted(self._request("req-x3")))
+
+        assert resp.success is True
+        assert llm.calls == 0
+        assert _new_facts(vs) == sorted(FACTS)
+
+
+class TestClientReceipt:
+    def test_reconcile_error_code_reaches_the_caller(self):
+        from hy_memory.client import _attach_write_diagnostics
+
+        resp: Dict[str, Any] = {}
+        _attach_write_diagnostics(resp, {
+            "agent_status": "failed",
+            "agent_error_code": "RECONCILE_FAILED",
+            "reconcile_error_code": "EMPTY_RESPONSE",
+            "l1_rollback": {"retry_safe": True},
+        })
+        assert resp["agent_error_code"] == "RECONCILE_FAILED"
+        assert resp["reconcile_error_code"] == "EMPTY_RESPONSE"
+        assert resp["l1_rollback"]["retry_safe"] is True
+
+
+class TestExtractorPromptRules:
+    """新增的两条 extractor 规则（线上 badcase 直接对应）。"""
+
+    def _prompts(self):
+        from hy_memory.agent.extractor import EXTRACT_SYSTEM_PROMPT
+        from hy_memory.agent import prompts_zh
+
+        zh = next(
+            v for k, v in vars(prompts_zh).items()
+            if isinstance(v, str) and "时间规则" in v and "质量要求" in v
+        )
+        return EXTRACT_SYSTEM_PROMPT, zh
+
+    def test_just_happened_events_rule_present_both_languages(self):
+        """"我刚旅行回来" 的返回日期就是 LoCoMo 时序题的答案，此前被丢掉。"""
+        en, zh = self._prompts()
+        assert "just got back" in en
+        assert "刚回来" in zh or "刚刚发生" in zh
+
+    def test_no_generalisation_rule_present_both_languages(self):
+        """Banff 被泛化成 "the Rocky Mountains"，具体地名丢失。"""
+        en, zh = self._prompts()
+        assert "Banff" in en and "Banff" in zh
+        assert "Rocky Mountains" in en
+
+    def test_cross_speaker_detail_rule_present(self):
+        """细节由对方话轮给出、用户确认 —— 主体规则不该导致丢细节。"""
+        en, zh = self._prompts()
+        assert "other speaker" in en
+        assert "对方话轮" in zh

tests/test_reconcile_fallback.py

@@ -1,146 +0,0 @@
-"""reconcile 失败时必须兜底为「全部 ADD」，不能丢弃抽取结果。
-
-一个暂存环境回归场景中，抽取已成功，但 reconcile 的 LLM 在重试后失败，
-导致整批完好事实被丢弃，`memory_ops` 返回 0 条。
-reconcile 只负责决定「新事实和旧记忆怎么合」，它不可用不代表抽取结果没价值。
-宁可留重复（后续 dedup / reconcile 还能收拾）也不要丢内容。
-"""
-
-from hy_memory.agent.reconciler import MemoryReconciler, ReconcileResult
-
-
-NEW = [
-    "The user took a road trip in July 2023 to Banff, where he went skiing.",
-    "The user is in a new romantic relationship as of August 2023.",
-    "Sam offered to give the user tips on organizing a book club.",
-]
-META = [
-    {"content": NEW[0], "layer": "L2_FACT", "tags": ["travel", "skiing"], "owner": "user"},
-    {"content": NEW[1], "layer": "L2_FACT", "tags": ["relationship"], "owner": "user"},
-    {"content": NEW[2], "layer": "L2_FACT", "tags": ["friends"], "owner": "agent"},
-]
-
-
-class TestFallbackAddOps:
-    def test_builds_one_add_per_extracted_memory(self):
-        ops = MemoryReconciler._fallback_add_ops(NEW, META)
-        assert len(ops) == 3
-        assert all(o.op == "ADD" for o in ops)
-        assert [o.content for o in ops] == NEW
-
-    def test_preserves_layer_tags_owner(self):
-        """兜底不能把 meta 丢了，否则 owner/tags 全部退化。"""
-        ops = MemoryReconciler._fallback_add_ops(NEW, META)
-        assert [o.layer for o in ops] == ["L2_FACT"] * 3
-        assert ops[0].tags == ["travel", "skiing"]
-        assert ops[2].owner == "agent"
-
-    def test_reason_marks_degradation(self):
-        ops = MemoryReconciler._fallback_add_ops(NEW, META)
-        assert "degraded" in ops[0].reason
-
-    def test_works_without_meta(self):
-        """meta 缺失时也要能兜底（只是没有 tags/owner）。"""
-        ops = MemoryReconciler._fallback_add_ops(NEW, None)
-        assert len(ops) == 3 and all(o.op == "ADD" for o in ops)
-
-    def test_skips_empty_content(self):
-        ops = MemoryReconciler._fallback_add_ops(["", None, "real fact"], None)
-        assert [o.content for o in ops] == ["real fact"]
-
-    def test_no_memories_yields_nothing(self):
-        assert MemoryReconciler._fallback_add_ops([], None) == []
-
-
-class TestReconcileResultDegradedFlag:
-    def test_default_is_not_degraded(self):
-        r = ReconcileResult()
-        assert r.degraded is False and r.degraded_reason is None
-
-    def test_degraded_result_still_succeeds(self):
-        """关键语义：兜底后 success 必须为 True，否则 writer 仍会丢弃全部内容。
-
-        writer 里是 `if not recon_result.success: return [], ...` —— 只要 success=False，
-        ops 里有什么都不会落库。
-        """
-        r = ReconcileResult(
-            ops=MemoryReconciler._fallback_add_ops(NEW, META),
-            success=True, degraded=True, degraded_reason="LLM_UNAVAILABLE",
-        )
-        assert r.success is True and r.degraded is True
-        assert len(r.ops) == 3
-
-
-class TestWriterHonorsDegradedResult:
-    def test_writer_stores_ops_when_success_true(self):
-        """从 writer 源码上钉住：success=True 时不会走提前 return 的丢弃分支。"""
-        import inspect
-
-        from hy_memory.pipelines.writer import MemoryWriter
-
-        src = inspect.getsource(MemoryWriter._reconcile_and_store_locked)
-        assert "if not recon_result.success:" in src
-        # 兜底路径必须可观测
-        assert "degraded" in src
-
-
-class TestLLMFailureIsNotPropagated:
-    """LLM 异常不能冒泡 —— 冒泡会让整批抽取结果连带丢弃。"""
-
-    def test_llm_call_is_wrapped(self):
-        import inspect
-
-        from hy_memory.agent.reconciler import MemoryReconciler as R
-
-        src = inspect.getsource(R.reconcile)
-        # llm.complete 必须在 try 内，且失败后走 ADD-all
-        assert "_llm_error" in src
-        assert "_fallback_add_ops" in src
-        assert "LLM_UNAVAILABLE" in src
-
-    def test_failure_log_tolerates_none_response(self):
-        """LLM 挂掉时 response 是 None；失败分支里若直接取 response.prompt_tokens
-        会抛 AttributeError，把兜底本身也废掉。"""
-        import inspect
-
-        from hy_memory.agent.reconciler import MemoryReconciler as R
-
-        src = inspect.getsource(R.reconcile)
-        # 只看失败分支（成功路径里 response 必然非 None，直接取属性是对的）
-        start = src.index("if not _parse_ok:")
-        end = src.index("_fallback_add_ops", start)
-        fail_branch = src[start:end]
-        assert 'getattr(response, "prompt_tokens", 0)' in fail_branch
-        assert "response.prompt_tokens," not in fail_branch
-
-
-class TestExtractorPromptRules:
-    """新增的两条 extractor 规则（线上 badcase 直接对应）。"""
-
-    def _prompts(self):
-        from hy_memory.agent.extractor import EXTRACT_SYSTEM_PROMPT
-        from hy_memory.agent import prompts_zh
-
-        zh = next(
-            v for k, v in vars(prompts_zh).items()
-            if isinstance(v, str) and "时间规则" in v and "质量要求" in v
-        )
-        return EXTRACT_SYSTEM_PROMPT, zh
-
-    def test_just_happened_events_rule_present_both_languages(self):
-        """"我刚旅行回来" 的返回日期就是 LoCoMo 时序题的答案，此前被丢掉。"""
-        en, zh = self._prompts()
-        assert "just got back" in en
-        assert "刚回来" in zh or "刚刚发生" in zh
-
-    def test_no_generalisation_rule_present_both_languages(self):
-        """Banff 被泛化成 "the Rocky Mountains"，具体地名丢失。"""
-        en, zh = self._prompts()
-        assert "Banff" in en and "Banff" in zh
-        assert "Rocky Mountains" in en
-
-    def test_cross_speaker_detail_rule_present(self):
-        """细节由对方话轮给出、用户确认 —— 主体规则不该导致丢细节。"""
-        en, zh = self._prompts()
-        assert "other speaker" in en
-        assert "对方话轮" in zh