diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md
index e0e7c7bc48..ecda2c6ddf 100644
--- a/loopx/capabilities/manager_context/README.md
+++ b/loopx/capabilities/manager_context/README.md
@@ -182,6 +182,16 @@ Goal visibility are checked before and after reading. Original-Chat delivery
 and explicit peer-return consumption remain different receipt paths. The list
 does not mark reads, accept work, start an Agent or certify an overall outcome.
 Unreadable/missing sources and incomplete scans keep counts visibly partial.
+Within one inventory read, requests sharing an exact Goal lifetime reuse its
+held guard in batches of at most one page. Every request retains its own lock,
+source/Turn read, grant and typed receipt/lifecycle checks; each new batch
+observes the current Goal lifetime. Core linked-work IO runs after those guards
+are released. This reduces repeated fence IO without caching authorization
+between requests or pages; pagination and search still scan the bounded history.
+The pure conversation-identity projection is reused only while the complete
+host Session facts are identical within that read. Changed facts are classified
+again; live provider, grant, receipt and closing-source checks are not cached.
+
 When the typed lifecycle proves a matching original source/route but finds a
 conflicting handoff receipt, coverage includes at most 12 diagnostic
 request/Goal/Agent identities. This is a refusal readback, not a readable brief,
@@ -197,6 +207,12 @@ checks or requests. Custom scope resolvers keep their existing validation path.
 管家标题栏提供常驻委派入口，按接收方查看待跟进请求，并可展开当前交办内容、关联
 工作和原对话。同一来源的较早会话不会因重开会话而丢失；其它用户、来源、群聊和
 项目会话不混入。结论已回传与成果已验收不是同一事实，历史记录仍可回看。
+同一次读取中，相同 Goal 实例按至多一页大小复用持有中的生命周期锁；逐请求锁、
+来源与 Turn 读回、授权及 typed 回执检查保留，每批重新观察当前实例。关联 Core 工作
+在释放这些锁后读取，不跨请求或分页缓存授权；分页与搜索仍扫描有界历史。
+同一次读取中，完整宿主 Session 字段相同时复用纯身份分类；字段变化即重新分类，
+provider、授权、回执及结束时的来源检查仍实时执行。
+
 来源与原路匹配但发送回执冲突时，保留不完整计数，以警示小符号按需提供至多 12 条
 具体请求身份和原因；不展示未获读取资格的交办正文，不推定已取消或完成。刷新重新
 核验原 owner，不修改历史回执，也不解除执行或返回权限；其它读取缺口继续保持未知。
diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py
index f24c334285..40d7f1bd8d 100644
--- a/loopx/control_plane/collaboration/conversation_inventory.py
+++ b/loopx/control_plane/collaboration/conversation_inventory.py
@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+from contextlib import ExitStack, contextmanager
 from datetime import datetime
 import hashlib
 import json
@@ -32,73 +33,47 @@ def _last_event(values):
     return max(dated, default=(None, None), key=lambda pair: pair[0])[1]
 
 
