fix: single-source dispatch routing and bounded client subscription buffer (#104 #108)

- Dispatcher::routeLifecycle/routeWork derive "handled" from a single match
  via a $handled flag instead of duplicating the type matrix in a parallel ||
  chain, so a new message type cannot silently fall through.
- ResponseRouter caps per-subscription pending events for not-yet-registered
  subscriptions (dropping with a warning past the cap) and clears the buffer on
  unregisterSubscriber, so a racy/bogus subscription id cannot grow unbounded.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: nficano

src/Internal/Client/ResponseRouter.php

@@ -28,6 +28,13 @@
  */
 final class ResponseRouter
 {
+    /**
+     * Upper bound on events buffered for a not-yet-registered subscription,
+     * so a never-registered (racy or bogus) subscription id cannot grow the
+     * buffer without limit.
+     */
+    private const int MAX_PENDING_SUBSCRIPTION_EVENTS = 1024;
+
     /** @var array<string, \Closure(Envelope): void> */
     private array $subscribers = [];
 
@@ -52,7 +59,8 @@ public function registerSubscriber(SubscriptionId $id, \Closure $onEvent): void
 
     public function unregisterSubscriber(SubscriptionId $id): void
     {
-        unset($this->subscribers[(string) $id]);
+        $key = (string) $id;
+        unset($this->subscribers[$key], $this->pendingSubscriptionEvents[$key]);
     }
 
     public function handle(Envelope $env): void
@@ -103,6 +111,13 @@ private function routeSubscribeEvent(Envelope $env, SubscribeEvent $msg): void
             $this->invokeSubscriber($subscriber, $inner);
             return;
         }
+        if (\count($this->pendingSubscriptionEvents[$key] ?? []) >= self::MAX_PENDING_SUBSCRIPTION_EVENTS) {
+            $this->deps->logger->warning(
+                'dropping subscription event; pending buffer full for unregistered subscription',
+                ['subscription_id' => $key, 'cap' => self::MAX_PENDING_SUBSCRIPTION_EVENTS],
+            );
+            return;
+        }
         $this->pendingSubscriptionEvents[$key][] = $inner;
     }
 

src/Internal/Runtime/Dispatcher.php

@@ -143,6 +143,7 @@ private function dispatch(Session $session, Envelope $env): void
 
     private function routeLifecycle(Session $session, Envelope $env, MessageType $msg): bool
     {
+        $handled = true;
         match (true) {
             $msg instanceof Ping => $this->lifecycle->handlePing($session, $env, $msg),
             $msg instanceof Pong, $msg instanceof Ack => null,
@@ -152,16 +153,9 @@ private function routeLifecycle(Session $session, Envelope $env, MessageType $ms
             $msg instanceof Resume => $this->lifecycle->handleResume($session, $env, $msg),
             $msg instanceof LeaseRefresh
                 => $this->lifecycle->handleLeaseRefresh($session, $env, $msg),
-            default => null,
+            default => $handled = false,
         };
-        return $msg instanceof Ping
-            || $msg instanceof Pong
-            || $msg instanceof Ack
-            || $msg instanceof SessionClose
-            || $msg instanceof Cancel
-            || $msg instanceof Interrupt
-            || $msg instanceof Resume
-            || $msg instanceof LeaseRefresh;
+        return $handled;
     }
 
     private function routeWork(Session $session, Envelope $env, MessageType $msg): bool
@@ -176,6 +170,7 @@ private function routeWork(Session $session, Envelope $env, MessageType $msg): b
             $this->lifecycle->nack($session, $env, 'UNIMPLEMENTED', 'subscribe not negotiated');
             return true;
         }
+        $handled = true;
         match (true) {
             $msg instanceof ToolInvoke => $this->toolInvocation->handle($session, $env, $msg),
             $msg instanceof ListJobs => $this->jobList->handle($session, $env, $msg),
@@ -184,15 +179,9 @@ private function routeWork(Session $session, Envelope $env, MessageType $msg): b
             $msg instanceof ArtifactPut => $this->artifacts->put($session, $env, $msg),
             $msg instanceof ArtifactFetch => $this->artifacts->fetch($session, $env, $msg),
             $msg instanceof ArtifactRelease => $this->artifacts->release($session, $env, $msg),
-            default => null,
+            default => $handled = false,
         };
-        return $msg instanceof ToolInvoke
-            || $msg instanceof ListJobs
-            || $msg instanceof Subscribe
-            || $msg instanceof Unsubscribe
-            || $msg instanceof ArtifactPut
-            || $msg instanceof ArtifactFetch
-            || $msg instanceof ArtifactRelease;
+        return $handled;
     }
 
     private function featureEnabled(Session $session, string $feature): bool