fix(operator-queue): one malformed entry no longer stops answer delivery to its agent (#3313)

src/backend/services/operator_queue_service.py

@@ -916,7 +916,9 @@ def _clamp_ingested_item(req: dict, agent_name: str = "") -> dict:
         if opt_bytes is None or opt_bytes > OPERATOR_QUEUE_OPTIONS_MAX_BYTES:
             out["options"] = [_OPTIONS_DROPPED_MARKER]
 
-    if out.get("priority") not in _VALID_PRIORITIES:
+    # #3313: isinstance first — a list/object priority is unhashable, and the
+    # set membership test raised out of this never-raises clamp.
+    if not isinstance(out.get("priority"), str) or out["priority"] not in _VALID_PRIORITIES:
         out["priority"] = "medium"
 
     # Ignore the agent-supplied created_at (a future date pins the item atop the
@@ -1080,7 +1082,8 @@ def _comparable_type(value) -> str:
 
 
 def _comparable_priority(value) -> str:
-    return value if value in _VALID_PRIORITIES else "medium"   # the clamp's default
+    # #3313: isinstance first, as in the clamp — an unhashable value raised here.
+    return value if isinstance(value, str) and value in _VALID_PRIORITIES else "medium"
 
 
 def _comparable_addressee(value) -> Optional[str]:
@@ -1771,7 +1774,15 @@ async def _poll_cycle(self):
         if running_agents:
             # Sync each agent concurrently (with a reasonable limit)
             tasks = [self._sync_agent(name) for name in running_agents]
-            await asyncio.gather(*tasks, return_exceptions=True)
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            # #3313: `return_exceptions=True` keeps one agent's failure from
+            # cancelling the others, but it also dropped the error unlogged.
+            for name, result in zip(running_agents, results):
+                if isinstance(result, Exception):
+                    logger.error(
+                        f"Operator queue sync failed for '{name}': "
+                        f"{type(result).__name__}: {result}"
+                    )
 
         if self._changed_this_cycle:
             await self._broadcast_sync()
@@ -2013,6 +2024,19 @@ async def _sync_agent(self, agent_name: str):
             req_id = req.get("id")
             if not req_id:
                 continue
+            if not isinstance(req_id, str):
+                # #3313: only a string id can name a row. A list or object id
+                # reached the unguarded `in open_by_rid` below and raised out of
+                # the whole sync, so none of this agent's answers were written
+                # back. A pending one is held as `invalid_id`, as the create-side
+                # shape check would hold it; any other status names nothing.
+                if req.get("status", "pending") == "pending":
+                    logger.warning(
+                        f"Rejecting malformed operator-queue id from {agent_name}: {req_id!r}"
+                    )
+                    held += 1
+                    hold_reason = _stronger_hold(hold_reason, HOLD_INVALID_ID)
+                continue
 
             # #1631: reject an agent-authored id that impersonates a platform id
             # prefix (hijack/suppress guard). Log once per (agent, id) so it
@@ -2473,6 +2497,11 @@ async def _write_responses_to_agent(
             if not isinstance(req, dict):
                 continue
             req_id = req.get("id")
+            if not isinstance(req_id, str):
+                # #3313: every map here is keyed on a row's string request_id,
+                # so a non-string id names no row — and a list/object one made
+                # the lookup below raise, aborting the whole write-back.
+                continue
             if req_id in response_map:
                 resp = response_map[req_id]
                 # #2989 review: "send again to answer anyway" must reach the file.

tests/registry.json

@@ -4944,6 +4944,16 @@
         "unit"
       ],
       "description": "MCP list_operator_queue completeness (trinity-enterprise#815); real tool execute() over a fake TrinityClient, node:test. A broad agent-key read takes its permits first through the strict permissions read and passes {self} ∪ permitted as agent_names; a failed read is permissions_unavailable (retryable, no list call, no self-read substitute); over 500 names or an 8 KB request target is permit_set_too_large, never an unfiltered read. The output always carries count, total, has_more, next_cursor, next_offset and items; an older backend's missing fields become null with a warning; backend warnings pass through; a belt drop nulls total and warns. User and system scopes and scoped reads make no extra permissions call. cursor is opt-in (no cursor → offset 0 as before), passed through unchanged, and a two-page walk re-reads and re-sends agent_names on every page. The description states the contract within the 1,800-character headroom."
+    },
+    {
+      "file": "unit/test_3313_operator_queue_sync_error_logged.py",
+      "feature": "abilityai/trinity#3313",
+      "added": "2026-10-07",
+      "categories": [
+        "backend",
+        "operator-queue"
+      ],
+      "description": "An exception from one agent's operator-queue sync must be logged with the agent name, not dropped by `_poll_cycle`'s gather(return_exceptions=True) (#3313). Drives the real `_poll_cycle` over two agents where one sync raises: both run, the failing one is logged at ERROR with its exception type, the healthy one is not. Mutation-proven: reverting the logging goes red. No Docker, no network."
     }
   ]
 }

tests/unit/test_3313_operator_queue_sync_error_logged.py

