perf(storage): bound ingestion and fence observation capture

[XuPeng-SH]

Summary

• Isolate ingestion transactions by authenticated owner/session, retain only unresolved retry groups, and schedule owners/sessions fairly within bounded event/byte and database-attempt budgets.
• Keep timeout/cancellation cleanup from leaking SQL pool capacity; unrelated sessions and foreground reads/writes can progress while other sessions are fenced.
• Fence observation identities with versioned canonical payload hashes and typed inserted/replayed/collision outcomes. Publish counters, causal edges, configuration side effects, manifests, snapshots, and transcript projections only under the corresponding accepted capture contract.
• Add bounded opt-in per-delivery observations and reproducible ingestion probes that distinguish offered, committed, rejected, and uncertain outcomes. Keep ordinary offline CI independent of live database credentials.

Related issue

None. This extends the existing PR through independently reviewed stages; no additional PR is needed for these commits. Explain snapshot discovery remains in #837.

Change type

• Feature
• Bug fix
• Refactor or performance improvement
• Test
• Build, CI, or maintenance

User and compatibility impact

Fresh-schema contract v83. This change does not provide migration of existing data or compatibility with the old observation schema. Deploy against a freshly provisioned schema; do not treat it as an in-place upgrade.

One blocked, malformed, deleted, or colliding session/event must not replay unrelated committed work or publish rejected derived content. Session queues are logical scheduling units, not one database connection or worker task per session. Production creates one ingestion worker per process-owned pool and shares its sender with root and delegated execution.

Architecture and complexity delta

• Canonical owners: EventIngestionWorker, durable session admission, observation capture, and existing runtime persistence/projection paths.
• Superseded behavior: whole-batch transaction/retry coupling and subset-based capture identity checks. No alternate agent lifecycle or scheduler ownership is introduced.
• Bounds cover deferred/channel/scheduled/retry/in-flight payloads; database attempts have deadlines and cancellation-safe connection cleanup. A shared limiter is available for multiple workers using one process pool.
• Process-local limits do not establish a cluster-wide ingestion budget. Database fences and identities remain authoritative across processes.

Verification

Every included implementation stage received independent GPT-6 review (Medium for the initial stages, High for the transaction-boundary reassessment); actionable findings were repaired and their affected checks rerun before commit/push.

Prior candidate (before the final review/cleanup follow-up below):

• cargo test -p astra-services --test event_ingestion_db_it -- --ignored --test-threads=1 with an explicitly configured disposable MatrixOne database: 25 entries passed in 13.31s (24 scenarios plus the internal child entrypoint).
• The suite includes finite 100-owner/1,000-session ingestion, all-slot timeouts, connection recovery, cancellation, shared-pool two-worker isolation, tenant separation, replay/collision behavior, and deletion fencing.
• A real two-child-process scenario verifies one insert plus one replay for the same frozen event, one session increment and exact parent edge, independent-session progress while fenced, and rejection of writes after canonical deletion. Earlier targeted run: 0.47s; also passed in the latest full suite.
• Targeted clippy with warnings denied, formatting, and diff checks passed. The preceding observer stage also passed services all-target clippy with capacity-probes enabled, 104 ingestion unit tests, eight offline probe tests, and seven report-gate tests.

Capture-stage evidence (semantically unchanged by subsequent instrumentation/test additions): services library 2,037 passed / 73 ignored; live observation capture 3/3, context manifest 6/6, schema checks 6/6, runtime persistence 9/9; schema inventory 47/47; services/runtime all-target clippy passed. Focused runtime collision tests verified that rejected snapshots/transcripts cannot be attributed to accepted captures.

Performance evidence and limits

Transaction-boundary reassessment (GPT-6 High)

An identical-source short probe compared the pre-PR global-transaction version (bf87d4918), independent session transactions (2094efbd1), and the revised independent-transaction concurrency budget. Each used a fresh dedicated database, pool size 8, default batch size 100 / flush interval 1s, 1,000 events with 256-byte content, and three repetitions per workload.

Workload	Old global transactions	Independent, pool/4	Independent, pool/2
1,000 sparse Sessions: median all-visible time	2.810s	6.731s	3.867s
10 dense Sessions: median all-visible time	6.479s	0.560s	0.370s

The revised budget lowers the measured sparse completion time by about 42.5% versus pool/4, but remains about 38% slower than the old global-transaction implementation in that workload. The old dense samples drifted substantially; do not generalize its apparent speedup ratio. This full-revision comparison includes capture/schema differences and initial session-fence creation, not only COMMIT cost. Foreground SELECT 1 sampling continued through completion; its noise and small dense sample counts do not establish a production SLO.

