Fix: keep dispatch marker in producer slot state

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

@@ -446,7 +446,7 @@ The orchestrator completes fanout wiring before publishing a task to the ready q
    - Checks `task_state >= COMPLETED` (early-finished optimization)
    - If not completed: prepends consumer to producer's `fanout_head` via `dep_pool.prepend`
    - **Releases** `fanout_lock`
-3. Atomically releases the +1 redundance + early_finished count via `fanin_refcount.fetch_add`
+3. Atomically releases the +1 redundance + completed-fanin count via `fanin_refcount.fetch_add`
 4. If all deps satisfied: pushes task to the routed ready queue
 
 Zero-fanin tasks and tasks whose claimed producers are already completed skip dep_pool entry allocation and publish directly to the routed ready queue.
@@ -683,11 +683,13 @@ reserved core slot and a launch-visible payload before a consumer can pre-occupy
 `next_block_idx` records reservation progress; `published_block_count` independently establishes
 publication and early-candidate readiness.
 
-The producer's `dispatch_propagated` claim and fanout snapshot are serialized under
-`fanout_lock` with consumer wiring. An edge already in the snapshot is counted by scheduler
-propagation; wiring seeds an edge added after the claim and enqueues the consumer if that seed
-completes `dispatch_fanin`. This gives each eligible producer-consumer edge exactly one
-early-candidate contribution regardless of which side acquires the lock first.
+The producer's slot-local dispatch-propagated bit in `ready_state` and fanout snapshot are
+serialized under `fanout_lock` with consumer wiring. Wiring already locks and reads the
+producer's 64-byte `PTO2TaskSlotState`, so testing this bit does not read a producer payload
+cache line. An edge already in the snapshot is counted by scheduler propagation; wiring seeds
+an edge added after the claim and enqueues the consumer if that seed completes
+`dispatch_fanin`. This gives each eligible producer-consumer edge exactly one early-candidate
+contribution regardless of which side acquires the lock first.
 
 #### MIX per-core placement
 

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp

@@ -389,7 +389,7 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32
     PTO2TaskPayload *payload = slot_state.payload;
     slot_state.fanin_count = wfanin + 1;
 
