audio: wait for source when active WSOLA cannot synthesize

Query bounded retained engine state before pacing an empty native pass. Keep pending output, EOF and transitions runnable, and require retired processing metadata to be applied before waiting.

Exercise readiness against zero-input processing in both native formats across rates and tempos, plus deferred control events. No sample conversion, allocation or new synchronization is added.

audio/audio_pipeline_stretch.c

@@ -59,8 +59,14 @@ void audio_pipeline_stretch_free(audio_pipeline_stretch_t *s)
 
 bool audio_pipeline_stretch_needs_input(const audio_pipeline_stretch_t *s)
 {
+   /* An event may be retired while its application waits for old output.
+    * Keep running until the stream has seen that processing request. */
    return s && !s->direct_frames && !s->draining_layout
-      && audio_stretch_stream_quiescent(s->stream);
+      && s->control == s->metadata->current_control
+      && s->cutoff == s->metadata->current_cutoff
+      && s->layout == s->metadata->current_layout
+      && s->seen_reset == s->metadata->reset_serial
+      && audio_stretch_stream_needs_input(s->stream);
 }
 
 static void apstretch_offer(audio_pipeline_stretch_t *s,

audio/audio_stretch.c

@@ -619,6 +619,22 @@ bool audio_stretch_stream_quiescent(const audio_stretch_stream_t *s)
          && !s->count && !s->eof);
 }
 
