perf: cache per-class argument specs; make worker logging and profiling reachable

extract_argument_specs derives a function class's arguments from its
annotations, its FunctionArguments, and the Arg descriptors on its MRO —
none of which change once the class exists. It was nonetheless re-derived
on every catalog listing and every overload match.

Profiling one run of the upstream integration suite through the Cedar
proxy: 58,058 calls costing 15.8s, driving 131,653 typing.get_type_hints
calls, 2.4M _eval_type calls and 648k compile() calls — about a quarter of
the worker's CPU. catalog_schema_contents_functions cost 38ms per call
where every other catalog method was 0.1-2ms, entirely because of this.
After: 16.7ms per call, and 179x on the function itself in isolation.

Memoized with functools.cache, which is thread safe (workers serve on a
pool) and unbounded — the keys are function classes the catalog already
holds for the process's lifetime, so it retains nothing new. The cached
value is a tuple and the public function returns a fresh list, so a caller
that mutates the result cannot corrupt the cache; ArgumentSpec is a frozen
dataclass, so sharing the specs themselves is safe.

Two supporting changes, both about reaching diagnostics from outside:

VGI_WORKER_LOG_LEVEL / _FORMAT / _LOGGERS, applied inside
configure_worker_logging rather than in one CLI. The vgi_rpc.access log
already records duration_ms, method and row/byte counts for every RPC, but
turning it on required --log-logger and --log-format on the command line,
which anything launching a worker without owning its argv cannot do. Placed
in the shared function because vgi-serve and the fixture servers had
byte-identical logging blocks and wiring it into one would be a trap.

vgi/profiling.py: VGI_WORKER_PROFILE=<dir> writes pstats on exit. It uses a
single profiler deliberately — under --http the worker runs on waitress, so
a main-thread-only profile would show time in select() and nothing else,
and since Python 3.12 cProfile is sys.monitoring-based and process-global,
so one profiler covers the pool threads. A profiler per thread raises
"Another profiling tool is already active" as each waitress thread starts,
which stops the server accepting connections at all. SIGTERM is handled
because its default action skips interpreter shutdown, losing the profile
on every normal stop.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: rustyconover

vgi/_test_fixtures/http_server.py

@@ -27,6 +27,7 @@
 from vgi._test_fixtures.worker import ExampleWorker
 from vgi.logging_config import LogFormat, LogLevel, configure_worker_logging
 from vgi.meta_worker import MetaWorker
+from vgi.profiling import maybe_start_profile
 
 if TYPE_CHECKING:
     from vgi_rpc.external import UploadUrl
