[Bug]: a subscriber that stops reading wedges the event broker, deadlocks queue creation process-wide, and leaks the execution's concurrency quota

### What happens

`inMemoryEventBroker`'s dispatch loop does a blocking channel send per registered queue, with no default, no deadline and no context. A subscriber that stops draining its 32-slot buffer therefore parks the broker goroutine indefinitely, and because that same goroutine also serves registration and unregistration, the stall is not confined to the data path.

The consequences compound. While the broker is parked it serves no `registerChan` and no `unregisterChan`, so `Reader.Close()` and any new subscription for that task block forever, and neither takes a context, so neither can be cancelled. `inMemoryManager.createReadWriter` calls `broker.connect()` while holding the process-wide manager mutex, so that blocked registration deadlocks `CreateReader`, `CreateWriter` and `Destroy` for every other task in the process as well. And in the default server configuration nothing ever breaks the wedge: the execution context is detached with `context.WithoutCancel`, and the inactivity watcher that would eventually cancel it only runs when the operator passes `WithAgentInactivityTimeout`. So the consumer stays blocked in `Write`, `runProducerConsumer` never returns, `cleanupExecution` never runs, and the execution's goroutines and its concurrency-quota slot are never released.

The reachable trigger is ordinary: a streaming client whose TCP receive window fills stops the `queue.Read` loop in `internal/taskexec/subscription.go` while staying connected. It does not need to disconnect, and disconnecting is in fact what would have saved the server.

### Where

`a2asrv/eventqueue/queue_in_memory_impl.go`, the broadcast case of the broker loop:

```go
case b := <-broker.broadcastChan:
for queue := range broker.registered {
if queue == b.sender {
continue
}
select {
case queue.eventsChan <- b.payload: // blocks here
case <-broker.destroySignal:
return
}
}
close(b.dispatched)
```

and `a2asrv/eventqueue/manager_in_memory_impl.go`, where the blocking handshake happens under the manager mutex:

```go
func (m *inMemoryManager) createReadWriter(ctx context.Context, taskID a2a.TaskID) (*inMemoryQueue, error) {
m.mu.Lock()
defer m.mu.Unlock()
...
return broker.connect()
}
```

`connect()` selects only on `b.destroyed` and `b.registerChan`. It takes no parameters at all, so there is no context for it to honour: createReadWriter receives one and never passes it down, which is what leaves the registration handshake uncancellable.

### Reproduction

Three assertions using only the public API, against `dda32ac`. A subscriber is created and then abandoned, and a writer fills its buffer:

```
healthy.Close() did not return within 3s, want nil (a stalled peer blocked unregistration)
healthy.Read() error = context deadline exceeded, want nil (event 3 of 5; a stalled peer starved this subscriber)
qm.CreateReader() did not return within 3s, want a reader (a stalled peer blocked registration)
--- FAIL: TestRepro_StalledSubscriberStarvesOthers (3.11s)
--- FAIL: TestRepro_StalledSubscriberBlocksClose (3.10s)
panic: test timed out after 10m0s
running tests:
TestRepro_StalledSubscriberBlocksNewSubscriber (10m0s)
```

The third did not merely fail, it deadlocked the binary. The dump shows the broker parked on the fan-out send at `queue_in_memory_impl.go:66`, a new subscriber blocked in `connect()` at `queue_in_memory_impl.go:97` while holding `m.mu` via `createReadWriter` at `manager_in_memory_impl.go:83`, and an unrelated task's `Destroy` blocked on that mutex at `manager_in_memory_impl.go:87`.

The server-level consequence reproduces too. With `MaxExecutions: 1`, starting one execution whose subscription is never read makes every subsequent execution fail with `concurrency quota exceeded: max concurrency limit reached`, indefinitely. A handful of non-reading clients is a full denial of service for new work.

### What I think the fix is

Two parts, and I have both working with tests if you would like a PR.

First, the broker should keep serving registration and unregistration while a send is pending, so that connect and disconnect are never hostage to a data-path stall. This needs no configuration and no time constant, and it removes the process-wide mutex deadlock at its source.

Second, a subscriber that stays full past a bounded grace period should be dropped rather than waited on forever, with its `Read` returning `ErrQueueClosed`. I would rather drop the subscriber than drop events, because a2a-go stamps messages with `TaskVersion` and subscriptions skip already-emitted versions, and `tasks/resubscribe` re-seeds from the task-store snapshot, so a dropped subscriber recovers cleanly while a dropped event is a gap the client cannot detect. That is deliberately the opposite of what the Python SDK does in the equivalent path, where the tap evicts the oldest event and keeps the subscriber, and the reason for the difference is this version and resubscribe machinery, which the Python layer does not have at that level.

I have kept the documented synchronous-write contract intact: `Write` still blocks until every still-registered queue has received the message, and `TestInMemoryQueue_WriteFull` passes unmodified. I know `docs/ai/CONCURRENCY_MODEL.md` records "a slow subscriber blocks the pipeline" as expected behaviour, and I am not proposing to change producer-side backpressure. What I am proposing to change is that a subscriber can hold the control plane, deadlock unrelated tasks, and leak quota with no bound and no way to cancel.

One related observation I have left alone: holding the manager mutex across `broker.connect()` is a hazard independent of this bug, since any `Manager` whose connect path can block will deadlock the whole manager. Fixing the broker makes it moot today. Happy to address it separately if you would like.

Edited 2026-08-20 for one detail. This said connect ignores the context it was passed; connect takes no parameters, and the context is the one createReadWriter receives and does not pass on. The consequence, a registration handshake nobody can cancel, is the same.