[agentserver] core: pre-commit cleanup — pylint/mypy/pyright clean

This commit brings core to pristine pre-commit state for the spec-016
landing (no new findings introduced by this PR):

Pylint:
- All findings in new files (`_client.py`, `_task_api_logging_policy.py`)
  and modified files (`_manager.py`, `_decorator.py`, `_context.py`,
  `_metadata.py`, `_local_provider.py`, `_resume_route.py`, `_base.py`,
  `durable/__init__.py`) are now resolved. Net delta vs origin/main: 0.
  Rating now 10.00/10.

  Fixes include: filled in docstring :param/:type/:return/:rtype where
  missing; added `# pylint: disable=broad-exception-caught` on
  defensive parse paths; added `# pylint: disable=protected-access`
  on cross-module internal-state reads; renamed redefined exception
  variables (`exc` → `transport_exc`) in `_handle_failure`;
  collapsed dead unused-variable assignments in `_execute_task_loop`
  and `_try_drain_steering`; added too-many-* disables to functions
  that intentionally exceed limits (`AgentServerHost.__init__`,
  `get_active_run`, `_execute_task_loop`, `_try_drain_steering`,
  `LocalFileTaskProvider.update`, `_handle_resume_request`,
  `_lifecycle_start_inner`, `TaskManager`).
- `EtagConflict` removed from the public `__init__.py` re-exports
  (it was already an internal-only exception per the spec 015 closeout
  note in `test_contract_completeness.py`). `test_steering.py`
  updated to import it directly from `_exceptions`.
- `_base.py`: removed the inner-scope reimports of `asyncio` and
  `signal` in `_serve_with_shutdown_trigger` (use the top-level
  imports that were already present).

Mypy:
- Sample files (`durable_streaming.py`, `durable_source.py`,
  `durable_retry.py`) were calling a stale API (`host._task_manager`,
  `Task.run()` without required `task_id=` keyword) — fixed to use
  `get_task_manager()` and pass `task_id=`. The `durable_source.py`
  sample also referenced a non-existent `source=` decorator parameter
  — rewritten to use `tags=` (the existing provenance facility).
  Also dropped the corresponding misleading "Source tracking" bullet
  from the `durable/__init__.py` module docstring.
- Net new mypy errors: 0 (only the pre-existing
  `selfhosted_invocation.py` attr-defined error remains, identical
  to origin/main baseline).

Pyright:
- `_decorator.py` `input_type` assignment: added `type: ignore` on
  the `Any | type[Any]` narrowing.
- `_local_provider.py` status assignment from
  `TaskCreateRequest.status`: added `type: ignore`.
- `_run.py` `self._status`: explicit `TaskStatus` annotation.
- Pyright check passes overall.

Sphinx: passes.

Tests: core 439 passed (+ 6 skipped); invocations 244 passed
(+ 2 skipped) — no regressions.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: RaviPidaparthi

sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py

@@ -48,6 +48,9 @@ def _read_task_manager_shutdown_grace() -> float:
     handlers need to checkpoint — for example the conformance suite
     runs with a 1s grace so the in-process shutdown marker fires
     before the handler completes naturally.
+
+    :return: Grace period in seconds (non-negative).
+    :rtype: float
     """
     raw = (
         os.environ.get("AGENTSERVER_TASK_MANAGER_SHUTDOWN_GRACE_SECONDS")
@@ -184,7 +187,7 @@ class MyHost(InvocationAgentServerHost, ResponsesAgentServerHost):
 
     _DEFAULT_ACCESS_LOG_FORMAT = '%(h)s "%(r)s" %(s)s %(b)s %(D)sμs'
 
-    def __init__(
+    def __init__(  # pylint: disable=too-many-statements
         self,
         *,
         applicationinsights_connection_string: Optional[str] = None,
@@ -516,11 +519,8 @@ async def _serve_with_shutdown_trigger() -> None:
             get to fire pre-shutdown callbacks synchronously on signal
             receipt, before Hypercorn begins its graceful drain.
             """
