fix(queue-state): parseable non-dict JSON no longer wedges scans and drains

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

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:

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}"

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

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:

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:]
 
 

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

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()

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

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"

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"}]

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

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]"

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")