Merge remote-tracking branch 'origin/next' into release-please--branches--main--changes--next

Author: declan-scale

.stats.yml

@@ -1,4 +1,4 @@
 configured_endpoints: 75
-openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/sgp/agentex-sdk-ee0c521f0612c31b874bd595b90cd9209545bab603983552a1a7a87f38ed931e.yml
-openapi_spec_hash: 917a1ffe9e353bed2740524dec786ed2
+openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/sgp/agentex-sdk-fbc0683871d6abb03588f30d9468eeeeacb2b8538eb0c9002813e6df68c5802b.yml
+openapi_spec_hash: 4ecd8d496f056dccf80826264ddf8fe1
 config_hash: 593e89b291976a5e84e4c3c3f8324354

src/agentex/lib/cli/debug/debug_handlers.py

@@ -16,6 +16,7 @@
     pass
 
 from agentex.lib.utils.logging import make_logger
+from agentex.lib.cli.utils.cli_utils import SUBPROCESS_STREAM_LIMIT
 
 from .debug_config import DebugConfig, resolve_debug_port
 
@@ -66,6 +67,7 @@ async def start_temporal_worker_debug(
         env=debug_env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 
@@ -119,6 +121,7 @@ async def start_acp_server_debug(
         env=debug_env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 

src/agentex/lib/cli/handlers/run_handlers.py

@@ -12,6 +12,7 @@
 from agentex.lib.cli.debug import DebugConfig, start_acp_server_debug, start_temporal_worker_debug
 from agentex.lib.utils.logging import make_logger
 from agentex.config.agent_manifest import AgentManifest
+from agentex.lib.cli.utils.cli_utils import SUBPROCESS_STREAM_LIMIT
 from agentex.lib.cli.utils.path_utils import (
     get_file_paths,
     calculate_uvicorn_target_for_local,
@@ -23,6 +24,11 @@
 logger = make_logger(__name__)
 console = Console()
 
+# How many consecutive unreadable lines to skip before giving up on the stream.
+# Skipping is only known-safe for the limit-overrun case; this bounds the damage
+# if some other error repeats without consuming anything.
+MAX_CONSECUTIVE_READ_ERRORS = 100
+
 
 class RunError(Exception):
     """An error occurred during agent run"""
@@ -215,6 +221,7 @@ async def start_acp_server(
         env=env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 
@@ -234,23 +241,68 @@ async def start_temporal_worker(
         env=env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 
 async def stream_process_output(process: asyncio.subprocess.Process, prefix: str):
-    """Stream process output with prefix"""
+    """Stream process output with prefix.
+
+    This loop is the only reader of the child's stdout pipe. If it ever stops
+    reading, the pipe fills and the child blocks forever inside ``write()``,
+    which presents as a silent freeze: 0% CPU, no further logs, no traceback.
+    So a single unreadable line must never end the loop.
+    """
     try:
         if process.stdout is None:
             return
+        consecutive_read_errors = 0
         while True:
-            line = await process.stdout.readline()
+            try:
+                line = await process.stdout.readline()
+            except ValueError as e:
+                # readline() raises ValueError when a line exceeds the stream limit.
+                # In *that* case it has already discarded the line and resumed the
+                # transport, so skipping it makes guaranteed progress. Any other
+                # ValueError carries no such guarantee, and retrying it forever would
+                # spin without draining. We cannot tell the two apart (readline
+                # flattens LimitOverrunError into a bare ValueError), so bound the
+                # retries and let the outer handler report the hang risk.
+                consecutive_read_errors += 1
+                if consecutive_read_errors > MAX_CONSECUTIVE_READ_ERRORS:
+                    raise
+                logger.warning(
+                    f"Skipping an unreadable line from {prefix}: {e!r} "
+                    f"(consecutive failure {consecutive_read_errors}/{MAX_CONSECUTIVE_READ_ERRORS}). "
+                    f"If this says the chunk exceeded the limit, raise limit= on this "
+                    f"process's create_subprocess_exec."
+                )
+                continue
+
+            consecutive_read_errors = 0
+
             if not line:
                 break
-            decoded_line = line.decode("utf-8").rstrip()
+
+            try:
+                decoded_line = line.decode("utf-8").rstrip()
+            except UnicodeDecodeError as e:
+                logger.warning(f"Dropped an undecodable log line from {prefix} ({e}).")
+                continue
+
             if decoded_line:  # Only print non-empty lines
                 console.print(f"[dim]{prefix}:[/dim] {decoded_line}")
     except Exception as e:
-        logger.debug(f"Output streaming ended for {prefix}: {e}")
+        # The escalation path, including for the re-raise above. Anything reaching
+        # here ends the loop, so the child is now at risk of blocking on a full pipe.
+        # Warning rather than debug: this used to be a debug() that make_logger could
+        # never emit, which is why three freezes produced no clue.
+        # CancelledError derives from BaseException, so the auto-reload path that
+        # cancels these tasks passes straight through and is unaffected.
+        logger.warning(
+            f"Output streaming for {prefix} stopped on {e!r}. "
+            f"Nothing is draining its stdout now, so {prefix} will hang once the pipe fills."
+        )
 
 
 async def run_agent(manifest_path: str, debug_config: "DebugConfig | None" = None):

src/agentex/lib/cli/utils/cli_utils.py

@@ -5,6 +5,18 @@
 
 console = Console()
 
+# asyncio's StreamReader defaults to 64 KiB, and a single log line above that makes
+# readline() raise. Agents legitimately emit large lines (serialized charts, payloads
+# echoed back by validation errors), so give the reader room before it has to drop one.
+#
+# Lives here rather than beside its users so that both the normal spawns in
+# cli/handlers/run_handlers.py and the debug spawns in cli/debug/debug_handlers.py can
+# import it: run_handlers imports cli.debug, so the constant cannot live in either one.
+# Keep the two in step. A subprocess left on the asyncio default overruns far more
+# easily, and enough consecutive overruns exhaust the reader's retry bound and stop it
+# draining, which is the deadlock the bound is there to avoid.
+SUBPROCESS_STREAM_LIMIT = 8 * 1024 * 1024
+
 
 def handle_questionary_cancellation(
     result: str | None, operation: str = "operation"

src/agentex/lib/core/temporal/workers/worker.py

@@ -178,6 +178,7 @@ def __init__(
         metrics_headers: dict[str, str] | None = None,
         metrics_use_http: bool = False,
         metrics_temporality_delta: bool = False,
+        agent_card: Any | None = None,
     ):
         self.task_queue = task_queue
         self.activity_handles = []
@@ -196,6 +197,7 @@ def __init__(
         self.metrics_temporality_delta = metrics_temporality_delta
         self.payload_codec = payload_codec
         self.data_converter = data_converter
+        self.agent_card = agent_card
 
     @overload
     async def run(
@@ -312,6 +314,6 @@ async def _register_agent(self):
             # the worker process never goes through the ACP server lifespan, so it needs its
             # own guard (mirrors base_acp_server.lifespan_context).
             await assert_backend_compatible(env_vars.AGENTEX_BASE_URL)
-            await register_agent(env_vars)
+            await register_agent(env_vars, agent_card=self.agent_card)
         else:
             logger.warning("AGENTEX_BASE_URL not set, skipping worker registration")

src/agentex/lib/types/agent_card.py

@@ -5,7 +5,7 @@
 from enum import Enum
 from typing import TYPE_CHECKING, Any, get_args, get_origin
 
-from pydantic import BaseModel
+from pydantic import Field, BaseModel
 
 if TYPE_CHECKING:
     from agentex.lib.sdk.state_machine.state import State
@@ -31,6 +31,11 @@ class AgentCard(BaseModel):
     data_events: list[str] = []
     input_types: list[str] = []
     output_schema: dict | None = None
+    # Free-form JSON object for opt-in self-description (e.g. protocol-specific
+    # capability flags). Not interpreted by the platform, but callers can filter
+    # agents on it with ``agents.list(agent_card_metadata=...)`` -- see
+    # ``agentex.lib.utils.metadata_filters.encode_metadata_filter``.
+    metadata: dict[str, Any] = Field(default_factory=dict)
 
     @classmethod
     def from_states(
@@ -40,6 +45,7 @@ def from_states(
         output_event_model: type[BaseModel] | None = None,
         extra_input_types: list[str] | None = None,
         queries: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
     ) -> AgentCard:
         """Build an AgentCard directly from a list[State] + initial_state.
 
@@ -81,6 +87,7 @@ def from_states(
             data_events=data_events,
             input_types=sorted(derived_input_types | set(extra_input_types or [])),
             output_schema=output_schema,
+            metadata=metadata or {},
         )
 
     @classmethod
@@ -90,6 +97,7 @@ def from_state_machine(
         output_event_model: type[BaseModel] | None = None,
         extra_input_types: list[str] | None = None,
         queries: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
     ) -> AgentCard:
         """Build an AgentCard from a StateMachine instance. Delegates to from_states()."""
         lifecycle = state_machine.get_lifecycle()
@@ -125,6 +133,7 @@ def from_state_machine(
             data_events=data_events,
             input_types=sorted(derived_input_types | set(extra_input_types or [])),
             output_schema=output_schema,
+            metadata=metadata or {},
         )
 
 

src/agentex/lib/utils/logging.py

@@ -11,6 +11,25 @@
 
 ctx_var_request_id = contextvars.ContextVar[str]("request_id")
 
+DEFAULT_LOG_LEVEL = logging.INFO
+
+
+def resolve_log_level() -> int:
+    """Read the log level from ``LOG_LEVEL``, falling back to INFO.
+
+    Read straight from the environment rather than through ``EnvVarKeys``, since
+    ``environment_variables`` imports this module and the reverse would be a cycle.
+
+    ``getLevelName`` returns the string ``"Level FOO"`` for anything it does not
+    recognise, so the isinstance check is what stops a typo in ``LOG_LEVEL`` from
+    silently turning logging off.
+    """
+    configured = os.getenv("LOG_LEVEL")
+    if not configured:
+        return DEFAULT_LOG_LEVEL
+    level = logging.getLevelName(configured.strip().upper())
+    return level if isinstance(level, int) else DEFAULT_LOG_LEVEL
+
 
 class CustomJSONFormatter(json_log_formatter.JSONFormatter):
     def json_record(self, message: str, extra: dict, record: logging.LogRecord) -> dict:  # type: ignore[override]
@@ -51,7 +70,7 @@ def make_logger(name: str) -> logging.Logger:
     """
     # Create a console object to print colored text
     logger = logging.getLogger(name)
-    logger.setLevel(logging.INFO)
+    logger.setLevel(resolve_log_level())
 
     environment = os.getenv("ENVIRONMENT")
     if environment == "local":

src/agentex/lib/utils/metadata_filters.py

@@ -0,0 +1,58 @@
+"""Helpers for the platform's JSON-encoded metadata filter query parameters.
+
+The containment filters on ``agents.list(agent_card_metadata=...)`` and
+``tasks.list(task_metadata=...)`` carry their filter as a JSON-encoded object
+inside a single query string value, so the generated clients type them as
+``str``. Encoding by hand is easy to get subtly wrong -- Python's ``json``
+happily emits ``NaN``/``Infinity``, which the server rejects with a 400 -- so
+these helpers do it once, here, in the hand-written layer where they survive
+SDK regeneration.
+
+    from agentex.lib.utils.metadata_filters import encode_metadata_filter
+
+    client.agents.list(
+        agent_card_metadata=encode_metadata_filter({"permits_capable": True}),
+    )
+
+The ``agent_card_metadata`` filter requires an Agentex server that includes
+scaleapi/scale-agentex#411. Older servers ignore the unknown query parameter
+and return the full unfiltered agent list rather than erroring, and the SDK's
+startup backend-contract check does not guard against this.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Mapping
+
+__all__ = ["encode_metadata_filter"]
+
+
+def encode_metadata_filter(metadata: Mapping[str, Any]) -> str:
+    """Encode a metadata filter mapping into the wire form the platform expects.
+
+    Args:
+        metadata: The key/value pairs the target's metadata object must contain.
+            Values may be any JSON type; matching is exact containment, so
+            ``{"permits_capable": True}`` matches a stored JSON ``true`` but not
+            the string ``"true"``. An empty mapping matches any target that has
+            a metadata object at all.
+
+    Returns:
+        A compact JSON object string, with keys sorted so the same filter always
+        produces the same query value.
+
+    Raises:
+        TypeError: If ``metadata`` is not a mapping, or contains a value that
+            isn't JSON-serializable.
+        ValueError: If a value is a non-finite float. ``NaN`` and ``Infinity``
+            aren't valid JSON and the server rejects them with a 400, so fail
+            here with a clearer message instead.
+    """
+    if not isinstance(metadata, Mapping):
+        raise TypeError(f"metadata must be a mapping, got {type(metadata).__name__}")
+
+    try:
+        return json.dumps(metadata, allow_nan=False, separators=(",", ":"), sort_keys=True)
+    except ValueError as exc:
+        raise ValueError(f"metadata filter is not encodable as JSON: {exc}") from exc

src/agentex/resources/agents/agents.py

@@ -116,6 +116,7 @@ def retrieve(
     def list(
         self,
         *,
+        agent_card_metadata: Optional[str] | Omit = omit,
         limit: int | Omit = omit,
         order_by: Optional[str] | Omit = omit,
         order_direction: str | Omit = omit,
@@ -132,6 +133,13 @@ def list(
         List all registered agents, optionally filtered by query parameters.
 
         Args:
+          agent_card_metadata: JSON-encoded object used to filter agents on
+              `registration_metadata.agent_card.metadata` via JSONB containment. Example:
+              {"permits_capable": true}. Only matches cards published through the direct
+              registration path: registrations that carry a `deployment_id` write the card to
+              the deployment record instead of `registration_metadata`, so those agents never
+              match this filter.
+
           limit: Limit
 
           order_by: Field to order by
@@ -159,6 +167,7 @@ def list(
                 timeout=timeout,
                 query=maybe_transform(
                     {
+                        "agent_card_metadata": agent_card_metadata,
                         "limit": limit,
                         "order_by": order_by,
                         "order_direction": order_direction,
@@ -777,6 +786,7 @@ async def retrieve(
     async def list(
         self,
         *,
+        agent_card_metadata: Optional[str] | Omit = omit,
         limit: int | Omit = omit,
         order_by: Optional[str] | Omit = omit,
         order_direction: str | Omit = omit,
@@ -793,6 +803,13 @@ async def list(
         List all registered agents, optionally filtered by query parameters.
 
         Args:
+          agent_card_metadata: JSON-encoded object used to filter agents on
+              `registration_metadata.agent_card.metadata` via JSONB containment. Example:
+              {"permits_capable": true}. Only matches cards published through the direct
+              registration path: registrations that carry a `deployment_id` write the card to
+              the deployment record instead of `registration_metadata`, so those agents never
+              match this filter.
+
           limit: Limit
 
           order_by: Field to order by
@@ -820,6 +837,7 @@ async def list(
                 timeout=timeout,
                 query=await async_maybe_transform(
                     {
+                        "agent_card_metadata": agent_card_metadata,
                         "limit": limit,
                         "order_by": order_by,
                         "order_direction": order_direction,

src/agentex/types/agent_list_params.py

@@ -9,6 +9,16 @@
 
 
 class AgentListParams(TypedDict, total=False):
+    agent_card_metadata: Optional[str]
+    """
+    JSON-encoded object used to filter agents on
+    `registration_metadata.agent_card.metadata` via JSONB containment. Example:
+    {"permits_capable": true}. Only matches cards published through the direct
+    registration path: registrations that carry a `deployment_id` write the card to
+    the deployment record instead of `registration_metadata`, so those agents never
+    match this filter.
+    """
+
     limit: int
     """Limit"""