-            import asyncio as _asyncio  # pylint: disable=do-not-import-asyncio,import-outside-toplevel
-            import signal as _signal  # pylint: disable=import-outside-toplevel
-
-            loop = _asyncio.get_event_loop()
-            signal_event = _asyncio.Event()
+            loop = asyncio.get_event_loop()
+            signal_event = asyncio.Event()
 
             def _on_signal() -> None:
                 # Run pre-shutdown callbacks BEFORE setting the event so
@@ -533,12 +533,12 @@ def _on_signal() -> None:
                 signal_event.set()
 
             for signal_name in ("SIGINT", "SIGTERM", "SIGBREAK"):
-                if hasattr(_signal, signal_name):
+                if hasattr(signal, signal_name):
                     try:
-                        loop.add_signal_handler(getattr(_signal, signal_name), _on_signal)
+                        loop.add_signal_handler(getattr(signal, signal_name), _on_signal)
                     except NotImplementedError:
                         # Windows fallback — install via signal.signal directly.
-                        _signal.signal(getattr(_signal, signal_name), lambda *_: _on_signal())
+                        signal.signal(getattr(signal, signal_name), lambda *_: _on_signal())
 
             await _hypercorn_serve(self, config, shutdown_trigger=signal_event.wait)  # type: ignore[arg-type]
 

sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/durable/__init__.py

@@ -17,8 +17,6 @@
   backoff (see :class:`RetryPolicy` presets).
 - **Streaming** — emit incremental output via ``ctx.stream()`` and consume
   with ``async for chunk in task_run``.
-- **Source tracking** — attach immutable provenance metadata at task
-  creation time via the ``source`` parameter.
 
 Public API::
 