A bounded cross-session microbatch prototype was rejected, not shipped. A real held-fence experiment showed that closing a timed-out client connection did not promptly release server-side locks on an unrelated healthy member: the independent healthy retry remained blocked until the peer fence was released. Therefore the final change retains independent session transactions and only changes the automatic attempt ceiling from one quarter to one half of the pool, still capped by configuration and the two-connection reserve. The one-second flush interval is unchanged.

GPT-6 High approved this narrower tradeoff after reviewing the evidence. Final affected validation: 105 ingestion unit tests and all 25 live integration entries passed (13.31s for the live suite); services all-target clippy with capacity-probes and warnings denied passed, and repository metadata validation passed. The reproducible short probe is opt-in through capacity-probes, not added to ordinary live CI.

Earlier hot-owner smoke evidence

A completed short hot-owner probe scheduled 35,000 arrivals at 500 events/s; 34,968 were submitted and durably reconciled, with 32 generator misses (0.0914%), complete measurements, and unchanged source fingerprints. This was 10s warmup plus 60s measurement in the recorded Cargo test profile, not proof of sustained deployment capacity.

The longer run was stopped at the user's request and produced no valid capacity result. No sustained-capacity or thousands-of-active-agent claim is made. Reproduction commands, accounting gates, quantized latency interpretation, and scope limits are documented in docs/guides/testing.md. Generated evidence and private connection configuration are not committed.

Final review and cleanup follow-up

• Fixed review comment #4058103685 in b7223ddc6: freeze canonical trace time bounds on first post-loop capture so delayed terminal replay and receipt-aware commit resolution compare the same full payload identity. Content and lineage conflict checks remain strict. The live regression covers buffered and unbuffered persistence, delayed replay, the lost-ack resolution path, and rejection of changed content; it does not inject a network-level lost COMMIT acknowledgement.
• Removed duplicate ingestion timestamp/readiness/owner bookkeeping, redundant delivery wrappers, forwarding persistence layers, repeated database-test queries/polling, and the obsolete Explain two-read timing comparison in c17ad1e5c. Retained FIFO, retry priority, fairness, bounded concurrency, cancellation, isolation, and production-entrypoint Explain assertions.
• Restored CLI/TUI cache summaries in 0fd235533. Partial observations display truthful lower bounds such as ≥19.9k tokens · ≥17.9k cached; unknown cache usage stays absent. Complete observations retain the cache percentage. Removed the confusing tokens known copy and shared the formatter across both surfaces.
• Combined follow-up: 20 files, 440 additions / 629 deletions, net 189 fewer lines. Independent GPT-6 High review approved the implementation subject to the final gates, which subsequently passed.

Final targeted verification:

• Ingestion unit tests: 106 passed. Offline capacity probe tests: 8 passed, sustained probe ignored. Python report tests: 7 passed.
• Actual MatrixOne ingestion integration: 25 entries passed in 13.75s, including finite 1,000-session and two-process scenarios.
• Runtime persistence: 34 offline tests and 9 live database tests passed. Explain/recovery: 5 live tests passed.
• CLI reporting: 21 tests passed. Turn summary: 17 tests passed.
• Services/runtime/CLI all-target Clippy with capacity probes enabled and warnings denied, formatting, diff checks, and repository validation passed.
• No long-duration load test was run in this follow-up. Earlier performance figures above remain historical evidence, not a new capacity claim. Remote CI must be evaluated separately for the new head.

Latest review follow-up (d9ff3d696)

• Preserve the admission-time root timestamp independently from the later execution-buffer origin. Exact replay now keeps the original user-query envelope without shifting response, round, or tool timestamps backward.
• Abort atomic terminal settlement when its required root or assistant canonical capture is rejected. The transaction rolls back before status, terminal events, or whole-turn projections can make rejected content authoritative; nonterminal mixed-outcome capture still preserves accepted siblings.
• Serialize current-residency publication with admission mutations under the existing state -> stats lock order, preventing stale snapshots from overwriting newer releases/acquisitions.
• GPT-6 Medium initially found the dual-time-origin issue, then approved the repaired candidate with no remaining P0/P1/P2 blockers.
• Final verification: 107 ingestion unit tests passed; 34 offline persistence tests passed; all 9 actual MatrixOne persistence tests passed; runtime/services/CLI all-target Clippy with warnings denied, formatting, and diff checks passed.

Final checklist

• Tests cover public ingestion/persistence boundaries and relevant unhappy paths.
• Design/public documentation reflects contracts and limits.
• No credentials, populated environment files, private endpoints, customer data, or raw traces are included.
• Conventional Commit title; feature-branch delivery only.