@@ -209,12 +210,15 @@ def _run(
 
         env_debug = os.environ.get("VGI_WORKER_DEBUG", "").lower() in ("1", "true", "yes")
         effective_debug = debug or env_debug
+        # VGI_WORKER_LOG_* overrides are applied inside configure_worker_logging,
+        # so this server and `vgi-serve` are configurable the same way.
         effective_level = configure_worker_logging(
             debug=effective_debug,
             log_level=log_level,
             log_loggers=log_logger,
             log_format=log_format,
         )
+        maybe_start_profile()
 
         bucket = s3_bucket or os.environ.get("VGI_HTTP_S3_BUCKET")
         external_location = None

vgi/argument_spec.py

@@ -18,6 +18,7 @@
 import warnings
 from collections.abc import Sequence
 from dataclasses import dataclass
+from functools import cache
 from typing import Annotated, Any, get_args, get_origin, get_type_hints
 
 import pyarrow as pa
@@ -519,6 +520,10 @@ def extract_argument_specs(
     2. Type annotation with `PYTHON_TO_ARROW` mapping
     3. Default to pa.null() with warning for unknown types
 
+    Memoized per class — see :func:`_extract_argument_specs_cached`. A fresh
+    list is returned each call so a caller that mutates it cannot corrupt the
+    cache; the `ArgumentSpec`s inside are frozen dataclasses and safe to share.
+
     Args:
         cls: Function class with `Arg` descriptors.
 
@@ -527,6 +532,52 @@ def extract_argument_specs(
         then named).
 
     """
+    return list(_extract_argument_specs_cached(cls))
+
+
+@cache
+def _extract_argument_specs_cached(cls: type) -> tuple[ArgumentSpec, ...]:
+    """Memoized :func:`_extract_argument_specs`, keyed on the class.
+
+    A function class's argument specs are fixed once the class exists — they are
+    derived from its annotations, its ``FunctionArguments``, and the `Arg`
+    descriptors on its MRO, none of which change at runtime. Deriving them was
+    nonetheless repeated on **every** catalog listing and every overload match:
+    profiling one run of the upstream integration suite through the proxy showed
+    58,058 calls costing 15.8s, which drove 131,653 ``typing.get_type_hints``
+    calls, 2.4M ``_eval_type`` calls and 648k ``compile`` calls — roughly a
+    quarter of the worker's CPU. ``catalog_schema_contents_functions`` was 38ms
+    per call against 0.1-2ms for every other catalog method, entirely because of
+    this.
+
+    Unbounded (``functools.cache``) is the right size: the keys are function
+    classes, which the catalog already holds for the process's lifetime, so the
+    cache adds no retention that did not already exist. It is also thread safe,
+    which matters because workers serve requests on a pool.
+
+    Returns a tuple so the cached value cannot be mutated through a caller's
+    reference.
+
+    Args:
+        cls: Function class with `Arg` descriptors.
+
+    Returns:
+        The class's argument specs, sorted.
+    """
+    return tuple(_extract_argument_specs(cls))
+
+
+def _extract_argument_specs(
+    cls: type,
+) -> list[ArgumentSpec]:
+    """Derive a class's argument specs, without caching.
+
+    Args:
+        cls: Function class with `Arg` descriptors.
+
+    Returns:
+        List of `ArgumentSpec` objects, sorted by position.
+    """
     specs: list[ArgumentSpec] = []
     seen_names: set[str] = set()
 

vgi/logging_config.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import sys
 from enum import StrEnum
 
@@ -46,6 +47,68 @@ class LogFormat(StrEnum):
 ]
 
 
+def _env_overrides(
+    log_level: LogLevel,
+    log_loggers: list[str] | None,
+    log_format: LogFormat,
+) -> tuple[LogLevel, list[str] | None, LogFormat]:
+    """Apply ``VGI_WORKER_LOG_*`` environment overrides to the CLI options.
+
+    The access log (``vgi_rpc.access``) already records ``duration_ms``,
+    ``method`` and row/byte counts for every RPC, but reaching it required
+    passing ``--log-logger vgi_rpc.access --log-format json`` on the command
+    line. Anything that launches a worker without owning its argv — a container
+    image, a test harness spawning workers from a dozen call sites — could not
+    turn it on. These variables make the same configuration reachable the way
+    ``VGI_SIGNING_KEY`` and ``VGI_ENABLE_DESCRIBE`` already are.
+
+    Applied here rather than in a single CLI so that *every* worker entry point
+    honours them identically — ``vgi-serve`` and the fixture servers had
+    byte-identical logging blocks, and an override wired into only one of them
+    would be a trap.
+
+    Each variable wins when set, matching ``VGI_ENABLE_DESCRIBE``: Typer cannot
+    distinguish "user passed the default" from "user passed nothing", so
+    deferring to an explicit environment setting is the only unambiguous rule.
+
+    Args:
+        log_level: The ``--log-level`` value.
+        log_loggers: The ``--log-logger`` values, if any.
+        log_format: The ``--log-format`` value.
+
+    Returns:
+        The effective ``(level, loggers, format)``.
+
+    Raises:
+        SystemExit: If a variable names a level or format that does not exist.
+            A typo would otherwise silently leave logging as it was, which is
+            indistinguishable from the feature not working.
+    """
+    raw_level = os.environ.get("VGI_WORKER_LOG_LEVEL")
+    if raw_level:
+        try:
+            log_level = LogLevel(raw_level.upper())
+        except ValueError:
+            valid = ", ".join(level.value for level in LogLevel)
+            sys.exit(f"VGI_WORKER_LOG_LEVEL={raw_level!r} is not a log level (expected one of: {valid})")
+
+    raw_format = os.environ.get("VGI_WORKER_LOG_FORMAT")
+    if raw_format:
+        try:
+            log_format = LogFormat(raw_format.lower())
+        except ValueError:
+            valid = ", ".join(fmt.value for fmt in LogFormat)
+            sys.exit(f"VGI_WORKER_LOG_FORMAT={raw_format!r} is not a log format (expected one of: {valid})")
+
+    raw_loggers = os.environ.get("VGI_WORKER_LOG_LOGGERS")
+    if raw_loggers:
+        names = [name.strip() for name in raw_loggers.split(",") if name.strip()]
+        if names:
+            log_loggers = names
+
+    return log_level, log_loggers, log_format
+
+
 def configure_worker_logging(
     *,
     debug: bool = False,
@@ -55,6 +118,10 @@ def configure_worker_logging(
 ) -> int:
     """Configure stdlib logging for a VGI worker process.
 
+    ``VGI_WORKER_LOG_LEVEL``, ``VGI_WORKER_LOG_FORMAT`` and
+    ``VGI_WORKER_LOG_LOGGERS`` override the corresponding arguments when set —
+    see :func:`_env_overrides`.
+
     Args:
         debug: If True, force DEBUG on all default loggers (overrides *log_level*).
         log_level: Logging level when *debug* is False.
@@ -65,6 +132,7 @@ def configure_worker_logging(
         The effective numeric log level.
 
     """
+    log_level, log_loggers, log_format = _env_overrides(log_level, log_loggers, log_format)
     effective_level = logging.DEBUG if debug else getattr(logging, log_level.value)
 
     handler = logging.StreamHandler(sys.stderr)

vgi/profiling.py

@@ -0,0 +1,93 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Opt-in CPU profiling for VGI worker processes.
+
+A worker is normally a subprocess someone else launched — a container, a test
+harness, a DuckDB client spawning it over stdio — so the usual ways of profiling
+Python (``python -m cProfile``, an interactive session) are not available to
+whoever needs the answer. Setting one environment variable is.
+
+Shared by every worker entry point rather than living in one CLI, so that
+``vgi-serve`` and the fixture servers profile identically.
+"""
+
+from __future__ import annotations
+
+import os
+import pathlib
+import sys
+
+__all__ = ["maybe_start_profile"]
+
+
+def maybe_start_profile() -> None:
+    """Start a ``cProfile`` profiler when ``VGI_WORKER_PROFILE`` names a path.
+
+    Set ``VGI_WORKER_PROFILE`` to a directory (or a file path) and the worker
+    writes ``pstats`` data on exit, readable with :mod:`pstats`, snakeviz, or
+    ``python -m pstats``. A directory gets one file per process
+    (``worker-<pid>.prof``), which is what you want when the thing under
+    investigation spawns more than one worker.
+
+    **All threads are covered by the single profiler.** That matters because
+    under ``--http`` a worker runs on waitress, which serves every request on a
+    pool thread — a main-thread-only profile would faithfully report time spent
+    in ``select()`` and nothing about the work being measured. Since Python 3.12
+    ``cProfile`` is built on :mod:`sys.monitoring`, whose tool registration is
+    process-global rather than per-thread, so one profiler observes the pool
+    threads too.
+
+    The corollary is that there can be only **one** active profiler in the
+    process: ``sys.monitoring`` allows a given tool id to be claimed once, so a
+    profiler-per-thread raises ``ValueError: Another profiling tool is already
+    active`` in every waitress thread — and because that happens as the thread
+    starts, the server never accepts a connection at all.
+
+    ``atexit`` alone would not be enough: a worker is normally stopped with
+    ``SIGTERM``, whose default action terminates without unwinding, so the
+    profile would be lost on every normal shutdown. A handler turns it into an
+    ordinary exit — but only if nothing else has claimed ``SIGTERM``, since
+    stealing another component's shutdown path would be a worse bug than a
+    missing profile.
+    """
+    dest = os.environ.get("VGI_WORKER_PROFILE")
+    if not dest:
+        return
+
+    import atexit
+    import cProfile
+    import signal
+
+    path = pathlib.Path(dest)
+    if path.is_dir() or dest.endswith(os.sep):
+        path.mkdir(parents=True, exist_ok=True)
+        path = path / f"worker-{os.getpid()}.prof"
+    else:
+        path.parent.mkdir(parents=True, exist_ok=True)
+
+    profiler = cProfile.Profile()
+    profiler.enable()
+    dumped = False
+
+    def _dump() -> None:
+        # atexit and the SIGTERM handler can both reach this; dumping twice
+        # would truncate the file the first call just wrote.
+        nonlocal dumped
+        if dumped:
+            return
+        dumped = True
+        try:
+            profiler.disable()
+            profiler.dump_stats(str(path))
+        except Exception as exc:  # noqa: BLE001 — never fail a shutdown over telemetry
+            sys.stderr.write(f"vgi: could not write profile: {exc}\n")
+            return
+        sys.stderr.write(f"vgi: wrote profile to {path}\n")
+
+    atexit.register(_dump)
+
+    def _on_sigterm(signum: int, frame: object) -> None:
+        raise SystemExit(128 + signum)
+
+    if signal.getsignal(signal.SIGTERM) in (signal.SIG_DFL, None):
+        signal.signal(signal.SIGTERM, _on_sigterm)

vgi/serve.py

@@ -35,6 +35,7 @@
 from typing import TYPE_CHECKING, Any, Literal
 
 from vgi.logging_config import LogFormat, LogLevel
+from vgi.profiling import maybe_start_profile
 
 if TYPE_CHECKING:
     import falcon
@@ -364,12 +365,17 @@ def serve(
     ) -> None:
         env_debug = os.environ.get("VGI_WORKER_DEBUG", "").lower() in ("1", "true", "yes")
         effective_debug = debug or env_debug
+        # VGI_WORKER_LOG_* overrides are applied inside configure_worker_logging,
+        # so every worker entry point honours them identically.
         effective_level = configure_worker_logging(
             debug=effective_debug,
             log_level=log_level,
             log_loggers=log_logger,
             log_format=log_format,
         )
+        # Before the worker class is imported, so module-import cost lands in
+        # the profile too — for many workers that is the bulk of startup.
+        maybe_start_profile()
 
         # Resolve env var overrides
         describe = _resolve_describe(describe)