@@ -0,0 +1,75 @@
+"""
+#3313 — an agent's sync failure is logged, not silently dropped.
+
+`_poll_cycle` runs every running agent's `_sync_agent` under
+`asyncio.gather(..., return_exceptions=True)`. That keeps one agent's failure
+from cancelling the others, but it also discarded the exception with no log
+line, so a sync that raised every 5 s (an unhashable id in the agent's file)
+stopped that agent's answer delivery with nothing in the logs.
+
+Driven through the real `OperatorQueueSyncService._poll_cycle`; only its
+collaborators (leadership, expiry, skill-gate sweep, Docker, db) are stubbed.
+The per-entry fixes are pinned in `test_ec_operator_queue_edges.py`
+(`TestUnhashableEntryValues`).
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import sys
+import types
+from unittest.mock import MagicMock
+
+import pytest
+
+pytest.importorskip("sqlalchemy")
+
+os.environ.setdefault("REDIS_URL", "redis://u:p@localhost:6379")
+os.environ.setdefault("SECRET_KEY", "test-secret")
+
+_BACKEND = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src", "backend"))
+if _BACKEND not in sys.path:
+    sys.path.insert(0, _BACKEND)
+
+import services  # noqa: E402
+import services.docker_service as docker_service  # noqa: E402
+import services.operator_queue_service as oqs  # noqa: E402
+
+
+def _wire(monkeypatch, synced):
+    svc = oqs.OperatorQueueSyncService()
+    monkeypatch.setattr(svc, "_try_acquire_leadership", lambda: True)
+    monkeypatch.setattr(oqs.ask_service, "expire", lambda: MagicMock(rows=[]))
+
+    async def _sweep():
+        return None
+    gate = types.SimpleNamespace(sweep=_sweep)
+    monkeypatch.setitem(sys.modules, "services.skill_gate_service", gate)
+    monkeypatch.setattr(services, "skill_gate_service", gate, raising=False)
+
+    monkeypatch.setattr(docker_service, "agent_container_states",
+                        lambda: {"bad": "running", "good": "running"})
+    db = MagicMock()
+    db.mark_operator_queue_unconfirmed.return_value = 0
+    db.mark_operator_queue_undeliverable_not_running.return_value = []
+    monkeypatch.setattr(oqs, "db", db)
+
+    async def _sync_agent(name):
+        synced.append(name)
+        if name == "bad":
+            raise TypeError("unhashable type: 'list'")
+    monkeypatch.setattr(svc, "_sync_agent", _sync_agent)
+    return svc
+
+
+def test_a_failing_agent_sync_is_logged_with_the_agent_name(monkeypatch, caplog):
+    synced = []
+    svc = _wire(monkeypatch, synced)
+    with caplog.at_level(logging.ERROR, logger=oqs.logger.name):
+        asyncio.run(svc._poll_cycle())
+
+    assert sorted(synced) == ["bad", "good"]  # one failure cancels no other agent
+    errors = [r.getMessage() for r in caplog.records if r.levelno >= logging.ERROR]
+    assert any("'bad'" in m and "TypeError" in m for m in errors), errors
+    assert not any("'good'" in m for m in errors), errors

tests/unit/test_ec_operator_queue_edges.py

@@ -565,9 +565,6 @@ def _harness():
         import test_2915_operator_queue_sync_honesty as h
         return h
 
-    @pytest.mark.xfail(strict=True, reason="#3313: " + BUG.format(
-        "an unhashable `priority` (list/object) makes `_clamp_ingested_item` raise TypeError, "
-        "breaking its never-raises contract; the ask is quarantined and never surfaced"))
     @pytest.mark.parametrize("priority", [
         pytest.param(["high"], id="r145-priority-list"),
         pytest.param({"level": "high"}, id="r146-priority-object"),
@@ -584,9 +581,6 @@ def test_the_clamp_never_raises_on_an_unhashable_priority(self, priority):
         out = oqs._clamp_ingested_item({"id": "u-1", "status": "pending", "priority": priority})
         assert out["priority"] == "medium"
 
-    @pytest.mark.xfail(strict=True, reason="#3313: " + BUG.format(
-        "one pending entry with an unhashable `id` raises out of `_sync_agent` before the "
-        "write-back, so NONE of that agent's answers are delivered, silently, every cycle"))
     def test_an_unhashable_id_does_not_stop_the_agents_answers(self, monkeypatch):
         """r147. REACHABILITY: in `_sync_agent`'s loop the reserved-prefix,
         native, seen and terminal checks are each `isinstance(req_id, str)`-
@@ -606,9 +600,6 @@ def test_an_unhashable_id_does_not_stop_the_agents_answers(self, monkeypatch):
         asyncio.run(svc._sync_agent("a"))
         client.write_file.assert_awaited_once()
 
-    @pytest.mark.xfail(strict=True, reason="#3313: " + BUG.format(
-        "rewriting an INGESTED entry's priority to a list raises out of `changed_fields`, "
-        "halting the agent's whole sync (no answers delivered) every cycle, silently"))
     def test_an_unhashable_priority_rewrite_is_a_change_not_a_crash(self, monkeypatch):
         """r148. REACHABILITY: for an entry whose id matches an open pending row,
         `_sync_agent` calls `changed_fields(row, req)` outside any try;