fix(batch): surface background stream failures on exit instead of returning partial results

[g-despot]

Why

Split out of #2056: a behaviour change for client.batch.stream() / collection.batch.stream() on every platform, so it gets its own review. (#2056 stays open for reference; siblings: #2140 token-refresh fixes, #2142 the trimmed WASM transport.)

Today a background failure in the stream (the recv/loop task or thread dying, or the server ending the stream early) is only noticed if the user calls add_object() / flush() afterwards. If it happens after the last add, leaving the with block returns normally with partial results — silent data loss. The sync __exit__ never raised at all.

What changes

• _wait() (both sync and async) raises the recorded background exception, or a WeaviateBatchStreamError saying how many objects/references were left unsent when the background tasks/threads are gone. Partial results are still copied first, so batch.results / batch.failed_objects can be inspected after catching.
• Leaving with / async with batch.stream() raises that failure on a clean block. If the block itself raised, that exception wins and the background failure is only logged.
• Async flush() checks the background tasks each tick (the sync one already did), so a task that died without recording an exception raises instead of spinning forever. "Died unexpectedly" is a WeaviateBatchStreamError instead of a bare Exception.
• Sync _start() raises the stored background error as soon as a thread has died, instead of polling for 60 s and then blaming the network — defect 4 of Batch stream recovery closes the shared connection and can leave the client permanently closed #2139 (the __reconnect shared-connection defects 1–3 there are not addressed here).
• _BatchStreamShutdownError (gRPC ABORTED) is now a WeaviateBatchStreamError subclass, since it can reach users on exit.

What users will notice

with client.batch.stream() as b:
    ...
print(client.batch.failed_objects)

may now raise WeaviateBatchStreamError at the end of the block where it previously returned silently with unsent data. That is the point, but it is a behaviour change.

Tests

test/test_batch_stream_async.py (4) / test/test_batch_stream_sync.py (5), unit tests over the private internals, no cluster: _wait raises and keeps partial results, unsent-data message, flush with a dead task, _start with a stored error (fails on main after the 60 s timeout), and the context-manager rules (clean block raises; the user's exception wins). test/ + mock_tests/test_batch.py pass locally; ruff / flake8 / pyright clean.

[orca-security-eu]

Orca Security Scan Summary

Status	Check	Issues by priority
Passed	Infrastructure as Code	0   0   0   0	View in Orca
Passed	SAST	0   0   0   0	View in Orca
Passed	Secrets	0   0   0   0	View in Orca
Passed	Vulnerabilities	0   0   0   0	View in Orca

[codecov-commenter]

Codecov Report

❌ Patch coverage is 98.29932% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 88.66%. Comparing base (95b5d76) to head (f2e3e11).
⚠️ Report is 79 commits behind head on main.

Files with missing lines	Patch %	Lines
weaviate/collections/batch/async_.py	90.47%	2 Missing ⚠️
test/test_batch_stream_async.py	99.31%	1 Missing ⚠️
weaviate/collections/batch/batch_wrapper.py	94.44%	1 Missing ⚠️
weaviate/exceptions.py	66.66%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2141      +/-   ##
==========================================
+ Coverage   86.64%   88.66%   +2.01%     
==========================================
  Files         300      306       +6     
  Lines       23172    23757     +585     
==========================================
+ Hits        20077    21063     +986     
+ Misses       3095     2694     -401     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[linhongyu510]

Blocking async edge case on f1dc88a: _BgTasks.gather(return_exceptions=True) currently retrieves and then discards each task result. Because loop_wrapper and recv_wrapper catch only Exception, a task-level BaseException (notably asyncio.CancelledError) can escape the wrapper; when both queues are already empty, _wait() then sees no __bg_exception and no unsent items and returns success.

I reproduced this deterministically with a custom BaseException: on the exact PR head, a completed background task raises it, _wait() is awaited with empty object/reference queues, and pytest.raises(...) fails with DID NOT RAISE. Returning the results from _BgTasks.gather() and checking them after copying the partial results closes that silent-success path.

There is a second arbitration requirement if gathered BaseExceptions are surfaced: _ContextManagerAsync.__aexit__ catches only Exception. With a user ValueError("user code") already leaving the block and _wait() raising the task BaseException, the background failure replaces the user's exception. The exit path should catch BaseException for arbitration only: re-raise it for a clean block, but log it and preserve the original user exception when exc_type is already set.

I validated a minimal patch with three regressions covering (1) empty-queue task BaseException, (2) user exception precedence, and (3) clean-block propagation: test/test_batch_stream_sync.py test/test_batch_stream_async.py => 12 passed. Targeted Ruff lint, Ruff format, and Flake8 also pass on the three changed files. The result-copy block remains before either background failure is raised, so partial diagnostics are preserved.

Separately, the current remote rollup has two real failures rather than a generic infrastructure failure: Flake8 reports B010 at test/test_batch_stream_sync.py:84,85,87 (reproduced locally on the exact head), and the Python 3.12 integration job exits 1 in Run integration tests with auth secrets. Please fix the deterministic Flake8 errors and inspect/re-run the 3.12 integration failure; the successful unit/security/proto checks do not discharge either gate.