Add: early-dispatch for require_sync_start SPMD cohorts (a2a3 tmr)

src/a2a3/platform/include/common/l2_swimlane_profiling.h

@@ -514,9 +514,18 @@ enum class L2SwimlaneSchedPhaseKind : uint32_t {
                   // push newly-ready successors, ring doorbells for
                   // early-dispatch hits. tasks_processed = # consumers visited.
     // Separate-lane (Worker View pid=4 AICPU_N)
-    DummyTask = 7,  // Per-dummy identity marker (zero-width). tasks_processed
-                    // = task_token_raw low 32 bits so deps.json flow arrows
-                    // can land on it.
+    DummyTask = 7,      // Per-dummy identity marker (zero-width). tasks_processed
+                        // = task_token_raw low 32 bits so deps.json flow arrows
+                        // can land on it.
+    Drain = 8,          // handle_drain_mode outer: the sync_start stop-the-world drain
+                        // (ack barrier + availability + parallel stage + finalize).
+                        // One bar per dispatch-loop iteration that enters the drain,
+                        // so retries show as multiple bars. Otherwise this time is a
+                        // swimlane blind spot (the loop `continue`s past all records).
+    DrainPrepare = 9,   // inner: this thread's drain_stage_cores prepare pass
+                        // (cluster scan + build_payload). tasks_processed = subtasks.
+    DrainPublish = 10,  // inner: this thread's drain_stage_cores publish pass
+                        // (MMIO write_reg per subtask). tasks_processed = subtasks.
 };
 
 /** Index layout of the queue-depth snapshot arrays below: AIC=0, AIV=1, MIX=2.
@@ -546,7 +555,7 @@ struct L2SwimlaneAicpuSchedPhaseRecord {
     uint32_t tasks_processed;       // Tasks processed in this phase batch
     uint32_t pop_hit;               // SCHED_DISPATCH delta since last emit (0 for Complete)
     uint32_t pop_miss;              // SCHED_DISPATCH delta since last emit (0 for Complete)
-    int16_t shared_depth_at_start[L2SWIMLANE_NUM_QUEUE_SHAPES];  // sched->ready_queues[shape].size()
+    int16_t shared_depth_at_start[L2SWIMLANE_NUM_QUEUE_SHAPES];  // ready_queues[shape] + ready_sync_queues[shape]
     int16_t shared_depth_at_end[L2SWIMLANE_NUM_QUEUE_SHAPES];
     uint32_t _pad[4];  // 64B alignment padding
 };

src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md

@@ -546,9 +546,11 @@ Each scheduler thread runs a tight loop with two main phases:
 - Poll register `COND` on each managed core
 - When `TASK_FIN_STATE` detected: record completion timestamps, call `on_subtask_complete(task_id, subslot)` to increment the completion counter; when `completed_subtasks == total_required_subtasks`, trigger `on_task_complete(task_id)` which marks `task_state[slot] = COMPLETED`, acquires fanout lock, traverses fanout list (incrementing consumers' `fanin_refcount`), marks `task_state[slot] = CONSUMED`, and advances `last_task_alive` watermark
 
-**Phase 2 — Dispatch**:
+**Phase 2 — Dispatch** (full model in §8.6):
 
-- For each idle core: pop a task from the matching shape-based ready queue (lock-free MPMC Vyukov queue, one per resource shape)
+- Drain each source (normal ready ▸ speculative early) in occupancy order — `sync_start`
+  Tier-0 ▸ MIX ▸ AIC/AIV, idle ▸ pending — popping from the matching shape-based ready queue
+  (lock-free MPMC Vyukov queue, one per resource shape)
 - Build `PTO2DispatchPayload` from `TaskDescriptor` with `task_id`, `subslot`, `kernel_id`, and `core_type`
 - Write task pointer to `Handshake.task`, signal AICore via register `DATA_MAIN_BASE`
 
@@ -558,8 +560,11 @@ After these phases, the scheduler updates profiling headers and checks for termi
 
 Ready queues use a lock-free bounded MPMC (Vyukov) design:
 
-- One `PTO2ReadyQueue` per resource shape (5 shapes: `AIC_ONLY`, `AIV_X1`, `AIV_X2`, `AIC_AIV_X1`, `AIC_AIV_X2`)
-- **Push**: any thread (orchestrator via `init_task`, or scheduler on completion) pushes newly-ready tasks to the queue matching `task->active_mask.to_shape()`
+- One `PTO2ReadyQueue` per resource shape — 3 shapes (`PTO2_NUM_RESOURCE_SHAPES`): `MIX`
+  (AIC+AIV cluster), `AIC`, `AIV`. Alongside `ready_queues[]` there is a per-shape
+  `ready_sync_queues[]` (sync_start Tier-0) and the speculative `early_dispatch_queues[]` /
+  `early_sync_start_queue` — see §8.6 for the full source × tier model.
+- **Push**: any thread (orchestrator via `init_task`, or scheduler on completion) pushes newly-ready tasks to the queue matching `task->active_mask.to_shape()` (sync_start cohorts to the sync lane)
 - **Pop**: scheduler threads pop from the queue matching the idle core's resource shape
 - Per-slot sequence counters prevent ABA problems
 - `enqueue_pos` and `dequeue_pos` are on separate cache lines to avoid false sharing
@@ -603,6 +608,88 @@ Private internals are split across three .cpp files by responsibility:
 
 `AicpuExecutor` calls neither `handshake_*`, `assign_*`, `reassign_*`, nor `emergency_shutdown` directly — they are private, invoked only by `init` and `on_orchestration_done`.
 
+### 8.6 Dispatch model — two sources, sync tiers, occupancy order
+
+`resolve_and_dispatch` places ready and speculative work onto AICore cores under one
+occupancy model. Two orthogonal axes decide *what* runs and *where*:
+
+- **Source** — `NORMAL` (all producers done; the task sits in a ready queue and launches on
+  pickup) vs `EARLY` (a *speculative* pre-stage of a not-yet-released task; its dispatch
+  payload carries a non-zero `src_payload` gate and launches later by a doorbell). Normal
+  strictly precedes early.
+- **Cohort** — `SYNC_START` (an SPMD cohort that must launch atomically) vs `REGULAR` (each
+  block launches independently). "is it ready" (source) and "does it need a rendezvous"
+  (cohort) are orthogonal.
+
+Within each source the occupancy order is **`sync_start` ▸ MIX ▸ AIC/AIV** (shape), and per
+shape **idle ▸ pending** (an idle core takes its running slot; a busy core takes its gated
+pending slot, promoted on completion). This order lives in one shared skeleton,
+`run_staging_order`; the normal and early sources differ only in the per-shape stage callback
+(pickup vs gated).
+
+#### Queues
+
+| Source | Regular lanes | sync_start lane |
+| ------ | ------------- | --------------- |
+| NORMAL (ready) | `ready_queues[MIX\|AIC\|AIV]` | `ready_sync_queues[MIX\|AIC\|AIV]` (per-shape) |
+| EARLY (speculative) | `early_dispatch_queues[MIX\|AIC\|AIV]` | `early_sync_start_queue` (single) |
+
+A task routes to the sync lane iff `active_mask.requires_sync_start()`. In each source the
+sync lane is drained as a strict **Tier-0** before the regular lane (`sync_start > MIX > C/V`),
+and early dispatch runs only once *both* normal lanes are empty (normal ▸ early).
+
+**Asymmetry (deliberate):** the normal sync lane is per-shape (3 queues) because a ready sync
+cohort can dispatch *inline* when it fits, reusing the per-shape `dispatch_shape`; the early
+sync lane is a single, shape-agnostic queue because an early cohort is *always* gated → always
+takes the drain path, whose rendezvous counts cores (not blocks) and is shape-agnostic. Both
+feed the same drain.
+
+#### sync_start drain + rendezvous
+
+A sync_start cohort of `block_num` cores must occupy all its cores before any of them run.
+When it cannot fit inline, `enter_drain_mode` arms a stop-the-world drain:
+
+1. **Single election** — a CAS on `sync_start_pending` (0 → −1) makes drains mutually
+   exclusive; only one cohort drains at a time, regardless of source.
+2. **All-or-nothing** — the elected thread checks `count_global_available >= block_num`
+   *before* staging; if short it aborts (stages nothing) and retries after completions free
+   cores. A cohort is fully staged or not at all — never partial.
+3. **Parallel stage** — all threads barrier, then each CAS-claims a block range and stages
+   its own cores with a non-zero `src_payload` gate: idle cores → running slots, busy cores →
+   pending slots.
+4. **Rendezvous launch** — `running_slot_count` counts staged running-slot cores; when it
+   reaches `popcount(staged_core_mask)` **and** the producer has released,
+   `maybe_rendezvous_ring` rings every gated core's doorbell together — the cohort starts as one.
+
+Single-election + all-or-nothing make the drain deadlock-free across multiple cohorts: at most
+one drains, and it fully stages or waits, so two cohorts can never each half-occupy the cluster
+set (see the completion path's `pending_gated` classification for why a promoted-but-still-gated
+block is not mistaken for a normal task).
+
+#### Early-candidate gate: producer must publish every block (deadlock avoidance)
+
+`propagate_dispatch_fanin` (the EARLY-source candidate trigger) no-ops until the producer is
+**fully published**: `published_block_count == logical_block_num`. Normal dispatch, regular
+early staging, and the sync drain increment this counter only after the claimed range's payloads
+and MMIO dispatch tokens are visible. A staged producer also waits for release and completion of
+its owned doorbell pass before exposing fanout.
+
+This is load-bearing: a flagged SPMD producer with more blocks than cores (for example, a
+50-block AIC projection on 24 AIC cores) dispatches in waves. If its first wave triggered a
+downstream MIX cohort to gate every running and pending slot, the remaining producer blocks
+would find no core, never complete, and the cohort rendezvous waiting for producer release would
+never ring. Full publication is stronger than full reservation: every producer block has both a
+reserved core slot and a launch-visible payload before a consumer can pre-occupy resources.
+`next_block_idx` records reservation progress; `published_block_count` independently establishes
+publication and early-candidate readiness.
+
+#### MIX per-core placement
+
+A MIX task spans a cluster (1 AIC + 2 AIV). `classify_mix_cluster` admits a cluster whenever
+every used core has a free slot; `prepare_block_for_dispatch` then places **per core**
+(`to_pending && !is_core_idle`): idle cores → running, busy cores → pending. Cross-core start
+skew within a block is tolerated by AICore incore synchronization.
+
 ---
 
 ## 9. AICore Worker Interaction

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h

@@ -229,6 +229,14 @@ enum PTO2EarlyDispatchLaunchState : uint8_t {
     PTO2_EARLY_DISPATCH_LAUNCH_COMPLETE = 2,
 };
 
+enum PTO2EarlySyncDrainState : uint8_t {
+    PTO2_EARLY_SYNC_DRAIN_NONE = 0,
+    PTO2_EARLY_SYNC_DRAIN_OWNER = 1 << 0,
+    PTO2_EARLY_SYNC_DRAIN_ARMED = 1 << 1,
+    PTO2_EARLY_SYNC_DRAIN_READY = 1 << 2,
+    PTO2_EARLY_SYNC_DRAIN_COMPLETE = 1 << 3,
+};
+
 // A pre-staged consumer occupies one core per gated subtask block. WHICH cores
 // it occupies is recorded as a bitmask (staged_core_mask, 1 bit per global
 // core_id); the completion-path release iterates the set bits and rings each
@@ -255,9 +263,10 @@ struct PTO2TaskPayload {
     // between the fanin array (offset 536) and the 64B-aligned tensors[] (offset
     // 576), so sizeof and tensors[] are unchanged.
     //
-    // Bitmask of global core_ids this consumer is pre-staged (gated) on. Set with
-    // atomic fetch_or by concurrent stagers, then destructively split between the
-    // release and late-stager paths. (Re)initialized in PTO2TaskPayload::init.
+    // Bitmask of global core_ids this consumer is pre-staged (gated) on. Concurrent
+    // stagers publish bits with atomic fetch_or. A regular consumer destructively
+    // splits them between release and late-stager owners; a sync_start drain keeps
+    // the completed mask stable for its single cohort launch owner.
     std::atomic<uint64_t> staged_core_mask[PTO2_EARLY_DISPATCH_CORE_MASK_WORDS]{};
     // Early-dispatch CANDIDATE detection (event-driven, dual of fanin_refcount):
     // seeded at wiring with producers already complete, then a flagged producer
@@ -276,14 +285,26 @@ struct PTO2TaskPayload {
     // 3=DISPATCHED (2=STAGED is unused now). STAGING is the STABLE gated state —
     // many threads stage blocks concurrently while it holds, each claiming a block
     // via the atomic next_block_idx and OR-ing its cores into staged_core_mask.
-    // Release does STAGING->DISPATCHED and claims the current mask; a thread that
-    // stages a block after that flip claims and rings only its remaining bits.
+    // Release does STAGING->DISPATCHED. For a regular consumer it claims the current
+    // mask and a late stager rings only its remaining bits. A sync_start consumer
+    // preserves the mask for rendezvous counting and its single launch pass.
     std::atomic<uint8_t> early_dispatch_state{0};
     std::atomic<uint8_t> dispatch_propagated{0};  // PRODUCER side: once-guard for fanout propagation
-    // The release owner publishes COMPLETE only after all doorbells it claimed
-    // are visible. Combined with published_block_count, this keeps fanout
-    // private until release-owned and late-owned blocks have both launched.
+    // The launch owner publishes COMPLETE only after all owned doorbells are
+    // visible, keeping fanout private until every gated block has launched.
     std::atomic<uint8_t> early_dispatch_launch_state{PTO2_EARLY_DISPATCH_LAUNCH_NONE};
+    // sync_start early-dispatch rendezvous: count of this task's gated CORES currently
+    // occupying a RUNNING slot (staged directly to an idle core, or promoted from a
+    // gated pending slot). Counted per-core (not per-block) so it is shape-agnostic: a
+    // MIX block spans a cluster whose cores promote independently. A sync_start task's
+    // doorbells are rung only once this reaches popcount(staged_core_mask) AND the
+    // producer released, so all cores launch atomically. Unused (0) for non-sync_start.
+    std::atomic<int16_t> running_slot_count{0};
+    // Ownership handshake between the early sync queue and final ready routing.
+    // A successful OWNER persists through ARMED and COMPLETE until payload
+    // reinitialization. READY records that producer release observed OWNER;
+    // only cancellation clears OWNER during the current task lifetime.
+    std::atomic<uint8_t> early_sync_drain_state{PTO2_EARLY_SYNC_DRAIN_NONE};
     // === Cache lines 9-72 (4096B) — tensors (alignas(64) forces alignment) ===
     Tensor tensors[MAX_TENSOR_ARGS];
     // === Cache lines 73-74 (128B) — scalars ===
@@ -368,6 +389,8 @@ struct PTO2TaskPayload {
         dispatch_propagated.store(0, std::memory_order_relaxed);
         published_block_count.store(0, std::memory_order_relaxed);
         early_dispatch_launch_state.store(PTO2_EARLY_DISPATCH_LAUNCH_NONE, std::memory_order_relaxed);
+        running_slot_count.store(0, std::memory_order_relaxed);
+        early_sync_drain_state.store(PTO2_EARLY_SYNC_DRAIN_NONE, std::memory_order_relaxed);
     }
 };