fix(mp): release stage thread pools so evaluate() stops leaking workers

[twang126]

Problem

MPRunner had no way to release its ThreadPoolExecutor. CPython pool workers have no idle timeout, so once started they park on the pool's internal work queue until the interpreter exits.

Every evaluator builds its pools inside evaluate(), and every orchestrator builds a fresh evaluator per call, so each call permanently added threads to a process that never restarts between requests in the server.

StreamingOrchestrator is the worst case. eval_service.py hands each arriving eval input to evaluate_item, which constructs a new Evaluator and therefore four new stage pools for a dataset of one item — four threads leaked per eval case, for the life of the pod.

Measurement

Simulating the streaming shape (500 cases × 4 stage pools × 1 submit each):

	threads	RSS
pools released	+0	+96 KB
pools leaked (before)	+2000	+47.5 MB

About 4 threads and 95 KB per eval case, growing without bound. RLIMIT_NPROC on the dev workstation is 131072, which one long-lived pod reaches at roughly 33k eval cases and then dies with can't start new thread.

Change

• MPRunner.shutdown(wait=False, cancel_futures=True) plus __enter__/__exit__, and a class docstring stating that the caller owns the pool's lifetime.
• Evaluator, AgentEvaluator, CortadoEvaluator, and DataEngineeringAgentEvaluator release their pools from a finally block, so a failing pipeline releases them too.
• The three agent evaluators now build their runner in evaluate() rather than __init__, because a runner is spent once shut down and these evaluators must tolerate being run again.
• In DataEngineeringAgentEvaluator the shutdown runs before _archive_workspace_to_gcs, so on the error path queued scenarios stop mutating the Dataform workspace while it is being zipped.

Evaluator's sqlexec pool is the one exception: it shuts down with cancel_futures=False. The dispatch loop takes a connection off db_queue before submitting each SQLExecWork, and only SQLExecWork.run returns it, so cancelling a queued item would strand a connection. StreamingOrchestrator shares one db_queue across inputs, where that would shrink the pool for every later input.

Scope

Deliberately limited to the pools that actually hold threads. InteractEvaluator and DataAgentEvaluator also construct runners, but their loops run every work item inline via work.run() and never submit, so those pools have started no threads and are left alone.

This does not raise concurrent scenarios per pod — the leaked threads are idle and hold no pool slot, DB connection, or GIL. It bounds thread and memory growth over a pod's lifetime. It also does not address timed-out work that keeps running, or the connection acquired on the dispatch thread; both are tracked separately.

Tests

New test/mprunner_test.py (8 tests) covers workers persisting until shutdown, context-manager release on both the normal and exception paths, shutdown not blocking on a hung work item, queued work being cancelled by default and kept under cancel_futures=False, and Evaluator.evaluate releasing all four stage pools with the sqlexec pool exempt from cancellation.

pytest test/mprunner_test.py test/evaluator_test.py test/robustness_test.py test/agentevaluator_test.py test/test_eval_case_timeout.py → 34 passed.

Review with git diff -w; most of the raw line count is re-indentation from wrapping Evaluator.evaluate's body in try:.

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[prernakakkar-google]

/gcbrun