refactor(core): decompose SessionBridge and harden transport/persistence

[teng-lin]

Summary

• decompose SessionBridge composition into dedicated plane composers under src/core/session-bridge/
• add focused composer unit tests and keep existing integration coverage green
• extract CLI early-message buffering into reusable BufferedWebSocket and migrate CliGateway
• add optional storage flush() contract and wire flush-on-bridge-close for shutdown durability
• update architecture/development docs to reflect new composition, transport buffering, and storage semantics

Validation

• pnpm typecheck
• pnpm vitest run src/adapters/file-storage.test.ts src/core/session-bridge.integration.test.ts src/core/session-bridge.adapter-consumer.integration.test.ts src/core/session/buffered-websocket.test.ts src/core/session/cli-gateway.test.ts src/core/session/session-transport-hub.test.ts src/core/session-bridge/compose-*.test.ts src/core/session-coordinator.wiring.test.ts

[gemini-code-assist]

Summary of Changes

Hello @teng-lin, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces a significant refactoring of the core SessionBridge component, breaking it down into more manageable and testable units. This decomposition enhances the system's architecture by clearly separating concerns into distinct planes for runtime, consumer interactions, message processing, and backend communication. Additionally, it improves the robustness of CLI interactions by buffering early messages and strengthens data persistence guarantees through an explicit storage flush mechanism, ensuring critical data is saved during application shutdown.

Highlights

• SessionBridge Decomposition: The monolithic SessionBridge has been refactored into dedicated, focused composition modules (compose-runtime-plane, compose-consumer-plane, compose-message-plane, compose-backend-plane) located in a new src/core/session-bridge/ directory, significantly improving modularity and maintainability.
• CLI Message Buffering: Early message buffering for CLI WebSocket connections has been extracted into a reusable BufferedWebSocket class, which ensures that initial messages are captured and replayed exactly once when the adapter's message handler is registered. The CliGateway has been updated to utilize this new buffering mechanism.
• Storage Durability Enhancement: An optional flush() contract has been added to the SessionStorage interface. This allows storage implementations, such as FileStorage, to immediately persist any pending debounced writes, enhancing data durability during critical events like SessionBridge shutdown.
• Architectural Documentation Updates: The DEVELOPMENT.md and docs/architecture.md files have been updated to reflect the new SessionBridge composition model, the introduction of BufferedWebSocket, and the enhanced storage semantics, providing clearer guidance for developers.

Changelog

• DEVELOPMENT.md
  • Updated architecture diagram link and SessionBridge description to reflect new composition.
  • Added session-bridge/ to core logic subdirectories list.
  • Updated FileStorage and SessionStorage descriptions to include flush() for shutdown durability.
  • Added session-bridge/ directory and its composition modules to the directory structure.
  • Added buffered-websocket.ts to session/ directory in the directory structure.
• docs/architecture.md
  • Added SessionBridge to Core Modules list and updated its line count in the diagram.
  • Expanded SessionBridge documentation with details on its composition model, responsibilities, and API.
  • Added a section detailing the inverted connection path for CLI, including BufferedWebSocket and CliGateway.
  • Updated SessionRepository description to clarify its role in owning live sessions and delegating persistence.
  • Modified SessionRepository responsibilities and persisted structure to align with new live session management.
  • Updated FileStorage and SessionStorage interface descriptions to include the new flush() method.
• src/adapters/file-storage.test.ts
  • Added test cases for the new flush() method, verifying immediate and coalesced persistence of debounced writes.
• src/adapters/file-storage.ts
  • Introduced pendingSaves map to track sessions awaiting debounced writes.
  • Implemented the flush() method to synchronously persist all pending debounced saves.
  • Modified save() to use pendingSaves for coalescing writes.
  • Added pendingSaves.delete() in remove() to clear pending saves for removed sessions.
• src/adapters/memory-storage.ts
  • Added a no-operation flush() method to conform to the updated SessionStorage interface.
• src/core/session-bridge.integration.test.ts
  • Added an integration test to confirm that SessionBridge.close() invokes the flush() method on the storage if available.
• src/core/session-bridge.ts
  • Removed direct imports and instantiations of numerous internal components, delegating their creation to new composition modules.
  • Refactored the constructor to utilize composeRuntimePlane, composeConsumerPlane, composeMessagePlane, and composeBackendPlane for building the bridge's functionality.
  • Added logic in the close() method to call storage.flush() for shutdown durability.
• src/core/session-bridge/compose-backend-plane.test.ts
  • Added unit tests to verify the correct creation and behavior of backend connector and API components.
• src/core/session-bridge/compose-backend-plane.ts
  • Added new file defining the composeBackendPlane function, responsible for composing backend-related services.
• src/core/session-bridge/compose-consumer-plane.test.ts
  • Added unit tests to verify the correct creation and routing behavior of consumer gateway components.
• src/core/session-bridge/compose-consumer-plane.ts
  • Added new file defining the composeConsumerPlane function, responsible for composing consumer-related services.
• src/core/session-bridge/compose-message-plane.test.ts
  • Added unit tests to verify the correct creation and message forwarding behavior of message processing services.
• src/core/session-bridge/compose-message-plane.ts
  • Added new file defining the composeMessagePlane function, responsible for composing message-related services.
• src/core/session-bridge/compose-runtime-plane.test.ts
  • Added unit tests to verify the correct creation and lazy resolution of runtime services.
• src/core/session-bridge/compose-runtime-plane.ts
  • Added new file defining the composeRuntimePlane function, responsible for composing core runtime infrastructure.
• src/core/session-bridge/types.ts
  • Added new file defining shared types for SessionBridge composition, including initialization options and accessor types.
• src/core/session/buffered-websocket.test.ts
  • Added unit tests for BufferedWebSocket, covering message buffering, replay logic, and proxying of WebSocket methods.
• src/core/session/buffered-websocket.ts
  • Added new file defining the BufferedWebSocket class for buffering and replaying early WebSocket messages.
• src/core/session/cli-gateway.ts
  • Integrated the new BufferedWebSocket class to handle early message buffering for CLI connections, replacing custom buffering logic.
• src/interfaces/storage.ts
  • Added an optional flush?() method to the SessionStorage interface to support explicit persistence of pending writes.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

The refactoring of SessionBridge into dedicated plane composers significantly improves the modularity and testability of the core logic. The introduction of BufferedWebSocket provides a robust solution for handling early CLI messages, and the addition of the flush() contract to SessionStorage ensures data durability during shutdown. The documentation updates accurately reflect these architectural changes. The suggested change for BufferedWebSocket is valid and addresses potential race conditions.

[gemini-code-assist]

To prevent any potential race conditions or duplicate message delivery if the underlying socket emits synchronously during listener registration, it is safer to disable buffering and capture the current buffer before registering the new handler on the socket. This ensures that the order of messages is strictly preserved and no message is both replayed and delivered via the live listener.

Suggested change

	socket.on("message", handler as (data: string | Buffer) => void);
	if (!this.replayed && !this.closedOrErrored) {
	this.replayed = true;
	for (const message of this.bufferedMessages) {
	handler(message);
	}
	this.bufferedMessages.length = 0;
	this.buffering = false;
	}
	if (!this.replayed && !this.closedOrErrored) {
	this.replayed = true;
	this.buffering = false;
	const toReplay = [...this.bufferedMessages];
	this.bufferedMessages.length = 0;
	socket.on("message", handler as (data: string | Buffer) => void);
	for (const message of toReplay) handler(message);
	} else {
	socket.on("message", handler as (data: string | Buffer) => void);
	}