fix: align coding authority with conversations

Co-authored-by: Cursor <cursoragent@cursor.com>

CHANGELOG.md

@@ -16,6 +16,14 @@ Hy Memory and MemHub are versioned and released independently.
 
 - Serialized each SDK SQLite store's shared connection lifecycle so concurrent
   history/cache operations cannot race with `close()` and crash the process.
+- Scoped Coding reconciliation targets to the current HY authority,
+  conversation, and workspace, preventing near-duplicate sessions from
+  mutating an experience or working fact owned by another evidence context.
+- Kept post-write authority verification failures available for read-only
+  reconciliation while preventing automatic or generic confirmed retries from
+  invoking the model again. Automatic reconciliation skips those sticky rows
+  before applying its batch limit, and linked durable jobs exclusively own
+  retry scheduling for their memory tasks.
 - Clarified that `add(metadata=...)` carries legacy SDK processing options,
   while persistent caller business context belongs in `meta`.
 - Added an opt-in `get(..., include_raw=True)` path for reading `L1_RAW`

hy_memory/coding/fact_reconciler.py

@@ -466,7 +466,13 @@ class WorkingFactReconciler:
             return [], []
 
         candidates = await self._retrieve_candidates(
-            drafts, user_id=user_id, source=source,
+            drafts,
+            user_id=user_id,
+            agent_id=agent_id,
+            session_id=session_id,
+            workspace_id=workspace_id,
+            branch=branch,
+            source=source,
         )
 
         try:
@@ -499,7 +505,15 @@ class WorkingFactReconciler:
     # ------------------------------------------------------------
 
     async def _retrieve_candidates(
-        self, drafts: List[WorkingFactDraft], *, user_id: str, source: str
+        self,
+        drafts: List[WorkingFactDraft],
+        *,
+        user_id: str,
+        agent_id: str,
+        session_id: Optional[str],
+        workspace_id: Optional[str],
+        branch: Optional[str],
+        source: str,
     ) -> List[List[WorkingFact]]:
         """每条 draft 用 statement 召回同 user 的 top-N active 事实。
 
@@ -512,6 +526,10 @@ class WorkingFactReconciler:
         关掉时整份召回原样交给 LLM。适用于 subject 不是稳定实体名的数据集（自由文本、
         位置标签），那里精确字符串相等会把同实体判成不同主体，反而拦掉真实更新。
         """
