fix(plugins): close BQAA fail-open privacy, GCS concurrency, and startup-loss gaps (#6356)

[caohy1988]

Addresses the four P1 findings and the mechanical P2s from #6356 (BigQuery Agent Analytics: fail-open privacy, GCS concurrency, and startup-loss gaps).

P1 fixes

1. content_formatter failure now fails closed. A raising formatter replaces content with a [FORMATTER_FAILED] sentinel instead of falling back to the unformatted payload; the exception is logged without the payload and counted as formatter_failed in get_drop_stats(). The existing test that codified fail-open (asserting the original "Secret message" was written) now asserts fail-closed.

2. Final sanitizer pass over the complete attributes tree. _log_event now runs _recursive_smart_truncate over the fully assembled attributes immediately before json.dumps, covering the producers that copied values in directly (state_delta via extra_attributes, custom_tags, labels, nested structures, temp:-scoped keys). Truncation from this pass propagates into is_truncated. New: _sanitize_json_blob redacts sensitive keys inside JSON-encoded string blobs (e.g. cached credential JSON) — the #5112-adjacent gap. It decodes container-shaped ({/[) strings FIRST (raw-substring prefilters are bypassable via JSON escapes), enforces max_content_length before json.loads, and fails closed to [UNPARSEABLE_JSON_BLOB] for anything container-shaped it cannot parse and verify (malformed/trailing-garbage/over-limit/too-deep blobs, duplicate members always reserialized). Producer-local sanitization is kept as an optimization; the final pass is the mandatory boundary.

3. GCS offload paths are call-local. HybridContentParser.parse() and _parse_content_object() take trace_id/span_id as keyword arguments and build every object path from those call-local values. _log_event passes them per call instead of mutating the shared parser. The constructor fields remain only as a backward-compatible default. Regression test: two overlapping two-part parses with a forced interleave assert 4 unique object paths with correct per-event ownership (previously: event A resumed with event B's identity and both wrote trace-b/span-b_p1, the later upload overwriting the earlier).

4. Table readiness is a startup requirement with bounded retry. _ensure_schema_exists now raises on table get/create failures (preserving NotFound → create and Conflict → concurrent-creation handling) instead of logging and returning. _ensure_started records the failure, keeps _started=False, and retries on a later event after exponential backoff (2s → 60s cap). Setup is coalesced ACROSS event loops and threads via a concurrent.futures.Future claimed under a briefly-held threading.Lock (waiters asyncio.wrap_future it from their own loops), so exactly one _lazy_setup runs per attempt and a persistent outage costs one setup RPC per backoff window, not one per event. Rows arriving while setup is unavailable are counted as setup_unavailable.

P2 hardening included

• enabled=False is a hard no-op at the single choke point (_ensure_started), covering before_run_callback, __aenter__, and _log_event: no ADC lookup, client creation, table RPCs, or background tasks. Test asserts zero side effects.
• Runtime settings validated at construction (_validate_runtime_config): batch_size, batch_flush_interval, shutdown_timeout, queue_max_size, max_content_length, and all RetryConfig fields. Notably max_retries < 0 used to skip the write loop entirely, dropping every batch without one attempt.
• Loss accounting before/outside the BatchProcessor: plugin-level drop counters (formatter_failed, setup_unavailable) merged into get_drop_stats(); they survive shutdown and loop cleanup.

Acceptance tests from #6356 covered here

• A raising content_formatter never writes the original content.
• Final attributes sanitization covers state deltas, custom tags, nested structures, temp: keys, and JSON-encoded secret blobs.
• Concurrent two-part GCS offloads retain call-local trace/span paths and never share an object name.
• Failed table get/create leaves _started=False, records loss, and succeeds on a later retry (coalesced, backed off).
• enabled=False causes zero auth/client/table/writer/background side effects through Runner callbacks and async context-manager use.
• Invalid batch, queue, duration, and retry settings fail during plugin construction.
• Drop stats remain queryable after shutdown and include pre-processor failure reasons.
• Bounding/decoupling first-use setup latency from agent callbacks (P2) — intentionally deferred; the retry/backoff change bounds the failure path, but the cold-start await is a larger design choice (explicit preflight vs. async readiness) best done as a follow-up.

Verification

• New TestIssue6356Hardening class (config validation, final-pass redaction incl. JSON blobs, concurrent-offload path isolation, setup retry, disabled-mode no-op, drop-stat persistence) + the rewritten formatter-failure test. Full-suite local run in progress — draft until it and CI confirm green.
• mypy (repo config) on the plugin file: no new errors vs. main baseline; pyink --check / isort --check-only clean.

Refs #6356 (kept open: the deferred cold-start latency P2 remains tracked there; closing is the maintainers' call on merge).

[caohy1988]

Fresh review at exact head 257c51b0 against base 9d306f5d.

Verdict: FAIL. Keep this draft. I found four P1 production issues, three P2 issues, and one deterministic red test.

1. [P1] Startup retry skips table readiness. _lazy_setup() caches _schema before _ensure_schema_exists() succeeds. On retry, the cache causes the table check to be skipped, then _started=True. Repro: after recovery, get_table.call_count remained 1. Run readiness on every startup attempt until it succeeds, and strengthen the test to assert the second RPC.

2. [P1] GCS keys still collide inside multi-message requests. idx restarts for each Content, while every message shares the same trace/span in the LlmRequest.contents loop. Two images at part 0 produced the same path, so the second overwrites the first. Add a per-parse unique ID plus content and part ordinals.

3. [P1] Formatter failures can leak the protected payload into application logs. The handler uses exc_info=True. If the formatter raises with content in its exception message, the traceback logs it. My repro printed TOPSECRET. Log only a generic message or the exception class, without its message or traceback.

4. [P1] Valid JSON encoding bypasses blob redaction. The raw substring prefilter runs before JSON decoding. {"access\u005ftoken":"SECRET"} remains unchanged, as do arrays containing credential objects. Parse {/[ candidates first, then inspect and sanitize decoded keys recursively.

5. [P2] Existing processor drop counters disappear during shutdown. get_drop_stats() reads active processors, but shutdown() clears every loop state. Repro: {"queue_full": 2} became {}. Fold processor totals into persistent plugin counters before clearing states.

6. [P2] Legacy pickles lack the new runtime fields. __setstate__() backfills only _init_pid. Pre-change state makes get_drop_stats() raise AttributeError for _local_drop_counts. Backfill all three new fields and reset retry-only state during pickling.

7. [P2] Runtime validation accepts non-finite and wrong-type values. Ordered comparisons in _validate_runtime_config() do not reject NaN. max_retries=float("nan") skips the write loop entirely, silently losing batches. Enforce integral count fields and finite real-valued durations/multipliers.

8. [P1/CI] The current test file is red. test_views_not_created_after_table_creation_failure still expects the now-intentional exception to be swallowed. Focused result: 7 passed, 1 failed. Update it to expect the propagated error while still asserting that views were not attempted.

Verification:

• pyink, isort, and git diff --check: clean.
• mypy: independently confirmed 64 errors on base and 64 on head.
• New/rewritten hardening tests pass.
• Independent adversarial and structured reviews confirmed the retry, JSON-redaction, pickle, GCS, formatter-log, and non-finite-config findings.

The cold-start latency deferral is reasonable and accurately disclosed. It is not part of this fail gate.

[caohy1988]

Pushed 75967014 addressing all eight findings from the review:

1. P1 retry readiness — _ensure_schema_exists now runs on every setup attempt until one succeeds; the cached _schema no longer gates it. The retry test asserts the second get_table RPC (call_count == failed_calls + 1).
2. P1 GCS key collisions within a request — object names now include a per-parse() uid plus a content ordinal alongside the part index (…/{span}_{uid}_c{n}_p{idx}). New regression: two single-image messages in one LlmRequest produce two unique paths.
3. P1 formatter log leak — the warning logs only the exception class (no message, no exc_info). New test: a formatter raising ValueError(f"…{content}") leaves no payload in caplog.text.
4. P1 blob redaction bypass — _sanitize_json_blob decodes first for {/[ prefixes and inspects decoded keys recursively; \u005f escapes and credential arrays are covered, and strings needing no redaction return byte-for-byte (no cosmetic re-serialization). New escape/array regression test.
5. P2 shutdown counter loss — shutdown() folds processor drop stats into the persistent plugin counters before clearing loop state; {"queue_full": 2} now survives shutdown (tested).
6. P2 legacy pickles — __setstate__ backfills _local_drop_counts/_setup_failures/_setup_retry_at; __getstate__ resets retry-only state. Tested by round-tripping a state dict with the new keys removed.
7. P2 non-finite/wrong-type config — _validate_runtime_config now enforces integral counts (bools rejected) and finite reals via math.isfinite; NaN/inf/str/float-typed values all raise at construction (tested, including max_retries=NaN).
8. CI red test — test_views_not_created_after_table_creation_failure updated to expect the intentional raise while still asserting views are skipped; the parser-reuse test now pins the no-mutation contract.

One additional fix the final sanitizer pass surfaced: _recursive_smart_truncate gained a recursion depth cap (_MAX_SANITIZE_DEPTH) — id()-based cycle detection can't catch object graphs that manufacture new objects per duck-typed access (model_dump() returning fresh wrappers, Mock-like), which previously recursed unboundedly.

Verification at 75967014: full BQAA suite 324 passed (~23s, faulthandler-guarded); mypy vs 9d306f5d baseline: 0 new errors (64→64); pyink/isort/git diff --check clean.

[caohy1988]

Fresh review at exact head 75967014 against base 9d306f5d.

Verdict: FAIL. Keep this draft. The eight findings from the previous review are addressed and their regressions are green, but this fresh pass found three P1 and five P2 gaps.

1. [P1] Duplicate JSON keys bypass credential redaction. _sanitize_json_blob() infers safety from sanitized == parsed, but json.loads() discards earlier duplicate members. Repro: {"access_token":"SECRET","access_token":"[REDACTED]"} was returned unchanged with SECRET still present. Track whether any sensitive member was encountered, or preserve duplicate pairs while parsing, and always reserialize that blob.

2. [P1] Schema-upgrade failures still mark setup ready. _maybe_upgrade_schema() swallows update_table() errors, after which _ensure_started() sets _started=True. If the remote table is missing required fields, every later row can fail with a non-retryable schema mismatch and readiness is never retried. The existing test explicitly pins “logged, not raised”. Propagate upgrade failures whenever fields are actually missing.

3. [P1] Non-dict mappings bypass the mandatory sanitizer. The walker recognizes only concrete dict containers, then stringifies unknown objects. Repros with MappingProxyType({"access_token": "SECRET-PROXY"}) and UserDict({"refresh_token": "SECRET-USERDICT"}) both retained the secret in output. Handle collections.abc.Mapping recursively and emit a plain sanitized dict.

4. [P2] Shutdown timeouts lose rows without a drop count. BatchProcessor.shutdown() cancels the worker without counting its in-flight batch or remaining queued rows; plugin shutdown then persists only the unchanged processor counters. In a two-row hung-writer repro, both rows were unwritten while every reported counter remained zero. Count a shutdown_timeout/shutdown-race reason for the in-flight batch and all non-sentinel queued rows before cancellation/cleanup.

5. [P2] Deep JSON defeats the sanitizer depth cap. json.loads() executes before the bounded traversal, while only TypeError and ValueError are caught. A 10,000-level nested array of roughly 20 KB raised RecursionError, despite fitting under the default 500 KB content limit; the callback wrapper then drops the entire analytics row. Bound/preflight JSON nesting or fail closed to a sentinel on parser resource errors, and add a regression test.

6. [P2] Closed-loop cleanup still erases drop statistics. _cleanup_stale_loop_states() deletes processor state directly. Repro: a closed-loop processor reporting {"write_failed": 7} became {} immediately after cleanup, so later shutdown cannot recover it. Fold processor counters into _local_drop_counts before deletion, just as shutdown now does.

7. [P2] Legacy pickles bypass the new runtime validation. __setstate__() restores configuration without calling _validate_runtime_config(). A pre-change state containing retry_config.max_retries = NaN restores successfully; the write-loop comparison is then false immediately, so every batch is skipped without a write or drop count. Validate restored configuration after backfilling compatibility fields.

8. [P2] GCS collision protection has only 32 random bits. uuid.uuid4().hex[:8] reaches about 50% birthday-collision probability around 77,000 parses sharing a date/trace/span prefix. Uploads overwrite without a generation precondition, so a collision makes an older BigQuery row resolve to another event’s bytes. Use the full UUID and preferably if_generation_match=0.

Scope note: the cold-start latency deferral remains reasonable and is not part of this fail gate. However, the PR body says Fixes #6356; merging into main will automatically close #6356, contradicting the statement that the deferred item remains tracked there. Change this to Refs #6356 or create and link a dedicated follow-up issue before merge.

Verification at the unchanged head:

• Full BQAA test file: 318 passed, 6 skipped in 26.86s, so 324 tests were collected; the exact result is not 324 passed.
• All prior review regressions are green.
• mypy: 64 errors on base, 64 on head, zero new.
• pyink --check, isort --check-only, and git diff --check: clean.
• GitHub currently shows only check-changes/CLA/header results, not the Python suite.
• Independent adversarial and structured reviews both confirmed the duplicate-key privacy bypass; the adversarial pass also independently confirmed schema-upgrade readiness, shutdown loss accounting, deep-JSON failure, closed-loop counter loss, legacy-pickle validation, and truncated-UUID collision risk.

[caohy1988]

Pushed 9b799403 addressing all eight round-2 findings from the review:

1. P1 duplicate-key bypass — _sanitize_json_blob parses with an object_pairs_hook that flags duplicate members; any blob containing duplicates is always reserialized from the sanitized (last-wins) structure, so {"access_token":"SECRET","access_token":"[REDACTED]"} can no longer ride through the sanitized == parsed shortcut. Regression test asserts the secret is gone.
2. P1 schema-upgrade readiness — _maybe_upgrade_schema re-raises update_table failures (fields are verifiably missing at that point), so _ensure_started keeps _started=False and the backoff/retry path re-runs readiness. Both pinned "logged, not raised" tests updated to the new contract.
3. P1 Mapping bypass — the walker now recurses collections.abc.Mapping (covering MappingProxyType, UserDict, …) and emits a plain sanitized dict; regression test pins both repros from the review.
4. P2 shutdown-timeout loss — new shutdown_timeout drop reason: the cancelled worker counts its in-flight batch, and shutdown() drains and counts remaining non-sentinel queued rows. The hung-writer repro is now a test (stats["shutdown_timeout"] == 2 for one in-flight + one queued row).
5. P2 deep-JSON resource errors — RecursionError/MemoryError during blob parsing fail closed to [UNPARSEABLE_JSON_BLOB] (the row keeps flowing); the 10k-nesting repro is the regression test.
6. P2 closed-loop counter loss — _cleanup_stale_loop_states folds the dead processor's counters into the persistent plugin counters before deletion, mirroring shutdown(); tested with the {"write_failed": 7} repro.
7. P2 legacy-pickle validation — __setstate__ runs _validate_runtime_config after backfilling; a restored max_retries=NaN now raises at unpickle instead of silently skipping the write loop. Tested.
8. P2 GCS collision margin — object names carry the full 128-bit uuid, and _upload_sync passes if_generation_match=0 so the (now astronomically unlikely) collision fails the upload — surfaced as [UPLOAD FAILED] — rather than rebinding an existing row to another event's bytes. Tested on both counts.

Scope note taken: PR body now says Refs #6356 instead of Fixes, so merging won't auto-close the issue while the cold-start latency P2 is still tracked there.

On the verification-count discrepancy you flagged: agreed — my "324 passed" was this env (no optional-dep skips); your 318+6 skipped is the same 324 collected. At 9b799403 this env reports 331 passed (7 new round-2 regressions), mypy 64→64 (0 new; the initially-untyped _pairs_hook was caught and annotated), pyink/isort/git diff --check clean.

[caohy1988]

Fresh review, round 3 (9b799403)

Verdict: 2 P1s, 3 P2s, 1 P3. All behavioral findings below were reproduced against the exact PR head.

P1 — JSON sanitizer still fails open on ValueError

_sanitize_json_blob() returns the original string for every ValueError. Python raises one when parsing integers over its digit limit, even though the input is syntactically valid JSON.

Repro: {"access_token":"SECRET","n":<5000 digits>} returns byte-for-byte unchanged and still contains SECRET. An inspection failure at this privacy boundary needs to produce a sentinel or otherwise guarantee redaction.

P1 — A label-only update failure disables a write-compatible table

_maybe_upgrade_schema() re-raises update_table failures even when new_fields and updated_records are both empty and only the governance label is stale.

Reproduced with an identical table schema and a forbidden label update: setup enters backoff and subsequent events become setup_unavailable, although writes could succeed. Only failures required to install missing schema fields should block readiness; label-only failures should be handled separately.

P2 — Setup coalescing is unsafe across supported event loops/threads

_ensure_started() shares one asyncio.Lock across all loops. Such locks are loop-bound and cannot safely wake waiters belonging to another thread's loop.

A deterministic two-loop startup produced RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one in one caller while the other remained stuck. Setup needs a cross-thread-safe coalescing mechanism or appropriately isolated per-loop initialization.

P2 — Concurrent stale-loop cleanup double-counts and raises

_cleanup_stale_loop_states() snapshots stale keys, then separately reads, folds, and deletes them without synchronization.

Two concurrent cleanups of one processor reporting seven drops produced {"write_failed": 14} and a KeyError. Claim ownership atomically before folding counters.

P2 — BatchProcessor.close() still loses queued rows silently

close() cancels the worker but neither drains nor counts remaining queued rows.

With one in-flight and one queued row, close() reported one shutdown_timeout drop while leaving queue size one. It should share the corrected shutdown() drain/accounting path.

P3 — Duplicate unreachable cancellation handler

_batch_writer() has two consecutive except asyncio.CancelledError blocks. The second is unreachable and should be removed or consolidated.

Verification

• 325 passed, 6 skipped for the complete BQAA test module.
• mypy baseline unchanged: 64 -> 64.
• pyink, isort, and git diff --check clean.
• Current upstream main does not modify either changed file.
• The duplicate-key bypass, escaped-key/array redaction, schema-field upgrade failure, Mapping handling, shutdown accounting, stale-state persistence, pickle validation, and full-UUID/create-only GCS changes were specifically rechecked and otherwise held up.

[caohy1988]

Pushed 5c5521d1 addressing all six round-3 findings from the review:

1. P1 blob ValueError fail-open — json.JSONDecodeError (genuinely not JSON) still passes through to normal string handling, but every other ValueError from json.loads — your big-int repro raises one on syntactically valid JSON — now fails closed to [UNPARSEABLE_JSON_BLOB], joining RecursionError/MemoryError. The 5000-digit repro is the regression test.
2. P1 label-only readiness block — _maybe_upgrade_schema re-raises only when new_fields or updated_records is non-empty; a label-only update_table failure logs and continues (table is write-compatible; the un-stamped label retries next run). Tested with an identical-schema table and a forbidden label update.
3. P2 cross-loop setup lock — _ensure_started now takes a per-event-loop asyncio.Lock from a threading.Lock-guarded dict; each loop coalesces independently and _lazy_setup re-checks _started inside, so a rare cross-loop overlap costs at most one duplicate idempotent RPC. Regression: two threads with fresh loops start concurrently with zero errors. Pickle/fork/reset paths updated for the new fields.
4. P2 cleanup race — _cleanup_stale_loop_states claims each state atomically with dict.pop before folding, making the fold exactly-once (no double count, no KeyError).
5. P2 close() loss — close() shares shutdown()'s drain/accounting; the one-in-flight + one-queued repro now reports shutdown_timeout == 2 with an empty queue (tested).
6. P3 duplicate handler — consolidated to a single CancelledError handler. One subtlety your finding surfaced: the kept handler must re-raise, not swallow-and-break — asyncio.wait_for treats suppressed cancellation as normal completion, so the swallow variant silently disabled shutdown()'s timeout-drain branch. The round-2 regression test caught this immediately when I first consolidated the wrong way.

Verification at 5c5521d1: 336 passed in this env (6 new round-3 regressions; your env should read 330 passed + 6 skipped), mypy 64→64 (0 new), pyink/isort/git diff --check clean.

[caohy1988]

Fresh review, round 4 (5c5521d1)

Verdict: 3 P1s, 5 P2s. All behavioral findings below were reproduced against the exact PR head. The six round-3 repros themselves now pass.

P1 — Malformed JSON-shaped credentials still leak

JSONDecodeError is passed through unchanged. Appending one character to valid credential JSON bypasses redaction, including with an escaped key:

{"access\u005ftoken":"SECRET"} trailing

Recommended fix: treat every container-shaped ({ / [) parse failure as unverified and replace it with [UNPARSEABLE_JSON_BLOB]. A raw sensitive-substring fallback is insufficient because JSON escapes recreate the same bypass. Add malformed-object, trailing-garbage, and escaped-key regressions.

P1 — JSON is materialized before applying max_content_length

json.loads() runs synchronously before the configured content limit is enforced. A user-controlled multi-megabyte attribute can block the callback loop and allocate far beyond the limit.

Recommended fix: pass the configured limit into _sanitize_json_blob and reject oversized container strings before json.loads. Fail closed to a fixed sentinel rather than truncating the raw JSON prefix, which could retain a secret and produce invalid JSON. Add a test that patches json.loads and proves it is never called for an over-limit blob.

P1 — Per-loop locks allow unsafe concurrent shared setup

_ensure_started() gives each loop a different lock, so two loops both enter _lazy_setup. Reproduced setup_calls == 2; a controlled success/failure race left _started=True and _startup_error=RuntimeError simultaneously.

_lazy_setup mutates shared clients, executor, parser, schema, views, and retry state across awaits, so the duplicate call is not idempotent.

Recommended fix: split setup into (1) process-wide shared initialization and (2) loop-local writer creation. Coalesce shared initialization with a concurrent.futures.Future or equivalent cross-loop future guarded by a short-held threading.Lock, then await it from each loop with asyncio.wrap_future. Do not hold a threading lock across an await. Keep loop-local _get_loop_state() separate. The regression test should assert setup_calls == 1, every thread terminated, and shared readiness invariants remain consistent when one participant fails.

P2 — Namedtuple attributes now drop the entire event

type(obj)(new_list) cannot reconstruct tuple subclasses whose constructors require positional fields. A namedtuple in custom_tags serialized as an array before the final pass; it now raises TypeError, and the safe callback drops the row.

Recommended fix: normalize tuple subclasses to plain lists. If preserving plain tuples matters internally, use tuple(new_list) only when type(obj) is tuple; JSON does not preserve tuple identity anyway. Add a row-level namedtuple regression.

P2 — _setup_locks permanently retains closed loops

_get_setup_lock() stores loops as strong keys. Neither stale cleanup nor normal shutdown removes them. Four fresh-loop start/shutdown cycles retained four closed loops.

Recommended fix: the shared-setup redesign above should remove the per-loop setup-lock map entirely. If the map remains, explicitly claim and remove closed-loop locks under _setup_locks_guard, and clear the map during shutdown. Do not rely only on weak keys if the lock value can itself retain its loop. Add a repeated asyncio.run() regression asserting no closed loops remain reachable.

P2 — Stale-state cleanup still races concurrent dictionary mutation

_cleanup_stale_loop_states() iterates the shared dictionary without synchronization before reaching the atomic pop. Inserting another loop during is_closed() reproduced RuntimeError: dictionary changed size during iteration.

The new "concurrent" regression invokes cleanup twice sequentially, so it does not exercise the race.

Recommended fix: guard _loop_state_by_loop ownership changes with a threading lock. Snapshot keys under the lock, evaluate closed state outside it, then reacquire the lock to pop each state exactly once. Fold counters after claiming, with synchronized updates to _local_drop_counts. Add the actual two-thread mutation repro and assert no exception or double fold.

P2 — Depth-based replacement reports is_truncated=False

[MAX_DEPTH_EXCEEDED] returns False. A finite 50-level mapping discards its remaining payload but the emitted row says it was not truncated.

Recommended fix: return True when inserting the sentinel, or expose a separate observable replacement/loss reason and set the row flag consistently. Add a row-level test, not only a direct helper assertion.

P2 — Validation rejects previously supported zero-delay retries

initial_delay and max_delay must now be strictly positive. Existing BQAA tests explicitly use max_retries=0, initial_delay=0, max_delay=0, and the write loop supports it. The same RetryConfig works with BatchProcessor but raises during plugin construction.

Recommended fix: validate retry delays as finite and non-negative. At minimum, allow zero delays when max_retries == 0; preserving zero-delay immediate retries is also backward-compatible because asyncio.sleep(0) is valid. Pin construction and write behavior with the existing zero-delay configuration.

Verification

• 330 passed, 6 skipped for the complete BQAA test module.
• mypy: no new errors, 64 at the pinned base versus 61 at the head in this fresh environment.
• pyink, isort, and git diff --check clean.
• Current upstream main does not modify either changed file.

The PR body should also be refreshed: it still describes a removed "cheap hint check" and claims one setup RPC per backoff window, which the per-loop-lock implementation does not provide.

[caohy1988]

Pushed 100e17ed addressing all eight round-4 findings from the review:

1. P1 malformed-blob leak — every container-shaped parse failure now fails closed to [UNPARSEABLE_JSON_BLOB] (the JSONDecodeError pass-through is gone; as you noted, a raw-substring fallback is escape-bypassable so there is none). Regressions: malformed object, trailing garbage on valid credential JSON, and the escaped-key variant.
2. P1 pre-parse size gate — _sanitize_json_blob takes max_content_length and rejects over-limit container strings before json.loads, failing closed to the sentinel (never a raw-prefix truncation). The regression patches json.loads with an AssertionError side effect to prove it is not called.
3. P1 cross-loop shared setup — redesigned per your recommendation: shared init is coalesced with a concurrent.futures.Future claimed under a briefly-held threading.Lock (never across an await); waiters asyncio.wrap_future it from their own loops; loop-local writer state stays in _get_loop_state(). Regressions: two threads/loops → setup_calls == 1, all threads terminate, _started/_setup_future consistent; a failing owner leaves _started=False + recorded error + cleared future for the next backoff window.
4. P2 namedtuples — tuple/list subclasses normalize to plain lists (type(obj)(new_list) only for exact tuple/list). Row-level regression: a namedtuple in extra_attributes lands as [1, 2] instead of dropping the row.
5. P2 _setup_locks retention — the map is gone entirely (absorbed by we need typescript version #3); regression runs four fresh-loop startups and asserts no per-loop structures remain (not hasattr(plugin, "_setup_locks"), _setup_future is None).
6. P2 cleanup iteration race — _loop_state_by_loop ownership changes are guarded by a threading.Lock; cleanup snapshots keys under the guard, evaluates is_closed() outside it, then claims each state atomically. Regression uses your repro shape: is_closed() inserting a new entry mid-scan — no RuntimeError, fold exactly once.
7. P2 depth-cap flag — [MAX_DEPTH_EXCEEDED] now reports is_truncated=True (it discards real payload, unlike [CIRCULAR_REFERENCE]), with a row-level test (60-deep real mapping → row flag set). To keep Mock-like graphs from falsely tripping it, duck-typed conversions (model_dump/dict/to_dict) now require progress to an actual dict/list before recursing — non-progressing objects settle at the stringify fallback.
8. P2 zero-delay retries — delays validated as finite and non-negative; the long-supported max_retries=0, initial_delay=0, max_delay=0 config constructs again (pinned).

PR body refreshed as requested: the removed "cheap hint check" description is replaced with the decode-first/fail-closed contract, and the coalescing claim now describes the cross-loop future (and is accurate again).

Verification at 100e17ed: 345 passed in this env (9 new round-4 regressions; expect 339 + 6 skipped in yours), mypy 0 new errors vs base, pyink/isort/git diff --check clean.