Fix: report an unusable spawn interpreter from UnsafeLocalCodeExecutor instead of leaking BrokenPipeError

[AmaadMartin]

Please ensure you have read the contribution guide before creating a pull request.

Link to Issue or Description of Change

1. Link to an existing issue (if applicable):
   N/A — no public issue tracks this defect.
2. Or, if no issue exists, describe the change:

Problem: UnsafeLocalCodeExecutor starts its worker by re-invoking the Python interpreter that multiprocessing holds. When that interpreter cannot be exec'd, execute_code leaks BrokenPipeError: [Errno 32] Broken pipe out of multiprocessing/resource_tracker.py, which reads like a fault in the executed snippet. The method is declared to return a CodeExecutionResult and the flow processor does not catch, so the whole invocation aborts. The model then retries the same snippet and fails the same way, because nothing in the error names the interpreter.

Solution: execute_code now returns a result whose stderr names the interpreter, states that no code was executed, and gives the remediation. The message reads multiprocessing.spawn.get_executable(), never sys.executable: those two differ exactly when a host has called multiprocessing.set_executable(), which is the situation being diagnosed. The catch stays on OSError, so a non-OSError setup failure still propagates, and an interpreter path that is falsy is caught by a guard before any multiprocessing object exists. The traceback is logged once at ERROR, so operators keep the cause.

Reporting in stderr follows the peer executors: GkeCodeExecutor reports infrastructure errors that way, and this executor already reports its own timeout that way.

Out of scope, and left alone: a worker that starts and then dies without a result. That is a different failure with its own open pull request.

Collision check: I listed the open pull requests on this fork and read the diff of every one that touches this file. Three of them change the same method. They form a stack whose middle part adds a public exception class that the tip deletes, they are based on a main that predates _kill_execution and the strict-mypy annotations, and their message names sys.executable. This change starts from current main, is self-contained, and does not carry that defect.

Testing Plan

Please describe the tests that you ran to verify your changes. This is required for all PRs that are not small documentation or typo fixes.
Unit Tests:
[x] I have added or updated unit tests for my change.
[x] All unit tests pass locally.

pytest tests/unittests/code_executors/test_unsafe_local_code_executor.py -q   # 19 passed
pytest tests/unittests/code_executors -q -n auto                             # 118 passed
mypy src/google/adk/code_executors                                           # no issues, 10 files
pre-commit run --files <the two changed files>                               # all hooks passed

Eight new tests. The eleven existing tests are unmodified and still pass.

1. test_execute_code_reports_the_unusable_interpreter — the reported bug: a host has pointed multiprocessing at a binary that is not there. It uses the real multiprocessing.set_executable(), and asserts that the bogus path appears in stderr and that sys.executable does not.
2. test_execute_code_reports_a_missing_interpreter_path — the falsy-interpreter guard, driven through the real multiprocessing global state with no test double at all.
3. and 4. — an OSError out of Process.start() and out of Queue(), the two origins the failure has.
4. ..._releases_the_queue_when_the_worker_cannot_start — a recorder over a real spawn queue proves close() and join_thread() each run once.
5. ..._does_not_tear_down_a_worker_that_never_started — no terminate(), join() or kill() on a worker that never started.
6. ..._propagates_a_non_oserror_setup_failure — pins the narrow catch.
7. ..._logs_the_launch_failure — one ERROR record, with exc_info.

Tests 3 to 8 inject a fake spawn context, because whether multiprocessing reports an unusable interpreter at all is a race, and a test cannot depend on it. multiprocessing learns the interpreter is unusable only when the write to a dead child's pipe returns EPIPE, so if the parent wins the race against the failing exec, no error is raised and the parent waits on a queue no worker will fill. I hit this: a first revision drove the real stack in a child interpreter, passed on Python 3.10, 3.11, 3.12 and 3.14, and timed out on 3.13 in CI while passing five times out of five on 3.13 locally. The tests now pin the interpreter through the real multiprocessing.set_executable() and inject only the OSError whose origin is racy.

That race is the neighbouring defect, not this one: when multiprocessing hides the failure, the worker dies and the parent waits, which is the out-of-scope case named above.

Coverage of the changed module is 100% line and branch on the new code (pytest --cov=google.adk.code_executors.unsafe_local_code_executor --cov-branch). The 21 lines the report still marks as missed are all pre-existing: they run inside the spawned child, or in error branches of the pre-existing teardown helpers.

Each new test was run against broken code and observed to fail:

Mutation	Test	Failure
Revert the whole source change	7 of the 8	BrokenPipeError: [Errno 32] Broken pipe escapes. Test 7 passes, since it pins a catch the old code does not have.
except OSError → except Exception	7	Failed: DID NOT RAISE ValueError
Delete close() / join_thread()	5	assert [] == ['close', 'join_thread']
Delete logger.exception(...)	8	assert 0 == 1
interpreter = sys.executable	1	the message names the running interpreter, not the bogus path the host set
Delete the falsy guard	2	BrokenPipeError: [Errno 32] Broken pipe

CI runs the unit tests and mypy green on Python 3.10 to 3.14. The Pre-commit Linter job is red for a reason that predates this branch: the update-constraints hook rewrites the five constraints-3.*.txt files, which this change does not touch. That job is red on the other open pull requests on this fork too.

Manual End-to-End (E2E) Tests:
Please provide instructions on how to manually test your changes, including any necessary setup or configuration.

On a POSIX host, from a checkout with the change applied:

python -c "
import multiprocessing
from unittest.mock import MagicMock
from google.adk.agents.base_agent import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.code_executors.code_execution_utils import CodeExecutionInput
from google.adk.code_executors.unsafe_local_code_executor import UnsafeLocalCodeExecutor
from google.adk.sessions.base_session_service import BaseSessionService
from google.adk.sessions.session import Session

multiprocessing.set_executable('/definitely/not/a/real/python')
ctx = InvocationContext(invocation_id='t', agent=MagicMock(spec=BaseAgent),
                        session=MagicMock(spec=Session),
                        session_service=MagicMock(spec=BaseSessionService))
print(UnsafeLocalCodeExecutor().execute_code(
    ctx, CodeExecutionInput(code='print(1)')).stderr)
"

Before the change the command dies with BrokenPipeError: [Errno 32] Broken pipe. After it, nothing escapes and the printed stderr names /definitely/not/a/real/python, says no code was executed, and points at ContainerCodeExecutor / VertexAiCodeExecutor.

Checklist

[x] I have read the CONTRIBUTING.md document.
[x] I have performed a self-review of my own code.
[x] I have commented my code, particularly in hard-to-understand areas.
[x] I have added tests that prove my fix is effective or that my feature works.
[x] New and existing unit tests pass locally with my changes.