diff --git a/batch_runner.py b/batch_runner.py
index 48551d0e2c99f..4c947831dcd44 100644
--- a/batch_runner.py
+++ b/batch_runner.py
@@ -490,13 +490,13 @@ def _load_dataset(self) -> List[Dict[str, Any]]:
 
                 try:
                     entry = json.loads(line)
-                    if 'prompt' not in entry:
-                        print(f"⚠️  Warning: Line {line_num} missing 'prompt' field, skipping")
-                        continue
-                    dataset.append(entry)
                 except json.JSONDecodeError as e:
                     print(f"⚠️  Warning: Invalid JSON on line {line_num}: {e}")
                     continue
+                if not isinstance(entry, dict) or 'prompt' not in entry:
+                    print(f"⚠️  Warning: Line {line_num} missing 'prompt' field, skipping")
+                    continue
+                dataset.append(entry)
 
         if not dataset:
             raise ValueError(f"No valid entries found in dataset file: {self.dataset_file}")
@@ -552,7 +552,7 @@ def _scan_completed_prompts_by_content(self) -> set:
                     for line in f:
                         try:
                             entry = json.loads(line.strip())
-                            if entry.get("failed", False):
+                            if not isinstance(entry, dict) or entry.get("failed", False):
                                 continue
                             prompt_text = _entry_prompt_text(entry)
                             if prompt_text:
@@ -722,6 +722,9 @@ def _combine_batch_files(self) -> Tuple[int, int]:
                         try:
                             data = json.loads(line)
 
+                            if not isinstance(data, dict):
+                                filtered_entries += 1
+                                continue
                             if data.get("discarded"):
                                 tombstone_entries += 1
                                 continue
diff --git a/cron/bot_chat_delivery.py b/cron/bot_chat_delivery.py
index 90d080378472e..f9f9735ad5fe9 100644
--- a/cron/bot_chat_delivery.py
+++ b/cron/bot_chat_delivery.py
@@ -37,6 +37,8 @@ def _records(root: Path) -> list[tuple[Path, dict]]:
     for path in root.glob("*.json"):
         try:
             record = json.loads(path.read_text(encoding="utf-8"))
+            if not isinstance(record, dict):
+                raise ValueError(f"expected a JSON object, got {type(record).__name__}")
         except (OSError, ValueError) as exc:  # ValueError: corrupt JSON and invalid UTF-8 alike
             # Keep damaged or unreadable receipts as evidence; never replay them or block peers
             # (same rule as tools/bot_live_delivery.py::_scan_read — one bad file must not wedge the dir).
@@ -56,7 +58,7 @@ def defer(key: str, job: dict, content: str, profile: str, home: Path) -> dict:
     with _FileLock(root / ".lock"):
         record = read_pending(key)
         if record is not None:
-            if record["content"] != content or record["home"] != str(home):
+            if not isinstance(record, dict) or record["content"] != content or record["home"] != str(home):
                 raise ValueError("delivery id already belongs to a different payload")
             return record
         sequence = max((record["sequence"] for _, record in _records(root)), default=0) + 1
@@ -84,7 +86,7 @@ def _drain(root: Path) -> None:
     for path, _ in records:
         with _FileLock(root / ".lock"):
             record = json.loads(path.read_text(encoding="utf-8"))
-            if record["status"] != "queued":
+            if not isinstance(record, dict) or record["status"] != "queued":
                 continue
             home = Path(record["home"])
             try:
diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py
index dee1741b22506..8daf862adb23f 100644
--- a/cron/scheduler_delivery.py
+++ b/cron/scheduler_delivery.py
@@ -751,7 +751,7 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str, *, deferred: Opt
             if pending is None and find_canonical_live_owner(home) is None and find_canonical_owner(home):
                 pending = defer(key, dict(job), content, profile, home)
             if pending is not None:
-                if pending["content"] != content or pending["home"] != str(home):
+                if not isinstance(pending, dict) or pending["content"] != content or pending["home"] != str(home):
                     raise ValueError("delivery id already belongs to a different payload")
                 status = pending["status"]
                 target = f"bot-chat:{profile_label}"
@@ -763,7 +763,7 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str, *, deferred: Opt
             if owner is not None:
                 receipt = deliver_to_live_owner(home, owner, message, delivery_id=key)
         if receipt is not None:
-            if receipt["message"] != message:
+            if not isinstance(receipt, dict) or receipt["message"] != message:
                 raise ValueError("delivery id already belongs to a different payload")
             status = receipt["status"]
             target = f"bot-chat:{profile_label}"
diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py
index b0af857d89143..beab98ed38b1c 100644
--- a/gateway/shutdown_flush.py
+++ b/gateway/shutdown_flush.py
@@ -149,6 +149,10 @@ def drain_transcript_spool(session_id: str, replay) -> tuple[int, int]:
             payload = json.loads(path.read_text(encoding="utf-8"))
         except Exception:
             continue
+        if not isinstance(payload, dict):
+            logger.warning("Removing structurally invalid transcript spool file %s", path)
+            path.unlink(missing_ok=True)
+            continue
         if (payload.get("reason") != TRANSCRIPT_CAP_DROP_REASON
                 or payload.get("session_key") != session_id):
             continue
@@ -216,7 +220,14 @@ def recover_pending_to_db(session_db=None) -> int:
     recovered = 0
     try:
         for path in flush_files:
-            payload = json.loads(path.read_text(encoding="utf-8"))
+            try:
+                payload = json.loads(path.read_text(encoding="utf-8"))
+                if not isinstance(payload, dict):
+                    raise ValueError(f"expected a JSON object, got {type(payload).__name__}")
+            except (OSError, ValueError) as exc:
+                logger.warning("Cannot recover unreadable flush file %s (%s); "
+                               "preserved for manual inspection", path, exc)
+                continue
             # Agent-history snapshots are for manual operator recovery, not automatic DB insertion.
             if payload.get("reason") == "shutdown-with-unpersisted-agent-history":
                 continue
diff --git a/hermes_cli/local_runtime/binaries.py b/hermes_cli/local_runtime/binaries.py
index c304f06ffa2f3..11844a6dabf19 100644
--- a/hermes_cli/local_runtime/binaries.py
+++ b/hermes_cli/local_runtime/binaries.py
@@ -66,9 +66,10 @@ def runtimes_root() -> Path:
 def manifest_verified(manifest: Path) -> bool:
     """True when an install manifest records a verified_version (missing/damaged -> False)."""
     try:
-        return bool(json.loads(manifest.read_text(encoding="utf-8")).get("verified_version"))
+        data = json.loads(manifest.read_text(encoding="utf-8"))
     except (json.JSONDecodeError, OSError):
         return False
+    return isinstance(data, dict) and bool(data.get("verified_version"))
 
 
 def _release_number(tag: str) -> int:
diff --git a/plugins/platforms/a2a/protocol.py b/plugins/platforms/a2a/protocol.py
index a4cbcfe083f2d..04484dc1765c0 100644
--- a/plugins/platforms/a2a/protocol.py
+++ b/plugins/platforms/a2a/protocol.py
@@ -456,9 +456,11 @@ def load_conversation(context_id: str, limit: int = 50) -> list[dict]:
     for line in lines:
         if line.strip():
             try:
-                out.append(json.loads(line))
+                entry = json.loads(line)
             except json.JSONDecodeError:
-                pass
+                continue
+            if isinstance(entry, dict):
+                out.append(entry)
     return out[-limit:]
 
 
diff --git a/tests/cron/test_bot_chat_pending.py b/tests/cron/test_bot_chat_pending.py
index 3f1aefca7cab8..99f0f8f6845e9 100644
--- a/tests/cron/test_bot_chat_pending.py
+++ b/tests/cron/test_bot_chat_pending.py
@@ -1,6 +1,8 @@
 """Only never-started cron delivery may wait for a CLI owner's release."""
 import importlib.util
+import json
 import subprocess
+import time
 from pathlib import Path
 from unittest.mock import Mock
 