-def read_conversation_requests(root, registry, store, session, *, goal_ids, core_goal_ids,
-                               validate_source_session, history=False, offset=0, limit=12, search=""):
-    """Read only this host's exact source Session, with fresh caller read scope.
+@contextmanager
+def _locked_request_batch(root, scope, requests):
+    """Retain each request version for one page-sized typed observation."""
+    available, lock_keys = set(), set()
+    with ExitStack() as locks:
+        for path, _, request_id in sorted(requests, key=lambda item: (item[2], str(item[0]))):
+            # Exact requests share their guard across legacy entry locations.
+            # Legacy records still use their individual entry lock.
+            key = (request_id, None if scope.exact else path)
+            if key not in lock_keys:
+                try:
+                    locks.enter_context(_request_lock(root, request_id, scope, path.with_suffix(".lock")))
+                except (OSError, ValueError, KeyError, TypeError):
+                    continue
+                lock_keys.add(key)
+            available.add(path)
+        yield available
 
-    The caller supplies provider-validated Goal visibility and rechecks it after
-    this read. A model/caller-supplied channel, path or source id is never used.
-    """
-    if type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= 12:
-        raise ValueError("invalid request page")
-    if not isinstance(search, str) or len(search) > 200:
-        raise ValueError("request search must be at most 200 characters")
-    query = search.strip().casefold()
-    paths = sorted((_root(root) / "entries").glob("*/*.json"))
-    rows, unreadable, cache, turn_readers, diagnostics = [], 0, {}, {}, []
-    for path in paths[:2000]:
+
+def _read_receipt_batch(root, rows):
+    if not rows:
+        return []
+    try:
+        observed = read_inbox_transaction(root, {"operation": "receipt_states", "rows": rows, "include_agenda": True})
+        states = list(zip(observed["items"], observed["agenda_items"], strict=True))
+        if len(states) != len(rows):
+            raise ValueError("Inbox receipt count does not match requests")
+        return states
+    except (OSError, ValueError, KeyError, TypeError):
+        # A failed batch is only a read. Retry its rows independently while
+        # all their guards remain held; one unavailable record must not hide
+        # its readable neighbors or become an empty successful observation.
+        return [None] if len(rows) == 1 else [_read_receipt_batch(root, [row])[0] for row in rows]
+
+
+def _project_request_observations(root, registry, observations, *, core_goal_ids, history):
+    """Combine captured receipt facts with current linked Core observations."""
+    rows, unreadable, cache = [], 0, {}
+    for row, source_session_id, turn, state, agenda, controlled, read, read_error, decision, decision_error, replies in observations:
         try:
-            candidate = _read(path)
-            if candidate.get("source_kind") == "peer":
-                continue
-            if candidate.get("source_channel") not in {None, session["channel_id"]}:
-                continue
-            if not isinstance(candidate.get("request_id"), str) or not BARE_SHA256_PATTERN.fullmatch(candidate["request_id"]):
-                raise ValueError("invalid request identity")
-            route = _read(_root(root) / "roundtrips" / (candidate["request_id"] + ".json"))
-            if route.get("channel_id") != session["channel_id"]:
-                continue
-            if not is_source_chat_host(root, route, store):
-                raise ValueError("original source host changed")
-            source_session = store.load_session(route["session_id"])
-            if not source_session or source_session.get("channel_id") != session["channel_id"]:
-                raise ValueError("original source Session unavailable")
-            validate_source_session(source_session)
-            gid, aid = route["goal_id"], route["agent_id"]
-            if goal_ids is not None and gid not in goal_ids:
-                continue
-            sid = source_session["session_id"]
-            if sid not in turn_readers:
-                turn_readers[sid] = original_turn_reader(store, sid)
-            turn = turn_readers[sid](route["client_turn_id"])
-            if turn is None:
-                raise ValueError("original source Turn unavailable")
-            with collaboration_goal_scope(registry, goal_id=gid, agents=(), caller_goal_ref=route.get("goal_ref")) as scope:
-                with _request_lock(root, candidate["request_id"], scope, path.with_suffix(".lock")):
-                    row = _entry(root, gid, aid, candidate["request_id"], scope=scope)
-                    if row.get("source_kind") == "peer":
-                        continue
-                    grant = source_context_target_authority(root, source_session, turn, {"goal_id": gid, "agent_id": aid})
-                    try:
-                        decide_collaboration_lifecycle(scope, operation="original_request_inspect", record=row, route=route,
-                            initial_delivery=original_delivery_facts(row, route, turn, source_id=grant["source_id"]))
-                    except CollaborationLifecycleRejected as exc:
-                        # Only this typed refusal proves the route and source
-                        # matched before the conflicting receipt was observed.
-                        # Withhold the brief, work and completion state as before.
-                        if exc.decision.get("delivery_proof_reason") == "handoff_receipt_conflict" and len(diagnostics) < 12:
-                            diagnostics.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": aid,
-                                "code": exc.decision["code"], "reason": exc.decision["delivery_proof_reason"]})
-                        raise
-                    observed = read_inbox_transaction(root, {"operation": "receipt_states", "rows": [row], "include_agenda": True})
-                    state, agenda = observed["items"][0], observed["agenda_items"][0]
-                    controlled = state.get("readback")
-                    read, read_error = _receipt(root, "reads", row)
-                    decision, decision_error = _receipt(root, "decisions", row)
-                    replies = reply_status(root, row)
-            # Request/return observations are version-consistent. Core work is
-            # a separate current observation; never hold its request lock over
-            # another authority's IO.
+            gid, aid = row["goal_id"], row["agent_id"]
             work = read_linked_work(root, registry, row, cache,
                 include_core_details=core_goal_ids is None or gid in core_goal_ids)
             warnings = list(state.get("warnings", [])) + work["warnings"]
