refactor(engine): drain tasks before eager WINDOW for accept-side child

Replace the conditional task-queue-defer approach (commit bcf7b03 +
Phase 1 cc83e5e/484fb4eb) with a simpler in-line task drain. The
accepted-side WINDOW is once again emitted eagerly inside
ZillaStreamFactory.Stream.onBegin (pre-Phase-1 behaviour). For
accept-side child channels, ZillaEngine.drainTasks() is called between
beginInputFuture.setSuccess() and the eager doWindow so any tasks
queued during fireChannelBound's ACCEPTED-event chain (e.g., an
AdviseInputTask from a `read advise` placed before `connected`) run
synchronously and write their frames to streamsBuffer ahead of the
WINDOW.

Wire order for trick scripts: ACCEPTED → script statements (queue
AdviseInputTask) → drainTasks (CHALLENGE written) → eager WINDOW
written. Same ordering as Phase 1's intended CHALLENGE→WINDOW; no
flag, no submitTask gymnastics.

Wire order for non-trick scripts (e.g., binding-http AdvisoryIT): same
as pre-Phase-1 — no tasks queued before drain runs (drain is a no-op),
eager WINDOW written immediately.

Removes:
- ZillaChannel.windowNeedsTask field + setWindowNeedsTask/windowNeedsTask methods
- ZillaEngine.adviseInput's flag-set call
- ZillaPartition windowFuture listener's doWindow + submitTask logic
  (reverts listener to pre-Phase-1: just fireChannelConnected)

Adds:
- ZillaEngine.drainTasks() — package-private, delegates to executeTasks

Tests: engine driver ITs (DuplexIT 72/72, HalfDuplexIT 64/64,
SimplexIT 35/35, EngineIT 14/14); binding-http 374/374 (both
client.AdvisoryIT.shouldSendRequestAndFlush and
server.AdvisoryIT.shouldSendRequestAndFlush restored); binding-mcp
138/138; binding-mcp.spec ApplicationIT/NetworkIT all green.

https://claude.ai/code/session_01Sm5BSDDaRNueem3xUeM2Li

Author: claude

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/ZillaChannel.java

@@ -76,7 +76,6 @@ public abstract class ZillaChannel extends AbstractChannel<ZillaChannelConfig>
 
     private int capabilities;
     private boolean flushable;
-    private boolean windowNeedsTask;
 
     private DefaultBudgetDebitor debitor;
     private long debitorIndex = -1L;
@@ -671,16 +670,6 @@ public void setFlushable()
         flushable = true;
     }
 
-    public void setWindowNeedsTask()
-    {
-        windowNeedsTask = true;
-    }
-
-    public boolean windowNeedsTask()
-    {
-        return windowNeedsTask;
-    }
-
     public boolean isFlushable()
     {
         return flushable;

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/ZillaEngine.java

@@ -128,9 +128,6 @@ public void adviseInput(
         ChannelFuture handlerFuture,
         Object value)
     {
-        // an inbound advise placed before `connected` (e.g., `read advise zilla:challenge`)
-        // signals the script wants its frames written ahead of the accepted-side WINDOW
-        channel.setWindowNeedsTask();
         submitTask(new AdviseInputTask(channel, handlerFuture, value));
     }
 
@@ -299,6 +296,11 @@ void submitTask(
         submitTask(task, false);
     }
 
+    void drainTasks()
+    {
+        executeTasks();
+    }
+
     private void submitTask(
         Runnable task,
         boolean immediateIfAligned)

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/ZillaPartition.java

@@ -310,33 +310,7 @@ private void handleBeginInitial(
             {
                 if (future.isSuccess())
                 {
-                    final ZillaChannelConfig childConfig = childChannel.getConfig();
-                    final boolean updateNeedsWindow =
-                        childConfig.getUpdate() == ZillaUpdateMode.HANDSHAKE ||
-                        childConfig.getUpdate() == ZillaUpdateMode.STREAM;
-
-                    if (updateNeedsWindow && childChannel.windowNeedsTask())
-                    {
-                        // submit via task queue so the WINDOW write is ordered
-                        // after any AdviseInputTask queued by a `read advise`
-                        // statement that the script placed before `connected`,
-                        // and before any tasks queued by statements that follow
-                        // `connected` (since those run inside fireChannelConnected
-                        // below)
-                        childChannel.engine.submitTask(() -> sender.doWindow(childChannel));
-                        fireChannelConnected(childChannel, childChannel.getRemoteAddress());
-                    }
-                    else
-                    {
-                        // no `read advise` ahead of `connected`: preserve the
-                        // pre-Phase-1 ordering so the script's post-`connected`
-                        // listeners are registered before WINDOW is written
-                        fireChannelConnected(childChannel, childChannel.getRemoteAddress());
-                        if (updateNeedsWindow)
-                        {
-                            sender.doWindow(childChannel);
-                        }
-                    }
+                    fireChannelConnected(childChannel, childChannel.getRemoteAddress());
                 }
             });
         }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/ZillaStreamFactory.java

@@ -201,14 +201,15 @@ private void onBegin(
             if (config.getUpdate() == ZillaUpdateMode.HANDSHAKE ||
                 config.getUpdate() == ZillaUpdateMode.STREAM)
             {
-                if (channel.getParent() == null)
+                if (channel.getParent() != null)
                 {
-                    sender.doWindow(channel);
+                    // accept-side child: drain any tasks queued during the
+                    // beginInputFuture listener chain (e.g., AdviseInputTask
+                    // from a `read advise` placed before `connected`) so their
+                    // frames land on the streamsBuffer ahead of this WINDOW
+                    channel.engine.drainTasks();
                 }
-                // accept-side child channels defer WINDOW to handleBeginInitial's
-                // windowFuture listener so any advised statements that the script
-                // places before `connected` (e.g. `read advise zilla:challenge`)
-                // get a chance to write their frames first.
+                sender.doWindow(channel);
             }
 
             beginFuture.setSuccess();