fix(query): implement structured concurrency to prevent query() deadlocks (fixes #1136)

[Tanush-Jain]

Description

Resolves #1136 by replacing decoupled task lifecycles and superficial exception handling with structured concurrency powered by AnyIO task groups.

Problem Addressed

Previous PR #1109 introduced a local try/except inside stream_input() to convert prompt generator exceptions into error dictionaries. As audited in MCE Audit Tcas1109, this left several deadlock scenarios unaddressed:

1. Orphaned Child Pipes: If the underlying CLI subprocess dies or exits unexpectedly, read_messages() hangs on stdout because untracked child processes hold pipe file descriptors open.
2. Decoupled Task Lifecycles: Independent tasks for reading, writing, and input streaming meant failure in one stream did not trigger cascading cancellation in the others.
3. Brittle Test Masking: Regression tests relied on anyio.fail_after(5) to forcibly kill hanging calls, masking race conditions rather than enforcing true fail-fast behavior.

---

Technical Solution

1. Structured Task Group Orchestration (InternalClient._process_query_inner)

   • Bound _run_read_messages, _monitor_process, and input streaming (stream_input / wait_for_result_and_end_input) inside async with anyio.create_task_group() as tg:.
   • If any task raises an exception, the task group automatically cancels all sibling tasks immediately.

2. Native Exception Propagation (Query.stream_input)

   • Reverted the local try/except inside stream_input() that converted exceptions into dictionaries. Exceptions from AsyncIterable prompts bubble up natively to trigger immediate cross-task cancellation.

3. Proactive Subprocess Lifecycle Monitoring (SubprocessCLITransport.wait)

   • Implemented wait() on Transport and SubprocessCLITransport.
   • Added _monitor_process inside the task group to await process termination. If the process exits non-zero while I/O streams are active, ProcessError is raised immediately to unblock reading loops.

4. Fail-Fast Test Suite (tests/test_query.py)

   • Added TestStructuredConcurrency testing both AsyncIterable prompt exceptions and subprocess crash scenarios.
   • Tests execute in ~1.5 seconds across asyncio and trio backends without using artificial timeouts (anyio.fail_after()).

---

Verification

• Unit Tests: 1236 passed, 5 skipped (pytest tests/)
• Targeted Concurrency Suite: Passed in 1.59s (pytest tests/ -k TestStructuredConcurrency)
• Linting: Passed (ruff check src/ tests/)
• Formatting: Passed (ruff format --check src/ tests/)
• Type Checker: Success, 0 errors (mypy src/)

Fixes #1136