@@ -122,11 +97,140 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core
                     "unavailable" if read_error or control_unavailable else "supplied" if read else "pending",
                 "decision": assessment["status"], "decision_reason": assessment.get("reason"),
                 "returns": replies, "linked_todos": work["linked_todos"], "evidence_refs": work["evidence_refs"],
-                "warnings": warnings, "source_session_id": source_session["session_id"], "source_turn_id": turn["turn_id"],
+                "warnings": warnings, "source_session_id": source_session_id, "source_turn_id": turn["turn_id"],
                 "last_event_at": _last_event([row.get("delivered_at"), read_at, assessment.get("at"),
                     *(reply.get(key) for reply in replies for key in ("created_at", "delivered_at"))])})
         except (OSError, ValueError, KeyError, TypeError):
             unreadable += 1
+    return rows, unreadable
+
+
+def _request_groups(root, store, session, goal_ids, validate_source_session):
+    """Discover source-validated routes before acquiring any Goal guard."""
+    paths = sorted((_root(root) / "entries").glob("*/*.json"))
+    unreadable, groups = 0, {}
+    for path in paths[:2000]:
+        try:
+            candidate = _read(path)
+            if candidate.get("source_kind") == "peer":
+                continue
+            if candidate.get("source_channel") not in {None, session["channel_id"]}:
+                continue
+            if not isinstance(candidate.get("request_id"), str) or not BARE_SHA256_PATTERN.fullmatch(candidate["request_id"]):
+                raise ValueError("invalid request identity")
+            route = _read(_root(root) / "roundtrips" / (candidate["request_id"] + ".json"))
+            if route.get("channel_id") != session["channel_id"]:
+                continue
+            if not is_source_chat_host(root, route, store):
+                raise ValueError("original source host changed")
+            source_session = store.load_session(route["session_id"])
+            if not source_session or source_session.get("channel_id") != session["channel_id"]:
+                raise ValueError("original source Session unavailable")
+            validate_source_session(source_session)
+            gid, aid = route["goal_id"], route["agent_id"]
+            if goal_ids is not None and gid not in goal_ids:
+                continue
+            if not isinstance(gid, str) or not isinstance(aid, str):
+                raise ValueError("invalid request target")
+            key = (gid, json.dumps(route.get("goal_ref"), sort_keys=True))
+            groups.setdefault(key, []).append((path, route, candidate["request_id"]))
+        except (OSError, ValueError, KeyError, TypeError):
+            unreadable += 1
+    return groups, unreadable, len(paths) <= 2000
+
+
+def read_conversation_requests(root, registry, store, session, *, goal_ids, core_goal_ids,
+                               validate_source_session, history=False, offset=0, limit=12, search=""):
+    """Read only this host's exact source Session, with fresh caller read scope.
+
+    The caller supplies provider-validated Goal visibility and rechecks it after
+    this read. A model/caller-supplied channel, path or source id is never used.
+    """
+    if type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= 12:
+        raise ValueError("invalid request page")
+    if not isinstance(search, str) or len(search) > 200:
+        raise ValueError("request search must be at most 200 characters")
+    query = search.strip().casefold()
+    groups, unreadable, scan_complete = _request_groups(root, store, session, goal_ids, validate_source_session)
+    turn_readers, diagnostics, work_reads = {}, [], []
+    # Reuse only the held lifetime guard, never an authorization or request
+    # snapshot. Resolve sources outside it first: their provider checks can
+    # read another authority. Each source/Turn and request is reread below.
+    # Keep each guard bounded to a page-sized batch so a long inventory does
+    # not exclude Goal writers for the entire history scan.
+    batches = [(key, items[start:start + limit]) for key, items in groups.items()
+               for start in range(0, len(items), limit)]
+    for (gid, _), requests in batches:
+        start, failed = len(work_reads), 0
+        try:
+            with (
+                collaboration_goal_scope(registry, goal_id=gid, agents=(),
+                                         caller_goal_ref=requests[0][1].get("goal_ref")) as scope,
+                _locked_request_batch(root, scope, requests) as available,
+            ):
+                prepared = []
+                for path, route, request_id in requests:
+                    if path not in available:
+                        failed += 1
+                        continue
+                    try:
+                        source_session = store.load_session(route["session_id"])
+                        if not source_session or source_session.get("channel_id") != session["channel_id"]:
+                            raise ValueError("original source Session unavailable")
+                        validate_source_session(source_session)
+                        sid = source_session["session_id"]
+                        if sid not in turn_readers:
+                            turn_readers[sid] = original_turn_reader(store, sid)
+                        turn = turn_readers[sid](route["client_turn_id"])
+                        if turn is None:
+                            raise ValueError("original source Turn unavailable")
+                        row = _entry(root, gid, route["agent_id"], request_id, scope=scope)
+                        if row.get("source_kind") == "peer":
+                            continue
+                        grant = source_context_target_authority(root, source_session, turn,
+                            {"goal_id": gid, "agent_id": route["agent_id"]})
+                        try:
+                            decide_collaboration_lifecycle(scope, operation="original_request_inspect", record=row, route=route,
+                                initial_delivery=original_delivery_facts(row, route, turn, source_id=grant["source_id"]))
+                        except CollaborationLifecycleRejected as exc:
+                            # Only this typed refusal proves the route and source
+                            # matched before the conflicting receipt was observed.
+                            # Withhold the brief, work and completion state as before.
+                            if exc.decision.get("delivery_proof_reason") == "handoff_receipt_conflict" and len(diagnostics) < 12:
+                                diagnostics.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": row["agent_id"],
+                                    "code": exc.decision["code"], "reason": exc.decision["delivery_proof_reason"]})
+                            raise
+                        # Do not retain full model responses while waiting for
+                        # linked-work IO; only the displayed question is used.
+                        source_turn = {"turn_id": turn["turn_id"], "message": turn.get("message", "")}
+                        prepared.append((row, source_session["session_id"], source_turn))
+                    except (OSError, ValueError, KeyError, TypeError):
+                        failed += 1
+                observations = _read_receipt_batch(root, [row for row, _, _ in prepared])
+                for (row, source_session_id, source_turn), observed in zip(prepared, observations, strict=True):
+                    if observed is None:
+                        failed += 1
+                        continue
+                    try:
+                        state, agenda = observed
+                        controlled = state.get("readback")
+                        read, read_error = _receipt(root, "reads", row)
+                        decision, decision_error = _receipt(root, "decisions", row)
+                        replies = reply_status(root, row)
+                        work_reads.append((row, source_session_id, source_turn, state, agenda, controlled,
+                                           read, read_error, decision, decision_error, replies))
+                    except (OSError, ValueError, KeyError, TypeError):
+                        failed += 1
+        except (OSError, ValueError, KeyError, TypeError):
+            del work_reads[start:]
+            unreadable += len(requests)
+        else:
+            unreadable += failed
+    # Request/return observations are version-consistent. Core work is a
+    # separate observation: release all lifetime/request guards before its IO.
+    rows, projection_failures = _project_request_observations(root, registry, work_reads,
+        core_goal_ids=core_goal_ids, history=history)
+    unreadable += projection_failures
     rows = list({row["request_id"]: row for row in rows}.values())
     pending_count = sum(row["unfinished"] for row in rows)
     visible = [row for row in rows if history or row["unfinished"]]
