Permanently failed Subscriber is never detected: binding stays "running", health stays UP, messages silently stop

**Issue Description**

We had a incident where our stream binder consumers stopped processing for approx 1 hr with no alert firing. The app looked healthy the entire time, binding state `running`, actuator health `UP`. The only trace was the pubsub client's own internal log:

```
ERROR c.g.c.p.v.StreamingSubscriberConnection : terminated streaming with exception
```

Root cause after digging: `Subscriber` is a Guava `ApiService`. When the streaming pull hits a non-retryable error it transitions to `FAILED` and stops permanently. Nothing in spring-cloud-gcp listens for that transition, as far as I can tell there is no `ApiService.Listener.failed()` override anywhere in the codebase:

- `PubSubSubscriberTemplate.subscribeAndConvert()` calls `subscriber.startAsync()` and returns; the lifecycle is dropped there
- `PubSubInboundChannelAdapter.addListeners()` only attaches a listener if a `HealthTrackerRegistry` is configured, and that listener (`HealthTrackerRegistryImpl`) only overrides `terminated()`
- `PubSubHealthIndicator` probes with a unary `pullAsync`, which is a different code path from streaming pull, so health stays UP even when every streaming subscriber in the JVM is dead

So a failed subscriber is completely invisible: `isRunning()` returns true, the binding reports `running`, health reports UP, throughput is zero, and the only recovery is restarting the app.

Version: 7.x, but the relevant code is unchanged on current main so I'm reporting against main.

**Sample**

This test passes on current main, which is exactly the problem, no listener is ever attached to the `Subscriber`, and even with health tracking on, a `failed()` transition changes nothing:

```java
package com.google.cloud.spring.pubsub.integration.inbound;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import com.google.api.core.ApiService;
import com.google.api.gax.core.FixedExecutorProvider;
import com.google.cloud.monitoring.v3.MetricServiceClient;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.cloud.spring.pubsub.core.health.HealthTrackerRegistry;
import com.google.cloud.spring.pubsub.core.health.HealthTrackerRegistryImpl;
import com.google.cloud.spring.pubsub.core.subscriber.PubSubSubscriberOperations;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;

@ExtendWith(MockitoExtension.class)
class SubscriberFailureReproTests {

private final TestUtils.TestApplicationContext context = TestUtils.createTestApplicationContext();

@Mock private PubSubSubscriberOperations subscriberOperations;
@Mock private MessageChannel outputChannel;

@SuppressWarnings("unchecked")
private PubSubInboundChannelAdapter adapterFor(Subscriber subscriber) {
when(this.subscriberOperations.subscribeAndConvert(
anyString(), any(Consumer.class), any(Class.class)))
.thenReturn(subscriber);

PubSubInboundChannelAdapter adapter =
new PubSubInboundChannelAdapter(this.subscriberOperations, "testSubscription");
adapter.setOutputChannel(this.outputChannel);
adapter.setBeanFactory(this.context);
return adapter;
}

// Default config: no listener at all on the Subscriber, so failure can't be observed.
@Test
void defaultConfig_noListenerIsAttached() {
Subscriber subscriber = mock(Subscriber.class);
PubSubInboundChannelAdapter adapter = adapterFor(subscriber);

adapter.start();

verify(subscriber, never()).addListener(any(), any());
assertThat(adapter.isRunning()).isTrue();
}

// With health tracking: the only listener overrides terminated(), so failed() is ignored.
@Test
void withHealthTracking_failedTransitionIsIgnored() {
Subscriber subscriber = mock(Subscriber.class);
when(subscriber.getSubscriptionNameString())
.thenReturn("projects/test-project/subscriptions/testSubscription");

HealthTrackerRegistry registry =
new HealthTrackerRegistryImpl(
"test-project",
mock(MetricServiceClient.class),
1, 1, 1,
FixedExecutorProvider.create(Executors.newSingleThreadScheduledExecutor()));

PubSubInboundChannelAdapter adapter = adapterFor(subscriber);
adapter.setHealthTrackerRegistry(registry);
adapter.start();

ArgumentCaptor captor = ArgumentCaptor.forClass(ApiService.Listener.class);
verify(subscriber).addListener(captor.capture(), any(Executor.class));

// what StreamingSubscriberConnection does on a non-retryable stream error
captor.getValue()
.failed(ApiService.State.RUNNING, new IllegalStateException("streaming pull terminated"));

// nothing reacted; adapter still claims to be running
assertThat(adapter.isRunning()).isTrue();
}
}
```

```
./mvnw test -pl spring-cloud-gcp-pubsub -am -Dtest=SubscriberFailureReproTests
```

Start with PubSubSubscriberTemplate.subscribeAndConvert(), PubSubInboundChannelAdapter.addListeners(), HealthTrackerRegistryImpl, and PubSubHealthIndicator to trace subscriber lifecycle and health reporting. Run SubscriberFailureReproTests with the provided Maven command, then define how a failed streaming subscriber should affect adapter state and health so the failure is visible instead of reporting running and UP.

Written by the indexing model from the issue text.