feat(streaming): bound incremental result records - #58
Conversation
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
seonghobae
left a comment
seonghobae
left a comment
There was a problem hiding this comment.
Exact-head lifecycle finding on a7af1ee08681edcf6e4e38f8fcf9dd85f340b0d9: the public operator documentation currently states that an early async for break closes the active response through normal asynchronous-generator cleanup, but Python's asynchronous iteration protocol does not call aclose() on break. Because iter_batch_records() delegates to _iter_jsonl_file(), an early consumer exit can leave the inner generator and its _request() response context open until explicit closure or nondeterministic finalization.
Fix test-first without weakening the streaming contract. Add a deterministic response-context regression that consumes one record, exits early while retaining the iterator, and proves the response remains open on the current implementation. Then provide a beginner-readable, supported deterministic-close API—such as an async context manager that owns and aclose()s the iterator in finally, or an explicitly documented contextlib.aclosing(...) usage contract—and prove early exit closes the active response exactly once. Ensure nested iterator closure is explicit, cancellation remains correct, exceptions are preserved, and docs/ADR/doctoring no longer claim implicit cleanup from a bare async for break. Do not add a write-capable workflow or mutate any other stacked branch.
seonghobae
left a comment
seonghobae
left a comment
There was a problem hiding this comment.
Two additional exact-head fail-closed findings on a7af1ee08681edcf6e4e38f8fcf9dd85f340b0d9:
-
The advertised body-free parser boundary is not complete.
_parse_jsonl_line()raisesGatewayErrorwithfrom excforUnicodeDecodeError,JSONDecodeError/ValueError, andRecursionError. A JSON decoding exception retains the full decoded provider line in fields such asJSONDecodeError.doc; a Unicode decoding exception retains the original provider bytes. Exception-chain logging or inspection can therefore recover content that the public error message andresponse_dataintentionally omit. Add RED tests that use distinctive secret bytes/text and assert the exportedGatewayErrorretains no cause or context object containing the payload. Refactor so the sanitized error is raised outside the activeexceptblock (merely usingfrom Nonestill leaves__context__attached), while preserving bounded error type/line metadata. -
The bounded-stream loop accepts zero-length byte chunks. A custom adapter can yield an unbounded sequence of
b""or empty memoryviews, making no byte progress and spinning indefinitely without reaching the line or total-download limits. Add a deterministic RED regression for an empty chunk and fail closed with a fixed body-freeNoForwardProgressdiagnostic before continuing. Preserve compatibility with real aiohttpiter_chunked()semantics and the 64 KiB maximum-chunk contract.
Implement test-first on this existing branch path only; no temporary write workflow or parallel branch writer is needed.
seonghobae
left a comment
seonghobae
left a comment
There was a problem hiding this comment.
Exact-head follow-up on 50a362a56e44a74fbc467177a2eff6a422643c63: the three prior lifecycle and confidentiality findings are addressed test-first.
open_batch_records()now owns and closes the public iterator infinally; the public iterator explicitly owns each nested provider-file iterator withcontextlib.aclosing.- Deterministic tests prove context-managed early break, explicit outer
aclose(), repeated close, consumer exception propagation, and cancellation all close the active response exactly once while preserving the original exception. - Zero-length chunks fail closed with fixed
NoForwardProgressmetadata before any line buffering. - UTF-8 and JSON decoder failures are translated outside their active exception handlers; tests prove the exported
GatewayErrorhas neither__cause__nor__context__retaining provider bytes or text. - README, architecture, ADR 0005, AGENTS, CLAUDE, operator documentation, doctoring, and CHANGELOG now state that a bare
async forbreak does not callaclose()and require explicit lifecycle ownership.
Exact-head CI run 31087306286 and Release Acceptance run 31087306374 both completed successfully; CodeRabbit is successful and unresolved inline review threads are zero. This is an author verification comment, not independent approval. The PR remains draft and stacked on exact base b6293ebf4c0f5957b3a5ba2de0f03c512cff7069; its evidence must be regenerated after prerequisite integration.
seonghobae
left a comment
seonghobae
left a comment
There was a problem hiding this comment.
Exact-head remediation verification for 50a362a56e44a74fbc467177a2eff6a422643c63 against stacked base b6293ebf4c0f5957b3a5ba2de0f03c512cff7069:
- deterministic RED coverage now proves context-managed early exit, explicit outer
aclose(), nested provider response closure exactly once, consumer exception preservation, cancellation propagation, zero-progress chunk rejection, and parser cause/context confidentiality; open_batch_records()owns the public iterator infinally, whileiter_batch_records()owns each nested file iterator throughcontextlib.aclosing;- empty byte and memoryview chunks fail closed as
NoForwardProgressbefore buffering; - invalid UTF-8 and malformed JSON are translated after their active decoder handlers, leaving exported
GatewayError.__cause__and.__context__unset; - AGENTS, CLAUDE, ARCHITECTURE, ADR 0005, doctoring, user documentation, CHANGELOG, and the PR contract now state the exact lifecycle and confidentiality boundaries;
- exact-head CI run
31087306286and Release Acceptance run31087306374completed successfully; CodeRabbit status is successful and no unresolved review thread exists.
The PR remains a stacked draft and is not approved or merge-ready. All evidence must be regenerated after prerequisite integration and retargeting to main.
|
Exact-head author verification for
This is author verification, not independent approval. The PR remains a stacked draft; all evidence must be regenerated after prerequisite integration and retargeting. |
seonghobae
left a comment
seonghobae
left a comment
There was a problem hiding this comment.
Exact-head transport lifecycle finding on 36bd1d3ad597d8d6cd6e7160198a6f622ec6d62f: BatchAPIClient._request() catches aiohttp.ClientError and asyncio.TimeoutError around the yield response boundary. A payload-stream failure raised by the consumer after the response has been handed off is therefore misclassified as a pre-handoff idempotent GET failure. With retry attempts remaining, the async context manager opens a second response while handling athrow(), which can produce RuntimeError: generator didn't stop after athrow() and may retry after records have already been yielded. That violates the streaming client's at-most-once record and deterministic body-failure boundary.
Fix test-first on the existing PR branch. Add a deterministic RED contract proving a post-handoff aiohttp.ClientPayloadError performs one GET, zero retry sleeps, and becomes one bounded GatewayError rather than reopening the request or exposing an async-contextmanager runtime error. Then separate request-acquisition/status retries from exceptions thrown by the caller body, preserve retries for failures before handoff, close the active response once, and add streaming-level coverage proving a mid-file payload failure never restarts from byte zero or duplicates records. Keep all current credential, timeout, no-redirect, Retry-After, status, coverage, docstring, and body-free diagnostic contracts unchanged.
|
Superseded exact-head review request. The branch advanced after this comment from |
🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
base="b6293ebf4c0f5957b3a5ba2de0f03c512cff7069"
head="0b49a40b1b75df29106a334673d96c8cb7d8180f"
echo "== Commit verification =="
git cat-file -e "${base}^{commit}"
git cat-file -e "${head}^{commit}"
git merge-base --is-ancestor "$base" "$head"
git diff --stat "$base" "$head"
echo
echo "== Changed files =="
git diff --name-status "$base" "$head"
echo
echo "== Python files in diff =="
git diff --name-only --diff-filter=AM "$base" "$head" | grep -E '\.py$' || true
echo
echo "== Python structural outlines =="
git diff --name-only --diff-filter=AM "$base" "$head" | grep -E '\.py$' | while IFS= read -r file; do
echo "--- $file"
ast-grep outline "$file" --items all --type class,function --view compact || true
done
echo
echo "== Targeted symbol references =="
rg -n -C 3 --glob '*.py' \
'StreamingBatchAPIClient|BatchResultRecord|open_batch_records|download_results|_request|ClientPayloadError|max_jsonl_physical_lines|iter_chunked|iter_any|aclose|close\s*\(' \
.Length of output: 102546 🏁 Script executed: #!/bin/bash
set -euo pipefail
base="b6293ebf4c0f5957b3a5ba2de0f03c512cff7069"
head="0b49a40b1b75df29106a334673d96c8cb7d8180f"
echo "== Supported structural outline: result streaming =="
ast-grep outline pg_llm_batch/result_streaming.py --items all --view signatures
echo
echo "== Modified request and shared bounded-read code =="
git diff --unified=100 "$base" "$head" -- pg_llm_batch/batch_api_client.py pg_llm_batch/exceptions.py
echo
echo "== Result streaming implementation, Lines 1-460 =="
nl -ba pg_llm_batch/result_streaming.py | sed -n '1,460p'
echo
echo "== Handoff regression tests =="
nl -ba tests/test_streaming_transport_handoff.py | sed -n '1,280p'
echo
echo "== Streaming ownership and adapter regression tests =="
nl -ba tests/test_bounded_jsonl_result_streaming_coverage.py | sed -n '120,280p'
echo
echo "== Physical-line regression tests =="
nl -ba tests/test_bounded_jsonl_physical_line_budget.py | sed -n '90,180p'
echo
echo "== Main streaming behavior tests =="
nl -ba tests/test_bounded_jsonl_result_streaming.py | sed -n '120,420p'Length of output: 17452 🏁 Script executed: #!/bin/bash
set -euo pipefail
base="b6293ebf4c0f5957b3a5ba2de0f03c512cff7069"
head="0b49a40b1b75df29106a334673d96c8cb7d8180f"
numbered() {
awk '{printf "%6d %s\n", NR, $0}' "$1"
}
echo "== Diff whitespace validation =="
git diff --check "$base" "$head"
echo
echo "== result_streaming.py =="
numbered pg_llm_batch/result_streaming.py
echo
echo "== batch_api_client.py Lines 330-455 (_request and bounded reader) =="
awk 'NR >= 330 && NR <= 455 {printf "%6d %s\n", NR, $0}' pg_llm_batch/batch_api_client.py
echo
echo "== test_streaming_transport_handoff.py =="
numbered tests/test_streaming_transport_handoff.py
echo
echo "== test_bounded_jsonl_result_streaming_coverage.py Lines 120-280 =="
awk 'NR >= 120 && NR <= 280 {printf "%6d %s\n", NR, $0}' tests/test_bounded_jsonl_result_streaming_coverage.py
echo
echo "== test_bounded_jsonl_physical_line_budget.py Lines 90-180 =="
awk 'NR >= 90 && NR <= 180 {printf "%6d %s\n", NR, $0}' tests/test_bounded_jsonl_physical_line_budget.py
echo
echo "== test_bounded_jsonl_result_streaming.py Lines 120-420 =="
awk 'NR >= 120 && NR <= 420 {printf "%6d %s\n", NR, $0}' tests/test_bounded_jsonl_result_streaming.pyLength of output: 58212 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "== Public record-mutability contract =="
rg -n -i -C 3 \
'BatchResultRecord|immutable|mutabl|mapping|record.*dict|dict.*record' \
README.md docs/result-streaming.md AGENTS.md ARCHITECTURE.md \
pg_llm_batch/result_streaming.py \
tests/test_bounded_jsonl_result_streaming.py \
tests/test_bounded_jsonl_result_streaming_coverage.py \
tests/test_streaming_transport_handoff.py || true
echo
echo "== Changed streaming code: exception chaining and catch paths =="
rg -n -C 4 \
'raise |except |__cause__|__context__|from None|aclose|aclosing|asynccontextmanager' \
pg_llm_batch/result_streaming.py pg_llm_batch/batch_api_client.py \
tests/test_bounded_jsonl_result_streaming.py \
tests/test_bounded_jsonl_result_streaming_coverage.py \
tests/test_streaming_transport_handoff.py
echo
echo "== Remaining main streaming test file =="
awk 'NR >= 1 && NR <= 135 {printf "%6d %s\n", NR, $0}' tests/test_bounded_jsonl_result_streaming.py
awk 'NR >= 393 && NR <= 500 {printf "%6d %s\n", NR, $0}' tests/test_bounded_jsonl_result_streaming.py
echo
echo "== Full coverage test file, for unreviewed cases =="
awk '{printf "%6d %s\n", NR, $0}' tests/test_bounded_jsonl_result_streaming_coverage.py
echo
echo "== New-file diff metadata and no generated artifacts =="
git diff --name-only b6293ebf4c0f5957b3a5ba2de0f03c512cff7069 0b49a40b1b75df29106a334673d96c8cb7d8180f \
| grep -E '(^|/)(dist|build|\.cache|__pycache__|\.pytest_cache)/|\.pyc$' \
&& exit 1 || trueLength of output: 80083
I found no other actionable defects in the reviewed streaming limits, iterator ownership, zero-progress handling, body-free parser diagnostics, or response-handoff boundary. This is a review-only result. The stacked draft remains not merge-ready until prerequisite integration, retargeting to
|
seonghobae
left a comment
seonghobae
left a comment
There was a problem hiding this comment.
Exact-head verification for the transport handoff finding on 7fcd0ad4b4d2beedf0f753783e59fb57bf35d38d against stacked base b6293ebf4c0f5957b3a5ba2de0f03c512cff7069:
- request acquisition and retryable HTTP-status decisions finish before response handoff;
- direct payload failure, response-close failure, and a streaming body that yields one record then fails each perform one GET, zero retry sleeps, one response close, and no replay;
- translated transport diagnostics retain neither provider-controlled text nor the original exception through
__cause__or__context__; - AGENTS, CLAUDE, ARCHITECTURE, ADR 0005, doctoring, README, operator documentation, and CHANGELOG now define the same pre-handoff retry and post-handoff at-most-once boundary;
- exact-head CI run
31098730542completed successfully on Python 3.10, 3.12, and 3.14 with 526 tests passed, 100% of 1,957 production statements and 522 branches covered, 100% public docstrings, clean lint/compile/lock/package/container gates; - exact-head Release Acceptance run
31098730612completed successfully; CodeRabbit status is successful and unresolved inline review threads are zero.
The valid finding is addressed. This remains an author verification comment, not independent approval. The PR remains a stacked draft and its evidence cannot be reused after prerequisite integration or retargeting.
|
@opencode-agent @cwl-noema-review @coderabbitai review Perform one fresh review-only comparison of exact current draft head Exact-head CI Review the bounded streaming memory/CPU budgets, deterministic iterator ownership, zero-progress rejection, body-free diagnostics, and the response-handoff boundary: pre-handoff acquisition/status failures may retry, while post-handoff payload or close failures must perform one GET, zero sleeps, close exactly once, expose no provider exception through cause/context/public text, and never restart or duplicate an already-yielded record. Verify the new protocol-idempotency versus application-delivery-idempotency, durable checkpoint/reconciliation, and operator recovery documentation. Report only current-head actionable findings. This is a stacked-stage review. Missing default-branch required workflows are not success, and no merge approval is valid until prerequisites are integrated, this PR is retargeted to |
🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
base="b6293ebf4c0f5957b3a5ba2de0f03c512cff7069"
head="7fcd0ad4b4d2beedf0f753783e59fb57bf35d38d"
git cat-file -e "${base}^{commit}"
git cat-file -e "${head}^{commit}"
git merge-base --is-ancestor "$base" "$head"
echo "== Changed files =="
git diff --name-status "$base" "$head"
echo "== Whitespace errors =="
git diff --check "$base" "$head"
echo "== Python outlines =="
git diff --name-only --diff-filter=AM "$base" "$head" |
grep -E '\.py$' |
while IFS= read -r file; do
echo "--- $file"
ast-grep outline "$file" --items all --view signatures || true
done
echo "== Streaming and transport implementation =="
for file in pg_llm_batch/result_streaming.py pg_llm_batch/batch_api_client.py pg_llm_batch/exceptions.py; do
if [ -f "$file" ]; then
echo "--- $file"
nl -ba "$file" | sed -n '1,620p'
fi
done
echo "== Focused regression tests =="
for file in \
tests/test_streaming_transport_handoff.py \
tests/test_bounded_jsonl_result_streaming.py \
tests/test_bounded_jsonl_result_streaming_coverage.py \
tests/test_bounded_jsonl_physical_line_budget.py; do
if [ -f "$file" ]; then
echo "--- $file"
nl -ba "$file" | sed -n '1,700p'
fi
done
echo "== Contract documentation references =="
rg -n -i -C 3 \
'open_batch_records|StreamingBatchAPIClient|BatchResultRecord|idempot|checkpoint|reconcil|recovery|physical.line|zero.progress|handoff|retry|close' \
README.md AGENTS.md CLAUDE.md ARCHITECTURE.md CHANGELOG.md docs pg_llm_batch \
|| trueLength of output: 11278 🏁 Script executed: #!/bin/bash
set -euo pipefail
base="b6293ebf4c0f5957b3a5ba2de0f03c512cff7069"
head="7fcd0ad4b4d2beedf0f753783e59fb57bf35d38d"
numbered() {
awk '{printf "%6d %s\n", NR, $0}' "$1"
}
echo "== Exact implementation diffs =="
git diff --unified=80 "$base" "$head" -- \
pg_llm_batch/result_streaming.py \
pg_llm_batch/batch_api_client.py \
pg_llm_batch/exceptions.py \
pg_llm_batch/__init__.py
echo "== result_streaming.py =="
numbered pg_llm_batch/result_streaming.py
echo "== batch_api_client.py: request and bounded-stream methods =="
rg -n -C 8 \
'asynccontextmanager|def _request|async def _request|yield response|ClientPayloadError|iter_chunked|iter_any|async for|await response\.close|response\.close|retry|sleep' \
pg_llm_batch/batch_api_client.py
echo "== Handoff regression tests =="
numbered tests/test_streaming_transport_handoff.py
echo "== Streaming coverage tests =="
numbered tests/test_bounded_jsonl_result_streaming_coverage.py
echo "== Core streaming tests =="
numbered tests/test_bounded_jsonl_result_streaming.py
echo "== Physical line tests =="
numbered tests/test_bounded_jsonl_physical_line_budget.py
echo "== Documentation contract excerpts =="
rg -n -i -C 5 \
'protocol.idempot|application.delivery|checkpoint|reconcil|operator recovery|handoff|one GET|zero retry|physical.line|max_jsonl_physical_lines|zero.progress|close exactly once' \
README.md AGENTS.md CLAUDE.md ARCHITECTURE.md CHANGELOG.md \
docs/result-streaming.md \
docs/adr/0005-bounded-jsonl-result-streaming.md \
docs/doctoring/bounded-jsonl-result-streaming.md \
|| trueLength of output: 160151
I found no other current-head actionable defects in the reviewed bounds, iterator closure, zero-progress rejection, body-free diagnostics, response-handoff behavior, or checkpoint and recovery documentation. This was a read-only review. I did not modify, retarget, or merge the branch. The stacked draft remains not merge-ready until prerequisite integration, retargeting to
|
b6293eb to
6a6fac2
Compare
7fcd0ad to
c0b26eb
Compare
|
@coderabbitai review |
|
1 participant
Product and reliability gap
Bounded aggregate provider-file downloads still materialize complete JSONL and parsed records. This Draft adds an opt-in incremental result path with explicit physical-line accounting, deterministic response ownership, provider-payload confidentiality, and a post-handoff at-most-once application-delivery boundary while preserving the aggregate API.
Bounded behavior
StreamingBatchAPIClient+ immutableBatchResultRecordare opt-in.open_batch_records()is the deterministic-close consumer boundary.Current exact staged state
c0b26eb8746820b13f7ef0f4d191827a8f7e2a94.6a6fac2a99134f2bf0271d0d550dffd36ddb6d9c.31284475447: success.31284475435: success.Dependency chain
Current order is:
.github#790 -> #53 -> #57 -> this PR -> #59 -> #60 -> #92 -> #94 -> #95 -> #96 -> #97.Keep Draft while predecessors are unintegrated. After protected integration of prerequisites, reconcile/retarget only for integration correctness and reacquire every required exact-head/exact-base CI, security, dependency, SBOM/provenance, packaging, semantic-review, branch-policy, and repository-policy gate.
Merge only when the unchanged final head has zero valid unresolved findings and every review/approval requirement actually imposed by live policy is satisfied. The protected code-owner gate is disabled/on hold for the current solo-maintainer state and must not be inferred as a universal approval requirement. Queued/pending/failed/cancelled/skipped/absent/stale/predecessor/synthetic/status-only/author-only/rate-limited evidence is not final success.