@@ -125,3 +127,50 @@ def test_unreadable_deferred_receipt_does_not_block_siblings(tmp_path, monkeypat
     assert [r for r in caplog.records
             if "Unreadable deferred Bot Chat receipt" in r.message and "Permission denied" in r.message] and \
         sum("Unreadable deferred Bot Chat receipt" in r.message for r in caplog.records) == 1
+
+
+def test_non_dict_deferred_receipt_does_not_block_siblings(tmp_path, monkeypatch, caplog):
+    """A receipt that parses as JSON but is not an object is the same wedge class as an
+    unreadable one: the drain's sequence sort must not crash on it."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    queue.defer("a" * 64, {"id": "job"}, "healthy", "", tmp_path)
+    (queue._root() / f"{'e' * 64}.json").write_text('"just a string"', encoding="utf-8")
+    seen = []
+    monkeypatch.setattr(delivery, "_deliver_to_bot_chat", lambda j, c, p, **kw: seen.append(c))
+    with caplog.at_level("ERROR", logger=queue.logger.name):
+        queue.drain()
+        queue.drain()
+    assert seen == ["healthy"]
+    assert sum("Unreadable deferred Bot Chat receipt" in r.message for r in caplog.records) == 1
+
+
+def test_non_dict_receipt_keeps_sequence_allocation_and_exact_id_fail_closed(tmp_path, monkeypatch):
+    """A non-dict file must not crash defer()'s sequence scan, and an exact-id hit on it
+    fails closed (conflict error, evidence preserved) rather than TypeError or overwrite."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    root = queue._root()
+    root.mkdir(parents=True)
+    bad = root / f"{'e' * 64}.json"
+    bad.write_text("42", encoding="utf-8")
+    record = queue.defer("a" * 64, {"id": "job"}, "healthy", "", tmp_path)
+    assert record["sequence"] == 1
+    with pytest.raises(ValueError, match="different payload"):
+        queue.defer("e" * 64, {"id": "job"}, "anything", "", tmp_path)
+    assert json.loads(bad.read_text(encoding="utf-8")) == 42
+
+
+def test_drain_in_background_survives_non_dict_receipt(tmp_path, monkeypatch):
+    """E2E through the real scheduler entry: drain_in_background spawns the drain
+    thread that used to die in _records' sort before touching any receipt."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    queue.defer("a" * 64, {"id": "job"}, "healthy", "", tmp_path)
+    bad = queue._root() / f"{'e' * 64}.json"
+    bad.write_text("[1, 2, 3]", encoding="utf-8")
+    seen = []
+    monkeypatch.setattr(delivery, "_deliver_to_bot_chat", lambda j, c, p, **kw: seen.append(c))
+    queue.drain_in_background()
+    deadline = time.monotonic() + 10
+    while not seen and time.monotonic() < deadline:
+        time.sleep(0.05)
+    assert seen == ["healthy"]
+    assert bad.exists()  # kept as evidence, never replayed
diff --git a/tests/gateway/test_shutdown_flush.py b/tests/gateway/test_shutdown_flush.py
index 2f79303b57bfa..d5f9145a3f6d3 100644
--- a/tests/gateway/test_shutdown_flush.py
+++ b/tests/gateway/test_shutdown_flush.py
@@ -238,3 +238,52 @@ def test_flushed_overflow_is_replayed_by_recover_pending_to_db(tmp_path, monkeyp
 def test_flush_overflow_noop_on_empty():
     assert flush_overflow_to_file({}) == 0
     assert flush_overflow_to_file({"k": []}) == 0
+
+
+def test_recover_skips_unreadable_and_non_dict_flush_files(tmp_path, monkeypatch):
+    """A corrupt or non-object flush file must not wedge recovery of its siblings:
+    it stays on disk for manual inspection and the healthy payload still replays."""
+    flush_dir = _make_flush_dir(tmp_path)
+    monkeypatch.setattr("gateway.shutdown_flush._get_flush_dir", lambda: flush_dir)
+    good = {
+        "session_key": "agent:main:telegram:dm:1",
+        "reason": "shutdown",
+        "ts": 1,
+        "data": {"text": "recovered", "session_id": "s1"},
+    }
+    good_path = flush_dir / "a_good.json"
+    good_path.write_text(json.dumps(good), encoding="utf-8")
+    non_dict = flush_dir / "b_scalar.json"
+    non_dict.write_text('"oops"', encoding="utf-8")
+    broken = flush_dir / "c_broken.json"
+    broken.write_text("{not json", encoding="utf-8")
+
+    mock_db = MagicMock()
+    assert recover_pending_to_db(session_db=mock_db) == 1
+    mock_db.append_message.assert_called_once()
+    assert not good_path.exists()
+    assert non_dict.exists() and broken.exists()
+
+
+def test_drain_transcript_spool_removes_non_dict_and_replays_valid(tmp_path, monkeypatch):
+    """A non-object spool payload used to crash drain_transcript_spool at
+    payload.get(); it is structurally invalid, so it is removed while the
+    valid sibling still replays."""
+    from gateway.shutdown_flush import (
+        TRANSCRIPT_CAP_DROP_REASON,
+        drain_transcript_spool,
+    )
+    flush_dir = _make_flush_dir(tmp_path)
+    monkeypatch.setattr("gateway.shutdown_flush._get_flush_dir", lambda: flush_dir)
+    (flush_dir / "pending-bad.json").write_text("42", encoding="utf-8")
+    good = flush_dir / "pending-good.json"
+    good.write_text(json.dumps({
+        "session_key": "sess-1", "reason": TRANSCRIPT_CAP_DROP_REASON,
+        "ts": 1, "seq": 1, "data": {"message": {"role": "user", "content": "hi"}},
+    }), encoding="utf-8")
+    replayed = []
+    replayed_n, remaining = drain_transcript_spool("sess-1", replayed.append)
+    assert (replayed_n, remaining) == (1, 0)
+    assert replayed == [{"role": "user", "content": "hi"}]
+    assert not good.exists()
+    assert not (flush_dir / "pending-bad.json").exists()
diff --git a/tests/hermes_cli/test_local_runtime.py b/tests/hermes_cli/test_local_runtime.py
index 93c5930cc6ca4..50fb9753f4b39 100644
--- a/tests/hermes_cli/test_local_runtime.py
+++ b/tests/hermes_cli/test_local_runtime.py
@@ -870,3 +870,16 @@ def boom(*a, **k):
         "hermes_cli.local_runtime.binaries.ensure_runtime_installed", boom)
     result = bootstrap.ensure_local_runtime({"local_runtime": {"enabled": True}})
     assert result is None  # no exception escaped
+
+
+def test_manifest_verified_tolerates_non_dict_manifest(tmp_path):
+    """A parseable-but-non-object manifest used to raise AttributeError out of
+    manifest_verified (the .get ran inside a try that only caught decode/OSError),
+    breaking any() scans over install dirs."""
+    from hermes_cli.local_runtime.binaries import manifest_verified
+
+    m = tmp_path / "manifest.json"
+    m.write_text('"oops"', encoding="utf-8")
+    assert manifest_verified(m) is False
+    m.write_text(json.dumps({"verified_version": "5015 (abc)"}), encoding="utf-8")
+    assert manifest_verified(m) is True
diff --git a/tests/plugins/test_a2a_plugin.py b/tests/plugins/test_a2a_plugin.py
index 959d12caa5f24..87e903e81f276 100644
--- a/tests/plugins/test_a2a_plugin.py
+++ b/tests/plugins/test_a2a_plugin.py
@@ -1746,3 +1746,15 @@ def test_default_profile_unscoped_keeps_env_precedence(
         assert adapter.agent_name == "default-profile-agent"
         assert adapter._agents[""]["description"] == "Default profile's own agent."
         assert adapter._public_url == "https://default-profile.example.com/"
+
+
+def test_load_conversation_skips_non_dict_lines(monkeypatch, tmp_path):
+    """A scalar line in a conversation file must not break replay or pollute
+    the list[dict] contract."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    protocol.persist_message("ctx-mixed", "user", "hello", "t1")
+    path = protocol._conv_path("ctx-mixed")
+    with open(path, "a", encoding="utf-8") as f:
+        f.write("42\n")
+    convo = protocol.load_conversation("ctx-mixed")
+    assert len(convo) == 1 and convo[0]["text"] == "hello"
diff --git a/tests/test_batch_runner_discard_resume.py b/tests/test_batch_runner_discard_resume.py
index 7f8c725e0fadf..a05a48f5fc707 100644
--- a/tests/test_batch_runner_discard_resume.py
+++ b/tests/test_batch_runner_discard_resume.py
@@ -173,3 +173,36 @@ def test_entry_prompt_text_shapes():
     assert _entry_prompt_text({"prompt": "  padded  ", "discarded": "x"}) == "padded"
     assert _entry_prompt_text({}) == ""
     assert _entry_prompt_text("not-a-dict") == ""
+
+
+def test_content_scan_skips_non_dict_lines(tmp_path):
+    """A parseable-but-non-object line used to raise AttributeError at
+    entry.get() and kill the resume scan."""
+    (tmp_path / "batch_1.jsonl").write_text(
+        '"scalar row"\n'
+        + json.dumps({"prompt": "ok q", "completed": True}) + "\n",
+        encoding="utf-8",
+    )
+    assert _scan_runner(tmp_path)._scan_completed_prompts_by_content() == {"ok q"}
+
+
+def test_combine_batch_files_skips_non_dict_lines(tmp_path):
+    (tmp_path / "batch_1.jsonl").write_text(
+        '42\n'
+        + json.dumps({"conversations": [{"from": "human", "value": "kept"}]}) + "\n",
+        encoding="utf-8",
+    )
+    runner = _scan_runner(tmp_path)
+    kept, _found = runner._combine_batch_files()
+    assert kept == 1
+    out = (tmp_path / "trajectories.jsonl").read_text(encoding="utf-8")
+    assert "kept" in out and "42" not in out
+
+
+def test_load_dataset_skips_non_dict_lines(tmp_path):
+    dataset = tmp_path / "dataset.jsonl"
+    dataset.write_text(
+        '"not an entry"\n' + json.dumps({"prompt": "real"}) + "\n", encoding="utf-8")
+    runner = _scan_runner(tmp_path)
+    runner.dataset_file = dataset
+    assert runner._load_dataset() == [{"prompt": "real"}]
diff --git a/tests/test_trajectory_compressor_async.py b/tests/test_trajectory_compressor_async.py
index 89c381b524295..1f52dce999d96 100644
--- a/tests/test_trajectory_compressor_async.py
+++ b/tests/test_trajectory_compressor_async.py
@@ -199,3 +199,14 @@ async def test_generate_summary_async_public_moonshot_cn_kimi_k2_5_omits_tempera
 
     assert result.startswith("[CONTEXT SUMMARY]:")
     assert "temperature" not in async_client.chat.completions.create.call_args.kwargs
+
+
+@pytest.mark.asyncio
+async def test_process_entry_async_passes_non_dict_through():
+    """A scalar JSONL line used to crash on '"conversations" not in entry';
+    unknown shapes pass through byte-faithful."""
+    from trajectory_compressor import TrajectoryCompressor
+
+    compressor = TrajectoryCompressor.__new__(TrajectoryCompressor)
+    entry, metrics = await compressor.process_entry_async(42)
+    assert entry == 42
diff --git a/tests/tools/test_bot_live_owner_delivery.py b/tests/tools/test_bot_live_owner_delivery.py
index ee56c31dcd749..b86a095e7ebf3 100644
--- a/tests/tools/test_bot_live_owner_delivery.py
+++ b/tests/tools/test_bot_live_owner_delivery.py
@@ -158,3 +158,34 @@ def test_unreadable_ticket_keeps_exact_id_reads_fail_closed(tmp_path):
         mailbox.deliver_to_live_owner(tmp_path, owner, "same id", delivery_id="e" * 32)
     with pytest.raises(PermissionError):
         mailbox.read_delivery_result(tmp_path, "e" * 32)
+
+
+def test_non_dict_ticket_does_not_wedge_bulk_scans(tmp_path):
+    """A ticket that parses as JSON but is not an object must be skipped by the scans
+    exactly like an unreadable one — never crash the sequence sweep or the claim."""
+    from tools import bot_live_delivery as mailbox
+
+    owner = dict(profile_home=str(tmp_path.resolve()), session_id="chat",
+                 lease_id="lease", live_session_id="live")
+    queued = mailbox.deliver_to_live_owner(tmp_path, owner, "readable", delivery_id="d" * 32)
+    root = tmp_path / "runtime" / mailbox.DELIVERY_DIR_NAME
+    (root / f"{'9' * 32}.json").write_text('"not a receipt"', encoding="utf-8")
+    admitted = mailbox.deliver_to_live_owner(tmp_path, owner, "second", delivery_id="f" * 32)
+    assert admitted["sequence"] > queued["sequence"]
+    assert mailbox.claim_pending_delivery(tmp_path, owner)["delivery_id"] == queued["delivery_id"]
+    assert mailbox.claim_pending_delivery(tmp_path, owner)["delivery_id"] == admitted["delivery_id"]
+
+
+def test_non_dict_ticket_keeps_exact_id_reads_fail_closed(tmp_path):
+    """A non-dict ticket at an exact id is a conflict, never an overwrite or a TypeError."""
+    from tools import bot_live_delivery as mailbox
+
+    owner = dict(profile_home=str(tmp_path.resolve()), session_id="chat",
+                 lease_id="lease", live_session_id="live")
+    root = tmp_path / "runtime" / mailbox.DELIVERY_DIR_NAME
+    root.mkdir(parents=True)
+    bad = root / f"{'9' * 32}.json"
+    bad.write_text("[1, 2]", encoding="utf-8")
+    with pytest.raises(ValueError, match="different payload"):
+        mailbox.deliver_to_live_owner(tmp_path, owner, "same id", delivery_id="9" * 32)
+    assert bad.read_text(encoding="utf-8") == "[1, 2]"
diff --git a/tests/tools/test_bot_relay.py b/tests/tools/test_bot_relay.py
index 0c584d08032f5..d56d6ea950c7b 100644
--- a/tests/tools/test_bot_relay.py
+++ b/tests/tools/test_bot_relay.py
@@ -124,6 +124,23 @@ def test_enqueue_claim_is_atomic_and_single_shot(root):
     assert bot_relay.claim_pending_envelopes(root) == []
 
 
+def test_claim_skips_non_dict_envelope(root):
+    """A parseable-but-non-object outbox file must not reach the Desktop consumer or
+    crash the claim sweep; the claim stays claimed so it is not re-queued."""
+    bot_relay.write_remote_roster(root, _rows())
+    roster = bot_relay.read_remote_roster(root)
+    target = bot_relay.resolve_remote_target("researcher", roster)
+    env = bot_relay.enqueue_envelope(
+        root, target=target, message="hi", sender_profile="work", sender_handle="work"
+    )
+    base = bot_relay.relay_root(root)
+    bad = base / bot_relay.OUTBOX_DIR / f"{'9' * 32}.json"
+    bad.write_text('"not an envelope"', encoding="utf-8")
+    claimed = bot_relay.claim_pending_envelopes(root)
+    assert [e["id"] for e in claimed] == [env["id"]]
+    assert not bad.exists() and (base / bot_relay.CLAIMED_DIR / bad.name).exists()
+
+
 def test_write_reply_validates_envelope_id(root):
     with pytest.raises(ValueError):
         bot_relay.write_reply(root, "../../etc/passwd", reply="x")
diff --git a/tests/tools/test_browser_lightpanda.py b/tests/tools/test_browser_lightpanda.py
index f5da12a691147..3b0afd51df98b 100644
--- a/tests/tools/test_browser_lightpanda.py
+++ b/tests/tools/test_browser_lightpanda.py
@@ -762,3 +762,16 @@ def test_orphan_reaper_sweeps_lightpanda_records(self, tmp_path):
              patch("tools.browser_tool._socket_safe_tmpdir", return_value=str(tmp_path)):
             bt_lifecycle._reap_orphaned_browser_sessions()
         reap.assert_called_once()
+
+    def test_orphan_reaper_survives_non_dict_record(self, tmp_path, monkeypatch):
+        """A parseable-but-non-object state file is swept like an unreadable one —
+        never an AttributeError that wedges the reaper."""
+        from tools import browser_lightpanda
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        state_dir = browser_lightpanda._state_dir()
+        bad = state_dir / "broken.json"
+        bad.write_text('"not a record"', encoding="utf-8")
+        dead_owner = state_dir / "dead.json"
+        dead_owner.write_text(json.dumps({"owner_pid": 2**22 + 7}), encoding="utf-8")
+        assert browser_lightpanda.reap_orphaned_lightpanda() == 0
+        assert not bad.exists() and not dead_owner.exists()
diff --git a/tests/tools/test_write_approval.py b/tests/tools/test_write_approval.py
index 44db64514921b..b99960d6907b3 100644
--- a/tests/tools/test_write_approval.py
+++ b/tests/tools/test_write_approval.py
@@ -48,6 +48,18 @@ def test_invalid_subsystem_is_off(hermes_home):
     assert wa.write_approval_enabled("bogus") is False
 
 
+def test_list_pending_skips_non_dict_record(hermes_home):
+    """A parseable-but-non-object pending file must be skipped, not crash the sort."""
+    from tools import write_approval as wa
+    wa.stage_write("memory", {"action": "add", "target": "user", "content": "ok"},
+                   summary="ok", origin="foreground")
+    pending_dir = wa._pending_path("memory", "").parent
+    (pending_dir / "bad.json").write_text('"not a record"', encoding="utf-8")
+    records = wa.list_pending("memory")
+    assert len(records) == 1 and records[0]["payload"]["content"] == "ok"
+    assert wa.get_pending("memory", "bad") is None
+
+
 def test_normalize_enabled_coerces_values():
     from tools import write_approval as wa
     # Real bools pass through.
diff --git a/tests/tui_gateway/test_spawn_tree_records.py b/tests/tui_gateway/test_spawn_tree_records.py
new file mode 100644
index 0000000000000..116300093558c
--- /dev/null
+++ b/tests/tui_gateway/test_spawn_tree_records.py
@@ -0,0 +1,35 @@
+"""Tests: spawn_tree.* JSON-RPC handlers (tui_gateway/methods_session.py).
+
+A parseable-but-non-object snapshot file must not wedge ``spawn_tree.list``
+(the legacy per-file scan called ``raw.get`` outside its suppress guard) and
+must not satisfy the ``spawn_tree.load`` result contract.
+"""
+
+import json
+
+import tui_gateway.server as srv
+
+
+def test_spawn_tree_list_survives_non_dict_snapshot(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    d = srv._spawn_tree_session_dir("sess-x")
+    (d / "bad.json").write_text('"not a snapshot"', encoding="utf-8")
+    good = d / "good.json"
+    good.write_text(json.dumps({"session_id": "sess-x", "label": "ok"}), encoding="utf-8")
+
+    envelope = srv._methods["spawn_tree.list"](1, {"session_id": "sess-x"})
+    assert "error" not in envelope, envelope
+    entries = envelope["result"]["entries"]
+    labels = {e["label"] for e in entries}
+    assert "ok" in labels and len(entries) == 2  # bad file degrades to a fallback entry
+
+
+def test_spawn_tree_load_rejects_non_dict_snapshot(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    d = srv._spawn_tree_session_dir("sess-y")
+    bad = d / "bad.json"
+    bad.write_text("[1, 2]", encoding="utf-8")
+
+    envelope = srv._methods["spawn_tree.load"](1, {"path": str(bad)})
+    assert "error" in envelope
+    assert "not a JSON object" in envelope["error"]["message"]
diff --git a/tools/bot_live_delivery.py b/tools/bot_live_delivery.py
index eecb329dc4be9..f7ead7463a81b 100644
--- a/tools/bot_live_delivery.py
+++ b/tools/bot_live_delivery.py
@@ -126,6 +126,8 @@ def _scan_read(path: Path) -> dict[str, Any] | None:
     """
     try:
         record = _read(path)
+        if record is not None and not isinstance(record, dict):
+            raise ValueError(f"expected a JSON object, got {type(record).__name__}")
     except (OSError, ValueError) as exc:  # ValueError: corrupt JSON and invalid UTF-8 alike
         level = logging.DEBUG if path in _warned_unreadable else logging.WARNING
         _warned_unreadable.add(path)
@@ -177,7 +179,8 @@ def deliver_to_live_owner(
         path = root / f"{key}.json"
         existing = _read(path)
         if existing is not None:
-            if existing["owner"] != pinned or existing["message"] != message or existing.get("author") != author:
+            if not isinstance(existing, dict) or existing["owner"] != pinned or \
+                    existing["message"] != message or existing.get("author") != author:
                 raise ValueError("delivery id already belongs to a different payload")
             return existing
         record = dict(delivery_id=key, id=key, owner=pinned, **pinned,
@@ -243,6 +246,8 @@ def complete_delivery(
         record = _read(path)
         if record is None:
             raise FileNotFoundError(f"delivery not found: {key}")
+        if not isinstance(record, dict):
+            raise ValueError("delivery id already belongs to a different payload")
         if record["status"] in _TERMINAL:
             if any(record.get(k) != v for k, v in outcome.items()):
                 raise ValueError("delivery already has a different terminal receipt")
diff --git a/tools/bot_mode_dm.py b/tools/bot_mode_dm.py
index 547756282bcba..f1f7201604685 100644
--- a/tools/bot_mode_dm.py
+++ b/tools/bot_mode_dm.py
@@ -470,6 +470,8 @@ def _admit_live_dm(profile_home: Path | None, dm_file: str, author: Optional[dic
     if record is None:
         record = deliver_to_live_owner(home, intent["owner"], intent["message"],
                                        delivery_id=intent["delivery_id"], author=intent.get("author"))
+    elif not isinstance(record, dict):
+        raise ValueError("delivery id already belongs to a different payload")
     return record
 
 
@@ -479,11 +481,11 @@ def _wait_live_dm(home: str, delivery_id: str, *, dm_file: "str | os.PathLike |
     deadline = time.monotonic() + _LIVE_WAIT_SECONDS
     while True:
         record = read_delivery_result(home, delivery_id)
-        status = record["status"] if record else "ambiguous"
+        status = record["status"] if isinstance(record, dict) else "ambiguous"
         if status not in ("queued", "claimed") or time.monotonic() >= deadline:
             break
         time.sleep(min(0.5, max(0, deadline - time.monotonic())))
-    payload = {key: record[key] for key in ("reply", "error", "reason") if record and record.get(key)}
+    payload = {key: record[key] for key in ("reply", "error", "reason") if isinstance(record, dict) and record.get(key)}
     payload.update(status=status, delivery_id=delivery_id)
     if status in ("queued", "claimed", "ambiguous"):
         payload["detail"] = "Delivery remains pending or its outcome is unknown. Do not resend; receipt is retained."
diff --git a/tools/bot_relay.py b/tools/bot_relay.py
index 63cb5a31f4df9..fc70687730507 100644
--- a/tools/bot_relay.py
+++ b/tools/bot_relay.py
@@ -241,6 +241,8 @@ def _expire_if_stale(root: Path | str, path: Path, ttl: float, now: float) -> bo
     reply so the sender's waiter resolves (best effort). Unreadable envelopes are left for the claim."""
     try:
         env = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(env, dict):
+            raise ValueError(f"expected a JSON object, got {type(env).__name__}")
         created = float(env.get("created_at") or path.stat().st_mtime)
     except (OSError, ValueError):
         return False
@@ -287,7 +289,10 @@ def claim_pending_envelopes(root: Path | str) -> list[dict]:
         claimed = base / CLAIMED_DIR / path.name
         with contextlib.suppress(OSError, ValueError):
             os.replace(path, claimed)  # atomic claim
-            out.append(json.loads(claimed.read_text(encoding="utf-8")))
+            envelope = json.loads(claimed.read_text(encoding="utf-8"))
+            if not isinstance(envelope, dict):
+                raise ValueError(f"expected a JSON object, got {type(envelope).__name__}")
+            out.append(envelope)
     return out
 
 
diff --git a/tools/browser_lightpanda.py b/tools/browser_lightpanda.py
index 90c563c1665ea..8fe03531e7f78 100644
--- a/tools/browser_lightpanda.py
+++ b/tools/browser_lightpanda.py
@@ -316,6 +316,8 @@ def reap_orphaned_lightpanda() -> int:
         session_name = record_path.stem
         try:
             record = json.loads(record_path.read_text(encoding="utf-8"))
+            if not isinstance(record, dict):
+                raise ValueError(f"expected a JSON object, got {type(record).__name__}")
         except (OSError, ValueError):
             record_path.unlink(missing_ok=True)
             continue
diff --git a/tools/write_approval.py b/tools/write_approval.py
index c519bf939a35c..d767256187ab7 100644
--- a/tools/write_approval.py
+++ b/tools/write_approval.py
@@ -93,7 +93,10 @@ def list_pending(subsystem: str) -> List[Dict[str, Any]]:
     records: List[Dict[str, Any]] = []
     for p in _pending_files(subsystem):
         try:
-            records.append(json.loads(p.read_text(encoding="utf-8")))
+            record = json.loads(p.read_text(encoding="utf-8"))
+            if not isinstance(record, dict):
+                raise ValueError(f"expected a JSON object, got {type(record).__name__}")
+            records.append(record)
         except Exception:
             logger.warning("Skipping unreadable pending record: %s", p)
     records.sort(key=lambda r: r.get("created_at", 0))
@@ -106,7 +109,8 @@ def get_pending(subsystem: str, pending_id: str) -> Optional[Dict[str, Any]]:
     if not path.exists():
         return None
     with suppress(Exception):
-        return json.loads(path.read_text(encoding="utf-8"))
+        data = json.loads(path.read_text(encoding="utf-8"))
+        return data if isinstance(data, dict) else None
     return None
 
 
diff --git a/trajectory_compressor.py b/trajectory_compressor.py
index 19892bb80fe1d..d7c4f1fc7085d 100644
--- a/trajectory_compressor.py
+++ b/trajectory_compressor.py
@@ -572,7 +572,7 @@ async def compress_trajectory_async(self, trajectory: List[Dict[str, str]]) -> T
 
     async def process_entry_async(self, entry: Dict[str, Any]) -> Tuple[Dict[str, Any], TrajectoryMetrics]:
         """Compress one JSONL entry's ``conversations``; attach metrics when compressed."""
-        if "conversations" not in entry:
+        if not isinstance(entry, dict) or "conversations" not in entry:
             return entry, TrajectoryMetrics()
         compressed_trajectory, metrics = await self.compress_trajectory_async(entry["conversations"])
         result = dict(entry, conversations=compressed_trajectory)
diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py
index 652697b7dc487..890ac56e763f3 100644
--- a/tui_gateway/methods_session.py
+++ b/tui_gateway/methods_session.py
@@ -2158,6 +2158,8 @@ def _legacy_spawn_tree_entry(p, session_dir_name: str) -> dict | None:
     raw = {}
     with contextlib.suppress(Exception):
         raw = json.loads(p.read_text(encoding="utf-8"))
+    if not isinstance(raw, dict):
+        raw = {}
     subagents = raw.get("subagents") or []
     return {"path": str(p), "session_id": raw.get("session_id") or session_dir_name,
             "finished_at": raw.get("finished_at") or stat.st_mtime, "started_at": raw.get("started_at"),
@@ -2196,6 +2198,8 @@ def _(rid, params: dict) -> dict:
         payload = json.loads(resolved.read_text(encoding="utf-8"))
     except (OSError, json.JSONDecodeError) as exc:
         return _err(rid, 5000, f"spawn_tree.load failed: {exc}")
+    if not isinstance(payload, dict):
+        return _err(rid, 5000, "spawn_tree.load failed: snapshot is not a JSON object")
     return _ok(rid, payload)