+bool audio_stretch_stream_needs_input(const audio_stretch_stream_t *s)
+{
+   const audio_stretch_t *engine;
+   unsigned needed;
+   if (!s || s->eof || s->bound_read < s->bound_count || s->read < s->count)
+      return false;
+   if (s->phase == ASTRETCH_STREAM_RAW) return true;
+   if (s->phase != ASTRETCH_STREAM_ACTIVE) return false;
+   engine = s->engine;
+   if (engine->pending || engine->draining) return false;
+   /* Dropping an exhausted search prefix reduces count and next equally. */
+   needed = (engine->started ? engine->next + engine->radius : 0)
+      + 2 * engine->hop;
+   return engine->count < needed;
+}
+
 static void astretch_stream_pump(audio_stretch_stream_t *s,
       struct audio_stretch_io *io, double tempo, bool active)
 {

audio/audio_stretch.h

@@ -121,6 +121,9 @@ void audio_stretch_stream_reset(audio_stretch_stream_t *state);
 /* True only in raw state with no retained output. The owner may then bypass
  * the adapter until activation. NULL is quiescent. EOF requires reset first. */
 bool audio_stretch_stream_quiescent(const audio_stretch_stream_t *state);
+/* Consumer only, with the current processing request unchanged. True means
+ * no output can be produced without source; exit/EOF remain conservative. */
+bool audio_stretch_stream_needs_input(const audio_stretch_stream_t *state);
 bool audio_stretch_stream_process(audio_stretch_stream_t *state,
       struct audio_stretch_io *io, double tempo, bool active);
 bool audio_stretch_stream_flush(audio_stretch_stream_t *state,

samples/audio/stretch/stretch_filter_test.c

@@ -172,10 +172,42 @@ static void discard_filter(bool floating)
    audio_pipeline_stretch_free(s); retro_spsc_free(&ring);
 }
 
+static void readiness_controls(void)
+{
+   retro_spsc_t ring;
+   audio_pipeline_layout_t q;
+   audio_pipeline_stretch_t *s;
+   struct audio_pipeline_stretch_block block;
+   float output[32];
+   unsigned event;
+   CHECK(retro_spsc_init(&ring, 256));
+   audio_pipeline_layout_init(&q, 3);
+   s = audio_pipeline_stretch_new(48000, 2, true, 1, &ring, &q, output, 16);
+   if (!s) abort();
+   CHECK(audio_pipeline_stretch_needs_input(s));
+   guarded = 1;
+   for (event = 0; event < 4; event++)
+   {
+      /* Retired metadata can precede application to retained stream state. */
+      CHECK(audio_pipeline_layout_publish_processing(&q, 0,
+               event >= 2 ? 1 : 3, 131072, true, event == 3,
+               event >= 1 ? 1000 : 0));
+      audio_pipeline_layout_limit_transport(&q, 0, 0, ring.capacity);
+      CHECK(!audio_pipeline_stretch_needs_input(s));
+      CHECK(audio_pipeline_stretch_next(s, 0, 16, &block));
+      CHECK(!block.frames);
+      CHECK(audio_pipeline_stretch_needs_input(s));
+   }
+   guarded = 0;
+   audio_pipeline_stretch_free(s);
+   retro_spsc_free(&ring);
+}
+
 int main(void)
 {
    unsigned channels, mode, active, reset, capacity, j, cases = 0;
    metadata_contracts();
+   readiness_controls();
    discard_filter(false); discard_filter(true);
    for (channels = 2; channels <= 11; channels += 9)
    {

samples/audio/stretch/stretch_test.c

@@ -1197,6 +1197,72 @@ static void stream_quiescence(void)
    }
 }
 
+static void stream_input_readiness(void)
+{
+   static const double tempos[] = { 0.25, 1.0, 1.5, 4.0, 32.0 };
+   static const unsigned rates[] = { 8000, 48000, 192000 };
+   unsigned native, rate, tempo, cases = 0;
+   CHECK(!audio_stretch_stream_needs_input(NULL));
+   for (native = 0; native < 2; native++)
+      for (rate = 0; rate < 3; rate++)
+         for (tempo = 0; tempo < 5; tempo++)
+         {
+            audio_stretch_stream_t *s = audio_stretch_stream_new(
+                  rates[rate], 8, native, 1);
+            const void *input = native ? (const void*)input_f : (const void*)input_i;
+            void *output = native ? (void*)output_f[0] : (void*)output_i[0];
+            size_t used = 0, count, taken, frame = 8 * (native ? sizeof(float) : sizeof(int16_t));
+            unsigned iteration = 0, starved = 0, retained = 0;
+            bool complete = false;
+            fill(8);
+            CHECK(s != NULL);
+            CHECK(audio_stretch_stream_bind(s, output, 17));
+            guarded = 1;
+            CHECK(audio_stretch_stream_needs_input(s));
+            while (used < FRAMES && iteration++ < 100000)
+            {
+               bool needs = audio_stretch_stream_needs_input(s);
+               /* Processing, rather than internal counters, is the oracle. */
+               CHECK(audio_stretch_stream_push(s, NULL, 0, &taken, tempos[tempo], true));
+               CHECK(!taken);
+               audio_stretch_stream_peek(s, &count);
+               CHECK(!needs || !count);
+               if (needs) starved++;
+               if (count)
+               {
+                  CHECK(!audio_stretch_stream_needs_input(s));
+                  retained++;
+                  CHECK(audio_stretch_stream_consume(s, count));
+               }
+               else
+               {
+                  size_t n = FRAMES - used;
+                  if (n > 71) n = 71;
+                  CHECK(audio_stretch_stream_push(s, (const char*)input + used * frame,
+                           n, &taken, tempos[tempo], true));
+                  used += taken;
+                  audio_stretch_stream_peek(s, &count);
+                  if (count) CHECK(!audio_stretch_stream_needs_input(s));
+                  CHECK(audio_stretch_stream_consume(s, count));
+               }
+            }
+            CHECK(iteration < 100000 && starved && retained);
+            while (!complete)
+            {
+               CHECK(audio_stretch_stream_finish(s, &complete));
+               CHECK(!audio_stretch_stream_needs_input(s));
+               audio_stretch_stream_peek(s, &count);
+               CHECK(audio_stretch_stream_consume(s, count));
+            }
+            audio_stretch_stream_reset(s);
+            CHECK(audio_stretch_stream_needs_input(s));
+            guarded = 0;
+            audio_stretch_stream_free(s);
+            cases++;
+         }
+   printf("stream input readiness: %u cases completed\n", cases);
+}
+
 int main(void)
 {
    contracts();
@@ -1217,6 +1283,7 @@ int main(void)
    adapter_direct_active();
    bound_budget_cases();
    stream_quiescence();
+   stream_input_readiness();
    CHECK(heap_calls == 0);
    printf("stretch: %u failures, %u processing/reset heap calls\n", failures, heap_calls);
    return failures != 0;