Session rollover

daemon/launcher/cli_parse.cpp

@@ -154,17 +154,17 @@ UploadParseResult parseUploadArgs(const std::vector<std::string>& argv) {
             return {std::nullopt,
                     "--session-id is no longer supported; point <LOG_PATH> at a "
                     "directory containing only that session"};
-        } else if (!tok.empty() && tok[0] == '-') {
+        }
+        if (!tok.empty() && tok[0] == '-') {
             return {std::nullopt, "unknown flag: " + key};
-        } else {
-            // Bare token → the positional <LOG_PATH>. Only one allowed.
-            if (have_log_path) {
-                return {std::nullopt, "unexpected extra argument: " + tok +
+        }
+        // Bare token → the positional <LOG_PATH>. Only one allowed.
+        if (have_log_path) {
+            return {std::nullopt, "unexpected extra argument: " + tok +
                                       " (only one <LOG_PATH> is accepted)"};
-            }
-            out.log_path = tok;
-            have_log_path = true;
         }
+        out.log_path = tok;
+        have_log_path = true;
     }
 
     if (!have_log_path) {
@@ -244,7 +244,18 @@ std::string validateTraceExecutionMode(const TraceArgs& args) {
 }
 
 bool segmentationRequested(const TraceArgs& args) {
-    return args.segment_every_ms > 0 || args.segment_max_rows > 0;
+    return args.segment_every_ms > 0 || args.segment_max_rows > 0 ||
+           args.run_roll_every_ms > 0 || args.run_roll_max_bytes > 0;
+}
+
+std::string segmentationWarning(const TraceArgs& args) {
+    if (args.run_roll_every_ms > 0 && args.segment_every_ms > 0 &&
+        args.segment_every_ms > args.run_roll_every_ms / 10) {
+        return "--segment-every is more than a tenth of --roll-every, so a run "
+               "part can overshoot its budget visibly; a run part ends only at "
+               "a segment boundary";
+        }
+    return {};
 }
 
 std::string validateTraceSegmentation(
@@ -258,6 +269,29 @@ std::string validateTraceSegmentation(
         return "--segment-every must be at least 60s; shorter cadences can "
                "create a session storm";
     }
+
+    if (args.run_roll_every_ms < 0) {
+        return "--roll-every cannot be negative";
+    }
+
+    if (args.run_roll_every_ms > 0 && args.segment_every_ms <= 0) {
+        return "--roll-every requires --segment-every. A run part ends at the "
+               "next segment boundary, and with no segment time trigger a "
+               "quiet period produces no boundary, so the part would grow "
+               "without bound";
+    }
+    if (args.run_roll_every_ms > 0 &&
+        args.run_roll_every_ms < args.segment_every_ms) {
+        return "--roll-every must be at least --segment-every; a run part "
+               "cannot be shorter than the segment carrying its boundary";
+        }
+    if (args.run_roll_max_bytes > 0 && args.segment_every_ms <= 0 &&
+        args.segment_max_rows == 0) {
+        return "--roll-max-bytes requires --segment-every or "
+               "--segment-max-rows; at least one segment trigger must be armed "
+               "for a run part to have a boundary to end on";
+    }
+
     if (!segmentationRequested(args)) return {};
 
     if (!inherited_analysis_id.empty()) {

daemon/launcher/cli_parse.hpp

@@ -56,6 +56,9 @@ struct TraceArgs {
     // both zero preserves the ordinary single-session path.
     int64_t segment_every_ms = 0;    // --segment-every
     uint64_t segment_max_rows = 0;   // --segment-max-rows
+
+    int64_t run_roll_every_ms = 0;
+    uint64_t run_roll_max_bytes = 0;
     // PC sampling period as a log2 exponent (2^N GPU cycles/sample, valid 5..31;
     // lower = more frequent → catches shorter kernels). 0 = leave the engine
     // default. Plumbed to the injected target via GPUFL_PC_SAMPLING_PERIOD.
@@ -150,6 +153,13 @@ constexpr int64_t kMinSegmentEveryMs = 60'000;
 /** True when at least one segmentation trigger is enabled. */
 bool segmentationRequested(const TraceArgs& args);
 
+/**
+ * Non-fatal configuration advice, empty when there is none. Separate from
+ * validation because validation runs twice (prase, then the execution
+ * boundary) and a warning must print exactly one.
+ */
+std::string segmentationWarning(const TraceArgs& args);
+
 /**
  * Validate segmentation-specific mode restrictions. inherited_analysis_id is
  * supplied by the execution boundary so an exported GPUFL_ANALYSIS_ID cannot

daemon/launcher/cli_parse_internal.cpp

@@ -65,6 +65,39 @@ bool parseDurationMs(const std::string& value, std::int64_t& out_ms) {
     return true;
 }
 
+bool parseByteSize(const std::string& value, std::uint64_t& out) {
+    const std::string text = trim(value);
+    std::size_t digits = 0;
+    while (digits < text.size() &&
+        std::isdigit(static_cast<unsigned char>(text[digits]))) {
+        ++digits;
+    }
+    if (digits == 0) return false;
+
+    std::uint64_t number = 0;
+    if (!parseUint64(text.substr(0, digits), number)) return false;
+
+    std::string unit = trim(text.substr(digits));
+    for (char& c : unit) {
+        c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
+    }
+
+    std::uint64_t multiplier = 1;
+    if (unit.empty() || unit == "b") multiplier = 1;
+    else if (unit == "k" || unit == "kb" || unit == "kib") multiplier = 1024ULL;
+    else if (unit == "m" || unit == "mb" || unit == "mib")
+        multiplier = 1024ULL * 1024;
+    else if (unit == "g" || unit == "gb" || unit == "gib")
+        multiplier = 1024ULL * 1024 * 1024;
+    else return false;
+
+    if (number > (std::numeric_limits<std::uint64_t>::max)() / multiplier) {
+        return false;
+    }
+    out = number * multiplier;
+    return true;
+}
+
 bool parseUint64(const std::string& value, std::uint64_t& out) {
     const char* begin = value.data();
     const char* end = begin + value.size();

daemon/launcher/cli_parse_internal.hpp

@@ -28,5 +28,6 @@ bool parseDurationMs(const std::string& value, std::int64_t& out_ms);
 bool parseUint64(const std::string& value, std::uint64_t& out);
 bool parseNonNegativeInt(const std::string& value, int& out);
 bool parsePositiveInt(const std::string& value, int& out);
+bool parseByteSize(const std::string& value, std::uint64_t& out);
 
 }  // namespace gpufl::launcher::detail

daemon/launcher/cli_trace_options.cpp

@@ -123,6 +123,26 @@ std::string parseUint64Option(const FlagBreak& flag,
     return {};
 }
 
+/** Byte budget into a slot; 0 disables the trigger. */
+template <std::uint64_t TraceArgs::*Slot>
+std::string parseByteSizeOption(const FlagBreak& flag,
+                                const std::vector<std::string>& argv,
+                                std::size_t& index,
+                                TraceArgs& args) {
+    std::string value;
+    if (const std::string error =
+            detail::takeFlagValue(flag, argv, index, value);
+        !error.empty()) {
+        return error;
+        }
+    if (!detail::parseByteSize(value, args.*Slot)) {
+        return "invalid " + flag.key + " value: " + value +
+               " (expected a byte count with an optional k/m/g suffix, "
+               "e.g. 50g; 0 disables it)";
+    }
+    return {};
+}
+
 // A flag that no longer exists: consume its value so the next token is not
 // mistaken for a command, then explain where it went. Two named handlers rather
 // than a template over the message: taking the address of an internal-linkage
@@ -476,7 +496,18 @@ const CliOptionManager<TraceArgs>& traceOptions() {
                  "segmented.",
                  kSection(TraceHelpSection::Segmentation),
                  &parseUint64Option<&TraceArgs::segment_max_rows, 0>)
-
+            .add({"--roll-every"}, "<DUR>",
+                 "End the run and start a new part after this long, restarting "
+                 "at segment 0. Requires --segment-every: the part ends at the "
+                 "next segment boundary. Default: off",
+                 kSection(TraceHelpSection::Segmentation),
+                 &parseDurationOption<&TraceArgs::run_roll_every_ms>)
+            .add({"--roll-max-bytes"}, "<SIZE>",
+                 "Also end the run part after this much serialized telemetry, "
+                 "e.g. 50g. Counts every channel across the whole part, not "
+                 "per segment. Default: off",
+                 kSection(TraceHelpSection::Segmentation),
+                 &parseByteSizeOption<&TraceArgs::run_roll_max_bytes>)
             // Window
             .add({"--warmup"}, "<DUR>",
                  "Skip cold start: defer capture by this long (e.g. 30s, "

daemon/launcher/segmentation_env.cpp

@@ -16,7 +16,9 @@ bool applySegmentationEnv(const TraceArgs& args, const std::string& run_id,
     if (!segmentationRequested(args)) {
         return unsetEnvOrPrint(platform, env::kRunId) &&
                unsetEnvOrPrint(platform, env::kSegmentEveryMs) &&
-               unsetEnvOrPrint(platform, env::kSegmentMaxRows);
+               unsetEnvOrPrint(platform, env::kSegmentMaxRows) &&
+               unsetEnvOrPrint(platform, env::kRunRollEveryMs) &&
+               unsetEnvOrPrint(platform, env::kRunRollMaxBytes);
     }
 
     if (run_id.empty()) {
@@ -44,6 +46,24 @@ bool applySegmentationEnv(const TraceArgs& args, const std::string& run_id,
         return false;
     }
 
+    if (args.run_roll_every_ms > 0) {
+        if (!setEnvOrPrint(platform, env::kRunRollEveryMs,
+                           std::to_string(args.run_roll_every_ms))) {
+            return false;
+        }
+    } else if (!unsetEnvOrPrint(platform, env::kRunRollEveryMs)) {
+        return false;
+    }
+
+    if (args.run_roll_max_bytes > 0) {
+        if (!setEnvOrPrint(platform, env::kRunRollMaxBytes,
+                           std::to_string(args.run_roll_max_bytes))) {
+            return false;
+        }
+    } else if (!unsetEnvOrPrint(platform, env::kRunRollMaxBytes)) {
+        return false;
+    }
+
     return true;
 }
 

daemon/launcher/trace_command_common.cpp

@@ -726,6 +726,12 @@ int runTraceCommon(const TraceArgs& args, const TracePlatform& platform) {
         std::fprintf(stderr, "gpufl: %s\n", segmentation_error.c_str());
         return 2;
     }
+
+    if (const std::string warning = segmentationWarning(args);
+        !warning.empty()) {
+        std::fprintf(stderr, "[gpufl] warning: %s\n", warning.c_str());
+    }
+
     if (segmented && !segmentationRuntimeReady()) {
         std::fprintf(
             stderr,

include/gpufl/backends/nvidia/cupti_backend.cpp

@@ -255,7 +255,7 @@ void CuptiBackend::start() {
     last_sync_flush_launch_count_.store(0, std::memory_order_relaxed);
     {
         std::lock_guard lock(capture_capabilities_mu_);
-        capture_capabilities_segment_index_ = UINT32_MAX;
+        capture_capabilities_session_id_.clear();
         capability_kernel_rows_baseline_ = 0;
         capability_memory_rows_baseline_ = 0;
         capability_mem_transfer_rows_baseline_ = 0;

include/gpufl/backends/nvidia/cupti_backend.hpp

@@ -406,7 +406,7 @@ class CuptiBackend : public IMonitorBackend {
     // process-cumulative, so these baselines turn them into segment-local
     // deltas without resetting backend state at a sink cutover.
     mutable std::mutex capture_capabilities_mu_;
-    mutable uint32_t capture_capabilities_segment_index_ = UINT32_MAX;
+    mutable std::string capture_capabilities_session_id_;
     mutable uint64_t capability_kernel_rows_baseline_ = 0;
     mutable uint64_t capability_memory_rows_baseline_ = 0;
     mutable uint64_t capability_mem_transfer_rows_baseline_ = 0;

include/gpufl/backends/nvidia/cupti_capture_capabilities.cpp

@@ -17,7 +17,7 @@ void CuptiBackend::EmitCaptureCapabilities_() const {
     const auto segment = rt ? rt->acquireSegmentContext() : nullptr;
     if (!segment || !segment->logger) return;
     std::lock_guard capability_lock(capture_capabilities_mu_);
-    if (capture_capabilities_segment_index_ == segment->segment_index) return;
+    if  (capture_capabilities_session_id_ == segment->session_id) return;
 
     const auto delta = [](const uint64_t value, const uint64_t baseline) {
         return value >= baseline ? value - baseline : value;
@@ -148,7 +148,7 @@ void CuptiBackend::EmitCaptureCapabilities_() const {
     capability_launch_count_baseline_ = launch_count;
     capability_scope_truncated_baseline_ = truncated_total;
     capability_pm_rows_baseline_ = pm_rows_total;
-    capture_capabilities_segment_index_ = segment->segment_index;
+    capture_capabilities_session_id_ = segment->session_id;
 }
 
 }  // namespace gpufl

include/gpufl/core/deep_window_rule.cpp

@@ -362,6 +362,39 @@ void RuleEvaluator::enterBlackout(const int64_t) {
     ++state_sequence_;
 }
 
+void RuleEvaluator::beginRunPart() {
+    // A permanently dead rule stays dead: a fresh part cannot make an invalid
+    // or unsupported rule valid, so leave it Inactive and untouched.
+    if (terminal_ == RuleOutcome::InvalidConfig ||
+        terminal_ == RuleOutcome::Unsupported) {
+        return;
+    }
+
+    // Each run part is its own job: reset the window budget and the per-part
+    // summary counters. The rate baseline (source_), the cooldown (owned by the
+    // coordinator), and the current metric reading are deliberately preserved,
+    // because the workload is continuous across a roll.
+    const bool was_exhausted = (terminal_ == RuleOutcome::Exhausted);
+    windows_opened_ = 0;
+    samples_seen_ = 0;
+    truncated_samples_ = 0;
+    open_was_attempted_ = false;
+    terminal_ = RuleOutcome::None;
+    terminal_emitted_ = false;
+    reason_.clear();
+
+    if (was_exhausted) {
+        // It was parked in Inactive after spending its budget; re-arm so the
+        // new part's budget can be used. toArmed clears the sustained span and
+        // advances the sequence.
+        toArmed();
+    } else {
+        // The live state machine and its in-flight sustained span carry across
+        // the roll untouched; advance the sequence so the next summary orders.
+        ++state_sequence_;
+    }
+}
+
 void RuleEvaluator::poll(const int64_t now_ns) {
     if (state_ == RuleState::Inactive) return;
 

include/gpufl/core/deep_window_rule.hpp

@@ -254,6 +254,18 @@ class RuleEvaluator {
     /** @brief Current summary without concluding, for a mid-run emit. */
     RuleSummary snapshot(int64_t now_ns) const;
 
+    /**
+     * @brief Start a new run part after a rollover.
+     *
+     * Resets the window budget, terminal outcome, and per-part summary counters,
+     * re-arming a rule that had spent its budget so the next part gets its own.
+     * The live condition tracking - the rate baseline (in the source), the
+     * cooldown (owned by the coordinator), and the in-flight sustained-condition
+     * span - is preserved, because the workload continues across a roll. A
+     * permanently dead rule (invalid / unsupported) is left disabled.
+     */
+    void beginRunPart();
+
     /**
      * @brief True once, when a terminal outcome is first reached.
      *

include/gpufl/core/deep_window_rules.cpp

@@ -492,6 +492,13 @@ void DeepWindowRules::SnapshotSegment() {
     EmitCounterQuality();
 }
 
+void DeepWindowRules::BeginRunPart() {
+    std::lock_guard lk(g_mu);
+    if (g_installed && !g_finished && g_eval) {
+        g_eval->beginRunPart();
+    }
+}
+
 void DeepWindowRules::Finish() {
     RuleSummary summary;
     std::string expression;

include/gpufl/core/deep_window_rules.hpp

@@ -70,6 +70,15 @@ class DeepWindowRules {
      */
     static void SnapshotSegment();
 
+    /**
+     * Reset the rule's window budget and per-part summary for a new run part,
+     * re-arming a rule that had spent its budget. Called on a rollover cut,
+     * after SnapshotSegment has emitted the retiring part's summary. The
+     * evaluator's live condition tracking is preserved; a permanently dead rule
+     * (invalid / unsupported) stays disabled.
+     */
+    static void BeginRunPart();
+
     /** @brief True when a rule is installed - valid or refused. */
     static bool Installed();
 

include/gpufl/core/env_vars.hpp

@@ -136,6 +136,12 @@ constexpr const char* kRunId          = "GPUFL_RUN_ID";
 constexpr const char* kSegmentEveryMs = "GPUFL_SEGMENT_EVERY_MS";
 constexpr const char* kSegmentMaxRows = "GPUFL_SEGMENT_MAX_ROWS";
 
+// Run rollover: when a run part exceeds either budget, the run ends at the next
+// segment boundary and a fresh run part starts over at segment_index 0.
+// Both budgets are per run PART, not per segment.
+constexpr auto kRunRollEveryMs = "GPUFL_RUN_ROLL_EVERY_MS";
+constexpr auto kRunRollMaxBytes = "GPUFL_RUN_ROLL_MAX_BYTES";
+
 // PC sampling knobs ───────────────────────────────────────────────────────
 // Kernel-timeline collection strategy for the standalone PcSampling pass:
 //   "none" - PC samples only; PC/SASS kernel rows come from launch callbacks