+        # Empty/legacy session metadata remains readable, but cannot prove
+        # conversation authority for a destructive SUPERSEDE.
+        if not session_id:
+            return [[] for _ in drafts]
         try:
             embeddings = await self.embed_service.embed_batch(
                 [d.statement for d in drafts]
@@ -525,6 +543,10 @@ class WorkingFactReconciler:
             try:
                 hits = await self.store.search_facts(
                     emb, user_id=user_id, top=self.candidates_per_draft,
+                    agent_id=agent_id,
+                    session_id=session_id,
+                    workspace_id=workspace_id,
+                    branch=branch,
                     # 写入侧同样接词面通道：候选选错了 LLM 就没有机会判对。这里不传
                     # min_score——写入侧要的是"有哪些可能的前身"，让 LLM 看更宽的名单，
                     # 阈值过滤是读取侧的事。
@@ -794,6 +816,15 @@ class WorkingFactReconciler:
                 old = await self.store.get_fact(old_id, user_id=user_id)
                 if old is None:
                     continue
+                if (
+                    old.agent_id != agent_id
+                    or old.session_id != session_id
+                    or old.workspace_id != workspace_id
+                ):
+                    logger.warning(
+                        f"[fact-reconcile] refused cross-authority SUPERSEDE: {old.fact_id}"
+                    )
+                    continue
                 # 闸门在候选阶段已经拦过一道，这里再拦一次：候选是按 statement 召回的，
                 # 而 LLM 只能覆盖候选名单里的 id，理论上到不了这儿。真到了就说明前面
                 # 哪一环漏了，宁可让这次覆盖不生效——多留一条旧事实是可恢复的，删掉

hy_memory/coding/reconciler.py

@@ -240,7 +240,7 @@ class CodingMemoryReconciler:
         )
 
         # 3. 校验 + 执行
-        executed = await self._execute(ops, drafts)
+        executed = await self._execute(ops, drafts, candidates_per_draft)
         logger.info(
             f"[coding-reconcile] drafts={len(drafts)} ops={len(executed)} "
             + ", ".join(f"{a}={sum(1 for o in executed if o.action == a)}"
@@ -263,6 +263,12 @@ class CodingMemoryReconciler:
         """
         result: List[List[CodingMemory]] = []
         for d in drafts:
+            # A coding mutation must be attributable to one conversation.
+            # Empty/legacy session metadata stays readable through ordinary
+            # search, but it cannot authorize UPDATE/DELETE.
+            if not d.session_id:
+                result.append([])
+                continue
             try:
                 # task 为主入口，search_keys 为附加入口（去重、去空）
                 queries = [d.task] + [k for k in (d.search_keys or []) if k and k != d.task]
@@ -274,6 +280,8 @@ class CodingMemoryReconciler:
                     hits = await self.store.search_by_query_embedding(
                         emb,
                         user_id=d.user_id,
+                        agent_id=d.agent_id,
+                        session_id=d.session_id,
                         workspace_id=d.workspace_id,
                         branch=d.branch,
                         top=self.candidates_per_draft * 4,
@@ -415,6 +423,7 @@ class CodingMemoryReconciler:
         self,
         ops: List[ReconcileOp],
         drafts: List[CodingMemoryDraft],
+        candidates_per_draft: List[List[CodingMemory]],
     ) -> List[ReconcileOp]:
         """
         校验 + 执行 ops。返回成功执行的 ops（filter 掉无效 op）。
@@ -454,6 +463,11 @@ class CodingMemoryReconciler:
                 if not op.target_experience_id:
                     logger.warning(f"[coding-reconcile] drop UPDATE missing target: {op.to_dict()}")
                     continue
+                if not self._target_was_offered(op, candidates_per_draft):
+                    logger.warning(
+                        f"[coding-reconcile] drop UPDATE outside authority scope: {op.to_dict()}"
+                    )
+                    continue
                 if op.draft_idx in consumed:
                     logger.warning(f"[coding-reconcile] drop UPDATE on consumed draft: {op.to_dict()}")
                     continue
@@ -476,6 +490,11 @@ class CodingMemoryReconciler:
                 if not op.target_experience_id:
                     logger.warning(f"[coding-reconcile] drop DELETE missing target: {op.to_dict()}")
                     continue
+                if not self._target_was_offered(op, candidates_per_draft):
+                    logger.warning(
+                        f"[coding-reconcile] drop DELETE outside authority scope: {op.to_dict()}"
+                    )
+                    continue
                 # DELETE 需要知道 user_id；通过 draft_idx（如有）或第一个 draft 推断
                 user_id = self._infer_user_id(op, drafts)
                 if not user_id:
@@ -512,6 +531,19 @@ class CodingMemoryReconciler:
 
         return executed
 
+    @staticmethod
+    def _target_was_offered(
+        op: ReconcileOp,
+        candidates_per_draft: List[List[CodingMemory]],
+    ) -> bool:
+        """Only mutate a target frozen into this draft's scoped candidate set."""
+        if op.draft_idx is None or not (0 <= op.draft_idx < len(candidates_per_draft)):
+            return False
+        return any(
+            candidate.experience_id == op.target_experience_id
+            for candidate in candidates_per_draft[op.draft_idx]
+        )
+
     @staticmethod
     def _infer_user_id(
         op: ReconcileOp, drafts: List[CodingMemoryDraft]

hy_memory/coding/store.py

@@ -529,10 +529,20 @@ class CodingMemoryStore:
         """覆盖式更新一条 coding memory。"""
         # 拿旧 keys 长度，先删旧 VDB rows
         prev = await self.get_by_id(memory.experience_id, user_id=memory.user_id)
-        prev_key_count = (1 + len(prev.search_keys)) if prev else 0
+        if prev is None:
+            raise ValueError("coding_memory_update_target_missing")
+        if (
+            not prev.session_id
+            or not memory.session_id
+            or prev.agent_id != memory.agent_id
+            or prev.session_id != memory.session_id
+            or prev.workspace_id != memory.workspace_id
+        ):
+            raise ValueError("coding_memory_update_authority_mismatch")
+        prev_key_count = 1 + len(prev.search_keys)
 
         memory.updated_at = datetime.utcnow()
-        if memory.created_at is None and prev is not None:
+        if memory.created_at is None:
             memory.created_at = prev.created_at
         await _run_in_pool(self._upsert_sql_sync, memory)
         # 删旧 VDB
@@ -671,6 +681,8 @@ class CodingMemoryStore:
         query_embedding: List[float],
         user_id: str,
         *,
+        agent_id: Optional[str] = None,
+        session_id: Optional[str] = None,
         workspace_id: Optional[str] = None,
         branch: Optional[str] = None,
         top: int = 40,
@@ -688,21 +700,28 @@ class CodingMemoryStore:
                 },
                 ...
             ]
-        workspace_id / branch 是作用域过滤，规则见 _scope_where：不传则纯语义召回
-        （行为与加过滤前完全一致）；传了则只保留 SQLite 里落在作用域内的经验，并按
-        作用域的选择率放大向量取数，避免"先取 top-N 再过滤"把召回吃掉。
+        agent_id / session_id 是写授权作用域，传入时精确匹配；workspace_id / branch
+        的规则见 _scope_where。不传任何作用域则维持纯语义召回；传了则只保留 SQLite
+        里落在作用域内的经验，并按选择率放大向量取数，避免"先取 top-N 再过滤"把
+        召回吃掉。
         """
         if self._vec is None:
             return []
 
         allowed: Optional[Set[str]] = None
-        if workspace_id:
+        if agent_id or session_id or workspace_id:
             allowed, total = await _run_in_pool(
-                self._scoped_experience_ids_sync, user_id, workspace_id, branch
+                self._authority_scoped_experience_ids_sync,
+                user_id,
+                agent_id,
+                session_id,
+                workspace_id,
+                branch,
             )
             if not allowed:
                 logger.info(
                     f"[coding-store] no experience in scope: "
+                    f"agent_id={agent_id!r} session_id={session_id!r} "
                     f"workspace_id={workspace_id!r} branch={branch!r} "
                     f"(user has {total}); skip vdb search"
                 )
@@ -778,6 +797,8 @@ class CodingMemoryStore:
         query_embedding: List[float],
         user_id: str,
         *,
+        agent_id: Optional[str] = None,
+        session_id: Optional[str] = None,
         workspace_id: Optional[str] = None,
         branch: Optional[str] = None,
         top: int = 5,
@@ -791,8 +812,9 @@ class CodingMemoryStore:
         SUPERSEDE 时向量行已被删掉，理论上召不到 superseded；这里再按 SQLite 的
         status 过一道，避免两边不一致时把过期结论放出去。
 
-        workspace_id / branch 是作用域过滤，规则见 _scope_where，语义与经验层一致：
-        不传不过滤，传了按 SQLite 里的值筛，并按选择率放大向量取数。
+        agent_id / session_id 是写授权作用域，传入时精确匹配；workspace_id / branch
+        的规则见 _scope_where。不传任何作用域则不过滤，传了按 SQLite 里的值筛，并
+        按选择率放大向量取数。
 
         向量库按 top 的 4 倍过采样：向量行和 SQLite 行不是一个事务里写的，中间挂掉
         就会留下"向量还在、SQLite 没了"的孤儿，软删时删向量失败也会留下 superseded
@@ -809,13 +831,19 @@ class CodingMemoryStore:
             return []
 
         fetch = top * 4
-        if workspace_id:
+        if agent_id or session_id or workspace_id:
             scoped, total = await _run_in_pool(
-                self._count_active_facts_sync, user_id, workspace_id, branch
+                self._count_active_facts_sync,
+                user_id,
+                agent_id,
+                session_id,
+                workspace_id,
+                branch,
             )
             if not scoped:
                 logger.info(
                     f"[coding-store] no active fact in scope: "
+                    f"agent_id={agent_id!r} session_id={session_id!r} "
                     f"workspace_id={workspace_id!r} branch={branch!r} "
                     f"(user has {total}); skip fact vdb search"
                 )
@@ -852,6 +880,8 @@ class CodingMemoryStore:
         alive = [
             (f, scores[f.fact_id]) for f in facts
             if f.status == "active"
+            and (not agent_id or f.agent_id == agent_id)
+            and (not session_id or f.session_id == session_id)
             and _scope_matches(f.workspace_id, f.branch, workspace_id, branch)
         ]
         kept = [(f, s) for f, s in alive if s >= min_score]
@@ -1050,14 +1080,25 @@ class CodingMemoryStore:
             by_id = {r["experience_id"]: _row_to_memory(r) for r in rows}
             return [by_id[eid] for eid in experience_ids if eid in by_id]
 
-    def _scoped_experience_ids_sync(
-        self, user_id: str, workspace_id: str, branch: Optional[str]
+    def _authority_scoped_experience_ids_sync(
+        self,
+        user_id: str,
+        agent_id: Optional[str],
+        session_id: Optional[str],
+        workspace_id: Optional[str],
+        branch: Optional[str],
     ) -> Tuple[Set[str], int]:
-        """返回 (落在作用域内的 experience_id 集合, 该 user 的经验总数)。
+        """返回 (落在授权作用域内的 experience_id 集合, 该 user 的经验总数)。
 
         一次查询同时拿到过滤结果和选择率分母：前者供向量命中做交集，后者定放大倍数。
         """
         scope_sql, scope_params = _scope_where(workspace_id, branch)
+        if agent_id:
+            scope_sql = f"agent_id = ? AND ({scope_sql})"
+            scope_params = [agent_id, *scope_params]
+        if session_id:
+            scope_sql = f"session_id = ? AND ({scope_sql})"
+            scope_params = [session_id, *scope_params]
         with self._lock:
             assert self._conn is not None
             cur = self._conn.execute(
@@ -1069,13 +1110,24 @@ class CodingMemoryStore:
         return {r["experience_id"] for r in rows if r["in_scope"]}, len(rows)
 
     def _count_active_facts_sync(
-        self, user_id: str, workspace_id: str, branch: Optional[str]
+        self,
+        user_id: str,
+        agent_id: Optional[str],
+        session_id: Optional[str],
+        workspace_id: Optional[str],
+        branch: Optional[str],
     ) -> Tuple[int, int]:
         """返回 (作用域内 active 事实数, 该 user 的 active 事实总数)。
 
         事实层的过滤在 get_facts 拿到对象之后用 _scope_matches 做，这里只要选择率。
         """
         scope_sql, scope_params = _scope_where(workspace_id, branch)
+        if agent_id:
+            scope_sql = f"agent_id = ? AND ({scope_sql})"
+            scope_params = [agent_id, *scope_params]
+        if session_id:
+            scope_sql = f"session_id = ? AND ({scope_sql})"
+            scope_params = [session_id, *scope_params]
         with self._lock:
             assert self._conn is not None
             row = self._conn.execute(

memhub/apps/core/src/one_memory/api.py

@@ -185,7 +185,12 @@ from one_memory.memory_queue import (
     resolve_extraction_state,
 )
 from one_memory.memory_server.config_state import ConfigStore
-from one_memory.memory_tasks import MemoryTask, MemoryTaskStatus, MemoryTaskStore
+from one_memory.memory_tasks import (
+    NON_RETRYABLE_POST_WRITE_CODES,
+    MemoryTask,
+    MemoryTaskStatus,
+    MemoryTaskStore,
+)
 from one_memory.model_presets import seed_preset_profile
 from one_memory.model_probe import ModelProber, ModelProbeResult
 from one_memory.model_profiles import (
@@ -1082,8 +1087,14 @@ def create_app(
             drain_budget = production_drain_budget()
 
             async def reconcile_unknown_writes() -> None:
-                await app.state.memory_processing_supervisor.reconcile_unknown_outputs()
-                unknown = await asyncio.to_thread(memory_write_queue.unknown_jobs, limit=128)
+                await app.state.memory_processing_supervisor.reconcile_unknown_outputs(
+                    exclude_error_codes=NON_RETRYABLE_POST_WRITE_CODES,
+                )
+                unknown = await asyncio.to_thread(
+                    memory_write_queue.unknown_jobs,
+                    limit=128,
+                    exclude_error_codes=NON_RETRYABLE_POST_WRITE_CODES,
+                )
                 for job in unknown:
                     await asyncio.to_thread(
                         memory_write_queue.resolve_unknown_from_linked_output,

memhub/apps/core/src/one_memory/memory_processing.py

@@ -1522,14 +1522,24 @@ class MemoryProcessingSupervisor:
                 await self._refresh_read_bridge(completed[-1].profile_snapshot)
             return tuple(completed)
 
-    async def reconcile_unknown_outputs(self, *, limit: int = 128) -> int:
+    async def reconcile_unknown_outputs(
+        self,
+        *,
+        limit: int = 128,
+        exclude_error_codes: tuple[str, ...] = (),
+    ) -> int:
         """Publish HY outputs that can be proven by an unknown task's request id.
 
         This inquiry is read-only on HY and never invokes the model. Rows with
         no exact match remain unknown for the person's explicit retry choice.
         """
         reconciled = 0
-        for task in await in_the_background(self.tasks.unknown_tasks, limit=limit):
+        tasks = await in_the_background(
+            self.tasks.unknown_tasks,
+            limit=limit,
+            exclude_error_codes=exclude_error_codes,
+        )
+        for task in tasks:
             runner = self.runner_factory()
             bridge: MemoryBridgeProtocol | None = None
             try:

memhub/apps/core/src/one_memory/memory_server/durable_queue.py

@@ -38,10 +38,11 @@ from one_memory.memory_server.migrations import memory_write_jobs_sqlite_schema
 from one_memory.memory_server.queue import build_reclaim_sqlite_statements
 from one_memory.memory_server.registry import WriteRecord
 from one_memory.memory_tasks import (
-    POLICY_STOPPED_CODES as _TASK_POLICY_STOPPED_CODES,
+    NON_RETRYABLE_POST_WRITE_CODES,
+    MemoryTaskStatus,
 )
 from one_memory.memory_tasks import (
-    MemoryTaskStatus,
+    POLICY_STOPPED_CODES as _TASK_POLICY_STOPPED_CODES,
 )
 from one_memory.service_levels import ServiceLevel, service_level_context
 
@@ -1542,9 +1543,13 @@ class DurableWriteQueue:
             connection,
         ):
             row = connection.execute(
-                "SELECT state FROM memory_write_jobs WHERE job_id=?", (job_id,)
+                "SELECT state,last_error FROM memory_write_jobs WHERE job_id=?", (job_id,)
             ).fetchone()
-            if row is None or str(row["state"]) != UNKNOWN:
+            if (
+                row is None
+                or str(row["state"]) != UNKNOWN
+                or str(row["last_error"]) in NON_RETRYABLE_POST_WRITE_CODES
+            ):
                 return 0
             connection.execute(
                 """
@@ -1701,8 +1706,15 @@ class DurableWriteQueue:
             rows = connection.execute(
                 "SELECT job_id FROM memory_write_jobs "
                 "WHERE state=? AND attempts=1 AND attempts < max_attempts "
+                f"AND COALESCE(last_error, '') NOT IN "
+                f"({','.join('?' for _ in NON_RETRYABLE_POST_WRITE_CODES)}) "
                 "AND updated_at <= ? ORDER BY updated_at, job_id LIMIT ?",
-                (UNKNOWN, older_than.isoformat(), limit),
+                (
+                    UNKNOWN,
+                    *NON_RETRYABLE_POST_WRITE_CODES,
+                    older_than.isoformat(),
+                    limit,
+                ),
             ).fetchall()
             job_ids = tuple(str(row["job_id"]) for row in rows)
             if not job_ids:
@@ -1766,6 +1778,7 @@ class DurableWriteQueue:
         *,
         limit: int = 128,
         after_cursor: tuple[str, str] | None = None,
+        exclude_error_codes: tuple[str, ...] = (),
     ) -> tuple[UnknownWriteJob, ...]:
         """List unknown rows that still need an inquiry, in stable order.
 
@@ -1782,6 +1795,9 @@ class DurableWriteQueue:
         """
         if type(limit) is not int or limit < 1:
             raise ValueError("limit must be positive")
+        excluded = tuple(dict.fromkeys(exclude_error_codes))
+        if not all(isinstance(code, str) and code for code in excluded):
+            raise ValueError("exclude_error_codes must contain non-empty strings")
         cursor_clause = ""
         params: tuple[Any, ...] = ()
         if after_cursor is not None:
@@ -1794,14 +1810,20 @@ class DurableWriteQueue:
             updated_at, job_id = after_cursor
             cursor_clause = "AND (updated_at, job_id) > (?, ?)"
             params = (updated_at, job_id)
+        if excluded:
+            placeholders = ",".join("?" for _ in excluded)
+            excluded_clause = f"AND COALESCE(last_error, '') NOT IN ({placeholders})"
+        else:
+            excluded_clause = ""
         with main_database(self.database_path).read(row_factory=sqlite3.Row) as connection:
             rows = connection.execute(
                 "SELECT job_id, idempotency_key, created_at, updated_at, last_error "
                 "FROM memory_write_jobs "
                 "WHERE state=? "
                 f"{cursor_clause} "
+                f"{excluded_clause} "
                 "ORDER BY updated_at, job_id LIMIT ?",
-                (UNKNOWN, *params, limit),
+                (UNKNOWN, *params, *excluded, limit),
             ).fetchall()
         return tuple(
             UnknownWriteJob(

memhub/apps/core/src/one_memory/memory_tasks.py

@@ -67,6 +67,12 @@ POLICY_STOPPED_CODES = (
     "hy_memory_evidence_changed",
 )
 
+#: HY completed a write, but Core could not prove that the selected mutation
+#: target belonged to this task's frozen authority. Reconciliation may inspect
+#: HY by request id, but neither automatic nor user-confirmed generic retry may
+#: invoke the model again for these codes.
+NON_RETRYABLE_POST_WRITE_CODES = ("hy_memory_write_verification_authority_invalid",)
+
 
 class MemoryTask(BaseModel):
     model_config = ConfigDict(frozen=True, extra="forbid")
@@ -116,6 +122,24 @@ PROFILE_STOP_CODES = (
 ACTIVE_WINDOW_DEFER_LIMIT = 2048
 
 
+def _task_only_unknown_clause(connection: sqlite3.Connection) -> str:
+    """Exclude tasks whose Evidence is already owned by the durable queue."""
+    linked_table = connection.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name='memory_write_job_evidence'"
+    ).fetchone()
+    if linked_table is None:
+        return ""
+    return """
+        AND NOT EXISTS (
+            SELECT 1
+            FROM memory_task_inputs AS inputs, json_each(inputs.refs_json) AS ref
+            JOIN memory_write_job_evidence AS linked
+              ON linked.evidence_id=json_extract(ref.value, '$.evidence_id')
+            WHERE inputs.task_id=memory_tasks.task_id
+        )
+    """
+
+
 @dataclass(frozen=True, slots=True)
 class ActiveMemoryTask:
     """One task plus only the evidence fields the scheduler needs."""
@@ -1314,7 +1338,7 @@ class MemoryTaskStore:
         live_clause = f"task.task_id IN ({live_placeholders})" if live_ids else "0"
         not_live_clause = f"task.task_id NOT IN ({live_placeholders})" if live_ids else "1"
         excluded_clause = f"task.task_id NOT IN ({excluded_placeholders})" if excluded else "1"
-        common = f"""
+        common_base = f"""
             task.status IN (?,?,?,?)
             AND coalesce(task.last_error_code, '') NOT IN (
                 'hy_memory_input_exceeds_window', 'model_content_outcome_unknown'
@@ -1333,7 +1357,33 @@ class MemoryTaskStore:
         """
         moment = "coalesce(evidence.created_at, evidence.imported_at)"
 
+        def common_for(connection: sqlite3.Connection) -> str:
+            has_durable_queue = connection.execute(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' "
+                "AND name IN ('memory_write_jobs','memory_write_job_evidence')"
+            ).fetchone()[0]
+            if int(has_durable_queue) != 2:
+                return common_base
+            return (
+                common_base
+                + """
+                AND NOT EXISTS (
+                    SELECT 1
+                    FROM memory_task_inputs AS inputs, json_each(inputs.refs_json) AS ref
+                    JOIN memory_write_job_evidence AS linked
+                      ON linked.evidence_id=json_extract(ref.value, '$.evidence_id')
+                    JOIN memory_write_jobs AS job ON job.job_id=linked.job_id
+                    WHERE inputs.task_id=task.task_id
+                      AND job.state IN (
+                          'pending', 'running', 'unknown',
+                          'waiting_for_permission', 'waiting_for_model'
+                      )
+                )
+            """
+            )
+
         def run(connection: sqlite3.Connection, *, defer: bool) -> list[sqlite3.Row]:
+            common = common_for(connection)
             in_window = (
                 "1"
                 if extraction_since is None
@@ -1430,6 +1480,7 @@ class MemoryTaskStore:
         def has_defer_candidate(connection: sqlite3.Connection) -> bool:
             if extraction_since is None:
                 return False
+            common = common_for(connection)
             row = connection.execute(
                 f"""
                 SELECT 1
@@ -2146,6 +2197,7 @@ class MemoryTaskStore:
         *,
         limit: int = 128,
         after_cursor: tuple[str, str] | None = None,
+        exclude_error_codes: tuple[str, ...] = (),
     ) -> tuple[MemoryTask, ...]:
         """List `unknown` tasks that still need an external inquiry.
 
@@ -2160,6 +2212,9 @@ class MemoryTaskStore:
         """
         if type(limit) is not int or limit < 1:
             raise ValueError("limit must be positive")
+        excluded = tuple(dict.fromkeys(exclude_error_codes))
+        if not all(isinstance(code, str) and code for code in excluded):
+            raise ValueError("exclude_error_codes must contain non-empty strings")
         if after_cursor is not None:
             if (
                 not isinstance(after_cursor, tuple)
@@ -2173,13 +2228,19 @@ class MemoryTaskStore:
         else:
             cursor_clause = ""
             params = ()
+        if excluded:
+            placeholders = ",".join("?" for _ in excluded)
+            excluded_clause = f"AND COALESCE(last_error_code, '') NOT IN ({placeholders})"
+        else:
+            excluded_clause = ""
         with self._connect() as connection:
             rows = connection.execute(
                 "SELECT * FROM memory_tasks "
                 "WHERE status=? "
                 f"{cursor_clause} "
+                f"{excluded_clause} "
                 "ORDER BY updated_at, task_id LIMIT ?",
-                (MemoryTaskStatus.UNKNOWN, *params, limit),
+                (MemoryTaskStatus.UNKNOWN, *params, *excluded, limit),
             ).fetchall()
         return tuple(self._to_task(row) for row in rows)
 
@@ -2211,20 +2272,26 @@ class MemoryTaskStore:
         """
         now = datetime.now(UTC).isoformat()
         with self._connect(writing=True) as connection:
+            task_only = _task_only_unknown_clause(connection)
+            blocked = ",".join("?" for _ in NON_RETRYABLE_POST_WRITE_CODES)
             changed = connection.execute(
-                """
+                f"""
                 UPDATE memory_tasks
                 SET status=?, runner_token=NULL, attempts=0,
                     last_error_code='model_content_manual_retry_requested', updated_at=?
                 WHERE task_id IN (
                     SELECT task_id FROM memory_tasks
-                    WHERE status=? ORDER BY updated_at, task_id LIMIT ?
+                    WHERE status=?
+                      AND COALESCE(last_error_code, '') NOT IN ({blocked})
+                      {task_only}
+                    ORDER BY updated_at, task_id LIMIT ?
                 )
                 """,
                 (
                     MemoryTaskStatus.PENDING,
                     now,
                     MemoryTaskStatus.UNKNOWN,
+                    *NON_RETRYABLE_POST_WRITE_CODES,
                     limit,
                 ),
             )
@@ -2246,14 +2313,18 @@ class MemoryTaskStore:
             raise ValueError("limit must be positive")
         now = datetime.now(UTC).isoformat()
         with self._connect(writing=True) as connection:
+            task_only = _task_only_unknown_clause(connection)
+            blocked = ",".join("?" for _ in NON_RETRYABLE_POST_WRITE_CODES)
             changed = connection.execute(
-                """
+                f"""
                 UPDATE memory_tasks
                 SET status=?, runner_token=NULL,
                     last_error_code='model_content_unknown_auto_retry', updated_at=?
                 WHERE task_id IN (
                     SELECT task_id FROM memory_tasks
                     WHERE status=? AND attempts=1 AND updated_at <= ?
+                      AND COALESCE(last_error_code, '') NOT IN ({blocked})
+                      {task_only}
                     ORDER BY updated_at, task_id LIMIT ?
                 )
                 """,
@@ -2262,6 +2333,7 @@ class MemoryTaskStore:
                     now,
                     MemoryTaskStatus.UNKNOWN,
                     older_than.isoformat(),
+                    *NON_RETRYABLE_POST_WRITE_CODES,
                     limit,
                 ),
             )

memhub/apps/core/tests/test_memory_processing.py

@@ -4,7 +4,7 @@ import logging
 import sqlite3
 import threading
 import time
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 
 import pytest
 
@@ -56,7 +56,12 @@ from one_memory.memory_processing import (
 )
 from one_memory.memory_queue import classify
 from one_memory.memory_server.config_state import ConfigStore
-from one_memory.memory_tasks import MemoryTaskKind, MemoryTaskStatus, MemoryTaskStore
+from one_memory.memory_tasks import (
+    NON_RETRYABLE_POST_WRITE_CODES,
+    MemoryTaskKind,
+    MemoryTaskStatus,
+    MemoryTaskStore,
+)
 from one_memory.model_profiles import (
     ModelCapabilities,
     ModelProfileInput,
@@ -2613,6 +2618,123 @@ def test_an_unknown_task_without_inputs_does_not_end_the_reconcile_pass(tmp_path
     assert [memory.memory_id for memory in governance.list(user_id="local")] == ["hy-recovered"]
 
 
+def test_sticky_authority_unknowns_do_not_starve_ordinary_reconciliation(tmp_path) -> None:
+    stores = _setup(tmp_path)
+    evidence, profiles, tasks, governance, secrets, profile = stores
+    record = _put(
+        evidence,
+        message_id="fair-unknown",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="ordinary unknown remains reconcilable",
+    )
+    sticky = [
+        tasks.create(
+            request_id=f"sticky-authority-{index}",
+            user_id=record.user_id,
+            kind=MemoryTaskKind.EXTRACT,
+            input_ref=record.evidence_id,
+            profile=profile,
+            input_records=(record,),
+        )
+        for index in range(129)
+    ]
+    ordinary = tasks.create(
+        request_id="ordinary-unknown-after-sticky",
+        user_id=record.user_id,
+        kind=MemoryTaskKind.EXTRACT,
+        input_ref=record.evidence_id,
+        profile=profile,
+        input_records=(record,),
+    )
+    anchor = datetime(2026, 1, 1, tzinfo=UTC)
+    with sqlite3.connect(tasks.database_path) as connection, connection:
+        connection.executescript(
+            """
+            CREATE TABLE source_grants (
+                user_id TEXT NOT NULL, source_id TEXT NOT NULL,
+                root_path TEXT NOT NULL, candidate_kind TEXT NOT NULL,
+                granted_at TEXT NOT NULL, revoked_at TEXT,
+                PRIMARY KEY (user_id, source_id)
+            );
+            """
+        )
+        connection.execute(
+            "INSERT INTO source_grants VALUES (?,?,?,?,?,NULL)",
+            (
+                record.user_id,
+                record.source_id,
+                "/authorized",
+                "default",
+                datetime.now(UTC).isoformat(),
+            ),
+        )
+        for index, task in enumerate(sticky):
+            connection.execute(
+                "UPDATE memory_tasks SET status='unknown',last_error_code=?,updated_at=? "
+                "WHERE task_id=?",
+                (
+                    "hy_memory_write_verification_authority_invalid",
+                    (anchor + timedelta(seconds=index)).isoformat(),
+                    task.task_id,
+                ),
+            )
+        connection.execute(
+            "UPDATE memory_tasks SET status='unknown',last_error_code=?,updated_at=? "
+            "WHERE task_id=?",
+            (
+                "model_content_outcome_unknown",
+                (anchor + timedelta(minutes=3)).isoformat(),
+                ordinary.task_id,
+            ),
+        )
+
+    class InquiryOnlyBridge(FakeBridge):
+        def __init__(self) -> None:
+            super().__init__(
+                outputs=(
+                    HyMemoryOutput(
+                        memory_id="ordinary-recovered",
+                        content="ordinary unknown remains reconcilable",
+                        layer="L2_FACT",
+                        owner="user",
+                        evidence_ids=(record.evidence_id,),
+                    ),
+                )
+            )
+            self.inquiries: list[str] = []
+
+        async def outputs_for_request(self, **kwargs) -> tuple[HyMemoryOutput, ...]:
+            self.inquiries.append(str(kwargs["request_id"]))
+            return self.outputs or ()
+
+    bridge = InquiryOnlyBridge()
+    supervisor = MemoryProcessingSupervisor(
+        tasks=tasks,
+        profiles=profiles,
+        secrets=secrets,
+        evidence=evidence,
+        governance=governance,
+        data_dir=tmp_path,
+        runner_factory=lambda: _runner(tmp_path, stores, FakeBridge()),
+        bridge_factory=lambda _profile, _secret, _data_dir: bridge,
+    )
+
+    assert (
+        asyncio.run(
+            supervisor.reconcile_unknown_outputs(
+                limit=128,
+                exclude_error_codes=NON_RETRYABLE_POST_WRITE_CODES,
+            )
+        )
+        == 1
+    )
+    assert bridge.inquiries == ["ordinary-unknown-after-sticky"]
+    assert bridge.requests == []
+    assert tasks.require(ordinary.task_id).status is MemoryTaskStatus.COMPLETED
+    assert all(tasks.require(task.task_id).status is MemoryTaskStatus.UNKNOWN for task in sticky)
+
+
 def test_two_workspaces_with_one_conversation_id_prepare_apart(tmp_path) -> None:
     """Cursor can reuse a conversation id across projects.
 
@@ -3441,6 +3563,58 @@ def test_digest_scopes_separate_applications_and_dates(tmp_path) -> None:
     assert len({_hy_agent_id((record,)) for record in records}) == 3
 
 
+def test_workspace_requests_carry_conversation_authority_beyond_agent_scope(tmp_path) -> None:
+    """Same-day workspace scopes collide by design; session_id must separate writes."""
+    evidence, _profiles, tasks, _governance, _secrets, profile = _setup(tmp_path)
+    first = _put(
+        evidence,
+        message_id="conversation-a",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="same workspace, first conversation",
+        conversation_id="conversation-a",
+        workspace="repo-a",
+        created_at=datetime(2026, 10, 18, 8, tzinfo=UTC),
+    )
+    second = _put(
+        evidence,
+        message_id="conversation-b",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="same workspace, second conversation",
+        conversation_id="conversation-b",
+        workspace="repo-a",
+        created_at=datetime(2026, 10, 18, 9, tzinfo=UTC),
+    )
+    next_day = _put(
+        evidence,
+        message_id="conversation-a-next-day",
+        role="user",
+        written_by=WrittenBy.PERSON,
+        content="same conversation on the next day",
+        conversation_id="conversation-a",
+        workspace="repo-a",
+        created_at=datetime(2026, 10, 19, 8, tzinfo=UTC),
+    )
+
+    requests = []
+    for record in (first, second, next_day):
+        task = tasks.create(
+            request_id=f"authority:{record.evidence_id}",
+            user_id=record.user_id,
+            kind=MemoryTaskKind.EXTRACT,
+            input_ref=record.evidence_id,
+            profile=profile,
+            input_records=(record,),
+        )
+        requests.append(MemoryTaskRunner._request(task, (record,)))
+
+    assert requests[0].agent_id == requests[1].agent_id
+    assert requests[0].session_id == "conversation-a"
+    assert requests[1].session_id == "conversation-b"
+    assert requests[2].agent_id != requests[0].agent_id
+
+
 def test_cancelled_multi_source_lock_acquisition_releases_earlier_source(tmp_path) -> None:
     evidence, profiles, tasks, governance, secrets, _profile = _setup(tmp_path)
     first = _put(

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -3516,6 +3516,41 @@ def test_unknown_jobs_does_not_resolve_or_make_a_row_claimable(tmp_path) -> None
     assert queue.resolve_unknown(found[0].job_id, found=False, detail="not in HY") == DEAD
 
 
+def test_unknown_jobs_can_exclude_sticky_authority_failures_before_limit(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    sticky = [
+        queue.submit(_request(key=f"sticky-{index}", evidence=(f"sticky-e{index}",)))
+        for index in range(129)
+    ]
+    ordinary = queue.submit(_request(key="ordinary", evidence=("ordinary-e",)))
+    anchor = datetime(2026, 1, 1, tzinfo=UTC)
+    with sqlite3.connect(path) as connection, connection:
+        for index, record in enumerate(sticky):
+            connection.execute(
+                "UPDATE memory_write_jobs SET state='unknown',attempts=1,hy_invoked=1,"
+                "last_error=?,updated_at=? WHERE job_id=?",
+                (
+                    "hy_memory_write_verification_authority_invalid",
+                    (anchor + timedelta(seconds=index)).isoformat(),
+                    record.job_id,
+                ),
+            )
+        connection.execute(
+            "UPDATE memory_write_jobs SET state='unknown',attempts=1,hy_invoked=1,"
+            "last_error='model_content_outcome_unknown',updated_at=? WHERE job_id=?",
+            ((anchor + timedelta(minutes=3)).isoformat(), ordinary.job_id),
+        )
+
+    found = queue.unknown_jobs(
+        limit=128,
+        exclude_error_codes=("hy_memory_write_verification_authority_invalid",),
+    )
+
+    assert [job.job_id for job in found] == [ordinary.job_id]
+    assert all(queue.get(record.request_id).state == UNKNOWN for record in sticky)
+
+
 def test_unknown_jobs_tolerates_a_row_resolved_between_pages(tmp_path) -> None:
     """Discovery is read-only and may overlap an inquiry from another worker.
 
@@ -4896,15 +4931,16 @@ def test_a_failure_after_the_call_still_parks_as_unknown(tmp_path) -> None:
     record = queue.submit(_request(key="after-call", source="opencode"))
     claimer = DurableWriteDrain(queue, execute=_never_execute, owner="after-call-owner")
     assert claimer.claim(limit=1) == (record.job_id,)
+    recorder = _Recorder(
+        error=HyMemoryBridgeError("hy_memory_write_verification_authority_invalid"),
+        invoke_hy=True,
+    )
 
     asyncio.run(
         execute_job(
             record.job_id,
             queue=queue,
-            supervisor=_Recorder(
-                error=HyMemoryBridgeError("hy_memory_write_verification_authority_invalid"),
-                invoke_hy=True,
-            ),
+            supervisor=recorder,
             evidence=_EvidenceStore({"e1"}),
             owner="after-call-owner",
         )
@@ -4918,6 +4954,16 @@ def test_a_failure_after_the_call_still_parks_as_unknown(tmp_path) -> None:
 
     assert state == UNKNOWN
     assert invoked == 1
+    assert len(recorder.seen) == 1
+    assert (
+        queue.retry_first_unknown_after_reconcile(
+            older_than=datetime.now(UTC) + timedelta(minutes=1)
+        )
+        == 0
+    )
+    assert queue.retry_unknown_confirmed(record.job_id) == 0
+    assert DurableWriteDrain(queue, execute=_never_execute).claim(limit=1) == ()
+    assert len(recorder.seen) == 1
 
 
 def test_event_time_not_enqueue_order_drives_the_window(

memhub/apps/core/tests/test_memory_tasks.py

@@ -1113,6 +1113,109 @@ def test_task_only_unknown_is_automatically_retried_only_once(tmp_path) -> None:
     assert store.require(task.task_id).status is MemoryTaskStatus.UNKNOWN
 
 
+def test_authority_invalid_task_is_never_reinvoked(tmp_path) -> None:
+    store, profile = _profile_and_tasks(tmp_path)
+    task = store.create(
+        request_id="authority-invalid",
+        user_id="local",
+        kind=MemoryTaskKind.EXTRACT,
+        input_ref="authority-evidence",
+        profile=profile,
+    )
+    with sqlite3.connect(store.database_path) as connection, connection:
+        connection.execute(
+            "UPDATE memory_tasks SET status=?,attempts=1,last_error_code=?,updated_at=? "
+            "WHERE task_id=?",
+            (
+                MemoryTaskStatus.UNKNOWN,
+                "hy_memory_write_verification_authority_invalid",
+                "2026-01-01T00:00:00+00:00",
+                task.task_id,
+            ),
+        )
+
+    cutoff = datetime(2026, 1, 1, 0, 5, tzinfo=UTC)
+    assert store.retry_first_unknown_after_reconcile(older_than=cutoff) == 0
+    assert store.retry_unknown_confirmed() == 0
+    retained = store.require(task.task_id)
+    assert retained.status is MemoryTaskStatus.UNKNOWN
+    assert retained.attempts == 1
+    assert retained.last_error_code == "hy_memory_write_verification_authority_invalid"
+
+
+def test_linked_unknown_is_retried_only_by_the_durable_job(tmp_path) -> None:
+    from one_memory.memory_server.contracts import BatchIdentity, WriteRequest
+    from one_memory.memory_server.durable_queue import DurableWriteQueue
+
+    store, profile = _profile_and_tasks(tmp_path)
+    evidence = EvidenceStore(store.database_path)
+    record = evidence.put(
+        EvidenceInput(
+            source_id="cursor",
+            conversation_id="linked-conversation",
+            message_id="linked-message",
+            role="user",
+            written_by=WrittenBy.PERSON,
+            verbatim="linked unknown",
+            created_at=datetime(2026, 1, 1, tzinfo=UTC),
+            origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+        )
+    ).record
+    task = store.create(
+        request_id="linked-task",
+        user_id="local",
+        kind=MemoryTaskKind.EXTRACT,
+        input_ref=record.evidence_id,
+        profile=profile,
+        input_records=(record,),
+    )
+    queue = DurableWriteQueue(store.database_path, evidence=evidence)
+    job = queue.submit(
+        WriteRequest(
+            internal_session_id="linked-session",
+            batch_identity=BatchIdentity(
+                internal_session_id="linked-session",
+                first_round=1,
+                last_round=1,
+            ),
+            evidence_ids=[record.evidence_id],
+            idempotency_key="linked-job",
+            source_id="cursor",
+        )
+    )
+    with sqlite3.connect(store.database_path) as connection, connection:
+        connection.execute(
+            "UPDATE memory_tasks SET status=?,attempts=1,last_error_code=?,updated_at=? "
+            "WHERE task_id=?",
+            (
+                MemoryTaskStatus.UNKNOWN,
+                "model_content_outcome_unknown",
+                "2026-01-01T00:00:00+00:00",
+                task.task_id,
+            ),
+        )
+        connection.execute(
+            "UPDATE memory_write_jobs SET state='unknown',attempts=1,hy_invoked=1,"
+            "last_error='model_content_outcome_unknown',updated_at=? WHERE job_id=?",
+            ("2026-01-01T00:00:00+00:00", job.job_id),
+        )
+
+    cutoff = datetime(2026, 1, 1, 0, 5, tzinfo=UTC)
+    assert store.retry_first_unknown_after_reconcile(older_than=cutoff) == 0
+    assert store.require(task.task_id).status is MemoryTaskStatus.UNKNOWN
+    assert queue.retry_first_unknown_after_reconcile(older_than=cutoff) == 1
+    assert store.require(task.task_id).status is MemoryTaskStatus.PENDING
+    assert queue.get(job.request_id).state == "pending"
+    assert (
+        store.active_window(
+            now=datetime(2026, 1, 2, tzinfo=UTC),
+            fresh_window=timedelta(days=1),
+            extraction_window=None,
+        )
+        == []
+    )
+
+
 def test_orphan_recovery_preserves_active_running_tasks(tmp_path) -> None:
     store, profile = _profile_and_tasks(tmp_path)
     active = store.create(

tests/coding/test_fact_reconciler.py

@@ -107,6 +107,7 @@ class TestSupersede:
     async def _seed_old(self, store, *, derived=None):
         old = WorkingFact(
             fact_id="f_old", user_id="u1",
+            session_id="test-session",
             statement=(
                 "用户要修复 CI 的 npm run build；检查 .github/workflows/ci.yml 后"
                 "定位到 Node 18 报 ERR_OSSL_EVP_UNSUPPORTED，尚未修改。"
@@ -129,7 +130,7 @@ class TestSupersede:
             [_draft(
                 "用户继续修复同一 CI 构建；把 .github/workflows/ci.yml 升到 Node 20，"
                 "npm run build 已通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         new_id = facts[0].fact_id
@@ -152,7 +153,7 @@ class TestSupersede:
             [_draft(
                 "用户继续修复同一 CI 构建；把 .github/workflows/ci.yml 升到 Node 20，"
                 "npm run build 已通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
         new_id = facts[0].fact_id
 
@@ -184,7 +185,9 @@ class TestSupersede:
                 coding_store, mock_embed,
                 _llm([{"draft_idx": 0, "supersedes": [target or prev_id]}]),
             )
-            facts, _ = await rec.reconcile([_draft(stmt)], user_id="u1")
+            facts, _ = await rec.reconcile(
+                [_draft(stmt)], user_id="u1", session_id="test-session"
+            )
             prev_id = facts[0].fact_id
 
         lineage = await coding_store.get_fact_ancestors(prev_id, user_id="u1")
@@ -208,6 +211,7 @@ class TestSupersede:
                 "配置完成但尚未发布。"
             )],
             user_id="u1",
+            session_id="test-session",
         )
 
         prompt = llm.calls[0]["prompt"]
@@ -234,7 +238,9 @@ class TestSupersede:
             coding_store, mock_embed, _llm([{"draft_idx": 0, "supersedes": ["f_old"]}]),
         )
         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",
         )
 
         # 拿旧说法原文去搜，只能落到新事实（靠两者本身相似），
@@ -256,6 +262,34 @@ class TestSupersede:
         old = await coding_store.get_fact("f_old", user_id="u1")
         assert old.status == "active"
 
+    async def test_near_duplicate_fact_cannot_cross_authority_scope(
+        self, coding_store, mock_embed
+    ):
+        await coding_store.insert_fact(WorkingFact(
+            fact_id="f_old",
+            user_id="u1",
+            agent_id="scope:2026-10-18",
+            session_id="conversation-a",
+            workspace_id="repo-a",
+            statement="CI build now passes after upgrading Node",
+        ))
+        llm = _llm([{"draft_idx": 0, "supersedes": ["f_old"]}])
+        rec = WorkingFactReconciler(coding_store, mock_embed, llm)
+
+        _, ops = await rec.reconcile(
+            [_draft("CI build now passes after upgrading Node")],
+            user_id="u1",
+            agent_id="scope:2026-10-19",
+            session_id="conversation-b",
+            workspace_id="repo-a",
+        )
+
+        old = await coding_store.get_fact("f_old", user_id="u1")
+        assert old is not None
+        assert old.status == "active"
+        assert [op.action for op in ops] == ["INSERT"]
+        assert len(llm.calls) == 0
+
     async def test_old_fact_superseded_only_once(self, coding_store, mock_embed):
         """两条新事实都声称覆盖同一条旧事实时，只让第一条生效。
 
@@ -274,6 +308,7 @@ class TestSupersede:
                 _draft("config 位于 /srv/app/config.yaml"),
             ],
             user_id="u1",
+            session_id="test-session",
         )
         assert len(facts) == 2
         assert sum(1 for o in ops if o.action == "SUPERSEDE") == 1
@@ -288,6 +323,7 @@ class TestChainSummary:
     async def _seed_old(self, store):
         await store.insert_fact(WorkingFact(
             fact_id="f_old", user_id="u1",
+            session_id="test-session",
             statement=(
                 "用户要迁移 app config；确认当前读取 /opt/app/config.yaml，"
                 "尚未修改配置加载逻辑。"
@@ -310,7 +346,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         stored = await coding_store.get_fact(facts[0].fact_id, user_id="u1")
@@ -326,7 +362,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         assert len(llm.calls) == 2, "SUPERSEDE 应触发第二次调用"
@@ -344,7 +380,11 @@ class TestChainSummary:
         llm = _llm([])
         rec = WorkingFactReconciler(coding_store, mock_embed, llm)
 
-        await rec.reconcile([_draft("另一个话题的结论")], user_id="u1")
+        await rec.reconcile(
+            [_draft("另一个话题的结论")],
+            user_id="u1",
+            session_id="test-session",
+        )
         assert len(llm.calls) == 1
 
     async def test_prompt_carries_full_lineage(self, coding_store, mock_embed):
@@ -357,7 +397,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         llm2 = _llm([{"draft_idx": 0, "supersedes": [facts[0].fact_id]}])
@@ -366,7 +406,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续完善同一 config 迁移；改由环境变量 APP_CONFIG 指定路径，"
                 "回归测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         prompt = llm2.calls[1]["prompt"]
@@ -393,7 +433,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         assert sum(1 for o in ops if o.action == "SUPERSEDE") == 1
@@ -426,7 +466,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
         assert facts[0].fact_chain_summary == ""
 
@@ -436,7 +476,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续完善同一 config 迁移；改由环境变量 APP_CONFIG 指定路径，"
                 "回归测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         prompt = llm2.calls[1]["prompt"]
@@ -450,7 +490,8 @@ class TestChainSummary:
         total = MAX_FACT_CHAIN_ENTRIES + 3
         for i in range(total):
             await coding_store.insert_fact(WorkingFact(
-                fact_id=f"g{i}", user_id="u1", statement=f"第 {i} 代结论",
+                fact_id=f"g{i}", user_id="u1", session_id="test-session",
+                statement=f"第 {i} 代结论",
                 created_at=datetime(2026, 1, i + 1),
             ))
             if prev:
@@ -459,7 +500,9 @@ class TestChainSummary:
 
         llm = _llm([{"draft_idx": 0, "supersedes": [prev]}])
         rec = WorkingFactReconciler(coding_store, mock_embed, llm)
-        await rec.reconcile([_draft("最新结论")], user_id="u1")
+        await rec.reconcile(
+            [_draft("最新结论")], user_id="u1", session_id="test-session"
+        )
 
         prompt = llm.calls[1]["prompt"]
         assert "第 0 代结论" not in prompt, "最老的几代该被截掉"
@@ -478,7 +521,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
         entries = llm.calls[1]["prompt"].split(
             "ENTRIES\n═══════════════════════════════════════════════════════════════════\n",
@@ -497,7 +540,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         rec2 = WorkingFactReconciler(
@@ -509,7 +552,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续完善同一 config 迁移；改由环境变量 APP_CONFIG 指定路径，"
                 "回归测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         stored = await coding_store.get_fact(facts2[0].fact_id, user_id="u1")
@@ -526,7 +569,7 @@ class TestChainSummary:
             [_draft(
                 "用户继续迁移 app config；将加载路径改为 /etc/app/config.yaml，"
                 "配置加载测试通过。"
-            )], user_id="u1",
+            )], user_id="u1", session_id="test-session",
         )
 
         assert len(llm.calls) == 1
@@ -564,7 +607,8 @@ 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", statement="config 位于 /opt/app/config.yaml",
+            fact_id="f_old", user_id="u1", session_id="test-session",
+            statement="config 位于 /opt/app/config.yaml",
         ))
         rec = WorkingFactReconciler(coding_store, mock_embed, FailingLLMProvider())
 
@@ -578,7 +622,8 @@ class TestFailSafe:
 
     async def test_unparsable_llm_output(self, coding_store, mock_embed):
         await coding_store.insert_fact(WorkingFact(
-            fact_id="f_old", user_id="u1", statement="config 位于 /opt/app/config.yaml",
+            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 的一段话")
@@ -592,7 +637,8 @@ class TestFailSafe:
 
     async def test_markdown_fence_parsed(self, coding_store, mock_embed):
         await coding_store.insert_fact(WorkingFact(
-            fact_id="f_old", user_id="u1", statement="config 位于 /opt/app/config.yaml",
+            fact_id="f_old", user_id="u1", session_id="test-session",
+            statement="config 位于 /opt/app/config.yaml",
         ))
         llm = MockLLMProvider([
             "```json\n" + _ops_resp([{"draft_idx": 0, "supersedes": ["f_old"]}]) + "\n```",
@@ -601,7 +647,9 @@ class TestFailSafe:
         rec = WorkingFactReconciler(coding_store, mock_embed, llm)
 
         _, 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 sum(1 for o in ops if o.action == "SUPERSEDE") == 1
 
@@ -618,7 +666,8 @@ class TestSubjectGateSwitch:
 
     async def _seed(self, store, fact_id, subject, statement):
         await store.insert_fact(WorkingFact(
-            fact_id=fact_id, user_id="u1", subject=subject, statement=statement,
+            fact_id=fact_id, user_id="u1", session_id="test-session",
+            subject=subject, statement=statement,
         ))
 
     async def test_gate_on_blocks_cross_subject_supersede(self, coding_store, mock_embed):
@@ -632,6 +681,7 @@ class TestSubjectGateSwitch:
         _, ops = await rec.reconcile(
             [WorkingFactDraft(subject="Bart", statement="Bart 的三日行程已排定，预算待定。")],
             user_id="u1",
+            session_id="test-session",
         )
         assert [o.action for o in ops] == ["INSERT"]
         assert (await coding_store.get_fact("f_emma", user_id="u1")).status == "active"
@@ -647,6 +697,7 @@ class TestSubjectGateSwitch:
         facts, ops = await rec.reconcile(
             [WorkingFactDraft(subject="Bart", statement="Bart 的三日行程已排定，预算待定。")],
             user_id="u1",
+            session_id="test-session",
         )
         assert [o.action for o in ops] == ["SUPERSEDE", "INSERT"]
         old = await coding_store.get_fact("f_emma", user_id="u1")
@@ -667,7 +718,8 @@ class TestSubjectGateSwitch:
             [WorkingFactDraft(subject="Bart", statement="Bart 的行程。")],
             [[old]],
             {0: ["f_emma"]},
-            user_id="u1", agent_id="a1", session_id=None, workspace_id=None,
+            user_id="u1", agent_id="default_agent",
+            session_id="test-session", workspace_id=None,
         )
         assert "SUPERSEDE" in [o.action for o in ops]
 
@@ -687,7 +739,8 @@ class TestSubjectGateSwitch:
                 [WorkingFactDraft(subject="deploy.sh", statement="新的部署脚本结论。")],
                 [[old]],
                 {0: [f"f_noneed_{gate}"]},
-                user_id="u1", agent_id="a1", session_id=None, workspace_id=None,
+                user_id="u1", agent_id="default_agent",
+                session_id="test-session", workspace_id=None,
             )
             assert "SUPERSEDE" in [o.action for o in ops], f"gate={gate} 时空 subject 被误拦"
 
@@ -714,6 +767,7 @@ class TestDetailsInheritance:
     async def test_ancestor_details_copied_verbatim(self, coding_store, mock_embed):
         await coding_store.insert_fact(WorkingFact(
             fact_id="f_old", user_id="u1", subject="Emma",
+            session_id="test-session",
             statement="Emma 的行程草案已出，预算待核。",
             details="日均餐费 $32.4-$57.6\n酒店评分 3.2-3.8",
         ))
@@ -726,6 +780,7 @@ class TestDetailsInheritance:
             [WorkingFactDraft(subject="Emma", statement="Emma 的行程已定稿。",
                               details="航班 CA1234")],
             user_id="u1",
+            session_id="test-session",
         )
         stored = await coding_store.get_fact(facts[0].fact_id, user_id="u1")
         # 本代的 details 在，祖先的具体取值也一字不差地在
@@ -742,6 +797,7 @@ class TestDetailsInheritance:
         """
         await coding_store.insert_fact(WorkingFact(
             fact_id="f_g1", user_id="u1", subject="T",
+            session_id="test-session",
             statement="第一代结论。", details="V1=" + "a" * 300,
         ))
         prev = "f_g1"
@@ -756,6 +812,7 @@ class TestDetailsInheritance:
                 [WorkingFactDraft(subject="T", statement=f"第{gen}代结论。",
                                   details=f"V{gen}=" + "a" * 300)],
                 user_id="u1",
+                session_id="test-session",
             )
             prev = facts[0].fact_id
             stored = await coding_store.get_fact(prev, user_id="u1")
@@ -772,6 +829,7 @@ class TestDetailsInheritance:
         """预算是硬的，但被丢掉的代数必须写出来，不静默截断。"""
         await coding_store.insert_fact(WorkingFact(
             fact_id="f_old", user_id="u1", subject="T",
+            session_id="test-session",
             statement="旧结论。", details="X=" + "b" * 8000,
         ))
         rec = WorkingFactReconciler(
@@ -782,6 +840,7 @@ class TestDetailsInheritance:
         facts, _ = await rec.reconcile(
             [WorkingFactDraft(subject="T", statement="新结论。", details="Y=1")],
             user_id="u1",
+            session_id="test-session",
         )
         stored = await coding_store.get_fact(facts[0].fact_id, user_id="u1")
         assert "…(truncated)" in stored.details
@@ -790,6 +849,7 @@ class TestDetailsInheritance:
     async def test_switch_off_keeps_details_untouched(self, coding_store, mock_embed):
         await coding_store.insert_fact(WorkingFact(
             fact_id="f_old", user_id="u1", subject="T",
+            session_id="test-session",
             statement="旧结论。", details="祖先取值 Z=9",
         ))
         rec = WorkingFactReconciler(
@@ -800,6 +860,7 @@ class TestDetailsInheritance:
         facts, ops = await rec.reconcile(
             [WorkingFactDraft(subject="T", statement="新结论。", details="Y=1")],
             user_id="u1",
+            session_id="test-session",
         )
         # 覆盖照样发生，只是不过继
         assert "SUPERSEDE" in [o.action for o in ops]
@@ -814,6 +875,7 @@ class TestDetailsInheritance:
         """
         await coding_store.insert_fact(WorkingFact(
             fact_id="f_old", user_id="u1", subject="T",
+            session_id="test-session",
             statement="旧结论。", details="无关噪声 " + "q" * 2000,
         ))
         rec = WorkingFactReconciler(
@@ -824,6 +886,7 @@ class TestDetailsInheritance:
         facts, _ = await rec.reconcile(
             [WorkingFactDraft(subject="T", statement="新结论。", details="Y=1")],
             user_id="u1",
+            session_id="test-session",
         )
         new_id = facts[0].fact_id
         emb = await coding_store.embed_service.embed("T 新结论。")

tests/coding/test_reconciler.py

@@ -35,7 +35,8 @@ class TestReconcilerActions:
     async def test_update(self, coding_store, mock_embed):
         # 种入 m1
         await coding_store.insert(CodingMemory(
-            experience_id="m1", user_id="u1", task="T1",
+            experience_id="m1", user_id="u1", agent_id="scope-day-1",
+            workspace_id="repo-a", session_id="conversation-1", task="T1",
             search_keys=["old"], solution="old",
         ))
         llm = MockLLMProvider(_llm_resp([
@@ -44,23 +45,146 @@ class TestReconcilerActions:
         ]))
         rec = CodingMemoryReconciler(coding_store, mock_embed, llm)
         d = CodingMemoryDraft(task="T1", search_keys=["new1"], solution="new",
-                              user_id="u1")
+                              user_id="u1", agent_id="scope-day-1",
+                              workspace_id="repo-a", session_id="conversation-1")
         ops = await rec.reconcile([d])
         assert len(ops) == 1 and ops[0].action == "UPDATE"
         got = await coding_store.get_by_id("m1", "u1")
         assert got.solution == "new"
         assert got.search_keys == ["new1"]
 
+    @pytest.mark.parametrize(
+        (
+            "old_agent",
+            "old_workspace",
+            "old_session",
+            "new_agent",
+            "new_workspace",
+            "new_session",
+        ),
+        [
+            (
+                "scope:same-day",
+                "repo-a",
+                "conversation-a",
+                "scope:same-day",
+                "repo-a",
+                "conversation-b",
+            ),
+            (
+                "scope:2026-10-18",
+                "repo-a",
+                "conversation-a",
+                "scope:2026-10-19",
+                "repo-a",
+                "conversation-a",
+            ),
+            (
+                "shared-scope",
+                "repo-a",
+                "conversation-a",
+                "shared-scope",
+                "repo-b",
+                "conversation-a",
+            ),
+        ],
+    )
+    async def test_update_target_must_match_frozen_authority_scope(
+        self,
+        coding_store,
+        mock_embed,
+        old_agent,
+        old_workspace,
+        old_session,
+        new_agent,
+        new_workspace,
+        new_session,
+    ):
+        """Near duplicates cannot cross conversation, date, or workspace authority."""
+        await coding_store.insert(CodingMemory(
+            experience_id="old-target", user_id="u1", agent_id=old_agent,
+            workspace_id=old_workspace, session_id=old_session,
+            task="fix flaky release", solution="old solution",
+        ))
+        llm = MockLLMProvider(_llm_resp([
+            {
+                "draft_idx": 0,
+                "action": "UPDATE",
+                "target_experience_id": "old-target",
+                "reason": "near duplicate",
+            },
+        ]))
+        rec = CodingMemoryReconciler(coding_store, mock_embed, llm)
+        draft = CodingMemoryDraft(
+            task="fix flaky release",
+            solution="new solution",
+            user_id="u1",
+            agent_id=new_agent,
+            workspace_id=new_workspace,
+            session_id=new_session,
+        )
+
+        ops = await rec.reconcile([draft])
+
+        old = await coding_store.get_by_id("old-target", "u1")
+        assert old is not None
+        assert old.solution == "old solution"
+        assert old.agent_id == old_agent
+        assert old.workspace_id == old_workspace
+        assert old.session_id == old_session
+        assert len(ops) == 1
+        assert ops[0].action == "ADD"
+        assert ops[0].target_experience_id != "old-target"
+
+    @pytest.mark.parametrize("legacy_agent", ["", "default_agent", "scope:legacy"])
+    async def test_legacy_target_without_conversation_falls_back_to_add(
+        self, coding_store, mock_embed, legacy_agent
+    ):
+        await coding_store.insert(CodingMemory(
+            experience_id="legacy-target",
+            user_id="u1",
+            agent_id=legacy_agent,
+            session_id=None,
+            workspace_id="repo-a",
+            task="upgrade dependency",
+            solution="legacy solution",
+        ))
+        llm = MockLLMProvider(_llm_resp([
+            {
+                "draft_idx": 0,
+                "action": "UPDATE",
+                "target_experience_id": "legacy-target",
+            },
+        ]))
+        rec = CodingMemoryReconciler(coding_store, mock_embed, llm)
+        draft = CodingMemoryDraft(
+            task="upgrade dependency",
+            solution="current solution",
+            user_id="u1",
+            agent_id=legacy_agent,
+            session_id="current-conversation",
+            workspace_id="repo-a",
+        )
+
+        ops = await rec.reconcile([draft])
+
+        legacy = await coding_store.get_by_id("legacy-target", "u1")
+        assert legacy is not None
+        assert legacy.solution == "legacy solution"
+        assert legacy.session_id is None
+        assert [op.action for op in ops] == ["ADD"]
+
     async def test_skip(self, coding_store, mock_embed):
         await coding_store.insert(CodingMemory(
             experience_id="m1", user_id="u1", task="X", solution="Y",
+            session_id="conversation-1",
         ))
         llm = MockLLMProvider(_llm_resp([
             {"draft_idx": 0, "action": "SKIP", "reason": "duplicate"},
         ]))
         rec = CodingMemoryReconciler(coding_store, mock_embed, llm)
         d = CodingMemoryDraft(task="X", solution="Y",
-                              user_id="u1")
+                              user_id="u1", session_id="conversation-1")
         ops = await rec.reconcile([d])
         assert len(ops) == 1 and ops[0].action == "SKIP"
         # store 不变
@@ -71,7 +195,7 @@ class TestReconcilerActions:
         # 旧记忆要被 DELETE
         await coding_store.insert(CodingMemory(
             experience_id="m_dep", user_id="u1", task="use old.sh",
-            solution="old script",
+            solution="old script", session_id="conversation-1",
         ))
         llm = MockLLMProvider(_llm_resp([
             {"draft_idx": 0, "action": "ADD"},
@@ -80,7 +204,7 @@ class TestReconcilerActions:
         ]))
         rec = CodingMemoryReconciler(coding_store, mock_embed, llm)
         d = CodingMemoryDraft(task="use new CLI", solution="new way",
-                              user_id="u1")
+                              user_id="u1", session_id="conversation-1")
         ops = await rec.reconcile([d])
         assert sorted(op.action for op in ops) == ["ADD", "DELETE"]
         # 旧的没了

tests/coding/test_store.py

@@ -67,7 +67,8 @@ class TestStoreCRUD:
 
     async def test_update_overrides(self, coding_store, mock_embed):
         m = CodingMemory(experience_id="e1", user_id="u1", task="T",
-                         search_keys=["old1", "old2"], solution="old")
+                         search_keys=["old1", "old2"], solution="old",
+                         session_id="conversation-1")
         await coding_store.insert(m)
 
         # 旧 aspect 应可命中
@@ -90,6 +91,59 @@ class TestStoreCRUD:
         assert got.solution == "new"
         assert got.search_keys == ["new1"]
 
+    async def test_update_rejects_authority_scope_change(self, coding_store):
+        await coding_store.insert(CodingMemory(
+            experience_id="e1", user_id="u1", agent_id="scope-a",
+            workspace_id="repo-a", session_id="conversation-1",
+            task="T", solution="old",
+        ))
+        replacement = CodingMemory(
+            experience_id="e1", user_id="u1", agent_id="scope-b",
+            workspace_id="repo-a", session_id="conversation-1",
+            task="T", solution="new",
+        )
+
+        with pytest.raises(ValueError, match="coding_memory_update_authority_mismatch"):
+            await coding_store.update(replacement)
+
+        stored = await coding_store.get_by_id("e1", user_id="u1")
+        assert stored is not None
+        assert stored.agent_id == "scope-a"
+        assert stored.solution == "old"
+
+    async def test_update_rejects_conversation_change_and_legacy_empty_session(
+        self, coding_store
+    ):
+        for experience_id, old_session in (
+            ("different-conversation", "conversation-1"),
+            ("legacy-empty-session", None),
+        ):
+            await coding_store.insert(CodingMemory(
+                experience_id=experience_id,
+                user_id="u1",
+                agent_id="scope-a",
+                workspace_id="repo-a",
+                session_id=old_session,
+                task="T",
+                solution="old",
+            ))
+            replacement = CodingMemory(
+                experience_id=experience_id,
+                user_id="u1",
+                agent_id="scope-a",
+                workspace_id="repo-a",
+                session_id="conversation-2",
+                task="T",
+                solution="new",
+            )
+
+            with pytest.raises(ValueError, match="coding_memory_update_authority_mismatch"):
+                await coding_store.update(replacement)
+
+            stored = await coding_store.get_by_id(experience_id, user_id="u1")
+            assert stored is not None
+            assert stored.solution == "old"
+
     async def test_delete(self, coding_store):
         m = CodingMemory(experience_id="e1", user_id="u1", task="T",
                          search_keys=["k"], solution="S")

tests/coding/test_writer_e2e.py

@@ -335,7 +335,7 @@ class TestDualLayerWrite:
         from hy_memory.coding.types import CodingMemory
         await coding_store.insert(CodingMemory(
             experience_id="e1", user_id="u1", task="老 task", solution="旧 solution",
-            source_fact_ids=["f_before"],
+            source_fact_ids=["f_before"], session_id="test-session",
         ))
         llm = MockLLMProvider([
             _extract_resp(
@@ -348,7 +348,9 @@ class TestDualLayerWrite:
         ])
         writer = self._writer(coding_store, mock_embed, llm)
 
-        resp = await writer.write(_coding_messages(), user_id="u1")
+        resp = await writer.write(
+            _coding_messages(), user_id="u1", session_id="test-session"
+        )
         exp = await coding_store.get_by_id("e1", user_id="u1")
         assert exp.solution == "新 solution"
         assert exp.source_fact_ids == ["f_before", resp["fact_ids"][0]]