diff --git a/src/backend/services/operator_queue_service.py b/src/backend/services/operator_queue_service.py index d706be57c..befb6eb2a 100644 --- a/src/backend/services/operator_queue_service.py +++ b/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. diff --git a/tests/registry.json b/tests/registry.json index ad7915e4c..1f637dadc 100644 --- a/tests/registry.json +++ b/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." } ] } diff --git a/tests/unit/test_3313_operator_queue_sync_error_logged.py b/tests/unit/test_3313_operator_queue_sync_error_logged.py new file mode 100644 index 000000000..5ffc86cf8 --- /dev/null +++ b/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 diff --git a/tests/unit/test_ec_operator_queue_edges.py b/tests/unit/test_ec_operator_queue_edges.py index a48881d70..5db4b8e63 100644 --- a/tests/unit/test_ec_operator_queue_edges.py +++ b/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;