Refactoring and cleanup.#1
Merged
Merged
Conversation
…e A2A interface - Remove AMQPStreamTaskStore and AMQPEventBus classes - Make QueuingRequestHandler self-contained with event publishing and task projection - Add WorkerEventBus for worker processes to publish events - Simplify AMQPAgentBackend to only manage connection and work queue - Fix naming inconsistency: rename QueueingRequestHandler to QueuingRequestHandler - Change QueuingRequestHandler to use private EventEmitter instead of extending it - Add proper queue cleanup in integration test teardown - Fix Vitest 4 compatibility issues (vi.mocked, timeout syntax) All 65 unit tests and 3 integration tests passing. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
- Convert QueuedTaskMessage from interface to Zod schema - Export QueuedTaskMessageSchema for runtime validation - Add zod as a dependency - Add comprehensive unit tests for schema validation - Use z.custom() for Message and Task types from @a2a-js/sdk - Maintain backward compatibility by exporting inferred type The schema enables runtime validation of work queue messages, providing better error handling and type safety. All 73 unit tests and 3 integration tests passing. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
… tests Changes to AMQPWorkQueue: - Replace callback-based consumer with async generator pattern - Make queueName required in constructor (no default) - Remove startConsumer/stopConsumer methods in favor of start() generator - Remove enqueue routingKey parameter (use instance property) - Remove close() method (cleanup handled by connection) - Store routingKey as instance property Test updates: - Update all unit tests to match new async generator API - Fix integration test to use generator pattern for worker - Add proper test isolation by running unit and integration tests separately - Improve integration test cleanup order (close backends first) - Remove queue deletion from cleanup (tests use unique timestamped names) - Add afterEach cleanup to QueuingRequestHandler unit tests Other improvements: - Update worker example to use async generator pattern - Fix QueuingRequestHandler projection consumer to use async iteration - Remove unused WorkerEventBus getter methods - Update package.json test script to run test:unit && test:integration All tests passing: 68 unit tests + 3 integration tests ✅ 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
Critical fixes: - Fix worker shutdown to close async generator before closing backend - Fix server shutdown to close requestHandler (projection consumer) first - Prevents "Consumer cancelled by the server" errors during shutdown Code cleanup: - Remove unnecessary workQueue null check in worker (always initialized) - Add missing semicolon for consistency Documentation: - Update README to match current simplified API - Remove references to old callback-based consumer API - Remove references to removed classes (EventBusManager, TaskStore) - Update worker example to show async generator pattern - Update configuration docs to reflect actual options - Add graceful shutdown details All examples and documentation now match the actual codebase API. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
Problem: Integration tests were leaving queues behind in LavinMQ broker
Solution: Add proper cleanup after closing all connections:
1. Close all backends and consumers first
2. Create fresh AMQP connection for cleanup
3. Delete both test queues (event stream + work queue)
4. Close cleanup connection
Each test run creates uniquely named queues (timestamp-based), so they
were accumulating in the broker without explicit deletion.
Now properly cleans up:
- Event stream queue: "Test Agent.events"
- Work queue: "test-agent-{timestamp}.work-queue"
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
Before: 5 specific routing patterns - a2a.*.*.task - a2a.*.*.message - a2a.*.*.status - a2a.*.*.artifact - a2a.*.*.event After: Single catch-all pattern - a2a.# Benefits: - Simpler code (1 binding instead of 5) - Less broker overhead - Easier to extend (new event types need no code changes) - Functionally equivalent (QueuingRequestHandler consumes all event types) The fine-grained patterns only add value if different consumers need different event types. Since we have one consumer that needs everything, the catch-all pattern is cleaner. Publisher confirms remain: - Default: enabled (confirmMode: true) - Configurable via AMQPAgentBackendConfig.publishing.confirmMode - Provides guaranteed delivery at cost of latency 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
- Add tests/**/* to tsconfig includes for type checking - Remove rootDir constraint to allow checking test files - Fix null safety issues in e2e tests with ! assertions - Fix optional property access in unit tests - Add proper type casts for vitest mock methods - Complete AgentCard fixture with all required properties - Add prefix type to mock logger 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Pull Request Overview
This PR refactors the codebase to remove the separate TaskStore and EventBus implementations, consolidating their functionality into a unified QueuingRequestHandler. The main changes include creating a self-contained request handler that manages in-memory task projection, AMQP event publishing, and SSE streaming, while introducing a simplified WorkerEventBus for worker processes.
Key Changes:
- Removed AMQPStreamTaskStore and AMQPEventBus classes, replacing them with QueuingRequestHandler (combines both responsibilities)
- Introduced WorkerEventBus for simplified worker-side event publishing
- Migrated AMQPWorkQueue from callback-based to async generator pattern for better control flow
- Added Zod schema validation for QueuedTaskMessage
Reviewed Changes
Copilot reviewed 25 out of 27 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| src/lib/QueuingRequestHandler.ts | New self-contained handler merging TaskStore projection and EventBus publishing |
| src/lib/WorkerEventBus.ts | New simplified event bus for workers to publish events |
| src/lib/QueuedTaskMessage.ts | Converted to Zod schema for runtime validation |
| src/lib/AMQPWorkQueue.ts | Changed from callback to async generator pattern for message consumption |
| src/lib/AMQPAgentBackend.ts | Removed taskStore and eventBusManager, simplified to connection + work queue |
| tests/unit/QueuingRequestHandler.test.ts | Renamed and updated tests for new QueuingRequestHandler |
| tests/unit/AMQPWorkQueue.test.ts | Updated tests for generator-based consumption |
| tests/integration/e2e-happy-path.test.ts | Updated to use new QueuingRequestHandler API with initialization |
| package.json | Added Zod dependency and updated AMQP client version |
| README.md | Updated documentation to reflect new simplified API |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
The REFACTORING-NOTES.md described a partial refactoring of AMQPEventBus, TaskScopedEventBus, and AMQPStreamTaskStore. These classes have since been completely removed and replaced with a simpler architecture: - WorkerEventBus for publishing events from workers - QueuingRequestHandler with built-in event projection - No task store or factory pattern needed The document is no longer relevant to the current codebase. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
…ypecheck configs - Fix "Exchange not found" error by calling getPublishChannel() before binding - Ensure exchange is declared in both QueuingRequestHandler and AMQPWorkQueue init - Create separate tsconfig.typecheck.json that includes tests for type checking - Main tsconfig.json now only compiles src/**/* to dist/ (excludes tests) - Prevents test files from being compiled and run twice This fixes the CI integration test failures caused by trying to bind queues to an undeclared exchange. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
- Replace deprecated substr() with slice() throughout codebase - QueuingRequestHandler: 3 occurrences - AMQPWorkQueue: 1 occurrence - WorkerEventBus: 1 occurrence - worker.ts example: 1 occurrence - Remove unused 'promise' variable in AMQPWorkQueue.test.ts - Update slice length from 9 to 11 for consistency Note: Kept logger 'prefix' property in mock as it's needed for TypeScript type compatibility in tests. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This pull request refactors the AMQP backend and worker/HTTP server examples to simplify the architecture, improve event handling, and enhance developer ergonomics. The main changes include replacing the previous event bus manager and task store abstraction with a more streamlined backend focused on connection and work queue management, updating examples to use new handler and event bus classes, and improving graceful shutdown and configuration options.