-    int32_t early_finished = 0;
+    int32_t completed_fanin = 0;
     int32_t early_propagated = 0;
     bool early_disqualified = false;
     for_each_fanin_slot_state(*payload, [&](PTO2TaskSlotState *producer) {
@@ -399,12 +399,12 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32
             early_disqualified = true;
         }
         if (pstate >= PTO2_TASK_COMPLETED) {
-            early_finished++;
+            completed_fanin++;
         } else {
             producer->fanout_head = rss.dep_pool.prepend(producer->fanout_head, &slot_state);
             // The marker shares fanout_lock with propagation's snapshot. A set
             // marker means this edge is outside that snapshot and needs a seed.
-            if (producer->payload->dispatch_propagated.load(std::memory_order_acquire) != 0) {
+            if (!early_disqualified && producer->has_dispatch_propagated()) {
                 early_propagated++;
             }
         }
@@ -414,17 +414,17 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32
     // Completed producers and edges outside a producer's one-shot propagation
     // snapshot both need wiring-time contributions. Seed only when every
     // producer is codegen-flagged; one unflagged producer disqualifies the task.
-    int32_t early_seed = early_finished + early_propagated;
+    int32_t early_seed = completed_fanin + early_propagated;
     if (!early_disqualified && early_seed != 0) {
         int32_t dispatch_fanin = payload->dispatch_fanin.fetch_add(early_seed, std::memory_order_acq_rel) + early_seed;
         // A fully pre-completed fanin routes normally. If any producer was live,
         // the exact-full increment must enqueue the early candidate.
-        if (early_finished != payload->fanin_actual_count && dispatch_fanin == payload->fanin_actual_count) {
+        if (completed_fanin != payload->fanin_actual_count && dispatch_fanin == payload->fanin_actual_count) {
             sched->try_enqueue_early_dispatch_candidate(slot_state);
         }
     }
 
-    int32_t init_rc = early_finished + 1;
+    int32_t init_rc = completed_fanin + 1;
     int32_t new_rc = slot_state.fanin_refcount.fetch_add(init_rc, std::memory_order_acq_rel) + init_rc;
     mark_dep_pool_position(slot_state);
     if (new_rc >= slot_state.fanin_count) {

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

@@ -306,7 +306,6 @@ struct PTO2TaskPayload {
     // 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 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};
@@ -403,13 +402,12 @@ struct PTO2TaskPayload {
         // one of ITS producers is flagged (propagate_dispatch_fanin bumps
         // dispatch_fanin and may CAS early_dispatch_state on any consumer, independent of the
         // consumer's own hint). So they MUST be zeroed here unconditionally.
-        // Publication, propagation, and launch fields share this same
-        // per-submit lifetime and are reset here too.
+        // Publication and launch fields share this same per-submit lifetime
+        // and are reset here too.
         early_dispatch_state.store(PTO2_EARLY_DISPATCH_NONE, std::memory_order_relaxed);
         for (int w = 0; w < PTO2_EARLY_DISPATCH_CORE_MASK_WORDS; w++)
             staged_core_mask[w].store(0, std::memory_order_relaxed);
         dispatch_fanin.store(0, std::memory_order_relaxed);
-        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);
@@ -442,7 +440,7 @@ static_assert(
  * Per-task slot scheduling state (scheduler-private, NOT in shared memory)
  *
  * Consolidates all hot-path scheduling fields into a single cache-friendly
- * structure (32 bytes = half a cache line). Accessing any field of a task's
+ * structure (64 bytes = one cache line). Accessing any field of a task's
  * slot state brings all related fields into the same cache line.
  *
  * Concurrency notes:
@@ -480,6 +478,12 @@ enum PTO2DeferredCompletionFlag : uint8_t {
     PTO2_SUBTASK_DEFERRED = 4,
 };
 
+enum PTO2DispatchPropagationFlag : uint8_t {
+    PTO2_DISPATCH_PROPAGATED = 8,
+};
+
+static_assert((PTO2_DISPATCH_PROPAGATED & (PTO2_READY_CLAIMED | PTO2_COMPLETION_DONE | PTO2_SUBTASK_DEFERRED)) == 0);
+
 struct alignas(64) PTO2TaskSlotState {
     // Fanout lock + list (accessed together under lock in on_task_complete)
     std::atomic<int32_t> fanout_lock;  // Per-task spinlock (0=unlocked, 1=locked)
@@ -514,6 +518,8 @@ struct alignas(64) PTO2TaskSlotState {
     // slot_state (not payload) so fanin walks read the already-hot producer
     // slot_state cache line.
     bool allow_early_resolve{false};
+    // Concurrent lifecycle updates preserve unrelated bits. Slot reuse clears
+    // the byte only after the previous task lifetime is quiescent.
     std::atomic<uint8_t> ready_state{PTO2_READY_UNCLAIMED};
     int32_t dep_pool_mark{0};  // Dep pool top after Orch-side wiring
 
@@ -560,6 +566,19 @@ struct alignas(64) PTO2TaskSlotState {
         task = t;
     }
 
+    // Lock-free callers use this only as a fast-path hint. A false result is
+    // rechecked by try_mark_dispatch_propagated() while holding fanout_lock.
+    bool has_dispatch_propagated() const {
+        return (ready_state.load(std::memory_order_acquire) & PTO2_DISPATCH_PROPAGATED) != 0;
+    }
+
+    // The propagation owner holds fanout_lock from this claim through its
+    // fanout snapshot so wiring can classify late edges exactly once.
+    bool try_mark_dispatch_propagated() {
+        return (ready_state.fetch_or(PTO2_DISPATCH_PROPAGATED, std::memory_order_acq_rel) & PTO2_DISPATCH_PROPAGATED) ==
+               0;
+    }
+
     void mark_completed() {
         task_state.store(PTO2_TASK_COMPLETED, std::memory_order_release);
         ready_state.fetch_or(PTO2_COMPLETION_DONE, std::memory_order_release);

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h

@@ -923,9 +923,9 @@ struct PTO2SchedulerState {
         // The once-claim and fanout snapshot share fanout_lock with wiring, so a
         // set marker tells wiring that its new edge is outside this snapshot.
         // The unlocked precheck avoids fanout-lock traffic after propagation.
-        if (p.payload->dispatch_propagated.load(std::memory_order_acquire) != 0) return;
+        if (p.has_dispatch_propagated()) return;
         p.lock_fanout();
-        if (p.payload->dispatch_propagated.exchange(1, std::memory_order_acq_rel) != 0) {
+        if (!p.try_mark_dispatch_propagated()) {
             p.unlock_fanout();
             return;
         }

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

@@ -446,7 +446,7 @@ The orchestrator completes fanout wiring before publishing a task to the ready q
    - Checks `task_state >= COMPLETED` (early-finished optimization)
    - If not completed: prepends consumer to producer's `fanout_head` via `dep_pool.prepend`
    - **Releases** `fanout_lock`
-3. Atomically releases the +1 redundance + early_finished count via `fanin_refcount.fetch_add`
+3. Atomically releases the +1 redundance + completed-fanin count via `fanin_refcount.fetch_add`
 4. If all deps satisfied: pushes task to the routed ready queue
 
 Zero-fanin tasks and tasks whose claimed producers are already completed skip dep_pool entry allocation and publish directly to the routed ready queue.

src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp

@@ -376,19 +376,19 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32
     PTO2TaskPayload *payload = slot_state.payload;
     slot_state.fanin_count = wfanin + 1;
 
-    int32_t early_finished = 0;
+    int32_t completed_fanin = 0;
     for_each_fanin_slot_state(*payload, [&](PTO2TaskSlotState *producer) {
         producer->lock_fanout();
         int32_t pstate = producer->task_state.load(std::memory_order_acquire);
         if (pstate >= PTO2_TASK_COMPLETED) {
-            early_finished++;
+            completed_fanin++;
         } else {
             producer->fanout_head = rss.dep_pool.prepend(producer->fanout_head, &slot_state);
         }
         producer->unlock_fanout();
     });
 
-    int32_t init_rc = early_finished + 1;
+    int32_t init_rc = completed_fanin + 1;
     int32_t new_rc = slot_state.fanin_refcount.fetch_add(init_rc, std::memory_order_acq_rel) + init_rc;
     mark_dep_pool_position(slot_state);
     if (new_rc >= slot_state.fanin_count) {

tests/ut/cpp/a2a3/test_wiring.cpp

@@ -174,7 +174,7 @@ TEST_F(WiringTest, WireTaskAllProducersEarlyFinished) {
 
     // fanin_count = 2 + 1 = 3
     EXPECT_EQ(task_slot.fanin_count, 3);
-    // early_finished = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count
+    // completed_fanin = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count
     EXPECT_GE(task_slot.fanin_refcount.load(), task_slot.fanin_count);
 
     // Task should be in ready queue
@@ -210,7 +210,7 @@ TEST_F(WiringTest, WireTaskProducersPendingTaskNotReady) {
 
     // fanin_count = 3 (2 + 1)
     EXPECT_EQ(task_slot.fanin_count, 3);
-    // early_finished = 0, init_rc = 1 -> not ready
+    // completed_fanin = 0, init_rc = 1 -> not ready
     EXPECT_EQ(task_slot.fanin_refcount.load(), 1);
     EXPECT_LT(task_slot.fanin_refcount.load(), task_slot.fanin_count);
 
@@ -226,6 +226,26 @@ TEST_F(WiringTest, WireTaskProducersPendingTaskNotReady) {
     EXPECT_EQ(producer_slots[1].fanout_head->slot_state, &task_slot);
 }
 
+TEST_F(WiringTest, DispatchPropagationMarkerPreservesReadyStateFlagsAndResets) {
+    alignas(64) PTO2TaskSlotState producer;
+    init_slot(producer, PTO2_TASK_PENDING, 1, 1);
+
+    EXPECT_FALSE(producer.has_dispatch_propagated());
+    EXPECT_TRUE(producer.try_mark_dispatch_propagated());
+    EXPECT_TRUE(sched.try_claim_ready_once(producer));
+    producer.mark_completed();
+    producer.mark_any_subtask_deferred();
+    EXPECT_FALSE(producer.try_mark_dispatch_propagated());
+    EXPECT_EQ(
+        producer.ready_state.load(std::memory_order_relaxed),
+        PTO2_READY_CLAIMED | PTO2_COMPLETION_DONE | PTO2_SUBTASK_DEFERRED | PTO2_DISPATCH_PROPAGATED
+    );
+
+    producer.reset_for_reuse();
+    EXPECT_FALSE(producer.has_dispatch_propagated());
+    EXPECT_EQ(producer.ready_state.load(std::memory_order_relaxed), PTO2_READY_UNCLAIMED);
+}
+
 // =============================================================================
 // Orch-side wiring: mixed early-finished and pending producers
 // =============================================================================
@@ -253,7 +273,7 @@ TEST_F(WiringTest, WireTaskMixedProducerStates) {
 
     // fanin_count = 4 (3 + 1)
     EXPECT_EQ(task_slot.fanin_count, 4);
-    // early_finished = 2 (COMPLETED + CONSUMED), init_rc = 3
+    // completed_fanin = 2 (COMPLETED + CONSUMED), init_rc = 3
     // Not yet 4 -> not ready (one producer still running)
     EXPECT_EQ(task_slot.fanin_refcount.load(), 3);
 
@@ -321,7 +341,7 @@ TEST_F(WiringTest, WireTaskUnflaggedPrecompletedProducerDoesNotSeed) {
     wire_fanin(task_slot, 1);
 
     EXPECT_EQ(payload.dispatch_fanin.load(), 0);  // NOT seeded (direct-only)
-    // Readiness is unaffected: early_finished(1) + 1 == fanin_count(2) -> ready.
+    // Readiness is unaffected: completed_fanin(1) + 1 == fanin_count(2) -> ready.
     EXPECT_GE(task_slot.fanin_refcount.load(), task_slot.fanin_count);
 }
 
@@ -375,12 +395,12 @@ TEST_F(WiringTest, EarlyDispatchWaitsForAllProducerBlocksPublished) {
     sched.record_published_blocks(producer, 2);
     sched.propagate_dispatch_fanin(producer);
     EXPECT_EQ(payload.dispatch_fanin.load(), 0);
-    EXPECT_EQ(producer.payload->dispatch_propagated.load(), 0);
+    EXPECT_FALSE(producer.has_dispatch_propagated());
 
     sched.record_published_blocks(producer, 1);
     sched.propagate_dispatch_fanin(producer);
     EXPECT_EQ(payload.dispatch_fanin.load(), payload.fanin_actual_count);
-    EXPECT_EQ(producer.payload->dispatch_propagated.load(), 1);
+    EXPECT_TRUE(producer.has_dispatch_propagated());
 
     sched.propagate_dispatch_fanin(producer);
     EXPECT_EQ(payload.dispatch_fanin.load(), payload.fanin_actual_count);
@@ -396,7 +416,7 @@ TEST_F(WiringTest, LateWiredFullyPublishedProducerStillSeedsEarlyDispatch) {
     producer.allow_early_resolve = true;
     sched.record_published_blocks(producer, producer.logical_block_num);
     sched.propagate_dispatch_fanin(producer);
-    ASSERT_EQ(producer.payload->dispatch_propagated.load(), 1);
+    ASSERT_TRUE(producer.has_dispatch_propagated());
 
     init_slot(consumer, PTO2_TASK_PENDING, 0, 1);
     consumer_payload.fanin_actual_count = 1;
@@ -582,18 +602,18 @@ TEST_F(WiringTest, EarlyDispatchFanoutWaitsForDoorbellPass) {
     producer.payload->early_dispatch_state.store(PTO2_EARLY_DISPATCH_STAGING, std::memory_order_relaxed);
     sched.propagate_dispatch_fanin(producer);
     EXPECT_EQ(consumer.payload->dispatch_fanin.load(), 0);
-    EXPECT_EQ(producer.payload->dispatch_propagated.load(), 0);
+    EXPECT_FALSE(producer.has_dispatch_propagated());
 
     producer.payload->early_dispatch_launch_state.store(PTO2_EARLY_DISPATCH_LAUNCH_RINGING);
     producer.payload->early_dispatch_state.store(PTO2_EARLY_DISPATCH_DISPATCHED, std::memory_order_release);
     sched.propagate_dispatch_fanin(producer);
     EXPECT_EQ(consumer.payload->dispatch_fanin.load(), 0);
-    EXPECT_EQ(producer.payload->dispatch_propagated.load(), 0);
+    EXPECT_FALSE(producer.has_dispatch_propagated());
 
     producer.payload->early_dispatch_launch_state.store(PTO2_EARLY_DISPATCH_LAUNCH_COMPLETE, std::memory_order_seq_cst);
     sched.propagate_dispatch_fanin(producer);
     EXPECT_EQ(consumer.payload->dispatch_fanin.load(), 1);
-    EXPECT_EQ(producer.payload->dispatch_propagated.load(), 1);
+    EXPECT_TRUE(producer.has_dispatch_propagated());
 }
 
 TEST_F(WiringTest, LateStagerRingsCapturedDoorbellAfterTableReuse) {
@@ -636,13 +656,13 @@ TEST_F(WiringTest, EarlyDispatchReleaseConsumesDoorbellMask) {
     EXPECT_EQ(task.payload->staged_core_mask[0].load(), 0);
     EXPECT_EQ(get_test_reg_stub_base_addr(), reg_addr);
     EXPECT_EQ(get_test_reg_stub_value(), (static_cast<uint64_t>(token) << 32) | token);
-    EXPECT_EQ(task.payload->dispatch_propagated.load(), 0);
+    EXPECT_FALSE(task.has_dispatch_propagated());
 
     sched.record_published_blocks(task, 1);
     // The staging path must retry even though its earlier state snapshot was
     // STAGING; release already missed this final publication count.
     sched.propagate_dispatch_fanin(task);
-    EXPECT_EQ(task.payload->dispatch_propagated.load(), 1);
+    EXPECT_TRUE(task.has_dispatch_propagated());
 
     sched.early_dispatch_doorbell_table[core_id].addr = 0x11110000;
     sched.early_dispatch_doorbell_table[core_id].token = 12;
@@ -711,7 +731,7 @@ TEST_F(WiringTest, SyncStartDrainFinalizeRetriesProducerFirstRendezvous) {
     sync_consumer.payload->running_slot_count.store(2, std::memory_order_seq_cst);
     EXPECT_TRUE(sched.retry_sync_start_rendezvous_after_drain(sync_consumer));
     EXPECT_EQ(sync_consumer.payload->early_dispatch_launch_state.load(), PTO2_EARLY_DISPATCH_LAUNCH_COMPLETE);
-    EXPECT_EQ(sync_consumer.payload->dispatch_propagated.load(), 1);
+    EXPECT_TRUE(sync_consumer.has_dispatch_propagated());
     EXPECT_EQ(downstream.payload->dispatch_fanin.load(), 1);
 
     EXPECT_FALSE(sched.retry_sync_start_rendezvous_after_drain(sync_consumer));
@@ -900,8 +920,8 @@ TEST_F(WiringTest, UnflaggedProducerDoesNotPropagate) {
 
     sched.propagate_dispatch_fanin(producer);
 
-    EXPECT_EQ(cons_payload.dispatch_fanin.load(), 0);       // no bump
-    EXPECT_EQ(prod_payload.dispatch_propagated.load(), 0);  // gate returned before once-guard
+    EXPECT_EQ(cons_payload.dispatch_fanin.load(), 0);  // no bump
+    EXPECT_FALSE(producer.has_dispatch_propagated());  // gate returned before once-guard
 }
 
 TEST_F(WiringTest, FlaggedPrecompletedCreatorTransparentToEarlyDispatch) {

tests/ut/cpp/a5/test_wiring.cpp

@@ -156,7 +156,7 @@ TEST_F(WiringTest, WireTaskAllProducersEarlyFinished) {
 
     // fanin_count = 2 + 1 = 3
     EXPECT_EQ(task_slot.fanin_count, 3);
-    // early_finished = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count
+    // completed_fanin = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count
     EXPECT_GE(task_slot.fanin_refcount.load(), task_slot.fanin_count);
 
     // Task should be in ready queue
@@ -192,7 +192,7 @@ TEST_F(WiringTest, WireTaskProducersPendingTaskNotReady) {
 
     // fanin_count = 3 (2 + 1)
     EXPECT_EQ(task_slot.fanin_count, 3);
-    // early_finished = 0, init_rc = 1 -> not ready
+    // completed_fanin = 0, init_rc = 1 -> not ready
     EXPECT_EQ(task_slot.fanin_refcount.load(), 1);
     EXPECT_LT(task_slot.fanin_refcount.load(), task_slot.fanin_count);
 
@@ -235,7 +235,7 @@ TEST_F(WiringTest, WireTaskMixedProducerStates) {
 
     // fanin_count = 4 (3 + 1)
     EXPECT_EQ(task_slot.fanin_count, 4);
-    // early_finished = 2 (COMPLETED + CONSUMED), init_rc = 3
+    // completed_fanin = 2 (COMPLETED + CONSUMED), init_rc = 3
     // Not yet 4 -> not ready (one producer still running)
     EXPECT_EQ(task_slot.fanin_refcount.load(), 3);