[XuPeng-SH]

Deep review verdict: REQUEST CHANGES for 594902e.

One confirmed P1: exact canonical terminal recovery hashes a newly generated response timestamp, so an unchanged, successfully committed terminal is rejected when resolving a lost COMMIT acknowledgement or replaying settlement. The inline comment includes the reproduction and suggested correction.

Reviewed session scheduling/admission, retry isolation, connection cancellation, event/manifest identity classification, tenant scoping, and downstream transcript/snapshot projection fencing. Fresh-schema-only deployment was assumed; no legacy compatibility findings.

Validation performed: an isolated Rust executable using extracted production time-bound/event-building/hash/comparison functions reproduced the mismatch with the same event ID and content after 25ms, both with and without a turn buffer; 7 capacity-report tests and 47 schema-inventory tests passed; git diff --check passed. No local live MatrixOne fault-injection test was run. GitHub Static Checks passed; the overall Test Suite was still running at the final check (runtime, services, and online integration jobs had passed).

GitHub rejected the formal REQUEST_CHANGES event because the authenticated reviewer is also the PR author (HTTP 422: cannot request changes on your own pull request). This COMMENT records the same REQUEST CHANGES verdict and blocking finding; it is not an approval.

[XuPeng-SH]

[P1] Freeze trace timestamps before validating an exact terminal replay

verify_canonical_append_evidence() rebuilds expected events by calling turn_trace_time_bounds(state) again, but that function includes TurnEventBuffer::offset_ms() (the current elapsed time), or Utc::now() when there is no buffer. trace_event_payload_hash() now hashes the entire TraceEvent, including created_at. Consequently the response hash changes merely because time passed: an isolated Rust reproduction using these production functions, the same state/event ID/content, and a 25ms delay returned accepted_replay=false for llm_response; without a buffer it also rejected user_query.

This comparison is used by resolve_existing_atomic_terminal_settlement() both after a lost COMMIT acknowledgement (line 1323) and when replaying an already committed settlement (line 1139). It therefore reports canonical trace evidence ... conflicts with replay even though the durable terminal and payload are correct, causing the lifecycle to take its persistence-error path and suppress the successful live terminal publication. Freeze the original event envelope/time bounds and reuse them for persistence and authoritative resolution; add a delayed exact-replay/lost-ack regression test instead of recomputing wall-clock-derived fields.

[XuPeng-SH]

Fixed in b7223dd. Canonical trace time bounds are frozen at the first post-loop capture and reused for persistence, receipt-aware commit resolution, and terminal replay; full content/lineage hash validation is retained. The actual MatrixOne regression exercises buffered and unbuffered terminal commits, delayed receipt resolution and replay, and rejection after changing content. This covers the lost-ack resolution path, not a network-level COMMIT-ack fault injection. All 9 live runtime persistence tests passed; the final GPT-6 High review approved the implementation subject to gates, which subsequently passed.

[XuPeng-SH]

Deep review verdict: REQUEST CHANGES for 0fd2355.

Three actionable findings: two P1 correctness issues in canonical capture/terminal settlement and one P2 race in residency accounting. The inline comments contain production call chains and regression scenarios.

The timestamp fix stabilizes repeated post-loop calls, but omits the earlier persist_turn_start() capture. Ordinary root execution can therefore still classify its accepted user input as a collision and omit its transcript row. Separately, rejected response captures still allow successful terminal settlement and downstream canonical-context/state projections.

Reviewed session scheduling/admission, retry isolation, connection cancellation, event/manifest identity classification, tenant scoping, downstream projections, the timestamp follow-up, and CLI/TUI usage formatting. Fresh-schema v83 deployment was assumed; no migration/upgrade requirements were imposed.

Validation: 7 capacity-report tests and 47 schema-inventory tests passed; git diff --check passed. Source-order assertions plus a deterministic Python model of the production hash/data flow confirmed different capture hashes when only admission/post-loop timestamps change. A deterministic two-thread model confirmed reversed residency publication. These models are not executions of the Rust runtime: no Rust toolchain is available here, and no local MatrixOne integration/fault-injection tests were run. At the last check, Static Checks passed, Test Suite was in progress, and the head was unchanged.

The connected account is the PR author (XuPeng-SH), so GitHub does not permit a formal REQUEST_CHANGES event from this identity. This COMMENT records the REQUEST CHANGES verdict; it is not an approval.

[XuPeng-SH]

[P1] Preserve the admission-time user-query timestamp before freezing replay

