proposal: expose consumer-close notification on StreamWriter (Closed() <-chan struct{}) — producers blocked outside Send cannot observe reader close

## Problem

`schema.Pipe`'s producer side can only learn that the consumer closed the reader through `Send` returning `closed=true`. A producer goroutine that is **blocked somewhere other than `Send`** — typically blocked reading a stalled network connection while waiting for the next SSE event — never calls `Send`, so it never observes the close. The goroutine and its HTTP connection then leak until the request context is cancelled externally or the server closes the connection.

This is the standard shape of every streaming chat-model provider built on eino (including the eino-ext OpenAI/Gemini implementations):

```go
reader, writer := schema.Pipe[*schema.Message](64)
go func() {
defer writer.Close()
defer resp.Body.Close()
for {
event, err := sse.Next() // <-- blocked here while the connection is stalled
...
if closed := writer.Send(msg, nil); closed {
return // <-- only reachable when events are flowing
}
}
}()
return reader, nil
```

If the consumer calls `reader.Close()` while events are flowing, the next `Send` reports it and the producer exits — fine. If the consumer closes **while the stream is stalled** (no events arriving), the producer stays parked in the network read indefinitely. Reader `Close` propagates `closeRecv` correctly through `StreamReaderWithConvert` / `MergeStreamReaders` wrappers, but nothing surfaces it to a producer that isn't actively sending.

## Proposal

Expose the existing internal close signal on the writer. The `stream[T]` struct already has a `closed chan struct{}` that `closeRecv()` closes and that `send()` selects on — the mechanism exists; it just isn't reachable from user code. For example:

```go
// Closed returns a channel that is closed when the stream's receiver
// has stopped receiving (StreamReader.Close was called).
func (sw *StreamWriter[T]) Closed() <-chan struct{}
```

A producer can then tie its request lifetime to the consumer with no polling:

```go
ctx, cancel := context.WithCancel(ctx)
go func() {
select {
case <-sw.Closed():
cancel() // aborts the blocked body read
case <-producerDone:
}
}()
```

Design notes / open questions, from reading `schema/stream.go` at HEAD:

- For `automaticClose` readers, `closeRecv` is CAS-guarded and may fire from a finalizer — `Closed()` semantics stay the same (channel closes at most once).
- The channel is per-`stream[T]`, so a writer obtained from `Pipe` always has one; this needs no change to array-backed readers, child readers, or the multi-stream reader, whose `close()` paths already call `closeRecv` on their source streams.
- An alternative shape is an `OnClose(func())` reader option, but the channel form composes better with `select` loops producers already have.

## Real-world cost of the workaround

We hit this in production review (a consumer abandoning a stalled provider stream) and worked around it at the application layer with a heartbeat probe: a watchdog goroutine that periodically `Send`s a sentinel message into a dedicated capacity-1 pipe (fanned in with the data pipe via `MergeStreamReaders`, stripped with `StreamReaderWithConvert` + `ErrNoValue`), cancelling a stream-scoped context when a probe reports `closed=true`. It works, but it costs an extra goroutine + pipe + merge per stream, detection latency equal to the probe interval, and some genuinely subtle invariants (one sender per pipe because `Send` racing `Close` panics; `SetAutomaticClose` so two paths can release a probe blocked on an idle consumer). A `Closed()` accessor would replace all of it with one `select`.

Happy to contribute a PR if the maintainers think this (or an alternative shape) is the right direction.

Start by reading schema/stream.go, especially stream.closeRecv, send, and the StreamWriter/reader close paths described in the issue. Check how StreamReaderWithConvert and MergeStreamReaders propagate closure, then review the existing stream tests. Done means the chosen close-notification API has consistent semantics across the listed reader types and covers a producer blocked outside Send.

Written by the indexing model from the issue text.