Session segmentation launcher

CMakeLists.txt

@@ -177,6 +177,8 @@ target_sources(gpufl PRIVATE
     include/gpufl/core/deep_window.cpp
     include/gpufl/core/sampler.cpp
     include/gpufl/core/runtime.cpp
+    include/gpufl/core/segment_coordinator.cpp
+    include/gpufl/core/segment_runtime.cpp
     include/gpufl/core/backend_factory.cpp
     include/gpufl/core/monitor_adapter.cpp
     include/gpufl/core/nvtx_counters.cpp

daemon/launcher/CMakeLists.txt

@@ -25,6 +25,7 @@ add_executable(gpufl_launcher
     info_command.cpp
     trace_command_common.cpp
     deep_window_env.cpp
+    segmentation_env.cpp
     ${GPUFL_LAUNCHER_TRACE_IMPL}
     monitor_command.cpp
     ../monitor/monitor_runner.cpp

daemon/launcher/cli_parse.cpp

@@ -1,8 +1,18 @@
 #include "cli_parse.hpp"
 
 #include <algorithm>
+#include <array>
+#include <cerrno>
+#include <chrono>
+#include <cmath>
 #include <cstdlib>
 #include <cstring>
+#include <iomanip>
+#include <limits>
+#include <random>
+#include <sstream>
+
+#include "gpufl/core/segmentation_config.hpp"
 
 namespace gpufl::launcher {
 
@@ -48,16 +58,26 @@ std::string trim(const std::string& s) {
 bool parseDurationMs(const std::string& s, int64_t& out_ms) {
     if (s.empty()) return false;
     char* end = nullptr;
+    errno = 0;
     const double v = std::strtod(s.c_str(), &end);
-    if (end == s.c_str() || v < 0) return false;
+    if (end == s.c_str() || errno == ERANGE || !std::isfinite(v) || v < 0) {
+        return false;
+    }
     std::string unit = trim(end);
     double mult_ms;  // value * mult_ms = milliseconds
     if (unit.empty() || unit == "s") mult_ms = 1000.0;
     else if (unit == "ms") mult_ms = 1.0;
     else if (unit == "m") mult_ms = 60.0 * 1000.0;
     else if (unit == "h") mult_ms = 60.0 * 60.0 * 1000.0;
     else return false;
-    out_ms = static_cast<int64_t>(v * mult_ms);
+    const double milliseconds = v * mult_ms;
+    if (!std::isfinite(milliseconds) ||
+        milliseconds >= static_cast<double>(
+            (std::numeric_limits<int64_t>::max)()) ||
+        (v > 0 && milliseconds < 1.0)) {
+        return false;
+    }
+    out_ms = static_cast<int64_t>(milliseconds);
     return true;
 }
 
@@ -159,6 +179,14 @@ const char* traceHelp() {
         "        --agent-drain-ms=<MS>\n"
         "                            Max wait for the agent to finish uploading before\n"
         "                            stopping it (it exits on its own when done). Default: 60000\n"
+        "        --segment-every=<DUR>\n"
+        "                            Split a long run on this cadence (minimum: 60s).\n"
+        "                            Example: --segment-every=5m. Default: off\n"
+        "        --segment-max-rows=<N>\n"
+        "                            Also split after this many logical telemetry rows.\n"
+        "                            The batch crossing N stays in the old segment.\n"
+        "                            Default: off. V1 supports one Trace or PM pass;\n"
+        "                            multi-pass analyses cannot be segmented.\n"
         "        --warmup=<DUR>      Skip cold start: defer capture by this long\n"
         "                            (e.g. 30s, 500ms, 5m; bare number = seconds)\n"
         "        --window=<DUR>      Bounded window: capture this long after warmup,\n"
@@ -351,6 +379,40 @@ TraceParseResult parseTraceArgs(const std::vector<std::string>& argv) {
                         "invalid --agent-drain-ms value: " + v +
                         " (expected a non-negative integer, milliseconds)"};
             }
+        } else if (key == "--segment-every") {
+            std::string v;
+            auto err = take_value(v);
+            if (!err.empty()) return {std::nullopt, err};
+            if (!parseDurationMs(v, out.segment_every_ms)) {
+                return {std::nullopt,
+                        "invalid --segment-every value: " + v +
+                        " (expected a duration like 60s, 5m, 1h, or a bare "
+                        "number of seconds)"};
+            }
+            if (out.segment_every_ms > 0 &&
+                out.segment_every_ms < kMinSegmentEveryMs) {
+                return {std::nullopt,
+                        "--segment-every must be at least 60s; shorter cadences "
+                        "can create a session storm"};
+            }
+        } else if (key == "--segment-max-rows") {
+            std::string v;
+            auto err = take_value(v);
+            if (!err.empty()) return {std::nullopt, err};
+            if (v.empty() || v.front() == '-') {
+                return {std::nullopt,
+                        "invalid --segment-max-rows value: " + v +
+                        " (expected a non-negative integer; 0 disables it)"};
+            }
+            char* end = nullptr;
+            errno = 0;
+            const unsigned long long n = std::strtoull(v.c_str(), &end, 10);
+            if (end == v.c_str() || (end && *end != '\0') || errno == ERANGE) {
+                return {std::nullopt,
+                        "invalid --segment-max-rows value: " + v +
+                        " (expected a non-negative integer; 0 disables it)"};
+            }
+            out.segment_max_rows = static_cast<uint64_t>(n);
         } else if (key == "--pc-sample-period") {
             std::string v;
             auto err = take_value(v);
@@ -815,16 +877,92 @@ InfoParseResult parseInfoArgs(const std::vector<std::string>& argv) {
 }
 
 std::string validateTraceExecutionMode(const TraceArgs& args) {
-    if (!args.deep_requested || args.passes.empty()) return {};
-    return "--passes cannot be combined with --deep-* flags.\n"
-           "\n"
-           "  --passes runs the engines you name, relaunching the target "
-           "once per pass.\n"
-           "  --deep-* runs ONE adaptive pass: gpufl selects a compatible "
-           "deep engine\n"
-           "  and arms it only inside the window.\n"
-           "\n"
-           "Drop --passes to use a deep window.";
+    if (args.deep_requested && !args.passes.empty()) {
+        return "--passes cannot be combined with --deep-* flags.\n"
+               "\n"
+               "  --passes runs the engines you name, relaunching the target "
+               "once per pass.\n"
+               "  --deep-* runs ONE adaptive pass: gpufl selects a compatible "
+               "deep engine\n"
+               "  and arms it only inside the window.\n"
+               "\n"
+               "Drop --passes to use a deep window.";
+    }
+    return validateTraceSegmentation(args);
+}
+
+bool segmentationRequested(const TraceArgs& args) {
+    return args.segment_every_ms > 0 || args.segment_max_rows > 0;
+}
+
+std::string validateTraceSegmentation(
+    const TraceArgs& args,
+    const std::string& inherited_analysis_id) {
+    if (args.segment_every_ms < 0) {
+        return "--segment-every cannot be negative";
+    }
+    if (args.segment_every_ms > 0 &&
+        args.segment_every_ms < kMinSegmentEveryMs) {
+        return "--segment-every must be at least 60s; shorter cadences can "
+               "create a session storm";
+    }
+    if (!segmentationRequested(args)) return {};
+
+    if (!inherited_analysis_id.empty()) {
+        return "session segmentation cannot be combined with an inherited "
+               "GPUFL_ANALYSIS_ID; unset GPUFL_ANALYSIS_ID before launching "
+               "the target";
+    }
+    if (args.passes.size() > 1) {
+        return "session segmentation cannot be combined with a multi-pass "
+               "--passes list; segmented runs concatenate time while analysis "
+               "passes overlay the same interval";
+    }
+    if (!args.passes.empty()) {
+        const std::string& pass = args.passes.front();
+        if (pass != "Trace" && pass != "PmSampling") {
+            return "session segmentation V1 supports only a single Trace or "
+                   "PmSampling pass. Unsupported pass: " + pass;
+        }
+    }
+
+    // No explicit pass plus --deep-* is the one supported composite: the
+    // launcher pins native Trace as the base and prepares window-only PM.
+    // No explicit pass and no deep flags is the ordinary single Trace pass.
+    return {};
+}
+
+std::string generateRunId() {
+    std::array<uint8_t, 16> bytes{};
+    static thread_local std::mt19937_64 rng([] {
+        std::random_device rd;
+        std::seed_seq seed{
+            rd(), rd(), rd(), rd(),
+            static_cast<unsigned>(
+                std::chrono::steady_clock::now().time_since_epoch().count())};
+        return std::mt19937_64(seed);
+    }());
+    for (size_t i = 0; i < bytes.size(); i += sizeof(uint64_t)) {
+        const uint64_t word = rng();
+        for (size_t j = 0; j < sizeof(uint64_t); ++j) {
+            bytes[i + j] = static_cast<uint8_t>(word >> (j * 8));
+        }
+    }
+
+    bytes[6] = static_cast<uint8_t>((bytes[6] & 0x0f) | 0x40);
+    bytes[8] = static_cast<uint8_t>((bytes[8] & 0x3f) | 0x80);
+
+    std::ostringstream out;
+    out << std::hex << std::setfill('0');
+    for (size_t i = 0; i < bytes.size(); ++i) {
+        if (i == 4 || i == 6 || i == 8 || i == 10) out << '-';
+        out << std::setw(2) << static_cast<unsigned>(bytes[i]);
+    }
+    return out.str();
+}
+
+bool segmentationRuntimeReady() {
+    return gpufl::segmentation::kRuntimeReady;
 }
 
 CaptureMode resolveCaptureMode(const TraceArgs& args) {

daemon/launcher/cli_parse.hpp

@@ -52,6 +52,10 @@ struct TraceArgs {
     std::string deep_when;
     int64_t  deep_cooldown_ms = 0;   // --deep-cooldown; quiet time between windows
     bool     deep_requested = false; // any --deep-* flag was given
+    // Long-running run segmentation. Zero disables the corresponding trigger;
+    // 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
     // 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.
@@ -138,6 +142,33 @@ TraceParseResult parseTraceArgs(const std::vector<std::string>& argv);
  */
 std::string validateTraceExecutionMode(const TraceArgs& args);
 
+// The shortest user-configurable time cadence. This protects the backend from
+// an accidental session storm; unit tests of the future coordinator use a fake
+// clock instead of weakening this production CLI bound.
+constexpr int64_t kMinSegmentEveryMs = 60'000;
+
+/** True when at least one segmentation trigger is enabled. */
+bool segmentationRequested(const TraceArgs& args);
+
+/**
+ * Validate segmentation-specific mode restrictions. inherited_analysis_id is
+ * supplied by the execution boundary so an exported GPUFL_ANALYSIS_ID cannot
+ * silently turn a segmented single-pass run into an invalid two-axis run.
+ */
+std::string validateTraceSegmentation(
+    const TraceArgs& args,
+    const std::string& inherited_analysis_id = std::string());
+
+/** Generate the launcher-owned UUIDv4 shared by every segment in one run. */
+std::string generateRunId();
+
+/**
+ * False until SegmentCoordinator cutover is implemented. Keeping this as an
+ * explicit execution-boundary gate lets the parser/wire contract land without
+ * exposing a flag that claims to split sessions but silently produces one.
+ */
+bool segmentationRuntimeReady();
+
 /**
  * How a run decides which engines to select - two modes, never mixed.
  *

daemon/launcher/segmentation_env.cpp

@@ -0,0 +1,50 @@
+// Publishing the long-running session-segmentation contract into the target's
+// environment. Kept separate from trace_command_common so ownership and stale
+// environment scrubbing are unit-testable without launching a process.
+
+#include <cstdio>
+#include <string>
+
+#include "cli_parse.hpp"
+#include "gpufl/core/env_vars.hpp"
+#include "trace_command_common.hpp"
+
+namespace gpufl::launcher {
+
+bool applySegmentationEnv(const TraceArgs& args, const std::string& run_id,
+                          const TracePlatform& platform) {
+    if (!segmentationRequested(args)) {
+        return unsetEnvOrPrint(platform, env::kRunId) &&
+               unsetEnvOrPrint(platform, env::kSegmentEveryMs) &&
+               unsetEnvOrPrint(platform, env::kSegmentMaxRows);
+    }
+
+    if (run_id.empty()) {
+        std::fprintf(stderr,
+                     "gpufl: internal error: segmented run has no GPUFL_RUN_ID\n");
+        return false;
+    }
+    if (!setEnvOrPrint(platform, env::kRunId, run_id)) return false;
+
+    if (args.segment_every_ms > 0) {
+        if (!setEnvOrPrint(platform, env::kSegmentEveryMs,
+                           std::to_string(args.segment_every_ms))) {
+            return false;
+        }
+    } else if (!unsetEnvOrPrint(platform, env::kSegmentEveryMs)) {
+        return false;
+    }
+
+    if (args.segment_max_rows > 0) {
+        if (!setEnvOrPrint(platform, env::kSegmentMaxRows,
+                           std::to_string(args.segment_max_rows))) {
+            return false;
+        }
+    } else if (!unsetEnvOrPrint(platform, env::kSegmentMaxRows)) {
+        return false;
+    }
+
+    return true;
+}
+
+}  // namespace gpufl::launcher

daemon/launcher/trace_command_common.cpp

@@ -537,6 +537,21 @@ int runTraceCommon(const TraceArgs& args, const TracePlatform& platform) {
         return 2;
     }
 
+    const bool segmented = segmentationRequested(args);
+    const char* inherited_analysis = std::getenv(env::kAnalysisId);
+    if (const std::string segmentation_error = validateTraceSegmentation(
+            args, inherited_analysis ? inherited_analysis : "");
+        !segmentation_error.empty()) {
+        std::fprintf(stderr, "gpufl: %s\n", segmentation_error.c_str());
+        return 2;
+    }
+    if (segmented && !segmentationRuntimeReady()) {
+        std::fprintf(
+            stderr,
+            "gpufl: this build does not include executable session segmentation\n");
+        return 2;
+    }
+
     const fs::path exe = platform.selfExe();
     if (exe.empty()) {
         std::fprintf(stderr, "gpufl: cannot resolve launcher path (%s)\n",
@@ -560,7 +575,9 @@ int runTraceCommon(const TraceArgs& args, const TracePlatform& platform) {
     const bool multipass = plan.size() > 1;
 
     const std::string analysis_id = multipass ? makeAnalysisId() : std::string();
-    const std::string dir_tag = multipass ? analysis_id : makeSessionId();
+    const std::string run_id = segmented ? generateRunId() : std::string();
+    const std::string dir_tag =
+        multipass ? analysis_id : (segmented ? run_id : makeSessionId());
 
     const std::string app_name = args.name.empty()
                                ? platform.defaultAppName(args.command.front())
@@ -632,6 +649,7 @@ int runTraceCommon(const TraceArgs& args, const TracePlatform& platform) {
     }
 
     if (!applyDeepWindowEnv(args, platform)) return 2;
+    if (!applySegmentationEnv(args, run_id, platform)) return 2;
 
     // A bounded window stops the target after warmup+window wall-clock;
     // run_ms == 0 keeps the historical "run until the target exits" behavior.
@@ -706,6 +724,19 @@ int runTraceCommon(const TraceArgs& args, const TracePlatform& platform) {
             for (const auto& e : plan) std::fprintf(stderr, " %s", e.c_str());
             std::fputc('\n', stderr);
         }
+        if (segmented) {
+            std::fprintf(stderr, "[gpufl] segmented run %s:", run_id.c_str());
+            if (args.segment_every_ms > 0) {
+                std::fprintf(stderr, " every=%lldms",
+                             static_cast<long long>(args.segment_every_ms));
+            }
+            if (args.segment_max_rows > 0) {
+                std::fprintf(stderr, " max_rows=%llu",
+                             static_cast<unsigned long long>(
+                                 args.segment_max_rows));
+            }
+            std::fputc('\n', stderr);
+        }
         if (args.verbose) {
             std::fprintf(stderr, "[gpufl] inject lib: %s\n",
                          inject_lib.string().c_str());

daemon/launcher/trace_command_common.hpp

@@ -82,6 +82,15 @@ bool unsetEnvOrPrint(const TracePlatform& platform, const char* key);
  */
 bool applyDeepWindowEnv(const TraceArgs& args, const TracePlatform& platform);
 
+/**
+ * @brief Publish or scrub the launcher-owned segmentation environment.
+ *
+ * A non-segmented invocation removes all three internal variables so stale
+ * parent-shell state cannot turn an ordinary trace into a segmented run.
+ */
+bool applySegmentationEnv(const TraceArgs& args, const std::string& run_id,
+                          const TracePlatform& platform);
+
 int runTraceCommon(const TraceArgs& args, const TracePlatform& platform);
 
 }  // namespace gpufl::launcher