Freezing only the first post-loop bounds is too late for root runs. Production state starts with turn_event_buffer=None (lifecycle/mod.rs:12808). Both root entrypoints call persist_turn_start() before entering the loop (14297 and 17548); that function persists user_query with Utc::now() when the buffer is absent (persistence.rs:293-297). The loop subsequently creates its buffer (agentic_loop/lifecycle.rs:3565-3571), and this cache uses that later buffer start as the same root event's created_at.

The full-envelope hash consequently classifies an unchanged user_query as Collision. The new accepts_projection filter at 2352 drops its user transcript item on normal completion; persist_turn_start only writes the trace, not the transcript. Lost-COMMIT-ack resolution also fails verify_canonical_append_evidence on this root event. Normal buffered execution is affected, not just an unbuffered edge case.

Preserve the original user-query timestamp/envelope before the first durable write and reuse it during terminal capture/resolution, while retaining the correct terminal offset. Add a regression that calls persist_turn_start with no buffer, subsequently starts the loop/buffer, and settles/replays the terminal. The new delayed-replay test starts directly at terminal persistence and misses this production ordering.

[XuPeng-SH]

Fixed in d9ff3d6. persist_turn_start now freezes a dedicated admission-time root timestamp before the first durable write. Terminal capture and replay reuse that value for user_query, while response, round, and tool offsets retain the later execution-buffer origin. The live regression executes admission without a buffer, creates the buffer after a delay, settles and replays, and asserts user_query_at < buffer_origin <= response_at. All 9 actual MatrixOne persistence tests passed.

[XuPeng-SH]

[P1] Propagate rejected canonical captures before successful terminal settlement

A rejected response now yields terminal_assistant_source_event_id=None, but this success branch still sets core_trace_result=Ok and owner_terminal_committed=true. Use the existing canonical_append_excludes_colliding_response_from_transcript_and_deferred_projection fixture (same response ID, original durable content, conflicting state.final_text) through persist_atomic_terminal_settlement instead of the nonterminal append helper: the capture filters out the assistant transcript, yet persist_server_loop_canonical_append_inner still calls settle_terminal_in_existing_transaction with the attempted text_done/run_finished batch and commits completed status, returning None for the response only afterward.

That None gates deferred transcript insertion, but does not gate commit_canonical_turn(&state.messages), live terminal publication, or run_after_core. The latter still publishes state.final_text as the final decision in persist_server_loop_projection_state. The response rejected by agent_events can thus escape as successful canonical/context/state output, and completed status becomes visible without its corresponding accepted assistant transcript. The same commit also fails strict evidence verification if its acknowledgement is lost.

Propagate required canonical capture rejection before committing a successful terminal and authorizing whole-turn projections. Preserve valid sibling observations, but do not infer that all derived state is valid merely because the observation transaction returned Ok. Extend the collision regression through atomic settlement and downstream projections for streaming and non-streaming roots.

[XuPeng-SH]

Fixed in d9ff3d6. Atomic terminal settlement now treats rejection of its required root or assistant canonical event as a transaction error and rolls back before durable status or events are committed. The existing mixed-outcome nonterminal path still keeps valid siblings. The collision regression attempts atomic completed settlement with streaming terminal events and verifies the run remains running with no added durable terminal events; both streaming and non-streaming lifecycle callers receive the same error and skip successful publication and projections. All 9 actual MatrixOne persistence tests passed.

[XuPeng-SH]

[P2] Serialize residency publication with admission mutations

The snapshot is read under self.state, but that lock is dropped before self.stats is acquired; try_acquire does the same. Concurrent releases can publish in reverse order: A changes actual residency 2→1 and pauses after drop(state); B changes 1→0 and publishes 0; A resumes and publishes its stale 1. Once both leases are gone, resident_events_current/resident_bytes_current can remain nonzero indefinitely. Acquire/release overlap can conversely report zero while work is resident.

These values drive more than dashboards: shutdown reads resident_events_current to classify unresolved deliveries, and the capacity probe/report uses current residency and resident_events_after_shutdown to assess backlog/draining. A drained worker can be reported as retaining unresolved work. Keep publication serialized with the admission mutation, or read current residency from the authoritative admission state, and cover concurrent release/admission. Actual admission limits still work; the exported current accounting is unreliable.

[XuPeng-SH]

Fixed in d9ff3d6. Admission mutation and current-residency publication now remain serialized under the existing state-to-stats lock order for both acquire and release, so an older snapshot cannot overwrite a newer one. A concurrent 32-lease release regression verifies final current event and byte residency is zero. 107 ingestion unit tests and all-target Clippy passed.