Merge pull request #16 from Accenture/chore/code-quality-review

Code quality round: lint and type gates adopted, wire contract typed end to end

Author: acn-ericlaw

.gitignore

@@ -9,6 +9,7 @@ build/
 .mypy_cache/
 .ruff_cache/
 uv.lock
+.idea/
 .DS_Store
 
 # === agent-memory: AI infrastructure (personal / per-machine — do not commit) ===

README.md

@@ -26,10 +26,10 @@ graph task calls a Python function exactly as if it were local.
 
 ```python
 # app.py
-from mercury_composable import AppException, platform, preload
+from mercury_composable import AppException, Body, platform, preload
 
 @preload(route="hello.python", instances=10)
-def handle_event(headers: dict, body):
+def handle_event(headers: dict[str, str], body: Body):
     if not isinstance(body, dict) or "text" not in body:
         raise AppException(400, "missing 'text'")
     return {"text": str(body["text"]).upper(), "language": "python"}
@@ -66,7 +66,7 @@ now executes the Python function, with trace context carried end to end.
 ## The function contract
 
 A handler receives the same two-part input as an engine `TypedLambdaFunction` —
-`(headers: dict, body)` — and returns the reply body (or an `EventEnvelope` for full
+`(headers: dict[str, str], body: Body)` — `Body` is any MsgPack value — and returns the reply body (or an `EventEnvelope` for full
 control of status and reply headers). `async def` and plain `def` are both supported;
 synchronous handlers run in a thread-pool executor so the event loop never blocks.
 
@@ -75,6 +75,8 @@ synchronous handlers run in a thread-pool executor so the event loop never block
   exception handler or the graph's `error.*` contract.
 - `get_trace()` exposes `trace_id` / `trace_path` / `cid`; `annotate_trace(k, v)` sends an
   annotation back on the reply envelope.
+- Outside a hosted function (batch jobs, tests), `trace_context(trace_id, trace_path)`
+  establishes the context your `PostOffice` calls inherit — the node `runWithTrace` twin.
 - Functions must be stateless; anything you must keep belongs to the caller's flow model
   or state machine.
 
@@ -125,6 +127,19 @@ orchestration** — those live in the engines. It provides functions plus the mi
 foundation utilities, keeping Python fast to prototype with while the composable core
 guarantees the architecture.
 
+## Development
+
+```bash
+uv venv .venv && uv pip install -e '.[dev]'   # environment (uv-managed python)
+.venv/bin/pytest -q                           # tests
+uvx ruff check .                              # lint (config in pyproject.toml)
+uvx basedpyright                              # type check (config in pyproject.toml)
+```
+
+PyCharm: use interpreter type **uv** pointing at the project `.venv`, and set
+*Settings → Tools → Python Integrated Tools → Package requirements file* to
+`pyproject.toml` so the requirements inspection reads `[project.dependencies]`.
+
 ## License
 
 Apache 2.0 — see [LICENSE.txt](LICENSE.txt).

examples/demo_app.py

@@ -10,14 +10,26 @@
         target: 'http://127.0.0.1:8086/api/event'
 """
 
-from mercury_composable import AppException, annotate_trace, get_logger, platform, preload
+from mercury_composable import (
+    AppException,
+    Body,
+    annotate_trace,
+    get_logger,
+    platform,
+    preload,
+)
 
 log = get_logger(__name__)
 
 
 @preload(route="hello.python", instances=10)
-def handle_event(headers: dict, body):
-    """Uppercase transform - the polyglot hello world."""
+def handle_event(_headers: dict[str, str], body: Body):
+    """Uppercase transform - the polyglot hello world.
+
+    The (headers, body) two-part signature is the function contract (the
+    TypedLambdaFunction mirror) - a handler that does not need headers keeps
+    the parameter, underscore-prefixed per Python convention.
+    """
     if not isinstance(body, dict) or "text" not in body:
         raise AppException(400, "missing 'text'")
     annotate_trace("language", "python")
@@ -26,7 +38,7 @@ def handle_event(headers: dict, body):
 
 
 @preload(route="hello.declarative", instances=10)
-async def declarative_echo(headers: dict, body):
+async def declarative_echo(headers: dict[str, str], body: Body):
     """Echo for the composable-example declarative Event-over-HTTP demo."""
     return {"body": body, "headers": headers, "language": "python"}
 

pyproject.toml

@@ -27,14 +27,35 @@ Homepage = "https://github.com/Accenture/mercury-python"
 Documentation = "https://accenture.github.io/mercury-composable"
 
 [project.optional-dependencies]
-dev = ["pytest>=8", "pytest-asyncio>=0.23"]
+dev = ["pytest>=8", "pytest-asyncio>=0.23", "ruff>=0.9"]
 
 [project.scripts]
 mercury-serve = "mercury_composable.cli:main"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/mercury_composable"]
 
+[tool.ruff]
+line-length = 100
+target-version = "py310"
+# the agent-skills layer is tool-managed by agent-memory (overwritten on upgrade);
+# style fixes for it belong upstream, not here
+extend-exclude = ["agent-skills"]
+
+[tool.ruff.lint]
+extend-select = ["I"]
+
+[tool.basedpyright]
+# standard mode plus the contract-relevant strictness: every function parameter
+# is annotated, unit tests included (the wire contract is typed - Body,
+# dict[str, str]); the reportUnknown* warning family of the editor's
+# "recommended" mode is deliberately not chased. agent-skills is tool-managed
+# by agent-memory and excluded.
+typeCheckingMode = "standard"
+reportMissingParameterType = "error"
+include = ["src", "examples", "tests"]
+exclude = ["agent-skills", ".venv"]
+
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]

src/mercury_composable/__init__.py

@@ -11,18 +11,36 @@
 
 from .client import PostOffice
 from .config import AppConfig, app_config, load_config
-from .envelope import EventEnvelope, iso_utc
+from .envelope import Body, EventEnvelope, iso_utc
 from .exceptions import AppException, CompactFormatError
 from .log import get_logger
-from .registry import FunctionRegistry, default_registry, preload
+from .registry import FunctionRegistry, Handler, default_registry, preload
 from .server import EventApiServer, Platform, platform
-from .trace import TraceInfo, annotate_trace, get_trace
+from .trace import TraceInfo, annotate_trace, get_trace, trace_context
 
 __version__ = "0.1.0"
 
 __all__ = [
-    "AppConfig", "AppException", "CompactFormatError", "EventApiServer",
-    "EventEnvelope", "FunctionRegistry", "Platform", "PostOffice", "TraceInfo",
-    "annotate_trace", "app_config", "default_registry", "get_logger",
-    "get_trace", "iso_utc", "load_config", "platform", "preload", "__version__",
+    "AppConfig",
+    "AppException",
+    "Body",
+    "CompactFormatError",
+    "EventApiServer",
+    "EventEnvelope",
+    "FunctionRegistry",
+    "Handler",
+    "Platform",
+    "PostOffice",
+    "TraceInfo",
+    "__version__",
+    "annotate_trace",
+    "app_config",
+    "default_registry",
+    "get_logger",
+    "get_trace",
+    "iso_utc",
+    "load_config",
+    "platform",
+    "preload",
+    "trace_context",
 ]

src/mercury_composable/cli.py

@@ -25,10 +25,15 @@ def main() -> int:
                         help="Configuration file (default: resources/application.yml|properties)")
     # -Dkey=value runtime overrides are consumed by AppConfig from sys.argv
     args, _unknown = parser.parse_known_args()
+    # argparse Namespace attributes are untyped - pin the types at the boundary
+    app_arg: str = args.app
+    port_arg: int | None = args.port
+    host_arg: str = args.host
+    config_arg: str | None = args.config
 
     from .config import DEFAULT_CANDIDATES, load_config
-    app_path = os.path.abspath(args.app)
-    config_path = args.config
+    app_path = os.path.abspath(app_arg)
+    config_path: str | None = config_arg
     if config_path is None and not any(os.path.isfile(c) for c in DEFAULT_CANDIDATES):
         # fall back to a resources folder next to the application file
         app_dir = os.path.dirname(app_path)
@@ -56,7 +61,7 @@ def main() -> int:
         print("No functions registered - use @preload(route=..., instances=...)",
               file=sys.stderr)
         return 1
-    platform.run(port=args.port, host=args.host)
+    platform.run(port=port_arg, host=host_arg)
     return 0
 
 

src/mercury_composable/client.py

@@ -15,7 +15,7 @@
 from __future__ import annotations
 
 import re
-from typing import Any, Dict, Optional
+from typing import Any
 
 import aiohttp
 
@@ -27,32 +27,52 @@
 _W3C_SPAN_ID = re.compile(r"^[0-9a-f]{16}$")
 
 
+def _build_event(route: str, body: Any, headers: dict[str, str] | None,
+                 from_route: str | None, cid: str | None) -> EventEnvelope:
+    """Build the outbound envelope, inheriting the current trace context."""
+    event = EventEnvelope(to=route, body=body, headers=headers or {})
+    if from_route:
+        event.set_from(from_route)
+    info = get_trace()
+    if info and info.trace_id:
+        event.set_trace(info.trace_id, info.trace_path or route)
+    effective_cid = cid or (info.cid if info else None)
+    if effective_cid:
+        event.set_correlation_id(effective_cid)
+    return event
+
+
 class PostOffice:
     """Event-over-HTTP client for calling functions on peer applications."""
 
-    def __init__(self, endpoint: Optional[str] = None,
-                 security_headers: Optional[Dict[str, str]] = None):
+    def __init__(self, endpoint: str | None = None,
+                 security_headers: dict[str, str] | None = None):
         self.endpoint = endpoint
         self.security_headers = dict(security_headers or {})
-        self._session: Optional[aiohttp.ClientSession] = None
+        self._session: aiohttp.ClientSession | None = None
 
-    async def _get_session(self) -> aiohttp.ClientSession:
-        if self._session is None or self._session.closed:
-            self._session = aiohttp.ClientSession()
-        return self._session
+    def _get_session(self) -> aiohttp.ClientSession:
+        # called from the running event loop only (inside request/send)
+        session = self._session
+        if session is None or session.closed:
+            session = aiohttp.ClientSession()
+            self._session = session
+        return session
 
     async def close(self) -> None:
         if self._session is not None and not self._session.closed:
             await self._session.close()
 
-    async def __aenter__(self) -> "PostOffice":
+    # PYI034 wants '-> Self', which needs python >= 3.11; switch when the
+    # floor moves past 3.10
+    async def __aenter__(self) -> PostOffice:  # noqa: PYI034
         return self
 
-    async def __aexit__(self, *_exc) -> None:
+    async def __aexit__(self, *_exc: object) -> None:
         await self.close()
 
     def _http_headers(self, timeout_ms: int, is_async: bool,
-                      event: EventEnvelope) -> Dict[str, str]:
+                      event: EventEnvelope) -> dict[str, str]:
         headers = {
             "content-type": "application/octet-stream",
             "accept": "*/*",
@@ -71,28 +91,15 @@ def _http_headers(self, timeout_ms: int, is_async: bool,
                 headers["traceparent"] = f"00-{event.trace_id}-{event.span_id}-01"
         return headers
 
-    def _build_event(self, route: str, body: Any, headers: Optional[Dict[str, str]],
-                     from_route: Optional[str], cid: Optional[str]) -> EventEnvelope:
-        event = EventEnvelope(to=route, body=body, headers=headers or {})
-        if from_route:
-            event.set_from(from_route)
-        info = get_trace()
-        if info and info.trace_id:
-            event.set_trace(info.trace_id, info.trace_path or route)
-        effective_cid = cid or (info.cid if info else None)
-        if effective_cid:
-            event.set_correlation_id(effective_cid)
-        return event
-
-    async def _call(self, route: str, body: Any, headers: Optional[Dict[str, str]],
-                    timeout_ms: int, endpoint: Optional[str], is_async: bool,
-                    from_route: Optional[str], cid: Optional[str]) -> EventEnvelope:
+    async def _call(self, route: str, body: Any, headers: dict[str, str] | None,
+                    timeout_ms: int, endpoint: str | None, is_async: bool,
+                    from_route: str | None, cid: str | None) -> EventEnvelope:
         url = endpoint or self.endpoint
         if not url:
             raise ValueError("Missing event endpoint - "
                              "e.g. PostOffice(endpoint='http://peer:8085/api/event')")
-        event = self._build_event(route, body, headers, from_route, cid)
-        session = await self._get_session()
+        event = _build_event(route, body, headers, from_route, cid)
+        session = self._get_session()
         # +100 ms cushion so the HTTP client does not time out before the target
         client_timeout = aiohttp.ClientTimeout(total=(max(100, timeout_ms) + 100) / 1000)
         async with session.post(url, data=event.to_bytes(),
@@ -106,21 +113,21 @@ async def _call(self, route: str, body: Any, headers: Optional[Dict[str, str]],
                                    f"Invalid event-over-http response - {e}") from e
 
     async def request(self, route: str, body: Any = None, *,
-                      headers: Optional[Dict[str, str]] = None,
+                      headers: dict[str, str] | None = None,
                       timeout_ms: int = 30000,
-                      endpoint: Optional[str] = None,
-                      from_route: Optional[str] = None,
-                      cid: Optional[str] = None) -> EventEnvelope:
+                      endpoint: str | None = None,
+                      from_route: str | None = None,
+                      cid: str | None = None) -> EventEnvelope:
         """RPC call: returns the target function's reply envelope."""
         return await self._call(route, body, headers, timeout_ms, endpoint,
                                 False, from_route, cid)
 
     async def send(self, route: str, body: Any = None, *,
-                   headers: Optional[Dict[str, str]] = None,
+                   headers: dict[str, str] | None = None,
                    timeout_ms: int = 30000,
-                   endpoint: Optional[str] = None,
-                   from_route: Optional[str] = None,
-                   cid: Optional[str] = None) -> EventEnvelope:
+                   endpoint: str | None = None,
+                   from_route: str | None = None,
+                   cid: str | None = None) -> EventEnvelope:
         """Drop-n-forget: returns the peer's 202 delivery acknowledgement envelope."""
         return await self._call(route, body, headers, timeout_ms, endpoint,
                                 True, from_route, cid)