@@ -138,7 +242,7 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core
             (row.get("brief") or {}).get("purpose"), (row.get("brief") or {}).get("context")))]
     visible.sort(key=lambda row: (row.get("last_event_at") or "", row["request_id"]), reverse=True)
     page = visible[offset:offset + limit]
-    coverage = {"scan_complete": len(paths) <= 2000, "unreadable": unreadable, "diagnostics": diagnostics}
+    coverage = {"scan_complete": scan_complete, "unreadable": unreadable, "diagnostics": diagnostics}
     # A page fence over observations, not an atomic snapshot or a new ledger.
     snapshot_digest = hashlib.sha256(json.dumps([visible, coverage, query], sort_keys=True, ensure_ascii=False).encode()).hexdigest()
     return {"ok": True, "session_id": session["session_id"], "rows": page,
diff --git a/loopx/presentation/request_control_api.py b/loopx/presentation/request_control_api.py
index be6c3aa2fc..3ed04c12e5 100644
--- a/loopx/presentation/request_control_api.py
+++ b/loopx/presentation/request_control_api.py
@@ -2,6 +2,8 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+from copy import deepcopy
+import json
 from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qs, urlparse
@@ -116,8 +118,18 @@ def _request_inventory_read(self) -> None:
                 raise ValueError("invalid request inventory view")
             session, goals, core = self._request_inventory_scope(query["session_id"][0])
             controller = self.server.runtime_controller
+            source_bases = {}
             def source_basis(source):
-                return conversation_scope(source), source.get("steward_context")
+                # Reuse only the pure identity projection for identical host
+                # facts in this read. Keep all Session fields in the key so
+                # changed or newly supported inputs reach the typed owner.
+                sid = source["session_id"]
+                signature = json.dumps(source, sort_keys=True, separators=(",", ":"))
+                previous = source_bases.get(sid)
+                if previous is None or previous[0] != signature:
+                    previous = (signature, (conversation_scope(source), deepcopy(source.get("steward_context"))))
+                    source_bases[sid] = previous
+                return previous[1]
             # Provider identity belongs to a source, not to every historical row.
             # Keep this proof within one read and fence every observed source
             # again before returning any content. Receipt/grant IO stays live.
diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py
index 7282fb9d07..0938f6cb1d 100644
--- a/tests/control_plane/test_chat_request_inventory.py
+++ b/tests/control_plane/test_chat_request_inventory.py
@@ -1,6 +1,8 @@
 """Original steward requests, not inferred worker activity or a second ledger."""
 from urllib.parse import urlencode
 
+import pytest
+
 from test_chat_goal_amendment import api as api
 from test_goal_amendment_authority import source_goal as source_goal
 from test_chat_request_control import _source
@@ -229,6 +231,205 @@ def observe_scope(reader, session_id):
     assert "rows" not in denied and "Check the current contract" not in str(denied)
 
 
+def test_inventory_bounds_lifetime_guards_and_keeps_core_io_outside_them(api, monkeypatch):
+    from contextlib import contextmanager
+    from loopx.control_plane.collaboration import conversation_inventory as inventory
+    from loopx.presentation import request_control_api
+
+    _, _, route = _source(api)
+    _, f, server = api
+    session = server.chat_store.load_session(route["session_id"])
+    for number in range(12):
+        turn, _ = server.chat_store.create_turn(session["session_id"], client_turn_id=f"batch-{number}",
+            message=f"Check artifact {number}", origin="web")
+        delivered = deliver(f["runtime"], f["registry"], session=session, turn=turn,
+            request={"goal_id": f["ref"]["goal_id"], "agent_id": "agent-a"}, source_store=server.chat_store)
+        server.chat_store.update_turn(session["session_id"], turn["turn_id"], status="completed",
+            response={"context_handoff_receipt": delivered})
+        server.chat_store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+    original_scope, original_work = inventory.collaboration_goal_scope, inventory.read_linked_work
+    original_lock, original_receipts = inventory._request_lock, inventory.read_inbox_transaction
+    original_classify = request_control_api.conversation_scope
+    classifications = []
+
+    def classify(source):
+        classifications.append(source["session_id"])
+        return original_classify(source)
+
+    monkeypatch.setattr(request_control_api, "conversation_scope", classify)
+    held, batches, locked, receipt_batches = [], [], set(), []
+
+    @contextmanager
+    def scope(*args, **kwargs):
+        with original_scope(*args, **kwargs) as current:
+            held.append(current)
+            batches.append(0)
+            try:
+                yield current
+            finally:
+                held.pop()
+
+    original_entry = inventory._entry
+    def entry(*args, **kwargs):
+        assert held, "A request must retain the Goal lifetime guard"
+        batches[-1] += 1
+        return original_entry(*args, **kwargs)
+
+    def work(*args, **kwargs):
+        assert not held, "Core IO must not nest under a Goal lifetime guard"
+        return original_work(*args, **kwargs)
+
+    @contextmanager
+    def lock(root, request_id, *args, **kwargs):
+        with original_lock(root, request_id, *args, **kwargs):
+            locked.add(request_id)
+            try:
+                yield
+            finally:
+                locked.remove(request_id)
+
+    def receipts(root, request, **kwargs):
+        ids = {row["request_id"] for row in request["rows"]}
+        assert held and ids <= locked, "Receipt versions need every corresponding request lock"
+        receipt_batches.append(len(ids))
+        return original_receipts(root, request, **kwargs)
+
+    monkeypatch.setattr(inventory, "collaboration_goal_scope", scope)
+    monkeypatch.setattr(inventory, "_entry", entry)
+    monkeypatch.setattr(inventory, "read_linked_work", work)
+    monkeypatch.setattr(inventory, "_request_lock", lock)
+    monkeypatch.setattr(inventory, "read_inbox_transaction", receipts)
+    code, value = _inventory(api, route["session_id"], view="history")
+    assert code == 200 and value["matched"] == value["pending_count"] == 13
+    assert value["coverage"] == {"scan_complete": True, "unreadable": 0, "diagnostics": []}
+    assert sum(batches) == 13 and len(batches) == 2 and max(batches) <= 12
+    assert sum(receipt_batches) == 13 and len(receipt_batches) == 2 and max(receipt_batches) <= 12
+    assert not locked, "A completed read must release every request lock"
+    # Two fresh boundaries and at most one additional pure projection for the
+    # unchanged source, independently of row count and on each HTTP read.
+    assert set(classifications) == {session["session_id"]} and 2 <= len(classifications) <= 3
+    first_count = len(classifications)
+    assert _inventory(api, route["session_id"], view="history")[0] == 200
+    assert 2 <= len(classifications) - first_count <= 3
+
+
+@pytest.mark.parametrize("failure", ["request_lock", "receipt"])
+def test_inventory_isolates_one_failed_request_and_rereads_on_recovery(api, monkeypatch, failure):
+    from contextlib import contextmanager
+    from loopx.control_plane.collaboration import conversation_inventory as inventory
+
+    _, _, route = _source(api)
+    _, f, server = api
+    store = server.chat_store
+    session = store.load_session(route["session_id"])
+    expected = {route["request_id"]: store.load_turn(route["session_id"], route["turn_id"])["message"]}
+    for number in range(2):
+        message = f"Review artifact {number}\n" + "Retain the complete source question.\n" * 80
+        turn, _ = store.create_turn(session["session_id"], client_turn_id=f"isolated-{number}", message=message, origin="web")
+        receipt = deliver(f["runtime"], f["registry"], session=session, turn=turn,
+            request={"goal_id": f["ref"]["goal_id"], "agent_id": "agent-b"}, source_store=store)
+        store.update_turn(session["session_id"], turn["turn_id"], status="completed", response={"context_handoff_receipt": receipt})
+        store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+        expected[receipt["request_id"]] = message
+    broken = route["request_id"]
+    original_lock, original_receipts = inventory._request_lock, inventory.read_inbox_transaction
+
+    @contextmanager
+    def lock(root, request_id, *args, **kwargs):
+        if request_id == broken:
+            raise OSError("request lock unavailable")
+        with original_lock(root, request_id, *args, **kwargs):
+            yield
+
+    def receipts(root, request, **kwargs):
+        if any(row["request_id"] == broken for row in request["rows"]):
+            raise OSError("request receipt unavailable")
+        return original_receipts(root, request, **kwargs)
+
+    monkeypatch.setattr(inventory, "_request_lock" if failure == "request_lock" else "read_inbox_transaction",
+        lock if failure == "request_lock" else receipts)
+    saved = {path: path.read_bytes() for path in _root(f["runtime"]).rglob("*.json")}
+    code, partial = _inventory(api, session["session_id"], view="history")
+    assert code == 200 and partial["matched"] == partial["pending_count"] == 2
+    assert partial["coverage"] == {"scan_complete": True, "unreadable": 1, "diagnostics": []}
+    assert {row["request_id"]: row["source_question"] for row in partial["rows"]} == {
+        key: value for key, value in expected.items() if key != broken}
+    assert {path: path.read_bytes() for path in saved} == saved
+    monkeypatch.setattr(inventory, "_request_lock", original_lock)
+    monkeypatch.setattr(inventory, "read_inbox_transaction", original_receipts)
+    recovered = _inventory(api, session["session_id"], view="history")[1]
+    assert recovered["coverage"]["unreadable"] == 0 and recovered["matched"] == 3
+    assert {row["request_id"]: row["source_question"] for row in recovered["rows"]} == expected
+
+
+def test_inventory_reclassifies_changed_facts_for_the_same_source_session(api, monkeypatch):
+    _, _, route = _source(api)
+    _, _, server = api
+    original_load = server.chat_store.load_session
+    reads = []
+
+    def load(session_id):
+        session = original_load(session_id)
+        if session_id == route["session_id"]:
+            reads.append(session_id)
+            if len(reads) == 3:
+                # The source changed after discovery, then recovered before
+                # the closing read. Its row must still withhold the content.
+                return {**session, "goal_id": None}
+        return session
+
+    monkeypatch.setattr(server.chat_store, "load_session", load)
+    code, value = _inventory(api, route["session_id"], view="history")
+    assert code == 200 and len(reads) >= 4
+    assert value["rows"] == [] and value["pending_count"] == 0
+    assert value["coverage"] == {"scan_complete": True, "unreadable": 1, "diagnostics": []}
+
+
+def test_inventory_rereads_goal_lifetime_between_batches(api, monkeypatch):
+    from contextlib import contextmanager
+    from loopx.control_plane.collaboration import conversation_inventory as inventory
+    from loopx.control_plane.goals.source_session_recreation import RecreateGoalRequest, recreate_goal_instance
+
+    _, _, route = _source(api)
+    _, f, server = api
+    session = server.chat_store.load_session(route["session_id"])
+    for number in range(12):
+        turn, _ = server.chat_store.create_turn(session["session_id"], client_turn_id=f"replace-{number}",
+            message=f"Check artifact {number}", origin="web")
+        delivered = deliver(f["runtime"], f["registry"], session=session, turn=turn,
+            request={"goal_id": f["ref"]["goal_id"], "agent_id": "agent-a"}, source_store=server.chat_store)
+        server.chat_store.update_turn(session["session_id"], turn["turn_id"], status="completed",
+            response={"context_handoff_receipt": delivered})
+        server.chat_store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+    original_scope = inventory.collaboration_goal_scope
+    original_decide = inventory.decide_collaboration_lifecycle
+    calls, modes = [], []
+
+    @contextmanager
+    def scope(*args, **kwargs):
+        if calls:
+            replaced = recreate_goal_instance(RecreateGoalRequest(registry_path=f["registry"],
+                goal_id=f["ref"]["goal_id"], goal_instance_id=f["ref"]["goal_instance_id"], operation_id="replace-during-read"))
+            assert replaced["ok"]
+        with original_scope(*args, **kwargs) as current:
+            calls.append(current.current_goal_ref)
+            yield current
+
+    def decide(*args, **kwargs):
+        result = original_decide(*args, **kwargs)
+        modes.append(result.get("mode"))
+        return result
+
+    monkeypatch.setattr(inventory, "collaboration_goal_scope", scope)
+    monkeypatch.setattr(inventory, "decide_collaboration_lifecycle", decide)
+    code, value = _inventory(api, route["session_id"], view="history")
+    assert code == 200 and len(calls) == 2
+    assert calls[0] == f["ref"] and calls[1] != f["ref"]
+    # The existing typed owner permits proven historical inspection after a
+    # replacement; it must observe the new lifetime rather than reuse the old.
+    assert modes == ["current_instance"] * 12 + ["historical_read"]
+    assert value["matched"] == 13 and value["coverage"] == {"scan_complete": True, "unreadable": 0, "diagnostics": []}
+
 def test_conflicting_receipt_is_diagnosable_but_does_not_authorize_content_or_controls(api):
     control, _, route = _source(api)
     _, f, server = api