Handling of trailing enqueues on realtime paths

libs/qec/include/cudaq/qec/decoder.h

@@ -245,6 +245,14 @@ class decoder
   std::size_t get_block_size() { return block_size; }
   std::size_t get_syndrome_size() { return syndrome_size; }
 
+  /// Total measurements enqueued per shot, including any trailing measurements
+  /// not referenced by any detector in D_sparse. The realtime decoder factory
+  /// sets this from decoder_config, defaulting to num_msyn_per_decode (no
+  /// trailing budget). Values above num_msyn_per_decode absorb that many
+  /// trailing bits before the session goes stale for the next volume.
+  void set_total_circuit_measurements(uint64_t n);
+  uint64_t get_total_circuit_measurements() const;
+
   // -- Begin realtime decoding API --
 
   // Note: all of the current realtime decoding API is designed to be used with

libs/qec/include/cudaq/qec/realtime/decoding_config.h

@@ -75,6 +75,12 @@ struct decoder_config {
   std::vector<std::int64_t> H_sparse;
   std::vector<std::int64_t> O_sparse;
   std::vector<std::int64_t> D_sparse;
+  /// Total measurements enqueued per shot, including any trailing measurements
+  /// not referenced by any detector in D_sparse. When unset (zero), the
+  /// realtime decoder factory defaults this to num_msyn_per_decode, giving no
+  /// trailing budget. Set above num_msyn_per_decode to silently absorb that
+  /// many trailing bits before the session goes stale for the next volume.
+  uint64_t total_circuit_measurements = 0;
   decoder_custom_args_t decoder_custom_args;
 
   bool operator==(const decoder_config &) const = default;

libs/qec/lib/decoder.cpp

@@ -56,6 +56,11 @@ struct decoder::rt_impl {
   /// The id of the decoder (for instrumentation)
   uint32_t decoder_id = 0;
 
+  /// Total measurements per shot set by set_total_circuit_measurements().
+  /// Initialized to 0; the realtime decoder factory sets it to at least
+  /// num_msyn_per_decode before the session goes live.
+  uint64_t total_circuit_measurements = 0;
+
   bool is_sliding_window = false;
 
   /// The number of syndromes per round.  Only used for sliding window decoder.
@@ -311,6 +316,14 @@ void decoder::set_decoder_id(uint32_t decoder_id) {
 
 uint32_t decoder::get_decoder_id() const { return pimpl->decoder_id; }
 
+void decoder::set_total_circuit_measurements(uint64_t n) {
+  pimpl->total_circuit_measurements = n;
+}
+
+uint64_t decoder::get_total_circuit_measurements() const {
+  return pimpl->total_circuit_measurements;
+}
+
 template <typename PimplType>
 void set_D_sparse_common(decoder *decoder,
                          const std::vector<std::vector<uint32_t>> &D_sparse,
@@ -365,12 +378,27 @@ void decoder::set_D_sparse(const std::vector<int64_t> &D_sparse_vec_in) {
 
 bool decoder::enqueue_syndrome(const uint8_t *syndrome,
                                std::size_t syndrome_length) {
-  if (pimpl->msyn_buffer_index + syndrome_length > pimpl->msyn_buffer.size()) {
+  // Effective total window: syndrome bits plus any trailing measurements.
+  // Falls back to num_msyn_per_decode when not set by the caller.
+  const auto total = pimpl->total_circuit_measurements > 0
+                         ? pimpl->total_circuit_measurements
+                         : static_cast<uint64_t>(pimpl->num_msyn_per_decode);
+
+  if (pimpl->msyn_buffer_index + syndrome_length > total) {
     // CUDA_QEC_WARN("Syndrome buffer overflow. Syndrome will be ignored.");
     printf("Syndrome buffer overflow. Syndrome will be ignored.\n");
     return false;
   }
 
+  // Trailing measurements (after decode fired, before the window resets) are
+  // absorbed without being buffered or triggering another decode.
+  if (pimpl->msyn_buffer_index >= pimpl->num_msyn_per_decode) {
+    pimpl->msyn_buffer_index += syndrome_length;
+    if (pimpl->msyn_buffer_index >= total)
+      pimpl->msyn_buffer_index = 0;
+    return false;
+  }
+
   pimpl->current_round++;
   bool did_decode = false;
   for (std::size_t i = 0; i < syndrome_length; i++) {
@@ -555,10 +583,14 @@ bool decoder::enqueue_syndrome(const uint8_t *syndrome,
         printf("%s\n", s.c_str());
     }
     did_decode = true;
-    // Prepare for more data.
-    pimpl->msyn_buffer_index = 0;
     pimpl->current_round = 0;
     pimpl->detector_layer_index = 0;
+    // Leave msyn_buffer_index at num_msyn_per_decode so trailing measurements
+    // are absorbed by the early-return path above up to
+    // total_circuit_measurements, at which point the index resets for the next
+    // shot.
+    if (pimpl->msyn_buffer_index >= total)
+      pimpl->msyn_buffer_index = 0;
   }
   return did_decode;
 }

libs/qec/lib/realtime/config.cpp

@@ -286,6 +286,8 @@ struct MappingTraits<cudaq::qec::decoding::config::decoder_config> {
     io.mapRequired("H_sparse", config.H_sparse);
     io.mapRequired("O_sparse", config.O_sparse);
     io.mapRequired("D_sparse", config.D_sparse);
+    io.mapOptional("total_circuit_measurements",
+                   config.total_circuit_measurements, uint64_t{0});
 
     // Validate that the number of rows in the H_sparse vector is equal to
     // syndrome_size.
@@ -611,6 +613,8 @@ std::string decoder_config_json_schema() {
       {"H_sparse", llvm::json::Object{{"$ref", "#/$defs/sparse_matrix"}}},
       {"O_sparse", llvm::json::Object{{"$ref", "#/$defs/sparse_matrix"}}},
       {"D_sparse", llvm::json::Object{{"$ref", "#/$defs/sparse_matrix"}}},
+      {"total_circuit_measurements",
+       llvm::json::Object{{"type", "integer"}, {"minimum", 0}}},
       {"decoder_custom_args", llvm::json::Object{{"type", "object"}}},
   };
 

libs/qec/lib/realtime/decoding-server-cqr/DecodingSession.cpp

@@ -147,9 +147,19 @@ void DecodingSession::enqueue_core(const slot::EnqueueView &req) {
     unpack_scratch_[i] = (req.packed_bits[i / 8] >> (i % 8)) & 1u;
 
   try {
-    // Any accepted input after a completed decode starts a new volume; the old
-    // correction vector must not be reported as the result of that volume.
-    shot_state = ShotState::collecting;
+    if (shot_state == ShotState::result_ready) {
+      // accepted_syndromes holds the total measurements absorbed this shot.
+      // Trailing measurements keep incrementing it up to
+      // total_circuit_measurements; once exhausted, the bits belong to the
+      // next volume and the result goes stale.
+      if (accepted_syndromes + req.num_syndromes <=
+          dec->get_total_circuit_measurements()) {
+        accepted_syndromes += req.num_syndromes;
+        return;
+      }
+      accepted_syndromes = 0;
+      shot_state = ShotState::collecting;
+    }
 
     const size_t expected_syndromes = dec->get_num_msyn_per_decode();
     if (accepted_syndromes > expected_syndromes ||
@@ -167,7 +177,6 @@ void DecodingSession::enqueue_core(const slot::EnqueueView &req) {
 
     if (did_decode) {
       ++decode_count;
-      accepted_syndromes = 0;
       shot_state = ShotState::result_ready;
     }
   } catch (const std::exception &e) {
@@ -232,6 +241,7 @@ RpcStatus DecodingSession::get_corrections_core(int64_t return_size_arg,
       // throw here must produce the single INTERNAL_ERROR status below, not
       // a second response after an already-delivered OK.
       dec->clear_corrections();
+      accepted_syndromes = 0;
       shot_state = ShotState::collecting;
     }
     out_len = result_len;

libs/qec/lib/realtime/qec_realtime_session.cpp

@@ -263,7 +263,7 @@ void get_corrections_host(const void *rx_slot, void *tx_slot,
         out[i >> 3] |= static_cast<std::uint8_t>(1u << (i & 7));
     }
     if (body->reset != 0)
-      decoder->clear_corrections();
+      decoder->reset_decoder();
     write_response(tx_slot, rx_slot, rpc::RpcStatus::OK,
                    static_cast<std::uint32_t>(result_len));
   } catch (const std::out_of_range &) {

libs/qec/lib/realtime/realtime_decoding.cpp

@@ -246,6 +246,11 @@ std::unique_ptr<cudaq::qec::decoder> create_realtime_decoder(
   decoder->set_decoder_id(decoder_config.id);
   decoder->set_O_sparse(decoder_config.O_sparse);
   decoder->set_D_sparse(decoder_config.D_sparse);
+  // Default to num_msyn_per_decode when not explicitly set: no trailing budget
+  decoder->set_total_circuit_measurements(
+      decoder_config.total_circuit_measurements > 0
+          ? decoder_config.total_circuit_measurements
+          : static_cast<uint64_t>(decoder->get_num_msyn_per_decode()));
 
   // Force plugin initialization before the caller publishes the decoder for
   // realtime work. This preserves configure_decoders()'s existing behavior.

libs/qec/python/bindings/py_decoding_config.cpp

@@ -192,6 +192,8 @@ void bindDecodingConfig(nb::module_ &mod) {
       .def_rw("H_sparse", &decoder_config::H_sparse)
       .def_rw("O_sparse", &decoder_config::O_sparse)
       .def_rw("D_sparse", &decoder_config::D_sparse)
+      .def_rw("total_circuit_measurements",
+              &decoder_config::total_circuit_measurements)
       .def_prop_rw(
           "decoder_custom_args",
           [](const decoder_config &self) -> nb::object {

libs/qec/unittests/test_decoders.cpp

@@ -1112,6 +1112,30 @@ TEST(EnqueueSyndrome, ObsFrameDecoderUsesResultDirectly) {
   EXPECT_EQ(corr[1], 0u);
 }
 
+// Trailing measurements (data-qubit readouts enqueued after the syndrome
+// window) must not trigger a second decode or discard the pending correction.
+TEST(EnqueueSyndrome, TrailingMeasurementPreservesCorrection) {
+  cudaqx::tensor<uint8_t> H_tensor({2, 4});
+  H_tensor.at({0, 0}) = 1;
+  H_tensor.at({1, 1}) = 1;
+  cudaqx::heterogeneous_map params;
+  params.insert("decode_to_obs", true);
+  auto dec = cudaq::qec::decoder::get("sample_decoder", H_tensor, params);
+  dec->set_D_sparse(std::vector<std::vector<uint32_t>>{{0}, {1}});
+  dec->set_O_sparse(std::vector<std::vector<uint32_t>>{{0}, {1}});
+  // 2 syndrome bits + 1 trailing data-qubit readout
+  dec->set_total_circuit_measurements(3);
+
+  EXPECT_TRUE(
+      dec->enqueue_syndrome(std::vector<uint8_t>{1, 0})); // decode fires
+  EXPECT_FALSE(
+      dec->enqueue_syndrome(std::vector<uint8_t>{0})); // trailing, absorbed
+
+  const uint8_t *corr = dec->get_obs_corrections();
+  EXPECT_EQ(corr[0], 1u); // correction from the decode is intact
+  EXPECT_EQ(corr[1], 0u);
+}
+
 // Verify that corrections XOR-accumulate correctly across multiple shots and
 // that clear_corrections() resets them between shots.
 TEST(EnqueueSyndrome, ObsFrameMultiShotAccumulation) {

libs/qec/unittests/test_decoding_server_core.cpp

@@ -439,6 +439,39 @@ void record_captured_syndromes(const uint8_t *data, size_t len) {
   g_captured_syndrome_bytes.assign(data, data + len);
 }
 
+// With total_circuit_measurements set, trailing bits (not referenced by any
+// detector) are absorbed silently after a decode fires, preserving the result.
+// Once the trailing budget is exhausted, the next enqueue marks it stale.
+TEST(DecodingSessionInline, TrailingMeasurementsPreserveResult) {
+  // ControlledDecoder: D_sparse = {{0,1}}, num_msyn_per_decode = 2.
+  // Circuit has 3 measurements total: 2 syndrome + 1 trailing data-qubit.
+  auto dec = std::make_unique<ControlledDecoder>();
+  dec->set_total_circuit_measurements(3); // 2 syndrome + 1 trailing data-qubit
+  auto session = DecodingSession::create(std::move(dec));
+
+  std::vector<uint8_t> tx(64, 0);
+  auto enq = [&](int ctr, uint8_t bit, uint32_t rid) {
+    auto eq = make_cqr_slot(kEnqueueSyndromesFunctionId, rid,
+                            make_enqueue_payload(ctr, {bit}));
+    session->handle_enqueue(eq.data(), tx.data(), eq.size());
+  };
+  auto gc = make_cqr_slot(kGetCorrectionsFunctionId, 9,
+                          make_get_corrections_payload(1, false));
+
+  enq(0, 1, 1); // bit 0 (syndrome)
+  enq(1, 0, 2); // bit 1 (syndrome) → decode fires → result_ready
+  enq(2, 0, 3); // bit 2 (trailing): absorbed without clearing result_ready
+  // Result is still available.
+  session->handle_get_corrections(gc.data(), tx.data(), gc.size());
+  expect_tx_status(tx, RpcStatus::OK, 9);
+
+  // The trailing budget (1 bit) is now exhausted. The next enqueue belongs
+  // to a new volume and must mark the result stale.
+  enq(3, 1, 4);
+  session->handle_get_corrections(gc.data(), tx.data(), gc.size());
+  expect_tx_status(tx, RpcStatus::NOT_READY, 9);
+}
+
 TEST(DecodingSessionInline, SaveSyndromeCaptureMatchesTheLegacyFormat) {
   auto [session, decoder] = make_session();
   (void)decoder;