diff --git a/EVALUATIONS.md b/EVALUATIONS.md index 1d08d1d..fa73e80 100644 --- a/EVALUATIONS.md +++ b/EVALUATIONS.md @@ -356,11 +356,11 @@ scorer = client.monitor.judge_scorers.builder( |---|---|---|---| | `live` | `bool` | `False` | Create the online profile and start scoring live traffic | | `sample_rate` | `float` | `0.1` | Fraction of traffic actually scored | -| `scope` | `str` | `"trace"` | `"trace"` scores individual traces at ingest; `"session"` scores whole conversations | +| `scope` | `str` | `"trace"` | `"trace"` scores individual traces at ingest; `"session"` scores whole conversations. Non-default values require `live=True` | | `alert_threshold` | `float \| None` | `5` | A score below this raises a signal. `None` scores without ever raising one | | `severity` | `str` | `"medium"` | `"low"`, `"medium"`, `"high"` or `"critical"`, applied to signals it raises | -| `agent_ids` | `list[str]` | `None` | Restrict scoring to specific agents instead of the whole workspace | -| `idle_seconds` | `int` | `120` | For `scope="session"`: how long a session must be quiet before it is judged | +| `agent_ids` | `list[str]` | `None` | Restrict scoring to specific agents instead of the whole workspace. Requires `live=True` | +| `idle_seconds` | `int` | `None` (server stores `120`) | For `scope="session"`: how long a session must be quiet before it is judged. Requires `live=True` and `scope="session"` - the builder raises otherwise | Calibration, tuning and the live-scoring history hang off the same scorer id - the SDK resolves the online profile for you: @@ -848,7 +848,7 @@ report = client.evaluations.run(...).execute(my_agent).finalize().analyze() # Top-level convenience accessors - return None when the metric was not # enabled on the dataset, or no case has a value yet. -report.average_rating # float | None - same as report.statistics.average_rating +report.average_rating # float | None - None only when no analysis exists; an analyzed run that scored nothing reads 0.0 report.cosine_similarity # float | None - averaged across cases (0-1) report.jaccard_similarity # float | None - averaged across cases (0-1) report.bleu_score # float | None - averaged across cases (0-1) diff --git a/README.md b/README.md index 47114d0..ca93a73 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,9 @@ Also see [SDK Developer Docs](https://developers.agentx.so), [API Reference Docs pip install --upgrade agentx-python ``` -Requires Python 3.9 or newer. +Requires Python 3.9 or newer for the core SDK. Some integration extras have higher floors set +by their upstream packages - `[crewai]`, `[autogen]`, and `[databricks]` need Python 3.10+ (`[all]` +therefore does too); the core tracer and every REST surface stay 3.9-compatible. #### Run the self-host governance suite locally diff --git a/TRACING.md b/TRACING.md index de42152..457dfb3 100644 --- a/TRACING.md +++ b/TRACING.md @@ -380,7 +380,7 @@ with tracer.trace("support-agent") as span: span.output = answer ``` -`tracer.record_memory(name, operation=..., query=..., output=..., duration_ms=...)` is the after-the-fact form. `operation` is free text - conventionally `"read"` or `"write"` - carried in the span's metadata, while the kind itself stays one value so dashboards and scorers can select all memory activity at once. With no active span the record is dropped (with a debug log) - the pending queue rides the next trace's retrieval steps, and memory content must never feed the RAG judges' retrieval context - so wrap the call in `tracer.trace()`. +`tracer.record_memory(name, operation=..., query=..., output=..., duration_ms=...)` is the after-the-fact form. `operation` is free text - conventionally `"read"` or `"write"` - carried in the span's metadata, while the kind itself stays one value so dashboards and scorers can select all memory activity at once. With no active span the record is dropped (warned once per process, then logged at debug), not queued: there are two pending queues (tool calls ride the next trace's `tool_calls`, retrievals its `retrieval_steps`) and neither fits - `retrieval_steps` feeds the engine's RAG `{context}` extraction, which recalled state must never reach, and no memory-shaped queue has been built yet - so wrap the call in `tracer.trace()`. --- @@ -520,7 +520,11 @@ print(pattern.id) client.monitor.patterns.get(pattern.id) # -> MonitorPattern client.monitor.patterns.list() # -> list[MonitorPattern] -client.monitor.patterns.update(pattern.id, enabled=False) # sparse update, wire camelCase keys -> MonitorPattern +client.monitor.patterns.update(pattern.id, enabled=False) # sparse update -> MonitorPattern +# update() takes snake_case kwargs (or wire camelCase); an unknown snake_case key raises. +# Sending regex / semantic_prompt / include_terms rebuilds the pattern's conditions for that +# detector kind; exclude_terms, match_mode, or match_target sent alone raise ValueError +# (the engine would silently ignore them) - pass conditions=[...] or a trigger field instead. client.monitor.patterns.delete(pattern.id) # historical signals remain as history ``` diff --git a/agentx/agentx.py b/agentx/agentx.py index 09eb04f..437521e 100644 --- a/agentx/agentx.py +++ b/agentx/agentx.py @@ -20,8 +20,8 @@ def __init__( # The api_key is NOT written back into os.environ (it used to be): every sub-client # below receives it explicitly, and mutating process-global state from a constructor # re-pointed unrelated code - the same leak the base_url write below had (deep-dive - # round 3, bug #1). Static flows that still read the env (AgentX.list_workforces, - # bare get_headers()) now require the caller to set AGENTX_API_KEY themselves. + # round 3, bug #1). Flows that still read the env (bare get_headers()) now require + # the caller to set AGENTX_API_KEY themselves. self.api_key = api_key or os.getenv("AGENTX_API_KEY") # base_url overrides AGENTX_API_BASE_URL env var (and the SDK default). It is @@ -147,7 +147,7 @@ def get_agent(self, id: str) -> Agent: response = requests.get(url, headers=get_headers(self.api_key)) # Check if response was successful if response.status_code == 200: - return Agent(**response.json()) + return Agent(**response.json())._bind(self.api_key, self.base_url) else: raise AgentXError( f"Failed to retrieve agent: {response.reason}. This endpoint is " @@ -165,22 +165,26 @@ def list_agents(self) -> List[Agent]: response = requests.get(url, headers=get_headers(self.api_key)) # Check if response was successful if response.status_code == 200: - return [Agent(**agent) for agent in response.json()] + return [Agent(**agent)._bind(self.api_key, self.base_url) for agent in response.json()] else: raise AgentXError( f"Failed to list agents: {response.reason}. This endpoint is " "hosted-platform only - on self-host use client.monitor.agents.list()." ) - @staticmethod - def list_workforces() -> List["Workforce"]: - """List all workforces/teams. Static, so it reads AGENTX_API_KEY from the environment - directly - the constructor no longer writes ``api_key`` into os.environ, so set the - env var yourself before calling this.""" - url = f"{api_base()}/access/teams" - response = requests.get(url, headers=get_headers()) + def list_workforces(self) -> List["Workforce"]: + """List all workforces/teams, each bound to this client's credentials - including each + workforce's ``manager`` and ``agents``, so their calls authenticate the same way. + + This used to be documented as a static call (``AgentX.list_workforces()``); that form + was broken (the old staticmethod body referenced ``self`` and raised NameError on any + non-empty response). Construct a client instead - ``AgentX().list_workforces()`` picks + up AGENTX_API_KEY / AGENTX_API_BASE_URL from the environment, which is what the static + form effectively did.""" + url = f"{self.base_url or api_base()}/access/teams" + response = requests.get(url, headers=get_headers(self.api_key)) if response.status_code == 200: - return [Workforce(**workforce) for workforce in response.json()] + return [Workforce(**workforce)._bind(self.api_key, self.base_url) for workforce in response.json()] else: raise Exception( f"Failed to list workforces: {response.status_code} - {response.reason}" diff --git a/agentx/integrations/crewai.py b/agentx/integrations/crewai.py index bddd845..3ef68a9 100644 --- a/agentx/integrations/crewai.py +++ b/agentx/integrations/crewai.py @@ -144,14 +144,14 @@ def _start_task_timing_capture(self, crew: Any = None): TaskFailedEvent, TaskStartedEvent, ) - except ImportError: + except Exception: # noqa: BLE001 - crewai import raises TypeError (PEP 604) on py3.9, not just ImportError from crewai.utilities.events import crewai_event_bus from crewai.utilities.events.task_events import ( TaskCompletedEvent, TaskFailedEvent, TaskStartedEvent, ) - except ImportError: + except Exception: # noqa: BLE001 - crewai import raises TypeError (PEP 604) on py3.9, not just ImportError if not _warned_no_event_bus: _warned_no_event_bus = True logger.warning( diff --git a/agentx/monitor/__init__.py b/agentx/monitor/__init__.py index ccc272b..8f7d2ad 100644 --- a/agentx/monitor/__init__.py +++ b/agentx/monitor/__init__.py @@ -10,6 +10,9 @@ from agentx.monitor.models import MonitorPattern, MonitorProfile, MonitorSignal, SignalOccurrence from agentx.monitor.patterns import MonitorPatternBuilder, MonitorPatternClient from agentx.monitor.profile import MonitorProfileClient +from agentx.monitor.review_queue import ReviewQueueClient, ReviewQueueItem +from agentx.monitor.rules import MonitorRule, MonitorRulesClient +from agentx.monitor.scorers import AgentXScorersError, ScorersClient from agentx.monitor.scorer_groups import AgentXScorerGroupsError, ScorerGroup, ScorerGroupsClient from agentx.monitor.sessions import MonitorSessionClient from agentx.monitor.signals import MonitorSignalClient @@ -19,6 +22,7 @@ "AgentXJudgeScorersError", "AgentXMonitorError", "AgentXScorerGroupsError", + "AgentXScorersError", "ImprovementGroupsClient", "JudgeScorer", "JudgeScorerBuilder", @@ -30,10 +34,15 @@ "MonitorPatternClient", "MonitorProfile", "MonitorProfileClient", + "MonitorRule", + "MonitorRulesClient", "MonitorSessionClient", "MonitorSignal", "MonitorSignalClient", + "ReviewQueueClient", + "ReviewQueueItem", "ScorerGroup", + "ScorersClient", "ScorerGroupsClient", "SignalOccurrence", ] diff --git a/agentx/monitor/_transport.py b/agentx/monitor/_transport.py new file mode 100644 index 0000000..b00e0f4 --- /dev/null +++ b/agentx/monitor/_transport.py @@ -0,0 +1,48 @@ +"""Shared HTTP transport for the monitor sub-clients that own their ``_request`` (scorers, +judge_scorers, scorer_groups, improvement_groups): one retry schedule mirroring +``MonitorClient._request``, so ``retry=False`` means the same thing everywhere the client.py +comment promises it ("retry=False for ANY non-idempotent write").""" + +from __future__ import annotations + +import logging +import time +from typing import Any, Optional + +import requests + +logger = logging.getLogger(__name__) + +# Same schedule as MonitorClient._request (agentx/monitor/client.py). +_RETRYABLE_STATUS = {429, 500, 502, 503, 504} +_RETRY_BACKOFF = [1.0, 2.0, 4.0] + + +def request_with_retries( + method: str, url: str, *, retry: bool = True, **kwargs: Any +) -> requests.Response: + """``requests.request`` with MonitorClient's transport posture: when ``retry`` is true, + connection errors and retryable statuses (429/5xx) walk the backoff schedule; the last + response (whatever its status) is returned for the caller's own error taxonomy. + + ``retry=False`` is single-shot - for non-idempotent writes (creates, deletes) and + judge-billing POSTs, where a client-side timeout must not fire the same work twice. + Transport errors keep their ``requests`` exception type (callers guard on + ``requests.Timeout`` for judge-billing endpoints).""" + schedule = [0.0] + _RETRY_BACKOFF if retry else [0.0] + last_exc: Optional[Exception] = None + for attempt, wait in enumerate(schedule): + if wait: + time.sleep(wait) + try: + resp = requests.request(method, url, **kwargs) + except requests.RequestException as e: + last_exc = e + logger.debug("Request error (attempt %d): %s", attempt + 1, e) + continue + if retry and resp.status_code in _RETRYABLE_STATUS and attempt < len(schedule) - 1: + logger.debug("Retryable status %d (attempt %d)", resp.status_code, attempt + 1) + continue + return resp + assert last_exc is not None # every non-raising path returned above + raise last_exc diff --git a/agentx/monitor/client.py b/agentx/monitor/client.py index 0850a37..b422406 100644 --- a/agentx/monitor/client.py +++ b/agentx/monitor/client.py @@ -205,9 +205,11 @@ def _api_root(self) -> str: def _request( self, method: str, path: str, timeout: int = 30, base: Optional[str] = None, retry: bool = True, **kwargs ) -> Any: - # retry=False for non-idempotent judge-spending POSTs (sweep, coherence, portability, - # tuning): a client-side timeout must not fire the same LLM-billing work a second time - # while the first invocation is still running server-side. Same precedent as + # retry=False for ANY non-idempotent write - duplicating creates, deletes (a lost + # response + retry turns success into a spurious 404), and judge-spending POSTs + # (sweep, coherence, portability, tuning: a client-side timeout must not fire the + # same LLM-billing work twice while the first invocation still runs server-side). + # The judge list is the example set, not the rule. Same precedent as # EvaluationsClient._request / analyze_run. url = f"{base or self._base_url}{path}" last_exc: Optional[Exception] = None diff --git a/agentx/monitor/improvement_groups.py b/agentx/monitor/improvement_groups.py index 2dabb2d..e1932b3 100644 --- a/agentx/monitor/improvement_groups.py +++ b/agentx/monitor/improvement_groups.py @@ -4,6 +4,7 @@ import requests +from agentx.monitor._transport import request_with_retries from agentx.util import api_base, get_headers from agentx.exceptions import AgentXError, AgentXAuthError, AgentXValidationError @@ -45,7 +46,9 @@ def __init__( self._workspace_id = workspace_id self._base_url = (base_url or api_base()).rstrip("/") - def _request(self, method: str, path: str, json: Any = None, timeout: int = 120) -> Any: + def _request( + self, method: str, path: str, json: Any = None, timeout: int = 120, retry: bool = True + ) -> Any: params = None if self._workspace_id: # Mirrors MonitorClient._workspace_params/_with_workspace: GETs (and DELETEs) @@ -57,9 +60,12 @@ def _request(self, method: str, path: str, json: Any = None, timeout: int = 120) json = {**json, "workspaceId": self._workspace_id} else: params = {"workspaceId": self._workspace_id} - resp = requests.request( + # retry=False for ANY non-idempotent write (member deletes, the report-generating + # POST) - MonitorClient._request's posture, via the shared monitor transport. + resp = request_with_retries( method, f"{self._base_url}/agent-monitoring{path}", + retry=retry, headers={**get_headers(self._api_key), "Content-Type": "application/json"}, json=json, params=params, @@ -92,7 +98,9 @@ def get(self, group_id: str) -> Dict[str, Any]: def remove_member(self, group_id: str, member_id: str) -> None: """Prune a member before spending the group (a confirm that turned out uninteresting).""" - self._request("DELETE", f"/improvement-groups/{group_id}/members/{member_id}") + # retry=False: a lost response + transport retry would turn a successful delete + # into a spurious 404. + self._request("DELETE", f"/improvement-groups/{group_id}/members/{member_id}", retry=False) def generate_report(self, group_id: str, model: Optional[str] = None) -> Dict[str, Any]: """Spend the group: one real LLM call clustering the confirmed failures into issues @@ -101,7 +109,11 @@ def generate_report(self, group_id: str, model: Optional[str] = None) -> Dict[st payload: Dict[str, Any] = {} if model is not None: payload["model"] = model - return self._request("POST", f"/improvement-groups/{group_id}/report", json=payload, timeout=300)["report"] + # retry=False: spends the group (real LLM billing) - a client-side timeout must not + # fire the same generation twice while the first still runs server-side. + return self._request( + "POST", f"/improvement-groups/{group_id}/report", json=payload, timeout=300, retry=False + )["report"] def list_reports(self) -> List[Dict[str, Any]]: return self._request("GET", "/improvement-reports").get("improvementReports", []) diff --git a/agentx/monitor/judge_scorers.py b/agentx/monitor/judge_scorers.py index 33bedcc..37f3e65 100644 --- a/agentx/monitor/judge_scorers.py +++ b/agentx/monitor/judge_scorers.py @@ -5,6 +5,7 @@ import requests +from agentx.monitor._transport import request_with_retries from agentx.util import api_base, get_headers from agentx.exceptions import AgentXError, AgentXAuthError, AgentXValidationError @@ -97,7 +98,9 @@ def __init__( # Captured once at construction so two clients with different bases can coexist. self._base_url = (base_url or api_base()).rstrip("/") - def _request(self, method: str, path: str, json: Any = None, timeout: int = 60) -> Any: + def _request( + self, method: str, path: str, json: Any = None, timeout: int = 60, retry: bool = True + ) -> Any: params = None if self._workspace_id: # Mirrors MonitorClient._workspace_params/_with_workspace: GETs (and DELETEs) @@ -109,9 +112,13 @@ def _request(self, method: str, path: str, json: Any = None, timeout: int = 60) json = {**json, "workspaceId": self._workspace_id} else: params = {"workspaceId": self._workspace_id} - resp = requests.request( + # retry=False for ANY non-idempotent write (creates, deletes) and judge-spending POST + # (tune/validate/publish) - MonitorClient._request's posture, via the shared monitor + # transport. + resp = request_with_retries( method, f"{self._base_url}/agent-monitoring{path}", + retry=retry, headers={**get_headers(self._api_key), "Content-Type": "application/json"}, json=json, params=params, @@ -263,7 +270,10 @@ def create( payload["offline"] = offline if online is not None: payload["online"] = online - return JudgeScorer(self._request("POST", "/judge-scorers", json=payload)["judgeScorer"]) + # Server-side create: a timeout + transport retry would create the scorer twice. + return JudgeScorer( + self._request("POST", "/judge-scorers", json=payload, retry=False)["judgeScorer"] + ) def get(self, scorer_id: str) -> JudgeScorer: return JudgeScorer(self._request("GET", f"/judge-scorers/{scorer_id}")["judgeScorer"]) @@ -300,7 +310,9 @@ def update( def delete(self, scorer_id: str) -> None: """Delete the scorer: rubric, version history, and online profile together. Irreversible; refused for the built-in Session Baseline Judge.""" - self._request("DELETE", f"/judge-scorers/{scorer_id}") + # retry=False: a lost response + transport retry would turn a successful delete + # into a spurious 404. + self._request("DELETE", f"/judge-scorers/{scorer_id}", retry=False) # ------------------------------------------------------------------ # Online-profile pass-throughs (calibration / tuning / ratings / events) @@ -334,8 +346,14 @@ def calibration(self, scorer_id: str, window: str = "7d") -> dict: def tune(self, scorer_id: str, window: str = "7d") -> dict: """Propose a rewrite of the rubric from calibration disagreements (LLM call, slow). ``window`` accepts the same values as :meth:`calibration`, including "rubric".""" + # retry=False (judge-spending POST, MonitorClient.propose_online_evaluator_tuning's + # posture): a client-side timeout must not fire the same LLM-billing work twice. data = self._request( - "POST", f"/online-evaluators/{self._profile_id(scorer_id)}/tune", json={"window": window}, timeout=300 + "POST", + f"/online-evaluators/{self._profile_id(scorer_id)}/tune", + json={"window": window}, + timeout=300, + retry=False, ) # The wire wraps the proposal ({"proposal": {...}}); unwrap like the legacy client so # proposal["reasoning"] / the criteria fields are directly addressable. @@ -344,11 +362,13 @@ def tune(self, scorer_id: str, window: str = "7d") -> dict: def validate_tuning(self, scorer_id: str, criteria: Dict[str, Any], window: str = "7d") -> dict: """Re-judge the disagreement + control cases with candidate criteria (LLM calls, slow).""" # The wire takes the criteria fields at the TOP level of the body, not nested. + # retry=False (judge-spending POST) - same posture as tune() above. return self._request( "POST", f"/online-evaluators/{self._profile_id(scorer_id)}/tune/validate", json={**criteria, "window": window}, timeout=600, + retry=False, ) def publish_tuning( @@ -382,7 +402,14 @@ def publish_tuning( payload["validation"] = validation_payload if force: payload["force"] = True - return self._request("POST", f"/online-evaluators/{self._profile_id(scorer_id)}/tune/publish", json=payload) + # retry=False: a non-idempotent write (each publish appends a rubric version) - + # MonitorClient.publish_online_evaluator_tuning's posture. + return self._request( + "POST", + f"/online-evaluators/{self._profile_id(scorer_id)}/tune/publish", + json=payload, + retry=False, + ) def ratings(self, scorer_id: str, window: str = "7d") -> "List[OnlineEvaluatorRatingPoint]": """Bucketed average-rating-over-time for this scorer's live checks - same typed points diff --git a/agentx/monitor/patterns.py b/agentx/monitor/patterns.py index 8e7a2b9..e04c007 100644 --- a/agentx/monitor/patterns.py +++ b/agentx/monitor/patterns.py @@ -137,16 +137,108 @@ def delete(self, pattern_id: str) -> None: retry=False, ) + # snake_case -> wire camelCase, same posture as rules.update: the engine reads only the + # camelCase key and silently keeps the stored value for anything it does not recognize - + # update(sample_rate=0.05) used to 200 with the rate unchanged. + _UPDATE_ALIASES = { + "sample_rate": "sampleRate", + "scope_mode": "scopeMode", + "agent_ids": "agentIds", + "detector_kind": "detectorKind", + "include_terms": "includeTerms", + "exclude_terms": "excludeTerms", + "regex": "regex", + "semantic_prompt": "semanticPrompt", + "match_mode": "matchMode", + "match_target": "matchTarget", + } + + # The engine's PUT rebuilds the pattern's WHOLE conditions array (legacyPayloadToConditions: + # "a full replace, not a sparse patch") exactly when the body carries one of these - its own + # sentConditionFields set, minus "conditions". Bodies without any of them keep the stored + # conditions untouched. + _TRIGGER_FIELDS = ("includeTerms", "regex", "semanticPrompt") + + # Sent alone, these look sparse but cannot land: without a trigger field the engine never + # rebuilds conditions (the values are silently ignored). For excludeTerms/matchMode there is + # also nothing to merge them over client-side - the wire GET reports display-only + # placeholders (includeTerms/excludeTerms always [], matchMode always "any"; conditions is + # the only truth). matchTarget IS reported faithfully, but the engine only reads it during a + # rebuild, so alone it is the same silent no-op. + _UNMERGEABLE_ALONE = ("excludeTerms", "matchMode", "matchTarget") + def update(self, pattern_id: str, **fields: Any) -> MonitorPattern: - """Update a pattern's fields in place (wire camelCase keys, passed through - verbatim - e.g. ``enabled=False``, ``conditions=[...]``) and return the - updated :class:`MonitorPattern`. retry=False: a non-idempotent server-side - write must not be re-fired on a lost response.""" + """Update a pattern and return the updated :class:`MonitorPattern`. Accepts snake_case + kwargs (``sample_rate=0.05``) or the wire's camelCase; an unrecognized snake_case key + raises instead of silently changing nothing. + + The real contract on self-host: the stored truth is the pattern's ``conditions`` array, + and the flat fields the wire GET returns are display-only placeholders (``includeTerms``/ + ``excludeTerms`` always ``[]``, ``matchMode`` always ``"any"``, ``regex``/ + ``semanticPrompt`` omitted). The engine's PUT rebuilds the WHOLE conditions array + whenever the body carries ``include_terms``, ``regex``, or ``semantic_prompt`` (or an + explicit ``conditions`` list, which wins outright). Consequences: + + - ``update(pid, regex=...)`` (or include_terms/semantic_prompt) is a full detector + rewrite; the detector kind follows the trigger field actually sent (``regex`` -> + ``"regex"``, ``semantic_prompt`` -> ``"semantic"``, ``include_terms`` -> + ``"contains"``), so cross-kind updates work, and this client back-fills only + ``matchTarget`` from the stored pattern so the rebuild keeps its target. It never + back-fills ``includeTerms``/``excludeTerms``/``matchMode`` - the GET values are + placeholders, and copying them in would destroy real conditions. + - ``exclude_terms=``, ``match_mode=``, or ``match_target=`` alone raises ValueError: + the engine silently ignores them without a rebuild. Pass ``conditions=[...]`` (built + from ``get(pattern_id).conditions``) instead, or pass them alongside the + include_terms/regex/semantic_prompt they should be rebuilt with. + - Everything else stays a sparse metadata edit that leaves the stored conditions + untouched.""" + payload: Dict[str, Any] = {} + for key, value in fields.items(): + wire_key = self._UPDATE_ALIASES.get(key, key) + if "_" in wire_key: + raise ValueError( + f"Unknown pattern field {key!r} - the engine reads camelCase keys and would " + "silently ignore this (see MonitorPattern for the field names)." + ) + payload[wire_key] = value + sends_conditions = "conditions" in payload + triggered = any(k in payload for k in self._TRIGGER_FIELDS) + if not sends_conditions and not triggered: + offending = [k for k in self._UNMERGEABLE_ALONE if k in payload] + if offending: + raise ValueError( + f"{' and '.join(offending)} cannot be updated on their own: the engine only " + "rebuilds a pattern's conditions when includeTerms/regex/semanticPrompt is " + "sent (alone they are silently ignored), and the wire GET returns display-" + "only placeholders (includeTerms/excludeTerms always [], matchMode always " + "'any'), so there is no stored value to merge them over. Pass " + "conditions=[...] built from get(pattern_id).conditions instead, or send " + "them alongside the trigger field they should be rebuilt with." + ) + if triggered and not sends_conditions: + stored = self.get(pattern_id) + # The rebuild's detector kind follows the trigger field actually sent - back-filling + # the STORED kind 400s a kind change (regex= on a "contains" pattern) and silently + # corrupts the inverse (include_terms= on a regex pattern would write phrase + # conditions while keeping detectorKind "regex"). + if "regex" in payload: + inferred_kind = "regex" + elif "semanticPrompt" in payload: + inferred_kind = "semantic" + else: + inferred_kind = "contains" + payload.setdefault("detectorKind", inferred_kind) + # matchTarget is the one field the wire reports faithfully - it rides along so the + # server-side full-replace rebuild keeps the stored target. NEVER includeTerms/ + # excludeTerms/matchMode - see _UNMERGEABLE_ALONE's comment. + payload.setdefault("matchTarget", stored.match_target) data = self._client._request( "PUT", f"/agent-monitoring/patterns/{pattern_id}", base=self._client._api_root(), - json=fields, + json=payload, + # An idempotent merge server-side (same payload, same result) - but the read- + # merge-write above is not atomic, so keep the single-shot posture. retry=False, ) return MonitorPattern(**data["pattern"]) diff --git a/agentx/monitor/rules.py b/agentx/monitor/rules.py index ebe5d84..60f18b1 100644 --- a/agentx/monitor/rules.py +++ b/agentx/monitor/rules.py @@ -70,9 +70,22 @@ def create( def update(self, rule_id: str, **fields: Any) -> MonitorRule: """Sparse update. snake_case keys are mapped to the wire (``sample_rate`` -> - ``sampleRate``, ``action_config`` -> ``actionConfig``).""" + ``sampleRate``, ``action_config`` -> ``actionConfig``); an unrecognized snake_case + key raises instead of 200ing with the rule unchanged (the engine reads only camelCase + and silently keeps the stored value for keys it does not know).""" aliases = {"sample_rate": "sampleRate", "action_config": "actionConfig"} - payload = {aliases.get(k, k): v for k, v in fields.items()} + payload: Dict[str, Any] = {} + for key, value in fields.items(): + wire_key = aliases.get(key, key) + if "_" in wire_key: + raise ValueError( + f"Unknown rule field {key!r} - the engine reads camelCase keys and would " + "silently ignore this (see MonitorRule for the field names)." + ) + payload[wire_key] = value + # PUT /rules/:id is an idempotent full-body merge (same payload, same result), so the + # transport's default retry is safe - and skipping it just drops legitimate edits on a + # transient failure. data = self._request("PUT", f"/agent-monitoring/rules/{rule_id}", json=payload) return MonitorRule(data.get("rule", data)) diff --git a/agentx/monitor/scorer_groups.py b/agentx/monitor/scorer_groups.py index d1b3983..f82cfee 100644 --- a/agentx/monitor/scorer_groups.py +++ b/agentx/monitor/scorer_groups.py @@ -10,6 +10,7 @@ import requests from agentx.exceptions import AgentXError, AgentXAuthError, AgentXValidationError +from agentx.monitor._transport import request_with_retries class AgentXScorerGroupsError(AgentXError): @@ -49,7 +50,9 @@ def __init__(self, api_key: str, base_url: str, workspace_id: Optional[str] = No self._workspace_id = workspace_id self._base = base_url.rstrip("/") + "/agent-monitoring/scorer-groups" - def _request(self, method: str, url: str, json: Optional[Dict[str, Any]] = None) -> Any: + def _request( + self, method: str, url: str, json: Optional[Dict[str, Any]] = None, retry: bool = True + ) -> Any: params = None if self._workspace_id: # Mirrors MonitorClient._workspace_params/_with_workspace: GETs (and DELETEs) @@ -61,9 +64,12 @@ def _request(self, method: str, url: str, json: Optional[Dict[str, Any]] = None) json = {**json, "workspaceId": self._workspace_id} else: params = {"workspaceId": self._workspace_id} - response = requests.request( + # retry=False for ANY non-idempotent write (creates, deletes) - + # MonitorClient._request's posture, via the shared monitor transport. + response = request_with_retries( method, url, + retry=retry, headers={"x-api-key": self._api_key, "content-type": "application/json"}, json=json, params=params, @@ -106,15 +112,33 @@ def create( payload["description"] = description if online is not None: payload["online"] = online - return ScorerGroup(self._request("POST", self._base, json=payload)["scorerGroup"]) + # Server-side create: a timeout + transport retry would create the group twice. + return ScorerGroup( + self._request("POST", self._base, json=payload, retry=False)["scorerGroup"] + ) def update(self, group_id: str, **fields: Any) -> ScorerGroup: - """Sparse update - pass any of name/description/members/online (online=None detaches - live scoring).""" + """Sparse update - pass any of name/description/members/online. ``online`` itself may be + partial: ``online={"enabled": False}`` pauses live scoring, the engine merges the patch + over the stored profile. ``online=None`` detaches live scoring entirely. A partial + ``online=`` patch on a group with NO stored live profile is rejected by the engine + (400) - send the full profile the first time (the shape ``create`` documents).""" + # Same guard patterns.update/rules.update carry: the engine's schema strips keys it + # does not recognize, so a snake_case key would 200 with the group unchanged. + for key in fields: + if "_" in key: + first, *rest = key.split("_") + camel = first + "".join(part.capitalize() for part in rest) + raise ValueError( + f"Unknown scorer group field {key!r} - the engine reads camelCase keys and " + f"would silently ignore this; send {camel!r} instead." + ) return ScorerGroup(self._request("PUT", f"{self._base}/{group_id}", json=fields)["scorerGroup"]) def delete(self, group_id: str) -> None: - self._request("DELETE", f"{self._base}/{group_id}") + # retry=False: a lost response + transport retry would turn a successful delete + # into a spurious 404. + self._request("DELETE", f"{self._base}/{group_id}", retry=False) def ratings(self, group_id: str, window: str = "7d") -> Dict[str, Any]: """Live score history for a group - ``{"window", "points": [{ts, averageRating, count}]}``, diff --git a/agentx/monitor/scorers.py b/agentx/monitor/scorers.py index 92bd649..5dda6b7 100644 --- a/agentx/monitor/scorers.py +++ b/agentx/monitor/scorers.py @@ -5,6 +5,7 @@ import requests +from agentx.monitor._transport import request_with_retries from agentx.util import api_base, get_headers from agentx.exceptions import AgentXError, AgentXAuthError, AgentXValidationError @@ -52,7 +53,9 @@ def __init__( # Captured once at construction (deep-dive round 3, bug #1). self._base_url = (base_url or api_base()).rstrip("/") - def _request(self, method: str, path: str, json: Any = None, params: Any = None) -> Any: + def _request( + self, method: str, path: str, json: Any = None, params: Any = None, retry: bool = True + ) -> Any: if self._workspace_id: # Mirrors MonitorClient._workspace_params/_with_workspace: GETs (and DELETEs) # carry workspaceId as a query param, write bodies carry it as a field. @@ -63,9 +66,12 @@ def _request(self, method: str, path: str, json: Any = None, params: Any = None) json = {**json, "workspaceId": self._workspace_id} elif not (params or {}).get("workspaceId"): params = {**(params or {}), "workspaceId": self._workspace_id} - resp = requests.request( + # retry=False for ANY non-idempotent write (creates, deletes, dry-run executions) - + # MonitorClient._request's posture, via the shared monitor transport. + resp = request_with_retries( method, f"{self._base_url}/agent-monitoring{path}", + retry=retry, headers={**get_headers(self._api_key), "Content-Type": "application/json"}, json=json, params=params, @@ -147,7 +153,8 @@ def create_code( ``None`` to skip; a score below ``alert_below`` raises a signal.""" if language not in ("python", "javascript"): raise AgentXScorersError('language must be "python" or "javascript"') - return self._request("POST", "/custom-evaluators", json={ + # Server-side create: a timeout + transport retry would deploy the scorer twice. + return self._request("POST", "/custom-evaluators", retry=False, json={ "name": name, "kind": "code", "language": language, @@ -173,7 +180,8 @@ def create_external( agent_ids: Optional[Sequence[str]] = None, ) -> Dict[str, Any]: """Register an external scorer endpoint (POSTed the v2 payload per sampled trace).""" - return self._request("POST", "/custom-evaluators", json={ + # Server-side create: a timeout + transport retry would register the scorer twice. + return self._request("POST", "/custom-evaluators", retry=False, json={ "name": name, "url": url, "sampleRate": sample_rate, @@ -191,7 +199,9 @@ def update(self, scorer_id: str, **fields: Any) -> Dict[str, Any]: return self._request("PUT", f"/custom-evaluators/{scorer_id}", json=wire)["evaluator"] def delete(self, scorer_id: str) -> None: - self._request("DELETE", f"/custom-evaluators/{scorer_id}") + # retry=False: a lost response + transport retry would turn a successful delete + # into a spurious 404. + self._request("DELETE", f"/custom-evaluators/{scorer_id}", retry=False) def events(self, scorer_id: str, window: str = "24h") -> List[Dict[str, Any]]: """The scorer's per-check history (score, matched, justification, trace ids).""" @@ -201,7 +211,9 @@ def dry_run(self, **payload: Any) -> Dict[str, Any]: """Execute a scorer against the built-in sample without persisting: pass either ``url=...`` (external) or ``kind="code", language=..., script=...`` (code).""" wire = {_SNAKE_TO_WIRE.get(k, k): v for k, v in payload.items()} - return self._request("POST", "/custom-evaluators/dry-run", json=wire) + # Executes real scorer work (and, for external, hits the user's endpoint) - a + # client-side timeout must not fire it twice. + return self._request("POST", "/custom-evaluators/dry-run", json=wire, retry=False) _SNAKE_TO_WIRE = { diff --git a/agentx/resources/agent.py b/agentx/resources/agent.py index aff597e..4b251bb 100644 --- a/agentx/resources/agent.py +++ b/agentx/resources/agent.py @@ -1,14 +1,12 @@ from typing import Optional, List -from pydantic import BaseModel, Field +from pydantic import BaseModel, PrivateAttr, Field import requests import os import logging -from dataclasses import dataclass from agentx.util import get_headers, api_base from .conversation import Conversation -@dataclass class Agent(BaseModel): id: str = Field(alias="_id") name: str @@ -16,16 +14,35 @@ class Agent(BaseModel): createdAt: Optional[str] = None updatedAt: Optional[str] = None + + # Hosted credentials threaded from the constructing AgentX client. The module-level + # api_base()/get_headers() read only env vars, and the client deliberately stopped + # writing its constructor args into os.environ - without binding, a client created with + # api_key=/base_url= issued these calls unauthenticated against the default host. + _api_key: Optional[str] = PrivateAttr(default=None) + _base_url: Optional[str] = PrivateAttr(default=None) + + def _bind(self, api_key: Optional[str], base_url: Optional[str]) -> "Agent": + self._api_key = api_key + self._base_url = base_url + return self + + def _api_base(self) -> str: + return self._base_url or api_base() + + def _headers(self): + return get_headers(self._api_key) + def __init__(self, **data): super().__init__(**data) def new_conversation(self) -> Conversation: - url = f"{api_base()}/access/agents/{self.id}/conversations/new" - response = requests.post(url, headers=get_headers(), json={"type": "chat"}) + url = f"{self._api_base()}/access/agents/{self.id}/conversations/new" + response = requests.post(url, headers=self._headers(), json={"type": "chat"}) if response.status_code == 200: data = response.json() data["agent_id"] = self.id - return Conversation(**data) + return Conversation(**data)._bind(self._api_key, self._base_url) else: raise Exception(f"Failed to create conversation: {response.reason}") @@ -37,8 +54,8 @@ def get_conversation(self, id: str) -> Conversation: ) def list_conversations(self) -> List[Conversation]: - url = f"{api_base()}/access/agents/{self.id}/conversations" - response = requests.get(url, headers=get_headers()) + url = f"{self._api_base()}/access/agents/{self.id}/conversations" + response = requests.get(url, headers=self._headers()) if response.status_code == 200: return [ Conversation( @@ -49,7 +66,9 @@ def list_conversations(self) -> List[Conversation]: agents=conv_res.get("bots"), createdAt=conv_res.get("createdAt"), updatedAt=conv_res.get("updatedAt"), - ) + # Bound to this agent's own credentials (threaded in via Agent._bind) so + # the conversation's calls authenticate the same way this listing did. + )._bind(self._api_key, self._base_url) for conv_res in response.json() ] else: diff --git a/agentx/resources/conversation.py b/agentx/resources/conversation.py index cd343e2..78a4605 100644 --- a/agentx/resources/conversation.py +++ b/agentx/resources/conversation.py @@ -1,7 +1,7 @@ import json import requests from typing import Optional, List, Any, Iterator -from pydantic import BaseModel, Field +from pydantic import BaseModel, PrivateAttr, Field from agentx.util import get_headers, api_base @@ -42,28 +42,49 @@ class Config: populate_by_name = True extra = "ignore" + + # Hosted credentials threaded from the constructing AgentX client. The module-level + # api_base()/get_headers() read only env vars, and the client deliberately stopped + # writing its constructor args into os.environ - without binding, a client created with + # api_key=/base_url= issued these calls unauthenticated against the default host. + _api_key: Optional[str] = PrivateAttr(default=None) + _base_url: Optional[str] = PrivateAttr(default=None) + + def _bind(self, api_key: Optional[str], base_url: Optional[str]) -> "Conversation": + self._api_key = api_key + self._base_url = base_url + return self + + def _api_base(self) -> str: + return self._base_url or api_base() + + def _headers(self): + return get_headers(self._api_key) + def __init__(self, **data): super().__init__(**data) def new_conversation(self) -> "Conversation": - url = f"{api_base()}/access/agents/{self.agent_id}/conversations/new" + url = f"{self._api_base()}/access/agents/{self.agent_id}/conversations/new" response = requests.post( url, - headers=get_headers(), + headers=self._headers(), json={"type": "chat"}, ) if response.status_code == 200: new_conv = response.json() new_conv["agent_id"] = self.agent_id - return Conversation(**new_conv) + # Bound to this conversation's own credentials - unbound, the sibling + # conversation fell back to env credentials against the default host. + return Conversation(**new_conv)._bind(self._api_key, self._base_url) else: raise Exception( f"Failed to create new conversation: {response.status_code} - {response.reason}" ) def list_messages(self) -> List[Message]: - url = f"{api_base()}/access/agents/{self.agent_id}/conversations/{self.id}" - response = requests.get(url, headers=get_headers()) + url = f"{self._api_base()}/access/agents/{self.agent_id}/conversations/{self.id}" + response = requests.get(url, headers=self._headers()) if response.status_code == 200: res = response.json() if res.get("messages"): @@ -83,18 +104,18 @@ def list_messages(self) -> List[Message]: ) def chat(self, message: str, context: Optional[int] = None): - url = f"{api_base()}/access/conversations/{self.id}/message" + url = f"{self._api_base()}/access/conversations/{self.id}/message" response = requests.post( url, - headers=get_headers(), + headers=self._headers(), json={"message": message, "context": context}, ) return response.json() def chat_stream(self, message: str, context: Optional[int] = None) -> Iterator[ChatResponse]: - url = f"{api_base()}/access/conversations/{self.id}/jsonmessagesse" + url = f"{self._api_base()}/access/conversations/{self.id}/jsonmessagesse" response = requests.post( - url, headers=get_headers(), json={"message": message, "context": context} + url, headers=self._headers(), json={"message": message, "context": context} ) result = "" if response.status_code == 200: diff --git a/agentx/resources/workforce.py b/agentx/resources/workforce.py index fc3b60b..0f66302 100644 --- a/agentx/resources/workforce.py +++ b/agentx/resources/workforce.py @@ -1,5 +1,5 @@ from typing import Optional, List, Dict, Any, Iterator -from pydantic import BaseModel, Field +from pydantic import BaseModel, PrivateAttr, Field import requests import os import json @@ -46,19 +46,44 @@ class Config: populate_by_name = True extra = "ignore" + + # Hosted credentials threaded from the constructing AgentX client. The module-level + # api_base()/get_headers() read only env vars, and the client deliberately stopped + # writing its constructor args into os.environ - without binding, a client created with + # api_key=/base_url= issued these calls unauthenticated against the default host. + _api_key: Optional[str] = PrivateAttr(default=None) + _base_url: Optional[str] = PrivateAttr(default=None) + + def _bind(self, api_key: Optional[str], base_url: Optional[str]) -> "Workforce": + self._api_key = api_key + self._base_url = base_url + # The nested Agent objects issue their own calls (new_conversation, + # list_conversations) - left unbound they silently fall back to env credentials + # against the default host, the exact leak _bind exists to close. + self.manager._bind(api_key, base_url) + for agent in self.agents: + agent._bind(api_key, base_url) + return self + + def _api_base(self) -> str: + return self._base_url or api_base() + + def _headers(self): + return get_headers(self._api_key) + def new_conversation(self) -> Conversation: """Create a new conversation for this workforce.""" - url = f"{api_base()}/access/teams/{self.id}/conversations/new" + url = f"{self._api_base()}/access/teams/{self.id}/conversations/new" response = requests.post( url, - headers=get_headers(), + headers=self._headers(), json={"type": "chat"}, ) if response.status_code == 200: conv_data = response.json() # Set the agent_id to the manager's ID since this is a workforce conversation conv_data["agent_id"] = self.manager.id - return Conversation(**conv_data) + return Conversation(**conv_data)._bind(self._api_key, self._base_url) else: raise Exception( f"Failed to create new conversation: {response.status_code} - {response.reason}" @@ -66,14 +91,14 @@ def new_conversation(self) -> Conversation: def list_conversations(self) -> List[Conversation]: """List all conversations for this workforce.""" - url = f"{api_base()}/access/teams/{self.id}/conversations" - response = requests.get(url, headers=get_headers()) + url = f"{self._api_base()}/access/teams/{self.id}/conversations" + response = requests.get(url, headers=self._headers()) if response.status_code == 200: conversations = [] for conv_data in response.json(): # Set the agent_id to the manager's ID since this is a workforce conversation conv_data["agent_id"] = self.manager.id - conversations.append(Conversation(**conv_data)) + conversations.append(Conversation(**conv_data)._bind(self._api_key, self._base_url)) return conversations else: raise Exception( @@ -85,10 +110,10 @@ def chat_stream( ) -> Iterator[ChatResponse]: """Send a message to a team conversation and stream the response.""" url = ( - f"{api_base()}/access/teams/conversations/{conversation_id}/jsonmessagesse" + f"{self._api_base()}/access/teams/conversations/{conversation_id}/jsonmessagesse" ) response = requests.post( - url, headers=get_headers(), json={"message": message, "context": context} + url, headers=self._headers(), json={"message": message, "context": context} ) result = "" if response.status_code == 200: diff --git a/agentx/tracing/tracer.py b/agentx/tracing/tracer.py index 9156789..4e7b1b4 100644 --- a/agentx/tracing/tracer.py +++ b/agentx/tracing/tracer.py @@ -931,9 +931,11 @@ def record_memory( Deliberately NOT a retrieval: retrieval spans feed the RAG judges' ``{context}`` (knowledge grounding), while memory is recalled state - see the engine's spanKind.ts. - With no active span the record is DROPPED (with a debug log), not queued: the only - pending queue rides the next trace's ``retrieval_steps``, and memory content must - never reach the engine's retrieval-context extraction for RAG judges. Wrap the call + With no active span the record is DROPPED (warns once per process, then logs at + debug), not queued. There are two pending queues (tool calls -> ``tool_calls``, + retrievals -> ``retrieval_steps``) and neither fits: ``retrieval_steps`` feeds the + engine's RAG ``{context}`` extraction, which recalled state must never reach, and no + memory-shaped queue has been built yet. Wrap the call in ``tracer.trace()`` to keep it - or, on a worker thread, wrap the worker body in ``tracer.use_span(span)`` - a bare thread starts with an empty span stack. (``record_tool_call``/``record_retrieval`` queue instead - see their docstrings.) @@ -980,7 +982,7 @@ def trace_memory( with tracer.trace_memory("user prefs", operation="read", query=user_id) as m: m.output = memory.search(user_id, question) - With no active span the record is DROPPED (with a debug log), not queued - see + With no active span the record is DROPPED (warns once per process, then logs at debug), not queued - see :meth:`record_memory`. An exception escaping the block records the operation as failed (error set, output ``ERROR: ...``) and then propagates unchanged - same posture as :meth:`trace_tool_call`. diff --git a/agentx/util.py b/agentx/util.py index 40eddae..043e2b0 100644 --- a/agentx/util.py +++ b/agentx/util.py @@ -1,6 +1,8 @@ import os from typing import Optional +from agentx.exceptions import AgentXAuthError + _DEFAULT_API_BASE = "https://api.agentx.so/api/v1" _EVALUATIONS_SUFFIX = "/custom-agent-evaluations" @@ -26,4 +28,10 @@ def api_base() -> str: def get_headers(api_key: Optional[str] = None): - return {"accept": "*/*", "x-api-key": api_key or os.getenv("AGENTX_API_KEY")} + key = api_key or os.getenv("AGENTX_API_KEY") + if not key: + # A None header serializes as the literal string "None" (or drops), turning a config + # mistake into an opaque 401 from the server - fail loud at the call site instead, + # with the SDK's canonical auth error so `except agentx.AgentXAuthError` catches it. + raise AgentXAuthError("No API key: pass api_key= or set AGENTX_API_KEY") + return {"accept": "*/*", "x-api-key": key} diff --git a/tests/test_judge_scorers.py b/tests/test_judge_scorers.py index 63dc64b..20fa1dd 100644 --- a/tests/test_judge_scorers.py +++ b/tests/test_judge_scorers.py @@ -302,7 +302,7 @@ def test_validate_and_publish_send_criteria_at_top_level(monkeypatch): monkeypatch.setattr(client, "_profile_id", lambda scorer_id: "prof-1") captured = {} - def fake_request(method, path, json=None, timeout=60): + def fake_request(method, path, json=None, timeout=60, retry=True): captured[path.rsplit("/", 1)[-1]] = json return {} diff --git a/tests/test_pattern_update_merge.py b/tests/test_pattern_update_merge.py new file mode 100644 index 0000000..36f06e0 --- /dev/null +++ b/tests/test_pattern_update_merge.py @@ -0,0 +1,200 @@ +"""patterns.update against the engine's REAL wire shape: the GET body's flat condition fields +are display-only placeholders (toWire hardcodes includeTerms/excludeTerms to [] and matchMode to +"any"; regex/semanticPrompt are omitted entirely - conditions is the only truth), and the PUT +rebuilds the WHOLE conditions array exactly when the body carries includeTerms/regex/ +semanticPrompt (the engine's sentConditionFields set, minus conditions). + +So update() must infer detectorKind from the trigger field actually sent (regex -> "regex", +semantic_prompt -> "semantic", else "contains" - back-filling the STORED kind 400s a kind +change and corrupts the inverse), back-fill ONLY matchTarget from the stored row, never copy +the placeholder includeTerms/excludeTerms/matchMode into the payload (the old merge destroyed +real conditions that way), and refuse exclude_terms=/match_mode=/match_target= sent alone +(alone the engine silently ignores them).""" + +from unittest.mock import MagicMock + +import pytest + +from agentx.monitor.patterns import MonitorPattern, MonitorPatternClient + + +def _wire_pattern(**overrides): + """A GET /patterns/:id body exactly as the self-host engine's toWire emits it: flat + condition fields are placeholders, the real rule lives in `conditions`.""" + body = { + "_id": "p1", + "workspaceId": "local", + "key": "ssn-in-response", + "name": "SSN in response", + "source": "custom", + "detectorKind": "regex", + "matchTarget": ["response"], + "matchMode": "any", # toWire hardcodes this + "includeTerms": [], # toWire hardcodes this + "excludeTerms": [], # toWire hardcodes this + # No "regex" / "semanticPrompt" keys at all - the wire omits them. + "conditions": [ + { + "connector": "and", + "negate": False, + "sources": ["response"], + "detector": "regex", + "value": r"\d{3}-\d{2}-\d{4}", + "caseSensitive": False, + } + ], + "severity": "high", + "polarity": "failure", + "enabled": True, + "sampleRate": 1.0, + "scopeMode": "all", + "agentIds": [], + "readOnly": False, + "createdAt": "2026-01-01T00:00:00.000Z", + "updatedAt": "2026-01-01T00:00:00.000Z", + } + body.update(overrides) + return body + + +CONTAINS_WIRE = _wire_pattern( + key="refund-promise", + name="Refund promise", + detectorKind="contains", + conditions=[ + { + "connector": "and", + "negate": False, + "sources": ["response"], + "detector": "phrase", + "value": "guaranteed refund", + "caseSensitive": False, + }, + { + "connector": "or", + "negate": False, + "sources": ["response"], + "detector": "phrase", + "value": "money back", + "caseSensitive": False, + }, + ], +) + + +def make_client(wire_body=None): + inner = MagicMock() + inner._api_root.return_value = "http://x/api/v1" + client = MonitorPatternClient(inner) + body = wire_body or _wire_pattern() + client.get = MagicMock(return_value=MonitorPattern(**body)) # type: ignore[method-assign] + inner._request.return_value = {"pattern": body} + return client, inner + + +def test_regex_only_update_sends_regex_detector_kind_and_match_target(): + client, inner = make_client() + client.update("p1", regex=r"\b\d{9}\b") + payload = inner._request.call_args.kwargs["json"] + assert payload == { + "regex": r"\b\d{9}\b", + "detectorKind": "regex", + "matchTarget": ["response"], + } + client.get.assert_called_once_with("p1") # type: ignore[union-attr] + + +def test_include_terms_update_never_sends_placeholder_siblings(): + client, inner = make_client(CONTAINS_WIRE) + client.update("p1", include_terms=["guaranteed refund", "money back", "full refund"]) + payload = inner._request.call_args.kwargs["json"] + assert payload["includeTerms"] == ["guaranteed refund", "money back", "full refund"] + assert payload["detectorKind"] == "contains" + assert payload["matchTarget"] == ["response"] + # The wire GET's excludeTerms/matchMode are display-only placeholders ([] / "any") - + # copying them in is exactly the merge that destroyed conditions. + assert "excludeTerms" not in payload + assert "matchMode" not in payload + + +def test_cross_kind_update_sends_the_sent_fields_kind_not_the_stored_one(): + """update(pid, regex=...) on a stored "contains" pattern must send detectorKind "regex" - + back-filling the stored kind 400s the change (and the inverse, include_terms= on a regex + pattern, would silently write phrase conditions under detectorKind "regex").""" + client, inner = make_client(CONTAINS_WIRE) + client.update("p1", regex=r"\b\d{9}\b") + payload = inner._request.call_args.kwargs["json"] + assert payload == { + "regex": r"\b\d{9}\b", + "detectorKind": "regex", + "matchTarget": ["response"], + } + + +def test_semantic_prompt_update_sends_semantic_kind(): + client, inner = make_client(CONTAINS_WIRE) + client.update("p1", semantic_prompt="The response promises a refund.") + payload = inner._request.call_args.kwargs["json"] + assert payload["detectorKind"] == "semantic" + + +def test_match_target_only_raises_instead_of_silently_doing_nothing(): + """matchTarget sent alone never lands: the engine only reads it during a conditions + rebuild, and a body without a trigger field never rebuilds.""" + client, inner = make_client() + with pytest.raises(ValueError, match="matchTarget"): + client.update("p1", match_target=["input", "response"]) + inner._request.assert_not_called() + client.get.assert_not_called() # type: ignore[union-attr] + + +def test_exclude_terms_only_raises_instead_of_silently_doing_nothing(): + client, inner = make_client() + with pytest.raises(ValueError, match="conditions"): + client.update("p1", exclude_terms=["as an AI"]) + inner._request.assert_not_called() + client.get.assert_not_called() # type: ignore[union-attr] + + +def test_match_mode_only_raises_too(): + client, inner = make_client() + with pytest.raises(ValueError, match="matchMode"): + client.update("p1", match_mode="all") + inner._request.assert_not_called() + + +def test_exclude_terms_alongside_a_trigger_field_is_allowed(): + """exclude_terms rides along fine when the rebuild actually happens (a trigger field is + present) - the engine folds them into the rebuilt conditions.""" + client, inner = make_client(CONTAINS_WIRE) + client.update("p1", include_terms=["guaranteed refund"], exclude_terms=["hypothetically"]) + payload = inner._request.call_args.kwargs["json"] + assert payload["includeTerms"] == ["guaranteed refund"] + assert payload["excludeTerms"] == ["hypothetically"] + assert "matchMode" not in payload + + +def test_explicit_conditions_win_outright_no_read_back(): + stored_conditions = [ + { + "connector": "and", + "negate": False, + "sources": ["response"], + "detector": "phrase", + "value": "wire transfer", + "caseSensitive": False, + } + ] + client, inner = make_client() + client.update("p1", conditions=stored_conditions, exclude_terms=["test"]) + payload = inner._request.call_args.kwargs["json"] + assert payload["conditions"] == stored_conditions + client.get.assert_not_called() # type: ignore[union-attr] + + +def test_non_condition_edit_stays_sparse(): + client, inner = make_client() + client.update("p1", sample_rate=0.05) + payload = inner._request.call_args.kwargs["json"] + assert payload == {"sampleRate": 0.05} + client.get.assert_not_called() # type: ignore[union-attr] diff --git a/tests/test_span_tree.py b/tests/test_span_tree.py index f71f0f4..8887dbc 100644 --- a/tests/test_span_tree.py +++ b/tests/test_span_tree.py @@ -12,6 +12,7 @@ """ from __future__ import annotations +import logging from unittest.mock import MagicMock import pytest @@ -770,7 +771,8 @@ def test_record_memory_with_no_active_span_drops_instead_of_queueing(): """Memory must never ride the pending retrieval queue: it lands in performance_summary.retrieval_steps, which feeds the engine's retrieval-context extraction for RAG judges - recalled state is not knowledge grounding. With no active - span the record is dropped (debug-logged), not attached to the next trace.""" + span the record is dropped (warns once per process, then debug), not attached to the + next trace.""" tracer = make_tracer() tracer.record_memory("orphan prefs", operation="read", query="u-1", output="secret") with tracer.trace("agent"): @@ -781,6 +783,33 @@ def test_record_memory_with_no_active_span_drops_instead_of_queueing(): assert_no_performance_summary(wires) +@pytest.fixture() +def reset_memory_warn_flag(): + """The warn-once flag is process-global; reset it around the test so the assertion holds + regardless of which earlier test (or test ordering) already tripped it.""" + import agentx.tracing.tracer as tracer_mod + + tracer_mod._WARNED_MEMORY_NO_SPAN = False + yield + tracer_mod._WARNED_MEMORY_NO_SPAN = False + + +def test_record_memory_no_span_warns_once_then_goes_quiet(reset_memory_warn_flag, caplog): + """A dropped memory record is a lost write: the FIRST orphan record_memory must log one + WARNING (visible by default), and every later one only debug-logs - otherwise a worker + thread without use_span read as 'memory spans don't work' with no log line anywhere.""" + tracer = make_tracer() + with caplog.at_level(logging.WARNING, logger="agentx.tracing.tracer"): + tracer.record_memory("orphan-1", operation="read", query="u-1") + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 + assert "no active span" in warnings[0].getMessage() + + caplog.clear() + tracer.record_memory("orphan-2", operation="read", query="u-2") + assert [r for r in caplog.records if r.levelno == logging.WARNING] == [] + + def test_merge_child_run_execution_steps_honor_their_stated_kind(): """crewai.py stamps its task steps "kind": "agent" - the merge loop must forward that instead of unconditionally stamping every step "llm" (autogen/llamaindex steps carry no diff --git a/tests/test_workforce_binding.py b/tests/test_workforce_binding.py new file mode 100644 index 0000000..c96cf86 --- /dev/null +++ b/tests/test_workforce_binding.py @@ -0,0 +1,71 @@ +"""Regressions for the workforce credential plumbing: + +- AgentX.list_workforces was a @staticmethod whose body referenced ``self`` - any non-empty + response raised NameError before a single Workforce was constructed. It is an instance + method now, binding the client's own credentials into each result. +- Workforce._bind used to bind only the workforce itself: its nested manager/agents issued + their calls with env-fallback credentials against the default host. +""" + +from unittest.mock import MagicMock, patch + +from agentx import AgentX + +_USER_WIRE = { + "_id": "u1", + "name": "Robin", + "email": "robin@example.com", + "deleted": False, + "createdAt": "2026-01-01T00:00:00.000Z", + "updatedAt": "2026-01-01T00:00:00.000Z", + "avatar": "", + "status": 1, + "customer": "c1", +} + +_WORKFORCE_WIRE = { + "_id": "wf1", + "name": "Support crew", + "image": "", + "description": "handles tickets", + "agents": [ + {"_id": "a1", "name": "Triage"}, + {"_id": "a2", "name": "Resolver"}, + ], + "manager": {"_id": "m1", "name": "Manager"}, + "creator": _USER_WIRE, + "context": 5, + "references": True, + "createdAt": "2026-01-01T00:00:00.000Z", + "updatedAt": "2026-01-01T00:00:00.000Z", +} + + +def _ok_response(payload): + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = payload + return resp + + +def test_list_workforces_no_nameerror_and_binds_workforce_and_children(): + with AgentX(api_key="k-wf", base_url="http://engine:9999/api/v1") as client: + with patch( + "agentx.agentx.requests.get", return_value=_ok_response([_WORKFORCE_WIRE]) + ) as mock_get: + workforces = client.list_workforces() # used to NameError on any non-empty body + + assert mock_get.call_args.args[0] == "http://engine:9999/api/v1/access/teams" + assert mock_get.call_args.kwargs["headers"]["x-api-key"] == "k-wf" + + assert len(workforces) == 1 + wf = workforces[0] + assert wf._api_key == "k-wf" + assert wf._base_url == "http://engine:9999/api/v1" + + # Children are bound too - they issue their own authenticated calls. + bound_children = [wf.manager, *wf.agents] + assert len(bound_children) == 3 + for child in bound_children: + assert child._api_key == "k-wf" + assert child._base_url == "http://engine:9999/api/v1"