Merge upstream #1746: Use lazy stream accessors for provider runtime events

[aaditagrawal]

Summary

Cherry-picks upstream commit 740d7a32 — "Use lazy stream accessors for provider runtime events (pingdotgg#1746)".

Changes

• Converts streamEvents from eager Stream.fromQueue() to a lazy getter on provider adapters (Codex + Claude), so the queue is only consumed when read
• Splits RuntimeReceiptBus into RuntimeReceiptBusLive (no-op) and RuntimeReceiptBusTest (PubSub-backed) for cleaner test isolation
• Adds SynchronizedRef concurrency safety to EventNdjsonLogger
• Updates integration tests to use it.live and lazy stream accessors

Conflict Resolution

1 conflict in ClaudeAdapter.ts — fork had restructured function positions in a prior refactor. Resolved by keeping fork structure and applying only the upstream lazy getter change.

Verification

• bun typecheck ✅
• bun fmt ✅
• bun lint ✅
• Targeted unit tests (70 passed, 3 skipped) ✅

Fork Preservation

All fork-specific multi-provider logic, guards, and imports verified intact.

Summary by CodeRabbit

Release Notes

• Bug Fixes

  • Enhanced concurrency safety in event logging operations to prevent data inconsistencies.

• Tests

  • Improved integration test framework for receipt event handling and validation.

• Refactor

  • Optimized event processing pipeline and stream handling across the orchestration layer for improved efficiency.

[coderabbitai]

Warning

Rate limit exceeded

@aaditagrawal has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 17 minutes and 34 seconds before requesting another review.

Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 17 minutes and 34 seconds.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: bc88908e-4df7-4fb4-afe1-f12e426d3ba8

📥 Commits

Reviewing files that changed from the base of the PR and between 2bb1082 and 0e9e240.

📒 Files selected for processing (2)

• apps/server/integration/providerService.integration.test.ts
• apps/server/src/provider/Layers/EventNdjsonLogger.ts

📝 Walkthrough

Walkthrough

This pull request refactors event streaming patterns across the orchestration system by converting eager stream property initialization to lazy getter-based evaluation, introduces a dedicated test implementation for the RuntimeReceiptBus with a renamed API, and restructures event processing in ProviderService from a queue-based worker to direct stream pipeline handling. State management in EventNdjsonLogger is also refactored to use synchronized references.

Changes

Cohort / File(s)	Summary
Stream Property Lazy Evaluation apps/server/src/orchestration/Layers/CheckpointReactor.test.ts, ProviderCommandReactor.test.ts, ProviderRuntimeIngestion.test.ts, apps/server/src/provider/Layers/ClaudeAdapter.ts, CodexAdapter.ts, ProviderRegistry.test.ts, ProviderService.test.ts	Converted streamEvents and streamChanges properties from eagerly-initialized values to getter functions that compute and return streams on access.
RuntimeReceiptBus Interface & Implementation apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts, apps/server/src/orchestration/Services/RuntimeReceiptBus.ts	Split receipt bus into live and test implementations; renamed stream property from stream to streamEventsForTest; live implementation now publishes no-op with empty stream; test implementation uses unbounded PubSub.
Integration Test Receipt Collection apps/server/integration/OrchestrationEngineHarness.integration.ts, providerService.integration.test.ts	Swapped test harness from RuntimeReceiptBusLive to RuntimeReceiptBusTest; added 50ms delay before turn action execution; changed test runner mode in some cases from it.effect to it.live.
Event Processing Architecture & Synchronization apps/server/src/provider/Layers/ProviderService.ts, EventNdjsonLogger.ts, EventNdjsonLogger.test.ts	Removed queue-based worker pattern in ProviderService, now directly processes events in stream pipeline; refactored EventNdjsonLogger state management from unsynchronized maps to single SynchronizedRef; added concurrent event write test case.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

Suggested labels

vouch:trusted, size:L

Poem

🐰 Streams now awake when eyes gaze near,
Not eager queues, but getters clear!
Receipts flow through the test lane bright,
Synchronized hops, events in flight!
A rabbit's refactor, sleek and tight. 🌿

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and concisely describes the main change: converting stream accessors to lazy evaluation for provider runtime events, matching the core objective of this upstream merge.
Description check	✅ Passed	The PR description provides comprehensive context: it clearly states the upstream commit being cherry-picked, details the main changes (lazy stream getters, RuntimeReceiptBus split, SynchronizedRef addition, integration test updates), documents conflict resolution, and includes verification results.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch merge/upstream-1746-lazy-stream-accessors

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

apps/server/integration/providerService.integration.test.ts (1)

89-90: Prefer an explicit collector-ready signal over the fixed 50ms delay.

Line 89 adds a wall-clock dependency that can still race on slower CI and adds latency to every turn capture. A ready handshake, or a single long-lived collector owned by the fixture, would make these tests deterministic without hard-coded timing.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@apps/server/integration/providerService.integration.test.ts` around lines 89
- 90, The test currently uses a fixed wall-clock delay (yield* Effect.sleep("50
millis")) before running yield* action, which causes flaky CI and added latency;
replace this with an explicit collector-ready handshake or a long-lived
collector owned by the test fixture: modify the test setup to start the
collector in the fixture (or expose a promise/event like collector.ready or
waitForCollectorReady()) and await that signal instead of using Effect.sleep, or
refactor the fixture to keep a single long-lived collector for all turns and
remove the sleep before calling action.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@apps/server/src/provider/Layers/EventNdjsonLogger.ts`:
- Around line 269-283: The current shutdown sequence in
SynchronizedRef.modifyEffect (operating on stateRef) clears threadWriters and
failedSegments but does not mark the logger as closed, so subsequent write calls
can recreate ThreadWriter instances and reopen files; modify the shared state
shape to include a terminal flag (e.g., closed or isShutdown) and in the close
path (where you call ThreadWriter.close on each writer) set that flag true; then
update the write/create path that accesses stateRef and threadWriters to check
the closed flag first and refuse or noop writes when closed, ensuring no new
ThreadWriter is created after shutdown and existing resources remain final.

---

Nitpick comments:
In `@apps/server/integration/providerService.integration.test.ts`:
- Around line 89-90: The test currently uses a fixed wall-clock delay (yield*
Effect.sleep("50 millis")) before running yield* action, which causes flaky CI
and added latency; replace this with an explicit collector-ready handshake or a
long-lived collector owned by the test fixture: modify the test setup to start
the collector in the fixture (or expose a promise/event like collector.ready or
waitForCollectorReady()) and await that signal instead of using Effect.sleep, or
refactor the fixture to keep a single long-lived collector for all turns and
remove the sleep before calling action.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: d0f67fe2-eed7-4f08-8f02-fbe2d3b77eb9

📥 Commits

Reviewing files that changed from the base of the PR and between 0196390 and 2bb1082.

📒 Files selected for processing (14)

• apps/server/integration/OrchestrationEngineHarness.integration.ts
• apps/server/integration/providerService.integration.test.ts
• apps/server/src/orchestration/Layers/CheckpointReactor.test.ts
• apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts
• apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts
• apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts
• apps/server/src/orchestration/Services/RuntimeReceiptBus.ts
• apps/server/src/provider/Layers/ClaudeAdapter.ts
• apps/server/src/provider/Layers/CodexAdapter.ts
• apps/server/src/provider/Layers/EventNdjsonLogger.test.ts
• apps/server/src/provider/Layers/EventNdjsonLogger.ts
• apps/server/src/provider/Layers/ProviderRegistry.test.ts
• apps/server/src/provider/Layers/ProviderService.test.ts
• apps/server/src/provider/Layers/ProviderService.ts