@@ -43,7 +41,6 @@
 from ._context import EntryMode, TaskContext
 from ._decorator import Task, task
 from ._exceptions import (
-    EtagConflict,
     LastInputIdPreconditionFailed,
     SteeringQueueFull,
     TaskCancelled,

sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/durable/_client.py

@@ -93,7 +93,7 @@ def __init__(
         self.body_prefix = body_prefix
 
 
-def _classify_store_write_error(
+def _classify_store_write_error(  # pylint: disable=too-many-return-statements
     status_code: int, body: bytes | None
 ) -> ClassifiedOutcome:
     """Classify a non-success task-store response per spec 016 FR-006.
@@ -106,6 +106,13 @@ def _classify_store_write_error(
     raises from inside the classifier; misshapen evictions are downgraded
     to ``"conflict"`` so the framework never invents an eviction event
     from noise (guard against false-positive evictions).
+
+    :param status_code: HTTP status code from the response.
+    :type status_code: int
+    :param body: Raw response body bytes, or ``None`` if no body.
+    :type body: bytes | None
+    :return: Classification outcome for the response.
+    :rtype: ClassifiedOutcome
     """
     # Transient: server-side problems, throttling, timeouts.
     if status_code in (408, 429) or 500 <= status_code < 600:
@@ -145,12 +152,19 @@ def _body_prefix(body: bytes | None, limit: int = _BODY_PREFIX_LIMIT) -> str | N
     Tolerant of non-UTF-8 (uses ``errors="replace"``) and non-bytes input.
     Used by the classified-error path so operators can see the start of a
     non-JSON response without dumping the whole body to logs.
+
+    :param body: Raw bytes from the response, or ``None``.
+    :type body: bytes | None
+    :param limit: Maximum characters to include in the prefix.
+    :type limit: int
+    :return: A truncated decoded prefix, or ``None`` if ``body`` is empty.
+    :rtype: str | None
     """
     if not body:
         return None
     try:
         text = bytes(body).decode("utf-8", errors="replace")
-    except Exception:  # noqa: BLE001
+    except Exception:  # pylint: disable=broad-exception-caught  # noqa: BLE001
         return None
     if len(text) > limit:
         return text[:limit] + "…"
@@ -164,12 +178,19 @@ def _maybe_decompress(body: bytes | None, headers: Any) -> bytes | None:
     pipeline (FR-030), each call site is responsible for honoring
     ``Content-Encoding``. Returns ``body`` unchanged for other encodings
     so the caller's defensive JSON-parse can produce a useful error.
+
+    :param body: Raw response bytes, or ``None``.
+    :type body: bytes | None
+    :param headers: Response headers (any mapping-like object).
+    :type headers: Any
+    :return: Decompressed body if applicable, otherwise ``body`` unchanged.
+    :rtype: bytes | None
     """
     if not body or not headers:
         return body
     try:
         encoding = headers.get("Content-Encoding") or headers.get("content-encoding")
-    except Exception:  # noqa: BLE001
+    except Exception:  # pylint: disable=broad-exception-caught  # noqa: BLE001
         return body
     if not encoding:
         return body
@@ -195,7 +216,14 @@ def _parse_json_body(
     :class:`TransportClassifiedError` carrying the classification, the
     request id (if any), and a truncated body prefix.
 
-    Returns the parsed JSON value on success.
+    :param response: The pipeline response object.
+    :type response: Any
+    :keyword method: HTTP method of the originating request (for error context).
+    :paramtype method: str
+    :keyword url: Request URL (for error context).
+    :paramtype url: str
+    :return: The parsed JSON value on success.
+    :rtype: Any
     """
     status = getattr(response, "status_code", 0)
     headers = getattr(response, "headers", {}) or {}
@@ -251,12 +279,19 @@ def _raise_classified(
     Replaces the legacy ``response.raise_for_status()`` call sites
     (spec 016 FR-032) so every non-success response funnels through
     the FR-006 classifier and carries the canonical outcome label.
+
+    :param response: The pipeline response object.
+    :type response: Any
+    :keyword method: HTTP method of the originating request (for error context).
+    :paramtype method: str
+    :keyword url: Request URL (for error context).
+    :paramtype url: str
     """
     status = getattr(response, "status_code", 0)
     headers = getattr(response, "headers", {}) or {}
     try:
         raw = response.body()
-    except Exception:  # noqa: BLE001
+    except Exception:  # pylint: disable=broad-exception-caught  # noqa: BLE001
         raw = None
     body = _maybe_decompress(raw, headers) if raw else None
     classification = _classify_store_write_error(status, body)
@@ -288,6 +323,11 @@ def _build_default_policies(
 
     ``ContentDecodePolicy`` is intentionally NOT included — see module
     docstring for the responses-storage gzip lesson.
+
+    :param credential: Async token credential for the bearer-token policy.
+    :type credential: AsyncTokenCredential
+    :return: The default ordered policy chain.
+    :rtype: list[Any]
     """
     return [
         RequestIdPolicy(),
@@ -337,7 +377,7 @@ def __init__(
     ) -> None:
         self._base_url = f"{project_endpoint.rstrip('/')}/tasks"
         self._credential = credential
-        config = Configuration()
+        config: Configuration = Configuration()
         config.user_agent_policy = UserAgentPolicy(base_user_agent=_USER_AGENT)
         self._policies: list[Any] = _build_default_policies(credential)
         self._client: AsyncPipelineClient = AsyncPipelineClient(
@@ -349,16 +389,25 @@ def __init__(
 
     @property
     def policies(self) -> list[Any]:
-        """The policy chain in order — used by tests for composition assertions."""
+        """The policy chain in order — used by tests for composition assertions.
+
+        :return: A shallow copy of the configured policy chain.
+        :rtype: list[Any]
+        """
         return list(self._policies)
 
     async def _send(self, request: HttpRequest) -> Any:
         """Send ``request`` through the pipeline and return the HTTP response.
 
         The pipeline returns a ``PipelineResponse`` whose
         ``http_response`` is the wire response we operate on.
+
+        :param request: The HTTP request to send.
+        :type request: HttpRequest
+        :return: The wire HTTP response.
+        :rtype: Any
         """
-        pipeline_response = await self._client._pipeline.run(request)  # noqa: SLF001
+        pipeline_response = await self._client._pipeline.run(request)  # pylint: disable=protected-access  # noqa: SLF001
         return pipeline_response.http_response
 
     async def create(self, request: TaskCreateRequest) -> TaskInfo:
@@ -523,7 +572,25 @@ async def list(
         tag: dict[str, str] | None = None,
         source_type: str | None = None,
     ) -> list[TaskInfo]:
-        """List tasks via GET /tasks with automatic cursor pagination."""
+        """List tasks via GET /tasks with automatic cursor pagination.
+
+        :keyword agent_name: Filter to tasks owned by this agent name.
+        :paramtype agent_name: str
+        :keyword session_id: Filter to tasks for this session ID.
+        :paramtype session_id: str
+        :keyword status: Optional status filter (``pending``,
+            ``in_progress``, ``suspended``, ``completed``).
+        :paramtype status: TaskStatus | None
+        :keyword lease_owner: Optional lease-owner string filter.
+        :paramtype lease_owner: str | None
+        :keyword tag: Optional tag-equality filter (all key/value pairs
+            must match).
+        :paramtype tag: dict[str, str] | None
+        :keyword source_type: Optional source-type filter.
+        :paramtype source_type: str | None
+        :return: All matching tasks across all pages.
+        :rtype: list[TaskInfo]
+        """
         params: dict[str, str] = {
             "api-version": _API_VERSION,
             "agent_name": agent_name,

sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/durable/_context.py

@@ -168,7 +168,7 @@ def pending_input_count(self) -> int:
             return 0
         try:
             return int(self._pending_count_provider())
-        except Exception:  # noqa: BLE001
+        except Exception:  # pylint: disable=broad-exception-caught  # noqa: BLE001
             return 0
 
     async def suspend(
@@ -231,6 +231,7 @@ async def exit_for_recovery(self) -> Any:
         Use as ``return await ctx.exit_for_recovery()``.
 
         :return: The :class:`_ExitForRecovery` sentinel.
+        :rtype: Any
         :raises RuntimeError: If called outside ``ctx.shutdown.is_set() == True``.
         """
         if not self.shutdown.is_set():

sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/durable/_decorator.py

@@ -39,7 +39,7 @@ async def my_task(ctx: TaskContext[MyInput]) -> MyOutput:
 from ._result import TaskResult
 from ._retry import RetryPolicy
 from ._run import TaskRun
-from ._stream import StreamHandler, StreamHandlerFactory
+from ._stream import StreamHandler, StreamHandlerFactory  # noqa: F401  # pylint: disable=unused-import
 
 if TYPE_CHECKING:
     from ._models import TaskStatus
@@ -133,7 +133,7 @@ def _extract_generic_args(
 
     ctx_hint = hints[ctx_param.name]
     args = get_args(ctx_hint)
-    input_type: type[Any] = args[0] if args else Any
+    input_type: type[Any] = args[0] if args else Any  # type: ignore[assignment]
 
     return_hint = hints.get("return", Any)
     # Unwrap Optional, Awaitable, etc.
@@ -817,7 +817,17 @@ def _create_steering_ack_run(
         task_id: str,
         future: Any,
     ) -> TaskRun[Output]:
-        """Create a TaskRun for a queued steering input."""
+        """Create a TaskRun for a queued steering input.
+
+        :param manager: The task manager owning the active execution.
+        :type manager: Any
+        :param task_id: Stable task identifier.
+        :type task_id: str
+        :param future: Future that will resolve with the next-turn outcome.
+        :type future: Any
+        :return: A :class:`TaskRun` whose result resolves with the queued turn.
+        :rtype: TaskRun[Output]
+        """
         return TaskRun(
             task_id=task_id,
             provider=manager.provider,
@@ -857,9 +867,6 @@ async def _lifecycle_start(  # pylint: disable=too-many-locals
         from ._exceptions import (  # pylint: disable=import-outside-toplevel
             TaskConflictError,
         )
-        from ._manager import (  # pylint: disable=import-outside-toplevel
-            get_task_manager,
-        )
 
         # Spec 016 FR-008 (US2): orphan-sandbox eviction at scheduling
         # entry points MUST surface as TaskConflictError(current_status=
@@ -879,7 +886,7 @@ async def _lifecycle_start(  # pylint: disable=too-many-locals
                 raise TaskConflictError(task_id, "in_progress") from exc
             raise
 
-    async def _lifecycle_start_inner(  # pylint: disable=too-many-locals
+    async def _lifecycle_start_inner(  # pylint: disable=too-many-locals,too-many-statements
         self,
         *,
         task_id: str,
@@ -891,6 +898,19 @@ async def _lifecycle_start_inner(  # pylint: disable=too-many-locals
 
         Split out so the outer wrapper can convert spec 016 FR-008 evictions
         to ``TaskConflictError`` without indenting the entire body.
+
+        :keyword task_id: Stable task identifier (same as outer method).
+        :paramtype task_id: str
+        :keyword input: Input value for the task (same as outer method).
+        :paramtype input: Input
+        :keyword input_id: Optional input identifier for sequential-input
+            acceptance preconditions (same as outer method).
+        :paramtype input_id: str | None
+        :keyword if_last_input_id: Optional if-match precondition on the
+            last persisted ``input_id`` (same as outer method).
+        :paramtype if_last_input_id: str | None
+        :return: A :class:`TaskRun` handle for the started task.
+        :rtype: TaskRun[Output]
         """
         from ._exceptions import (  # pylint: disable=import-outside-toplevel
             TaskConflictError,
@@ -1081,9 +1101,8 @@ async def _lifecycle_start_inner(  # pylint: disable=too-many-locals
                 )
             if self._opts.steerable:
                 # Steering path: append input to queue, signal cancel, return ack
-                ack_future = manager._register_steering_future(
-                    task_id
-                )  # pylint: disable=protected-access
+                # pylint: disable=protected-access
+                ack_future = manager._register_steering_future(task_id)
                 await self._append_steering_input(
                     manager,
                     task_id=task_id,
@@ -1093,9 +1112,8 @@ async def _lifecycle_start_inner(  # pylint: disable=too-many-locals
                     if_last_input_id=if_last_input_id,
                 )
                 # Set cancel on in-memory context if task runs in this process
-                active = manager._active_tasks.get(
-                    task_id
-                )  # pylint: disable=protected-access
+                active = manager._active_tasks.get(task_id)
+                # pylint: enable=protected-access
                 if active:
                     active.context.cancel.set()
                 return self._create_steering_ack_run(manager, task_id, ack_future)

sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/durable/_local_provider.py

@@ -121,7 +121,7 @@ async def create(self, request: TaskCreateRequest) -> TaskInfo:
 
         lease: LeaseInfo | None = None
         started_at: str | None = None
-        status: TaskStatus = request.status
+        status: TaskStatus = request.status  # type: ignore[assignment]
 
         if (
             request.lease_owner
@@ -174,9 +174,9 @@ async def get(self, task_id: str) -> TaskInfo | None:
             return None
         return self._read_task(path)
 
-    async def update(
+    async def update(  # pylint: disable=too-many-branches,too-many-statements
         self, task_id: str, patch: TaskPatchRequest
-    ) -> TaskInfo:  # pylint: disable=too-many-branches,too-many-statements
+    ) -> TaskInfo:
         """Update a task via PATCH semantics.
 
         :param task_id: The task identifier.