fix(bigquery): BQAA P1 preview blockers — branch-safe tracing, lifecycle bounds, boundary redaction, HITL pause/resume

[caohy1988]

Summary

Resolves the plugin-local P1 items from the BQAA Java preview readiness tracker (#1316) — the correctness, lifecycle, privacy-boundary, and duplicate-telemetry blockers that can be fixed inside the agentanalytics package. All changes verified against the tracker's pinned commit 08f4fdc and mirrored, where applicable, on the Python plugin's contract at google/adk-python@9d306f5.

Intentionally excluded: "Add terminal agent/run failure analytics" (the last P1) requires new framework plugin API (onAgentErrorCallback / onRunErrorCallback, mirroring Python's RFC #5044) and should land as its own framework change; this PR only covers defects local to the plugin.

Fixes (tracker item → change)

1. Trace state safe for concurrent ParallelAgent branches

TraceManager now keeps span records in per-branch stacks keyed by InvocationContext.branch() (distinct per ParallelAgent sub-agent via BaseAgent.createInvocationContext). Agent and model spans within a branch are sequential and use per-branch LIFO; tool spans carry an operation identity and push-time parent because ADK runs an event's function calls concurrently by default; a branch completing first can no longer pop another branch's span. Pops are additionally kind-checked (invocation, agent:*, llm_request, tool), so an error callback firing without its matching push cannot pop an unrelated record. Parent resolution walks the branch ancestor chain, so a branch's first span parents to the invocation root.

2. No plugin-owned OpenTelemetry spans

TraceManager no longer calls tracer.spanBuilder(...).startSpan() — records are ID-only (16-hex local IDs). Ambient OTel context is still consulted for trace_id (and the invocation root's span_id) so rows stay joinable to Cloud Trace, but a host with an SDK exporter configured receives zero BQAA-owned spans. Regression test asserts this with an in-memory exporter.

3. Sensitive-key redaction at the final attributes boundary

logEvent now passes the fully assembled attributes map through the redaction walk immediately before serialization, regardless of producer — state_delta, custom tags, labels, and nested extra attributes included. Redaction-only (no length truncation; does not set is_truncated, matching Python).

4. Per-invocation processor/task lifecycle

BatchProcessor.start() stores its ScheduledFuture; an idempotent close() cancels it, so completed invocations no longer leave periodic tasks retaining closed processors/writers until plugin-wide shutdown.

5. Bounded append, drain, and timeout cleanup

writer.append(...).get(shutdownTimeout) bounds each Storage Write RPC (the future is cancelled on timeout); the final drain in close() is bounded by a shutdownTimeout deadline instead of looping until empty. Rows abandoned at the deadline are counted (shutdown_timeout); rows appended after close are rejected and counted (after_close).

6. Writer-construction and late-continuation loss accounting

New PluginState.appendRow gates every append on the invocation lifecycle: a late parse/offload continuation completing after ensureInvocationCompleted finalized the invocation is dropped and counted (late_after_finalize) instead of silently recreating a processor nothing will close. StreamWriter construction failures are logged SEVERE and counted (writer_create_error); the mapping stays unpopulated so the next event retries construction.

7. HITL and long-running-tool pause/resume semantics

• A synthetic adk_request_* FunctionCall now emits the plain HITL_*_REQUEST event (previously mislabeled _COMPLETED).
• Any long-running FunctionCall (HITL or ordinary) additionally emits a pairable TOOL_PAUSED row with pause_kind (hitl_credential / hitl_confirmation / hitl_input / tool, derived from the call name) and function_call_id.
• User-message handling now inspects FunctionResponse parts (where resumed input actually arrives): HITL responses route to HITL_*_COMPLETED (never TOOL_COMPLETED); non-HITL responses — by construction the resume side of a paused long-running tool — emit TOOL_COMPLETED carrying the pause pair keys.
• Adds the TOOL_PAUSED view and pause_kind / function_call_id columns on the TOOL_COMPLETED view so the pause↔resume join works end-to-end in SQL.

Verification

• Focused BQAA suite: 175 passed, 0 failed (upstream baseline: 135).
• Full core main profile: 1,631 passed, 0 failures/errors, 24 skipped (upstream baseline: 1,591).
• Acceptance-list regression tests: same-invocation ParallelAgent overlap with out-of-order completion; same-branch CONCURRENT TOOLS popped by operation identity out of start order; recording OTel exporter receives zero BQAA spans; 100 completed invocations leave no retained processors/trace managers; scheduled flush task cancelled on close; close() waits for a real blocked in-flight append and stays bounded; deadline-expired close defers teardown and final-stats delivery to the in-flight flush; post-finalize append cannot recreate a processor even after processed-cache eviction (durable captured token); writer-construction failure counted and retryable; nested secret keys redacted in state_delta, in unserializable-sibling attribute trees, and in session state ahead of truncation; HITL request/pause/complete rows correctly named and pairable with pause_kind/function_call_id on completions.

Review rounds addressed

• Round 1 (5 findings): close/flush coordination, HITL completion pair keys, fail-closed redactTree boundary, atomic finalize gate, per-operation tool spans. Note: within one branch, ADK executes an event's function calls CONCURRENTLY by default — tool spans therefore carry an operation identity (ToolContext.functionCallId) and a push-time parent; only non-tool kinds use top-of-stack semantics.
• Round 2 (3 findings): teardown ownership transfer with closeAndFold stats delivery at true teardown completion; session-state redaction before smartTruncate's whole-object fallback; durable per-invocation lifecycle tokens captured by every append continuation, immune to tombstone-cache eviction.
• Round 3 (2 findings): atomic append admission via the lifecycle token's monitor (runIfActive shared with markFinalized); removed the pre-close flush so closeAndFold owns the complete drain.
• Round 4 (2 P1 + 1 P2): ONE absolute deadline shared across pending-task waits, every processor drain, and executor termination (invocation finalization and plugin-wide close are each bounded by a single shutdownTimeout); collision-proof tool operation identity for id-less calls (the framework materializes absent IDs as "") via a weak-keys per-ToolContext synthetic ID; HITL completion content normalized to result on both producer paths (Python parity).
• Round 5 (2 P1): completion-ordered finalization (cleanup runs via andThen(fromAction) BEFORE the returned Completable completes, since RxJava's doFinally notifies downstream first; disposal covered by the same idempotent action); deadline-safe Storage Writer usage (LimitExceededBehavior.ThrowException makes append admission nonblocking with accounted rejections, and StreamWriter.close() is detached to a daemon thread so teardown honors the absolute deadline).
• Round 6 (1 P1): detached writer closes moved from raw per-invocation daemon threads to a plugin-owned BOUNDED closer service (2 threads, 256-task backlog) with saturation accounting (writer_close_saturated), shut down within the shared shutdown deadline — a Storage outage becomes a bounded backlog instead of native-thread exhaustion.
• Round 7 (1 P1): writer cleanup ownership bounded END-TO-END — live StreamWriters (each owning an internal client + NON-DAEMON append thread that only its own close() stops) are capped by admission permits acquired BEFORE construction and released only when the writer's close has run; rows refused at the cap are accounted (writer_permit_exhausted); at plugin shutdown the closer's unstarted queue drains to a bounded reclaim owner WITHOUT interrupting active closes. Branch squashed to a single commit per repo requirement.
• Round 8 (1 P1 + P3): writer cleanup ownership now covers the full interval from permit acquisition through processor publication — a WriterLease is registered BEFORE createWriter() and owns the permit and (once attached) the writer; construction/start() failures and plugin close racing admission all dispatch the writer through the lease's dispatch-once protocol; plugin close publishes a closing gate, drains every live lease, and identity-removes published processors (creators racing publication self-close via a post-publication recheck — exactly one party wins, never neither). Stale round-7 shutdown comments and test naming reworded to the queue-drain/no-interruption behavior.

Tracker: #1316

🤖 Generated with Claude Code

[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.

[caohy1988]

Code Review Results

Scope: PR #1344, 08f4fdc -> e9b4073 (9 files, +932/-216)
Intent: resolve the seven plugin-local P1 items in tracker #1316 while explicitly deferring framework-level agent/run error callbacks
Mode: report-only, solo full review
Review lenses: correctness, testing, maintainability, privacy, concurrency/performance, API contract, reliability, adversarial failure paths, project standards, and prior comments

Triage Groups

Group	Findings	Context	Preferred Resolution	Why
Concurrency and lifecycle	#1, #4, #5	Queue emptiness, processed-cache state, and branch identity are each being used as ownership signals when work can still overlap	Fix operation ownership (#5), then make invocation finalization atomic (#4), then coordinate active flush shutdown (#1)	All three otherwise produce incorrect telemetry or silent loss only under real overlap, while isolated tests stay green
Privacy and HITL contract	#2, #3	The new output contracts omit correlation on one path and fail open on one serialization path	Preserve HITL response IDs (#2) and make redaction fallback fail closed (#3)	These are local, testable fixes that complete the advertised contract

P1 -- High

#	File	Issue	Reviewer	Confidence
1	BatchProcessor.java:316	Close ignores an in-flight flush after it drains the queue	solo correctness/reliability	100
2	BigQueryAgentAnalyticsPlugin.java:575	HITL completion drops the ID needed to pair its TOOL_PAUSED row	solo correctness/API contract	100
3	JsonFormatter.java:80	Serializer fallback bypasses the new final redaction boundary	solo privacy/correctness	100
4	PluginState.java:226	Late-continuation gate races finalization and expires	solo correctness/reliability	100
5	TraceManager.java:61	Branch-keyed stack still cross-pops parallel tools	solo concurrency/correctness	100

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 - flush() drains before its bounded append, but close() waits only while the queue is non-empty. A scheduled flush can hold flushLock, drain the queue, and block in Storage Write; close then skips the drain loop and closes the writer/Arrow objects underneath it. Coordinate close with the active flush and account its in-flight batch before folding counters.
• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 - TOOL_PAUSED records the HITL call ID, but the HITL response path emits empty EventData. Since the generic content parser also does not retain FunctionResponse IDs, the pause and completion are not pairable when multiple HITL requests share an invocation. Stamp function_call_id/pause_kind on the HITL completion without emitting a duplicate TOOL_COMPLETED row.
• ADK changes #3 - The new final boundary calls smartTruncate, but smartTruncate converts the entire object first and falls back to safeToString(obj). One unsupported custom-tag/session-state object therefore makes the complete attributes map textual, including sibling api_key/token values. Walk raw containers before conversion or return a safe sentinel on whole-object conversion failure.
• ADK changes #4 - appendRow checks isProcessed and creates the processor separately, while finalization marks processed and removes the processor. A continuation can pass the check before finalization and create a fresh processor afterward. The cache tombstone also expires after 10 minutes or 10,000 entries. Use an atomic lifecycle object captured by continuations and test the actual overlap.
• ADK changes #5 - TraceManager assumes one branch is sequential, but ADK executes tools concurrently by default, all with the same branch. Both records have kind tool, so the kind check cannot distinguish ownership: the second start is parented to the first tool and reverse completion pops the wrong span. Isolate by task/function-call identity and store explicit parents.

Requirements Completeness

• Partial: ParallelAgent branches no longer share one global stack, but the replacement ownership model still fails for default parallel tools (ADK changes #5).
• Met: plugin-owned OpenTelemetry spans are removed and ambient trace correlation remains.
• Partial: normal maps are redacted at the final attributes boundary, but serializer fallback can expose the full map (ADK changes #3).
• Partial: the periodic ScheduledFuture is cancelled, but close does not coordinate a currently running flush (Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1).
• Partial: append calls have a deadline, but final drain does not cover an already in-flight batch (Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1).
• Partial: writer construction failures are counted/retryable; the late-continuation gate is not atomic and its tombstone expires (ADK changes #4).
• Partial: ordinary long-running resumes carry pair keys; HITL completions drop their response ID and no Runner-level two-turn test was added (Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2).
• Intentionally deferred: terminal agent/run error callbacks require framework API work.

Actionable Findings

#	File	Issue	Route	Notes
1	BatchProcessor.java:316	Close races in-flight flush	gated_auto -> downstream-resolver	Add lifecycle synchronization and overlap test
2	BigQueryAgentAnalyticsPlugin.java:575	HITL completion not joinable	gated_auto -> downstream-resolver	Preserve response pair keys; add Runner-level test
3	JsonFormatter.java:80	Redaction fallback fail-open	gated_auto -> downstream-resolver	Raw-container walk or safe sentinel plus unsupported-value test
4	PluginState.java:226	Finalization gate check/act race	manual -> downstream-resolver	Requires an atomic per-invocation state design
5	TraceManager.java:61	Parallel tools share one stack	manual -> downstream-resolver	Use operation identity/explicit parent rather than branch-only LIFO

Coverage

• Reviewed the complete GitHub diff and every changed production/test file at head e9b4073 against base 08f4fdc; no checkout or project mutation.
• Directly traced adjacent framework paths for ParallelAgent, default parallel tool execution, Runner user-message/event routing, and parser behavior.
• git diff --check is clean.
• Static test annotation count is 135 at base and 153 at head, matching the reported focused-suite increase.
• I could not independently run Maven because this shell has no Java runtime (java -version fails). I am treating the author's reported 153 focused and 1,609 core passes as the execution signal.
• GitHub currently has no Maven result on the PR; cla/google fails and the remaining source-analysis jobs are skipped. Agent triage passes.
• Testing gaps: no default-parallel-tool trace test; no close-vs-active-flush test; the late test is sequential rather than a timed continuation crossing finalization; redaction does not test serializer fallback; HITL tests call callbacks directly rather than a Runner-level two-turn flow.
• No prior technical review comments were present. No repository docs/solutions learnings apply. No agent-accessibility gap applies to this passive analytics plugin.
• Review and validation were solo, per user instruction; no subagents or independent validator were used. Each finding was verified directly against source control flow.

---

Verdict: Not ready.

Reasoning: The PR fixes substantial parts of all seven local items, but five P1 gaps remain: parallel tool spans can still cross-pop, late continuations can still recreate processors, close can race an active flush, HITL pause/resume rows are not ID-pairable, and redaction can fail open on serializer fallback.

Fix order: #5 operation-owned tracing -> #4 atomic invocation lifecycle -> #1 active-flush shutdown -> #2 HITL pair keys -> #3 fail-closed redaction fallback.

Actionable Findings Recap

• ADK changes #5 P1 TraceManager.java:61 - replace branch-only LIFO ownership; manual design, suggested fix present, confidence 100.
• ADK changes #4 P1 PluginState.java:226 - make append/finalize atomic and durable for late continuations; manual design, suggested fix present, confidence 100.
• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 P1 BatchProcessor.java:316 - wait/account for active flush before resource close; gated auto, suggested fix present, confidence 100.
• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 P1 BigQueryAgentAnalyticsPlugin.java:575 - retain HITL response pair keys and test two turns; gated auto, suggested fix present, confidence 100.
• ADK changes #3 P1 JsonFormatter.java:80 - make whole-attributes fallback fail closed; gated auto, suggested fix present, confidence 100.

[caohy1988]

All five findings are addressed in 3e61e57. Per-finding resolution:

#1 — close() races an in-flight flush → Fixed.
After the bounded drain loop, close() now waits (within the same shutdownTimeout deadline) for the flush lock to be released before tearing down the writer and Arrow resources, instead of treating an empty queue as proof no flush is active. If the deadline expires with a flush still in flight, that is logged SEVERE, and the in-flight batch is accounted by its own bounded append's append_error path. Regression test: close_waitsForInFlightFlushBeforeTeardown (holds the flush lock from another thread, asserts close() waits and stays bounded).

#2 — HITL completion drops its pair keys → Fixed.
Both HITL completion paths (event FunctionResponse and user-message FunctionResponse) now stamp pause_kind (derived from the synthetic call name) and function_call_id (from FunctionResponse.id(), when present) into attributes via a shared hitlPairKeys helper — no duplicate TOOL_COMPLETED row is emitted. Tests assert the keys on both paths (onEventCallback_hitlFunctionResponse_completionCarriesPairKeys, updated onUserMessageCallback_hitlFunctionResponse_emitsCompleted). Worth noting the Python plugin has the same gap on its HITL completion path; Java now exceeds parity here.

#3 — redaction fallback fails open → Fixed.
New JsonFormatter.redactTree replaces the smartTruncate-based boundary in logEvent: raw Map/Iterable containers are walked natively with sensitive keys redacted before any Jackson conversion, and only leaf values are converted individually — an unserializable leaf becomes [UNSERIALIZABLE] without affecting siblings. One implementation subtlety the new test caught: JsonNode must be handled before the Iterable branch, since ObjectNode iterates its values and would have been flattened into an array. Tests: redactTree_unserializableValue_failsClosedPerLeaf, redactTree_redactsNestedContainersAndLists, redactTree_redactsInsideConvertedPojoLeaves.

#4 — finalization gate check/act race + expiring tombstone → Fixed (with documented residual).
The isProcessed check now runs inside computeIfAbsent's mapping function, i.e. under the map's per-key lock, paired with finalization's mark-processed-before-remove ordering: a continuation that finds no mapping after removal is guaranteed to observe isProcessed == true and installs nothing (a null mapping-function result adds no entry). If a continuation wins the key lock first, finalization closes the very processor it created and the late row is accounted by the processor's after_close gate — no leak in either interleaving. The residual window is the bounded tombstone cache (size/TTL eviction): a continuation arriving after eviction could still recreate a processor, which is drained at plugin-wide shutdown. This is documented on getOrCreateProcessorIfActive; if reviewers want the cache made unbounded-with-explicit-eviction instead, happy to follow up.

#5 — parallel tools cross-pop within one branch → Fixed.
Tool spans now carry an operation identity (ToolContext.functionCallId()) and a parent captured at push time. TOOL_STARTING rows are stamped directly from the pushed record (a sibling pushing in between can no longer divert the row's span IDs); afterToolCallback/onToolErrorCallback pop by kind + identity rather than stack position; and parent resolution at push skips concurrent same-kind siblings, so a second tool starting mid-flight parents to the enclosing agent span rather than to its sibling. Non-tool kinds keep top-of-stack semantics (model calls are sequential within a branch). Tests: concurrentToolsInOneBranch_popByOperationIdentity (out-of-order completion, parent assertions), popSpan_unknownOperationId_popsNothing.

Verification: focused BQAA suite 160 passed / 0 failed (previous head: 153); full core main profile 1,616 passed / 0 failures / 24 skipped. git diff --check clean; formatter applied.

🤖 Generated with Claude Code

[caohy1988]

Code Review Results

Scope: PR #1344, 08f4fdc -> 3e61e57 (11 files, +1,375/-233)
Intent: resolve the seven plugin-local P1 items in tracker #1316 and the five findings from the first PR review; framework-level agent/run error callbacks remain explicitly deferred
Mode: report-only, solo full review
Review lenses: correctness, testing, maintainability, project standards, privacy, concurrency/performance, event/API contracts, reliability, adversarial failure paths, and prior-comment verification

Triage Groups

Group	Findings	Context	Preferred Resolution	Why
Invocation shutdown ownership	#1, #3	Both failures come from using bounded observations (a timeout or an expiring tombstone) as durable lifecycle ownership	Introduce durable per-invocation lifecycle state first (#3), then make active flush cleanup/folding part of that ownership model (#1)	One lifecycle design can prevent processor resurrection and ensure shutdown never tears resources down before their last owner releases them

P1 -- High

#	File	Issue	Reviewer	Confidence
1	BatchProcessor.java:340	Timeout path still tears resources down under an active flush and loses its later counters	solo correctness/reliability/concurrency	100
2	BigQueryAgentAnalyticsPlugin.java:498	Session state can be stringified before the fail-closed redaction boundary	solo privacy/correctness	100
3	PluginState.java:259	Expiring tombstones still allow timed-out continuations to recreate processors	solo correctness/reliability	100

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 - close() stops waiting at its deadline but then closes Arrow and writer resources even when flushLock remains held. The comment says the active batch will be counted by its append path, but invocation finalization removes the processor and folds its counters immediately after close() returns. A failure recorded by the still-running flush is therefore too late to reach plugin-level drop stats. Resource access also remains concurrent on the deadline/interrupt path. Use an active-flush completion primitive and transfer deferred cleanup/counter folding to the active flush when the caller's deadline expires; pass the remaining close budget into each drain append. Test a real blocked append rather than setting flushLock manually, including the deadline-expired case and aggregate counters.
• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 - The new redactTree helper is fail-closed when it receives the raw map, but getAttributes() first sends session.state() through smartTruncate. On one unsupported state value, smartTruncate converts the entire state map to safeToString. The final boundary then sees an ordinary string, so keys such as api_key embedded in it cannot be redacted. Redact the raw state before truncating it (for example, truncate the JsonNode returned by redactTree) and add a plugin-path test with logSessionMetadata=true, a sibling secret, and an unserializable value.
• ADK changes #3 - The immediate computeIfAbsent/finalize TOCTOU is fixed, but the code explicitly retains a 10-minute/10,000-entry processed cache and documents that a continuation after eviction recreates a processor. That is the original late-continuation failure shape: the new writer, allocator, and periodic task survive until plugin-wide shutdown. Keep a terminal lifecycle token captured by each continuation (and remove it only after those continuations settle), or cancel/disable the append continuation on invocation timeout. Add TTL and high-cardinality eviction tests that complete the old future afterward.

Actionable Findings

#	File	Issue	Route	Notes
1	BatchProcessor.java:340	Active-flush timeout teardown is still unsafe/unaccounted	manual -> downstream-resolver	Coordinate ownership, remaining deadline, deferred cleanup, and counter folding
2	BigQueryAgentAnalyticsPlugin.java:498	Production session-state path bypasses fail-closed redaction	gated_auto -> downstream-resolver	Redact raw state before truncation and add callback-level regression
3	PluginState.java:259	Tombstone eviction reopens processor creation	manual -> downstream-resolver	Durable lifecycle token or cancellation tied to pending continuation

Coverage

• Reviewed the full PR again at head 3e61e57, not only the follow-up commit. The merge base remains 08f4fdc; current main advanced to fc95ce7 in a non-overlapping flows change.
• Re-read every changed production/test file and traced adjacent Runner, Functions, ToolContext, branch creation, and parser paths.
• Prior-round status: HITL completion pair keys and per-operation tool span ownership are fixed. The close/flush, redaction, and late-continuation findings are improved but only partially closed by Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1-ADK changes #3 above.
• git diff --check is clean. Static focused-suite annotation count is 135 at base and 160 at head, matching the reported focused result.
• I could not independently run Maven because this shell has no Java runtime (java -version fails). I am treating the author's reported 160 focused and 1,616 core passes as the execution signal.
• GitHub currently exposes no Maven check for this head. check-changes passes; cla/google fails; zizmor jobs are skipped.
• Testing gaps: the close test toggles flushLock rather than blocking a real append and does not cover deadline expiry; the redaction tests call redactTree directly and miss the session-state pre-stringification path; the late gate test is sequential and does not cover cache eviction; HITL remains callback-unit-tested rather than the tracker-requested Runner-level two-turn flow.
• The PR body is stale after 3e61e57: it still reports 153/1,609 tests and still describes within-branch execution as sequential without documenting operation-owned tool spans.
• No applicable repository docs/solutions artifacts or agent-accessibility gaps were found. Review and validation were solo per user instruction; no subagents or independent validator were used.

---

Verdict: Not ready.

Reasoning: Two of the five prior findings are fully closed, but three P1 paths remain: shutdown can still race an active flush and lose its counters, session-state fallback can still expose secrets before final redaction, and cache eviction can still resurrect a finalized invocation processor.

Fix order: #2 fail-closed production redaction -> #3 durable invocation lifecycle -> #1 active-flush ownership and deadline-safe teardown.

Actionable Findings Recap

• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 P1 BigQueryAgentAnalyticsPlugin.java:498 - redact raw session state before any whole-map fallback; gated auto, suggested fix present, confidence 100.
• ADK changes #3 P1 PluginState.java:259 - make finalized state durable for outstanding continuations beyond cache eviction; manual design, suggested fix present, confidence 100.
• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 P1 BatchProcessor.java:340 - never tear down/fold while an active flush can still mutate resources or counters; manual design, suggested fix present, confidence 100.

[caohy1988]

All three round-2 findings are addressed in 2cdd638, in the suggested fix order (#2 → #3 → #1). The PR body is also refreshed (test counts and the tool-span ownership description the previous body still misstated).

#2 — session state stringified before the redaction boundary → Fixed.
getAttributes now routes session.state() through the fail-closed redactTree FIRST, then length-bounds the resulting JsonNode with smartTruncate — which can no longer hit the whole-object textual fallback because its input is already a JsonNode. New plugin-path regression (logEvent_sessionState_redactedBeforeTruncationFallback) exercises the real callback with logSessionMetadata on: a sibling secret, an unserializable value, and a plain value — asserting the state stays structured, api_key is [REDACTED], and the bad value becomes [UNSERIALIZABLE].

#3 — tombstone eviction resurrects processors → Fixed with a durable token.
Exactly the "terminal lifecycle token captured by each continuation" you suggested: logEvent captures a per-invocation InvocationLifecycle while the invocation is active; every append continuation closes over it. Finalization marks the token BEFORE removing the processor mapping, and the token map entry is removed at finalize for memory bounds — outstanding continuations keep the token reachable through their captured reference, so it cannot be evicted the way the bounded cache can. The computeIfAbsent gate now checks the token first (cache second, for callers whose token postdates an evicted finalization). New regression appendRow_finalizedToken_dropsEvenAfterTombstoneEviction invalidates the processed cache via reflection after finalization and asserts the late append is still dropped with no processor resurrected.

#1 — deadline-expired teardown under an active flush, late counters lost → Fixed with ownership transfer.
The flush path now uses a ReentrantLock instead of a CAS flag, so close() can genuinely wait rather than infer from queue emptiness:

• close() publishes its deadline up front; appendTimeoutMillis() caps every drain and in-flight append to the remaining close budget (your "pass the remaining close budget into each drain append").
• close() sets teardownRequested, then waits for the mutex within the deadline. Acquired → teardown runs with no flush active. Expired → ownership transfers: the in-flight flush performs the teardown in its finally after releasing the mutex (with a re-check in close() covering the release-vs-flag visibility window, and a tornDown CAS making the handoff exactly-once). Resources are never destroyed underneath an active flush on any path.
• Counter folding moved to teardown completion: closeAndFold(consumer) delivers the FINAL drop-stat snapshot when teardown actually happens — possibly after close() returns — so an append failure recorded by that last flush reaches plugin-level drop stats instead of being folded away early. PluginState finalization and plugin-wide shutdown both fold through this callback now.

Both requested test shapes are in: close_waitsForInFlightFlushBeforeTeardown uses a real blocked append (the writer call parks; no manual lock toggling), and close_deadlineExpired_defersTeardownAndStatsToInFlightFlush blocks the writer past the close deadline and asserts close returns bounded, teardown+final stats are delivered by the deferred flush afterward, and the aggregate includes that flush's append_error.

Verification: focused BQAA suite 163 passed / 0 failed (previous head: 160); full core main profile 1,619 passed / 0 failures / 24 skipped. Formatter applied; git diff --check clean.

Remaining known gap from your testing-gaps list: the Runner-level two-turn HITL flow is still covered at callback level rather than through a real Runner; happy to add that as a follow-up if you consider it blocking for this PR.

🤖 Generated with Claude Code

[caohy1988]

Code Review Results

Scope: PR #1344, 08f4fdc -> 2cdd638 (11 files, +1,636/-251)
Intent: resolve the seven plugin-local P1 items in tracker #1316 and close both prior PR review rounds; framework-level terminal agent/run errors remain explicitly deferred
Mode: report-only, solo full review
Review lenses: correctness, testing, maintainability, project standards, privacy, concurrency/performance, event/API contracts, reliability, adversarial shutdown behavior, and prior-comment verification

Triage Groups

Group	Findings	Context	Preferred Resolution	Why
Finalization boundary	#1, #2	The new token and close deadline are sound inside their own methods, but work can still cross the boundary before either takes ownership	Make append admission atomic with token finalization (#1), then remove the pre-close drain outside the deadline (#2)	These two changes make one invocation-finalization deadline and one append ownership gate authoritative end-to-end

P1 -- High

#	File	Issue	Reviewer	Confidence
1	PluginState.java:272	A continuation already past the token gate can append after teardown/final stats	solo correctness/reliability/concurrency	100
2	PluginState.java:408	The explicit pre-close flush runs outside the newly published shutdown deadline	solo correctness/reliability	100

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 - appendRow releases the processor-map/token gate before calling processor.append, while timeout finalization can concurrently mark the token, remove the processor, close it, and deliver the final stats. A continuation that obtained the processor just before finalization can resume afterward. If it reaches BatchProcessor.append after close, it increments after_close after the supposedly final snapshot was delivered, so plugin-level stats miss the loss. If it read closed=false before close and resumes at queue.offer afterward, the row is accepted after the drain and can be stranded or schedule a flush against torn-down resources. Add an atomic lifecycle appendIfActive critical section shared with markFinalized, held only through the processor append/queue offer; add a latch test that pauses after processor lookup and crosses finalization.
• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 - Invocation finalization calls processor.flush() before closeAndFold(). closeDeadline is published only inside close(), so that first flush gets a full shutdownTimeout; closeAndFold then starts a fresh deadline for the remaining batches. With default batchSize=1 and two queued rows, a slow first append can consume nearly one timeout and the second can consume another, leaving the final drain outside the tracker-required total bound. Remove the explicit flush and let closeAndFold own the complete drain. Add an ensureInvocationCompleted multi-batch test where the first append consumes most of the budget and the second never completes.

Actionable Findings

#	File	Issue	Route	Notes
1	PluginState.java:272	Append admission is not atomic with lifecycle finalization	manual -> downstream-resolver	Share a short token critical section between finalization and processor append
2	PluginState.java:408	Pre-close flush resets the final-drain budget	gated_auto -> downstream-resolver	Remove redundant flush and add multi-batch total-deadline regression

Coverage

• Reviewed the full PR again at head 2cdd638, not only the round-3 delta. The merge base remains 08f4fdc; current main has no overlapping plugin change.
• Re-read all changed production/test files and traced adjacent Runner, parallel tool execution, ToolContext, parser, branch, and plugin-finalization paths.
• Round-3 verification: session state is now redacted before truncation; captured lifecycle tokens survive cache eviction; flush teardown ownership and deferred counter delivery work for an already active flush. Findings Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1-Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 are boundary interleavings outside those covered shapes.
• git diff --check is clean. Static focused-suite annotation count is 135 at base and 163 at head, matching the reported focused result.
• I could not independently run Maven because this shell has no Java runtime (java -version fails). I am treating the author's reported 163 focused and 1,619 core passes as the execution signal.
• GitHub currently exposes no Maven check. Agent triage and check-changes pass; cla/google fails; zizmor jobs are skipped.
• Runner-level HITL verdict: the missing two-turn test is a real tracker acceptance gap, but I would not block this code PR solely on it. Direct callback tests cover routing/pair keys, and Runner source confirms user FunctionResponse messages traverse onUserMessageCallback once rather than duplicating through onEventCallback. It should still land before checking off that acceptance item or claiming end-to-end Preview readiness.
• Additional cleanup: PluginState Javadoc at lines 288-290 still describes tombstone resurrection as a residual limitation even though the captured-token design addresses that shape. The PR body also retains the old blanket statement that a branch is sequential before later correctly describing concurrent tools.
• No applicable repository docs/solutions artifacts or agent-accessibility gaps were found. Review and validation were solo per user instruction; no subagents or independent validator were used.

---

Verdict: Not ready.

Reasoning: The three round-2 implementations substantially improve the design, and the redaction fix is complete. Two P1 lifecycle boundaries remain: a continuation can cross finalization after passing the token gate, and the pre-close flush can consume a separate full timeout before the bounded close drain begins.

Fix order: #1 atomic append/finalize admission -> #2 single close-owned drain deadline.

Actionable Findings Recap

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 P1 PluginState.java:272 - make lifecycle finalization and the final processor append admission atomic; manual design, suggested fix present, confidence 100.
• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 P1 PluginState.java:408 - remove the flush outside closeAndFold and test the total multi-batch deadline; gated auto, suggested fix present, confidence 100.

[caohy1988]

Both round-3 findings are addressed in 73c6233, plus the two cleanup notes.

#1 — append admission not atomic with finalization → Fixed with a shared token critical section.
Exactly the suggested shape: InvocationLifecycle.runIfActive(action) executes the processor append inside the token's monitor, shared with markFinalized. The interleaving you described is now impossible in both directions: a continuation that passed the processor-map gate either completes its (non-blocking queue.offer) admission before finalization proceeds — in which case the close-owned drain flushes that row — or observes the finalized token and is refused with late_after_finalize accounting at the plugin level, which is outside the folded snapshot and therefore never missed. Finalization blocks only for the short admission section; no lock is held across CHM operations on either side, so there is no ordering cycle. The requested latch test is in: appendRow_admissionIsAtomicWithFinalization parks a thread inside the admission section, asserts ensureInvocationCompleted waits for it (≥250ms of a 300ms hold), then asserts the next admission is atomically refused and counted.

#2 — pre-close flush escapes the close deadline → Fixed.
The explicit processor.flush() in invocation finalization is gone; closeAndFold now owns the complete drain under one published shutdownTimeout deadline, with each append capped to the remaining budget. The requested multi-batch regression is in: ensureInvocationCompleted_multiBatchDrain_boundedByOneShutdownTimeout queues two batches (batchSize=1) against never-completing appends with a 500ms timeout and asserts total finalization stays within a single bound (<1.2s observed ceiling; measured ~600ms), where the old code would consume ~2x.

Cleanup notes:

• The stale "residual limitation" javadoc on getOrCreateProcessorIfActive now states the durable-token guarantee (cache eviction is no longer a correctness hole).
• The TraceManager class javadoc and the PR body no longer claim within-branch execution is sequential; both now state the split explicitly (agent/model spans: sequential, top-of-stack; tool spans: concurrent by default, operation-identity owned).

On the Runner-level HITL verdict: agreed with your assessment — I'll land the two-turn Runner test before checking off that tracker acceptance item, as a follow-up to this PR.

Verification: focused BQAA suite 165 passed / 0 failed (previous head: 163); full core main profile 1,621 passed / 0 failures / 24 skipped. Formatter applied; git diff --check clean.

🤖 Generated with Claude Code

[caohy1988]

Fresh full review at 73c6233

The two round-3 fixes are correct: append admission is now atomic with finalization, the pre-close flush is removed, and both documentation cleanups landed. The earlier redaction, OTel, active-flush teardown, drop-accounting, lifecycle, and HITL pair-key fixes also remain sound for their covered inputs.

I found two remaining P1s and one P2 in the complete PR:

P1 - Missing function-call IDs collapse concurrent tool-span ownership

The framework creates ToolContext with functionCall.id().orElse(""), and the plugin uses that value as the tool span's operation identity. Two concurrent calls without IDs both become ""; when the first-started call completes first, the newest matching record is the sibling's, so it is popped instead. This recreates the same cross-pop the operation-identity design is intended to prevent.

Use a unique internal key per ToolContext/call, or have the framework synthesize and consistently propagate a unique function-call ID. Extend the existing out-of-order concurrent-tool test with both source IDs absent.

P1 - shutdownTimeout is restarted across phases and processors

The config defines shutdownTimeout as the maximum time to wait for shutdown, but invocation finalization can spend one complete timeout waiting for pending tasks and then start another complete processor deadline. Plugin-wide close also closes every processor sequentially, each with a fresh deadline, before granting the executors another timeout.

Two processors with queued rows and never-completing appends therefore take about 2 * shutdownTimeout; a pending parse plus one queued processor does the same. Establish one absolute deadline at the PluginState operation boundary and pass the remaining budget into every processor drain and executor wait.

The new regression uses a 500 ms timeout but accepts <1200 ms, so the old approximately 1000 ms two-deadline behavior can still pass. Assert append count/drop counters or use a deterministic shared-deadline test. Add a pending-task-plus-drain case and a multi-processor close case.

P2 - HITL completion content differs between callback paths

User-message completions emit content.result, while event-callback completions for the same HITL_*_REQUEST_COMPLETED event type emit content.response. The pinned Python implementation uses result on both paths. Normalize Java to result and assert identical content keys across both producers.

Coverage and verdict

• Reviewed the entire 08f4fdc -> 73c6233 PR, not only the latest delta. Current main has no overlapping agentanalytics change.
• git diff --check is clean. Static focused test count is 135 at base and 165 at head.
• I could not independently run Maven because this environment has no Java runtime; the reported 165 focused and 1,621 core passes remain the execution signal.
• The Runner-level two-turn HITL test remains required before checking off that tracker acceptance item.
• The PR body still reports 163 focused / 1,619 core and lists review rounds only through round 2.
• Agent triage and check-changes pass; cla/google fails; no Maven check is exposed.

Verdict: not ready. Suggested order: shared absolute shutdown deadline -> collision-proof tool operation identity -> normalize HITL completion content.

[caohy1988]

All three round-4 findings are addressed in b941ac7, in your suggested order. The PR body is refreshed (counts + rounds 3–4).

P1 — shared absolute shutdown deadline → Fixed.
ensureInvocationCompleted and plugin-wide close() each compute ONE absolute deadline at subscription (Completable.defer, so the budget starts when the operation actually runs) and thread it through everything downstream: the pending-task wait, every processor's drain (BatchProcessor.closeAndFold/close gained deadline-accepting variants that consume only the remaining budget), and both executor terminations. A stuck parse plus a queued processor, or N stuck processors, are all bounded by a single shutdownTimeout now.
On your test critique — agreed, the <1200ms bound was too loose to catch the old ~1000ms two-deadline behavior. It's tightened to <900ms (old behavior now fails it), and both requested cases are added: ensureInvocationCompleted_pendingTaskPlusDrain_shareOneDeadline (never-completing pending task + queued never-completing append) and close_multipleStuckProcessors_shareOneDeadline (two stuck processors under plugin-wide close, <1100ms vs the old ~2×500ms + executor budget).

P1 — id-less tool calls collide on "" → Fixed.
toolOperationId uses the real function-call ID when non-empty; otherwise a synthetic ID keyed by the ToolContext instance (a Guava weak-keys cache: identity semantics, GC cleanup, and the same instance flows through the before/after/error callbacks of one call — so push and pop agree without any framework change). hitlPairKeys also filters empty-string IDs, which are useless as join keys. The requested test is in: concurrentIdLessTools_keepSpanOwnership runs two tools with functionCallId = Optional.of("") (exactly what Functions.java materializes), completes them out of start order, and asserts each TOOL_COMPLETED row's span_id matches its own TOOL_STARTING row with distinct spans per tool.

P2 — HITL completion content divergence → Fixed.
The event path now emits content.result like the user-message path and the pinned Python implementation; the event-path completion test asserts the key.

Verification: focused BQAA suite 168 passed / 0 failed (previous head: 165); full core main profile 1,624 passed / 0 failures / 24 skipped. Formatter applied; git diff --check clean.

Standing follow-up unchanged: the Runner-level two-turn HITL test before the tracker acceptance item is checked off.

🤖 Generated with Claude Code

[caohy1988]

Code Review Results

Scope: PR #1344, 08f4fdc -> b941ac7 (11 files, +1,938/-260)
Intent: resolve the seven plugin-local P1 items in tracker #1316 and all four prior review rounds; framework-level terminal agent/run errors remain explicitly deferred
Mode: report-only, solo full review
Reviewers: correctness, testing, maintainability, project standards, privacy, concurrency/performance, API contract, reliability, adversarial failure paths, and prior comments

Triage Groups

Group	Findings	Context	Preferred Resolution	Why
Shutdown completion contract	#1, #2	The new absolute deadline is propagated through plugin code, but downstream completion and blocking Storage Writer calls can still cross it	Make finalization part of the returned Completable first (#1), then decide how Storage Writer admission and teardown become deadline-safe (#2)	The current timing tests cannot prove a hard public shutdown bound until both ownership layers are included

P1 -- High

#	File	Issue	Reviewer	Confidence
1	PluginState.java:425	doFinally signals completion before invocation/plugin cleanup runs	correctness, API contract, reliability, testing	100
2	BatchProcessor.java:181	Storage Writer entry and teardown can block outside the absolute deadline	reliability, performance, adversarial	100

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 - Both ensureInvocationCompleted and close put lifecycle marking, processor draining, resource teardown, and map cleanup inside doFinally (PluginState.java:438-466, 536-591). RxJava 3.1.12 implements CompletableDoFinally.onComplete as downstream.onComplete(); runFinally();; its blocking observer releases its latch in that downstream callback. On an asynchronous pending-task timeout/completion, blockingAwait() and ordinary subscribers can therefore observe success before markFinalized, closeAndFold, writer/client close, or executor shutdown has finished. The new pending-task-plus-drain test measures only that early notification and can pass while the drain is still running. Chain cleanup before completion, for example with onErrorComplete().andThen(Completable.fromAction(...)), and cover disposal separately if it must also finalize. Add a test whose pending future completes asynchronously and whose cleanup is latch-blocked; assert the returned Completable does not complete until cleanup releases.
• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 - The deadline only bounds appendFuture.get(...); writer.append(recordBatch) itself runs first (BatchProcessor.java:181-185), and teardownOnce() calls writer.close() inline after the deadline wait (BatchProcessor.java:432-455). The resolved BigQuery Storage dependency is 3.21.0. Its StreamWriter defaults flow control to LimitExceededBehavior.Block; ConnectionWorker.append waits for inflight quota for up to 300,000 ms before returning the future. Its close path joins the append thread, then can await the client for 150 seconds and its callback pool for 3 minutes. Thus quota saturation or a stuck Storage connection can exceed shutdownTimeout even with the shared deadline. Make append admission nonblocking (for example, configure ThrowException and account the rejection), and move writer teardown behind an explicitly bounded/asynchronous ownership mechanism. Add regressions where append() blocks before returning a future and where close() blocks; the public operation must still honor one deadline without destroying resources under active work.

Actionable Findings

#	File	Issue	Route	Notes
1	PluginState.java:425	Cleanup happens after the returned Completable notifies completion	gated_auto -> downstream-resolver	Concrete operator-ordering fix and deterministic regression shape are available
2	BatchProcessor.java:181	Storage Writer calls escape the configured shutdown bound	manual -> downstream-resolver	Requires an explicit flow-control and asynchronous teardown ownership decision

Coverage

• Reviewed the complete 08f4fdc -> b941ac7 diff and every changed production/test file, not only the round-five delta. Latest main is 760c8da; it has no overlapping agentanalytics production or test changes.
• Round-four verification: the shared deadline is threaded across pending tasks, processor drains, and executor waits; id-less concurrent tools now receive stable per-ToolContext synthetic identities; populated FunctionCall IDs are contractually unique in google-genai; event-path HITL completions now use content.result. Those fixes are sound within their stated boundaries.
• git diff --check is clean. The PR body is current through round four and reports 168 focused tests and 1,624 full-core tests passing with 24 skipped.
• I could not independently run Maven because this shell still has no Java runtime (java -version fails). The reported test results remain the execution signal. GitHub exposes no Maven check; agent triage and check-changes pass, cla/google fails, and zizmor jobs are skipped.
• Testing gaps: ensureInvocationCompleted_pendingTaskPlusDrain_shareOneDeadline is invalidated by Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 because doFinally notifies first. close_multipleStuckProcessors_shareOneDeadline accepts <1100ms for two 500ms processors, so the old roughly 1000ms per-processor behavior can still pass. All writer mocks return a future immediately and close immediately, so Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 is uncovered.
• The Runner-level two-turn HITL test remains a non-blocking follow-up and should land before the tracker acceptance item is checked off.
• No applicable docs/solutions/CONCEPTS.md learnings or agent-native gaps were found. Review and validation were solo per user instruction; no subagents were used.

---

Verdict: Not ready.

Reasoning: The three round-four code changes themselves are correct, but the advertised shutdown contract still has two P1 ownership holes: callers can observe completion before cleanup, and the real Storage Writer can block outside the propagated deadline.

Fix order: #1 make cleanup completion-ordered -> #2 bound Storage Writer admission and teardown -> add the real Runner HITL follow-up -> resolve CLA.

Actionable Findings Recap

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 P1 PluginState.java:425 - move finalization ahead of downstream completion; gated_auto, suggested fix present, confidence 100.
• Revert "Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory" #2 P1 BatchProcessor.java:181 - make Storage Writer entry/teardown obey the public bound; manual, suggested direction present, confidence 100.

[caohy1988]

Both round-5 findings are addressed in 544690e, in your suggested order. PR body updated through round 5.

#1 — doFinally signals completion before cleanup → Fixed with completion-ordered cleanup.
Exactly the suggested operator shape: both finalizeInvocation and closeInternal now build an idempotent cleanup Action and chain it as onErrorComplete().andThen(Completable.fromAction(cleanup)) — the returned Completable cannot complete until lifecycle marking, processor draining (closeAndFold), writer/client close initiation, and executor shutdown have run. Disposal is covered by a trailing doFinally(cleanup) sharing the same run-once action, so the normal path gets strict ordering and a disposed subscription still finalizes. The requested regression is in: ensureInvocationCompleted_doesNotCompleteBeforeCleanupFinishes completes the pending future asynchronously, latch-blocks the cleanup inside a mock processor's closeAndFold, asserts the subscriber has NOT observed completion while cleanup is blocked, then releases the latch and asserts completion. This also re-validates the round-4 pendingTaskPlusDrain timing test you correctly flagged as invalidated — blockingAwait now includes the drain.

#2 — Storage Writer entry/teardown escape the bound → Fixed on both sides.

• Admission: the writer is now built with setLimitExceededBehavior(ThrowException) (confirmed available on StreamWriter.Builder in the resolved 3.21.0), so inflight-quota saturation surfaces as an immediate append failure — caught by the existing flush error path and accounted as dropped rows — instead of parking append() for up to five minutes under the default Block behavior.
• Teardown: StreamWriter.close() is detached to a named daemon thread inside teardownOnce. The public close()/deferred-flush path no longer inherits its thread-join and client/callback-pool waits; the detachment is safe because teardown only runs after appends have stopped (closed gate + flush-mutex ownership) and the drop-stat snapshot is finalized before delivery, so nothing else touches the writer. Regression: close_blockingWriterClose_doesNotBlockTeardown blocks writer.close() for 1.5s and asserts the public close() returns well under that while the writer is still closed eventually.

Test-gap items from your coverage notes: the multi-processor close bound is tightened from <1100ms to <900ms, so the old per-processor-deadline behavior now fails it; the writer-close verifications use Mockito.timeout() now that teardown is detached.

Verification: focused BQAA suite 170 passed / 0 failed (previous head: 168); full core main profile 1,626 passed / 0 failures / 24 skipped. Formatter applied; git diff --check clean.

Standing follow-ups unchanged: Runner-level two-turn HITL test before the tracker acceptance item is checked off, and the CLA on this fork account.

🤖 Generated with Claude Code

[caohy1988]

Code Review Results

Scope: PR #1344, 08f4fdc -> 544690e (11 files, +2,092/-288)
Intent: resolve the seven plugin-local P1 items in tracker #1316 and all five prior review rounds; framework-level terminal agent/run errors remain intentionally deferred
Mode: report-only, solo full review
Reviewers: correctness, testing, maintainability, project standards, privacy/security, concurrency/performance, event/API contract, reliability, adversarial failure paths, agent accessibility, learnings, and prior comments

P1 -- High

#	File	Issue	Reviewer	Confidence
1	BatchProcessor.java:457	One raw closer thread per invocation is unbounded under slow writer shutdown	reliability, performance, adversarial, testing	100

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 - Every BatchProcessor teardown now creates and starts a new daemon thread for StreamWriter.close(). PluginState creates a distinct processor/writer per invocation, and there is no executor, semaphore, queue bound, or tracking for these closer threads. The resolved BigQuery Storage 3.21.0 implementation creates its own non-daemon append thread and close() joins it, then may wait 150 seconds for its client and three minutes for its callback pool. During a Storage outage, timed-out append requests can therefore leave one blocked closer plus one writer thread per completed invocation; the new nonblocking finalization converts invocation throughput directly into live-thread growth until native-thread exhaustion. At that point Thread.start() can fail before final stats are delivered. The single blocking-close test proves the individual call returns quickly but does not exercise multiple invocations or assert a resource bound. Replace raw per-writer thread creation with a bounded lifecycle design. The cleanest option is a plugin-scoped/shared StreamWriter (or the client's supported connection-pool mode) closed once at plugin shutdown; otherwise use a plugin-owned bounded daemon closer service with explicit saturation accounting and a policy for writers it cannot admit. Add a many-invocation test with blocked writer.close() calls that asserts a fixed upper bound on closer threads/tasks and verifies plugin shutdown behavior.

Actionable Findings

#	File	Issue	Route	Notes
1	BatchProcessor.java:457	Detached writer close creates unbounded raw threads	manual -> downstream-resolver	Requires choosing shared-writer/connection-pool ownership or a bounded closer with saturation semantics

Coverage

• Reviewed the complete 08f4fdc -> 544690e diff and every changed production/test file, not only the round-six delta. Current main remains 760c8da and has no overlapping agentanalytics change.
• Round-five verification: normal completion now runs cleanup before downstream completion; the idempotent trailing doFinally covers disposal; LimitExceededBehavior.ThrowException is present in the resolved 3.21.0 API and removes the five-minute inflight-quota wait; the multi-processor timing bound is now discriminating. Those mechanics are correct.
• All prior review findings remain fixed for their covered interleavings. The Runner-level two-turn HITL test remains a non-blocking follow-up before checking off the tracker acceptance item.
• git diff --check is clean. Static focused-suite annotation count is 135 at base and 170 at head, matching the reported focused result. The PR body is current through round five.
• Maven could not be rerun because this environment still has no Java runtime (java -version fails). The reported 170 focused and 1,626 full-core passes remain the execution signal.
• GitHub exposes no Maven result. Agent triage and check-changes pass; cla/google fails; zizmor jobs are skipped.
• Testing gap: close_blockingWriterClose_doesNotBlockTeardown covers one writer only. No test applies sustained blocked closes across many completed invocations or asserts bounded closer resources. The ThrowException builder setting is source-verified but has no regression assertion.
• No applicable docs/solutions/CONCEPTS.md learnings or agent-native gaps were found. Review and validation were solo per user instruction; no subagents or independent validator were used. Finding Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 was directly verified against the PR and the resolved BigQuery Storage source.

---

Verdict: Not ready.

Reasoning: The two round-five fixes work individually, but detached teardown introduces one unbounded raw thread per invocation. Under the exact slow-close failure shape this change is meant to contain, the plugin can exhaust process threads and still fail to complete resource shutdown.

Fix order: #1 establish bounded/shared writer-close ownership -> add the many-invocation blocked-close regression -> land the Runner HITL follow-up -> resolve CLA.

Actionable Findings Recap

• Bump com.github.tomakehurst:wiremock-jre8 from 2.35.0 to 2.35.1 in /core in the maven group across 1 directory #1 P1 BatchProcessor.java:457 - replace unbounded per-invocation closer threads with bounded/shared ownership; manual, suggested design options present, confidence 100.

[caohy1988]

The round-6 P1 is addressed in ad83e20. PR body updated through round 6.

#1 — unbounded per-invocation closer threads → Fixed with a plugin-owned bounded closer service.
Went with your second suggested design (the shared/plugin-scoped StreamWriter would change the per-invocation writer ownership model this whole PR is built around, so it felt like a separate discussion): PluginState owns one bounded ThreadPoolExecutor for detached closes — 2 daemon threads, 256-task ArrayBlockingQueue backlog, core-thread timeout — shared by every BatchProcessor. teardownOnce submits the close instead of spawning a thread.

• Saturation policy (explicit, as requested): AbortPolicy rejection never blocks teardown and never spawns a thread; the rejected writer is counted under a new writer_close_saturated drop-stat key (unit: writers, documented) and left for BigQueryWriteClient shutdown to reclaim at the transport level.
• Plugin shutdown: the closer service is shut down inside the same shared absolute deadline as everything else, with shutdownNow() interrupting closes still blocked past the bound — a close backlog cannot extend plugin shutdown.
• Under the outage shape you described: invocation throughput now feeds a bounded backlog (each blocked close is individually bounded by the Storage client's own internal waits), with worst-case live growth of exactly 2 closer threads, not one per invocation.

Requested regression is in: manyInvocationsWithBlockedWriterClose_boundedCloserThreads runs 25 completed invocations whose writer.close() blocks indefinitely on a latch, and asserts closer-thread growth ≤ the pool bound (delta-based so sibling-test plugin instances in the same JVM don't pollute the count — the first run of the absolute version failed on exactly that, which was itself a nice confirmation the count catches real threads), no retained processors, and plugin close() completing within its bound with the backlog still pending.

Verification: focused BQAA suite 171 passed / 0 failed (previous head: 170); full core main profile 1,627 passed / 0 failures / 24 skipped. Formatter applied; git diff --check clean.

On the remaining coverage note: agreed the ThrowException builder setting has no regression assertion — it is source-verified against 3.21.0 and exercised only indirectly; asserting it would require reflecting on StreamWriter internals or a real connection, so I'd prefer to leave it to the gated real-BigQuery smoke test the tracker already lists (P2). Standing follow-ups unchanged: Runner-level two-turn HITL test, and the CLA.

🤖 Generated with Claude Code

[caohy1988]

Round 7 fresh review — 1 P1 remains

I re-reviewed the complete PR at ad83e20, including the round-6 closer change. The executor now bounds the closer service, but it does not bound or reclaim the StreamWriter resources when a close task is rejected or discarded.

P1 — saturation and forced shutdown abandon live StreamWriters

BatchProcessor.teardownOnce() marks the processor torn down, then handles RejectedExecutionException by incrementing writer_close_saturated; it neither calls writer.close() nor transfers the writer to another cleanup owner.

The comment that plugin BigQueryWriteClient shutdown will reclaim it is not true for the resolved Storage Writer 3.21.0 implementation:

• A non-pooling StreamWriter constructs its own ConnectionWorker.
• That worker creates a separate BigQueryWriteClient and starts a non-daemon append thread.
• Those resources are signaled, joined, and closed by ConnectionWorker.close().

Closing PluginState.writeClient therefore cannot stop an abandoned writer's internal client/thread. Once 2 closes are active and 256 are queued, every subsequent rejection leaves another non-daemon writer thread and client live, so sustained outage still produces unbounded process-level resource growth.

There is a second path to the same leak: closeInternal() ignores the tasks returned by writerCloseExecutor.shutdownNow(). Those queued close tasks never start, so their writers never receive close() either.

Suggested fix

Move the resource bound to writer ownership/admission before StreamWriter construction, so a constructed writer can never lose its cleanup owner. In preference order:

1. Use one plugin-scoped/shared writer, or Storage Writer's supported connection-pool/multiplexing mode, and close that bounded resource set once at plugin shutdown.
2. If per-invocation writers must remain, introduce a capped writer pool/permit acquired before createWriter(). Refuse or account new analytics rows when capacity is exhausted; release the permit only after that writer's close has actually started/completed. This bounds the internal clients and non-daemon append threads, not merely the executor queue.
3. A pending-close registry can preserve ownership for rejected and shutdownNow()-returned tasks, but it is insufficient by itself if it can grow without bound. Do not set the writer's lifecycle terminal until cleanup has either completed or transferred to a durable, bounded owner.

Requested regressions:

• Saturation: block both close workers, fill all 256 queue slots, then submit at least one more writer. Assert every constructed writer is eventually closed exactly once, or that its construction was refused before internal resources were allocated.
• Deadline shutdown: call plugin close() while close workers and queued tasks are still blocked, force shutdownNow(), then release the latch and assert no queued writer lost cleanup ownership.
• Keep the closer-thread bound assertion, but do not use it as the resource invariant; mock close() ownership directly and add the existing real-BigQuery smoke test for internal-thread/process-exit behavior.

The current 25-writer regression does not reach the 258-task capacity, releases its latch before plugin shutdown, and only counts bq-analytics-writer-close-* threads, so it cannot exercise either failure path above.

All findings from rounds 1–6 remain fixed for their covered interleavings. Separate merge gates: cla/google still fails, and the repository requires a single-commit PR while this branch currently has seven commits.

Verdict: not ready until writer cleanup ownership is bounded end-to-end.

[caohy1988]

The round-7 P1 is addressed in e7537a7 — which is now the single squashed commit for the whole branch, per the repo's single-commit requirement (the per-round SHAs referenced in earlier comments remain fetchable). PR body updated through round 7.

Your source trace was right and my round-6 comment was wrong: a non-pooling StreamWriter's internal client and non-daemon append thread are only stopped by its own close, so both the saturation path and shutdownNow() leftovers were genuine leaks. Fix follows your option 2 — the bound moved to writer admission, before construction:

• Admission permits: tryCreateProcessor() acquires a permit from a Semaphore(MAX_LIVE_WRITERS = 64) BEFORE createWriter(); on exhaustion the row is refused with accounting (writer_permit_exhausted) and no internal resources are ever allocated. The permit is released only in the close task's finally — i.e. when that writer's close has actually run. Construction failure releases the permit immediately.
• Saturation designed out, not handled: MAX_LIVE_WRITERS (64) <= closer queue capacity (256) is a stated invariant, so every constructed writer always has a queue slot and pre-shutdown RejectedExecutionException is impossible. A rejection can only mean the closer was already shut down (teardown racing plugin close), where ownership transfers to a daemon reclaim thread — bounded by the permit cap. The round-6 writer_close_saturated key is gone since the state it counted is unreachable.
• shutdownNow() leak closed — with a twist found by the new test: the first implementation reclaimed the returned queue but the test failed 3/5, because shutdownNow() also interrupts the two ACTIVE closes mid-join, abandoning partially-closed writers with no retry. Plugin shutdown now drains the unstarted queue via getQueue().drainTo(...) to the bounded reclaim owner and lets active closes run to natural completion on their daemon workers — no interruption, no loss.

Requested regressions:

• writerPermitCap_boundsLiveWritersAndPreservesCleanupOwnership — exhausts all 64 permits with blocked closes, asserts the 65th row is refused before construction (creation counter unchanged, accounted), then drains the backlog and asserts every constructed writer is closed exactly once.
• close_shutdownNowLeftovers_retainWriterCleanupOwnership — plugin close() with 2 active + 3 queued blocked closes, past the deadline; asserts all 5 writers are eventually closed exactly once (this is the test that caught the interruption bug above).
• The round-6 closer-thread bound assertion is retained but, per your note, ownership is asserted via mocked close() counts, not thread names.

Verification: focused BQAA suite 173 passed / 0 failed (previous head: 171); full core main profile 1,629 passed / 0 failures / 24 skipped. Formatter applied; git diff --check clean; branch is one commit.

Remaining gates: cla/google (account-level, being handled outside this PR) and the Runner-level two-turn HITL follow-up before the tracker acceptance item is checked off.

🤖 Generated with Claude Code

[caohy1988]

Round 8 fresh review — 1 P1 remains

I re-reviewed the complete squashed PR at e7537a7. The round-7 permit cap correctly bounds the steady-state writer backlog, but cleanup ownership still begins too late: after writer construction and processor startup/publication.

P1 — processor construction can still abandon a permitted writer

tryCreateProcessor() acquires a permit before createWriter(), but the catch path only releases that permit. If new BatchProcessor(...) or p.start() throws after createWriter() returned, the constructed StreamWriter is never passed to closeWriterDetached().

p.start() has a concrete failure shape: its scheduled task is rejected when the shared scheduler has concurrently shut down. Releasing the permit while leaving that writer's internal client/non-daemon append thread alive defeats the resource cap — subsequent writers can reuse the permit while the abandoned writer still exists.

There is a second interleaving with the same root cause. closeInternal() marks the lifecycles it currently sees, closes only processors already published in batchProcessors, and then clears the map. There is no plugin-wide closing gate. A computeIfAbsent mapping already inside tryCreateProcessor() can therefore be absent from the close iteration and later be cleared or published after it, without any teardown owner.

This is reachable before pending-task coordination protects it: for an already-completed parseFuture, thenRun(...) may execute processor creation inline before addPendingTask(...) registers the future.

Suggested fix

Establish a cleanup owner immediately after permit acquisition, before createWriter():

1. Introduce a small WriterLease/admission record registered in a concurrent live-lease set before construction. It owns the permit and, once available, the writer.
2. Plugin close first publishes a global closing state and marks/drains every registered lease. A creator completing after that point observes the lease's close request and sends its writer directly to the detached closer instead of publishing a processor.
3. Transfer the lease to BatchProcessor only after construction and start() succeed. On any exception after the writer exists, close through the lease; do not release the permit directly. The close task's finally remains the only permit-release point.
4. Avoid blind batchProcessors.clear(). Remove each published processor with identity (remove(id, processor)) while closing it; a creator that races publication must re-check closing after computeIfAbsent returns and remove/close its own processor. The protocol should guarantee that either plugin close owns the processor or the creator self-closes it — never neither.

A simpler implementation can use a global closing gate plus post-publication recheck, provided processor creation and the close iteration have an explicit atomic handoff. Do not hold a global mutex across potentially blocking createWriter() merely to close the race; that would reintroduce unbounded plugin shutdown.

Requested regressions:

• Construction vs close: block createWriter() after permit acquisition, start plugin close(), then release construction. Assert plugin close stays bounded, the writer is closed exactly once, no processor remains, and capacity is restored only after the close task runs.
• Startup rejection: return a writer, force p.start() to reject via a shut-down/injected scheduler, and assert the writer is still closed exactly once and the permit is not directly released ahead of cleanup.

P3 — stale shutdown comments

PluginState.java:703-704 still says blocked closes are interrupted and the shared write client reclaims them, while the implementation deliberately does neither. The close_shutdownNowLeftovers_retainWriterCleanupOwnership test name and comments also describe the removed shutdownNow() design. Please rename/reword these to the queue-drain/reclaim behavior; this is the exact ownership assumption corrected in round 7.

All findings from rounds 1–7 remain fixed for their covered interleavings. The branch is correctly squashed to one commit. cla/google and the Runner-level two-turn HITL follow-up remain the separate gates.

Verdict: not ready until cleanup ownership covers the interval from permit acquisition through processor publication.

[caohy1988]

Both round-8 items are addressed in the amended single commit 5de0d93 (branch stays one commit; the previous head e7537a7 remains fetchable). PR body updated through round 8.

P1 — ownership gap from permit acquisition to publication → Fixed with the suggested lease design.

• WriterLease registered before construction: tryCreateProcessor() acquires the permit, registers a lease in a concurrent live-lease set, and only then calls createWriter(). The lease's monitor-guarded protocol (attachWriter / requestClose / takeForClose) guarantees dispatch-exactly-once no matter whether the creator, the processor's teardown, or plugin close gets there first.
• Construction/startup failures never abandon a writer: if new BatchProcessor(...) throws, the attached writer dispatches through the lease; if p.start() is rejected (shared scheduler concurrently shut down — your concrete shape), p.close() runs the processor teardown, which closes the Arrow resources AND routes the writer through the lease. The permit is never released directly once a writer is attached — the close task's finally remains the single release point (creation-failure-before-attach is the only direct-release, writer-less case).
• Closing gate + atomic handoff: closeInternal's cleanup publishes closing first, drains every registered lease (requestClose dispatches constructed writers; mid-construction writers dispatch when their creator attaches and observes closeRequested), then identity-removes each published processor (remove(id, processor)) while closing it — the blind clear() is gone. A creator racing publication re-checks closing after computeIfAbsent returns and self-closes the processor it still owns: exactly one party wins the remove, never neither. No global mutex is held across createWriter(), per your constraint.
• The pre-pending-task inline-execution path you identified (thenRun on an already-completed parseFuture) is covered by the same gate, since it goes through tryCreateProcessor() like every other creation.

Both requested regressions are in:

• close_racingWriterConstruction_writerClosedExactlyOnce — blocks createWriter() after permit acquisition, runs plugin close() (asserted bounded), releases construction, and asserts the writer is closed exactly once with no processor ever published.
• startupRejection_writerStillClosedExactlyOnce — shuts down the shared scheduler so p.start() throws, and asserts the constructed writer is still closed exactly once (via the lease, not a direct permit release), with writer_create_error accounted and no processor retained.

P3 — stale comments → Fixed. The shutdown comment now describes the actual drain-without-interruption/reclaim-owner behavior, and the round-7 test is renamed to close_pastDeadline_queuedWriterClosesRetainCleanupOwnership with its comments reworded.

Verification: focused BQAA suite 175 passed / 0 failed (previous head: 173); full core main profile 1,631 passed / 0 failures / 24 skipped. Formatter applied; git diff --check clean; single commit.

Remaining gates unchanged: cla/google and the Runner-level two-turn HITL follow-up.

🤖 Generated with Claude Code

[caohy1988]

Round-9 P3 addressed in the amended single commit 70d7d54: both stale shutdownNow() comments (drainQueuedWriterCloses javadoc and the close_pastDeadline_queuedWriterClosesRetainCleanupOwnership test comment) now describe the actual behavior — the unstarted closer queue drains to the bounded reclaim owner while active closes finish uninterrupted on their daemon workers. Documentation-only change; PluginStateTest re-run green (25/25), no runtime code touched.

With the round-9 runtime verdict clean across all nine review rounds, the PR is moving out of draft. Remaining items tracked outside this PR: cla/google for this account, and the Runner-level two-turn HITL test before tracker #1316's acceptance item is checked off.

🤖 Generated with Claude Code