fix the stream issue

.gitignore

@@ -47,7 +47,7 @@ bin/
 .DS_Store
 
 ## Local config
-config
+/config/
 
 cursor.json
 run-local.sh

src/main/java/com/gpuflight/agent/CursorPosition.java

@@ -14,8 +14,8 @@
  *
  * <p>Two identity signals, because the client both renames AND compresses on
  * rotation:
- *   - {@code fileKey} (inode on POSIX) survives a rename/shift (active →
- *     {@code .1.log} → {@code .2.log}) but NOT compression — gzip writes a new
+ *   - {@code fileKey} (inode on POSIX) survives a rename/shift (active ->
+ *     {@code .1.log} -> {@code .2.log}) but NOT compression - gzip writes a new
  *     file ({@code .1.log.gz}) with a different inode. Fast path on Linux while
  *     the rotated file is still uncompressed.
  *   - {@code headSig} (CRC32 of the first bytes of the *uncompressed* content)
@@ -24,7 +24,7 @@
  *     {@code .gz}.
  *
  * <p>Identity fields are nullable/zero and default that way, so cursor files
- * written before this change still deserialize cleanly (identity unknown →
+ * written before this change still deserialize cleanly (identity unknown ->
  * legacy behavior).
  */
 public record CursorPosition(
@@ -34,7 +34,7 @@ public record CursorPosition(
     @JsonProperty("fileSize")  long   fileSize,   // size at last update (sanity check)
     @JsonProperty("headSig")   long   headSig     // CRC32 of first uncompressed bytes; survives compression. 0 = unset.
 ) {
-    /** Backward-compatible constructor — identity unknown. */
+    /** Backward-compatible constructor - identity unknown. */
     public CursorPosition(int fileIndex, long offset) {
         this(fileIndex, offset, null, 0L, 0L);
     }

src/main/java/com/gpuflight/agent/LogArchiver.java

@@ -32,7 +32,7 @@ public void archive(Path file, String objectKey) {
             .build();
         // Blocking S3 upload is fine on a virtual thread.
         s3.putObject(request, file);
-        System.out.println("[archiver] Uploaded " + file.getFileName() + " → s3://" + config.bucket() + "/" + objectKey);
+        System.out.println("[archiver] Uploaded " + file.getFileName() + " -> s3://" + config.bucket() + "/" + objectKey);
         if (config.deleteAfterUpload()) {
             file.toFile().delete();
             System.out.println("[archiver] Deleted " + file.getFileName());

src/main/java/com/gpuflight/agent/config/HttpConfig.java

@@ -2,22 +2,24 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import java.util.Locale;
+
 /**
  * Configuration for the HTTP publisher.
  *
- * <p>The destination is split across {@code hostUrl} + {@code apiVersion} —
+ * <p>The destination is split across {@code hostUrl} + {@code apiVersion};
  * the full path is built internally via {@link #endpointFor(String)} so
- * the user only specifies what changes per-deployment (the host) and
- * occasionally per-rollout (the version). The {@code /api/{version}/events/}
+ * the user only specifies what changes per deployment (the host) and
+ * occasionally per rollout (the version). The {@code /api/{version}/events/}
  * path is a structural contract baked into this record.
  *
  * <p>Examples:
  * <pre>
  *   hostUrl="https://api.gpuflight.com"     apiVersion="v1"
- *     → endpointFor("init") = "https://api.gpuflight.com/api/v1/events/init"
+ *     endpointFor("init") = "https://api.gpuflight.com/api/v1/events/init"
  *
  *   hostUrl="http://localhost:8080"         apiVersion="v2"
- *     → endpointFor("init") = "http://localhost:8080/api/v2/events/init"
+ *     endpointFor("init") = "http://localhost:8080/api/v2/events/init"
  * </pre>
  *
  * <p><strong>Migration note (May 2026):</strong> previous releases used a
@@ -33,23 +35,53 @@ public record HttpConfig(
     @JsonProperty("hostUrl")        String hostUrl,
     @JsonProperty("apiVersion")     String apiVersion,
     @JsonProperty("authToken")      String authToken,
-    @JsonProperty("timeoutSeconds") long   timeoutSeconds
+    @JsonProperty("timeoutSeconds") long   timeoutSeconds,
+    @JsonProperty("uploadMode")     String uploadMode,
+    @JsonProperty("streamMaxLines") int    streamMaxLines,
+    @JsonProperty("streamMaxBytes") long   streamMaxBytes
 ) implements PublisherConfig {
 
     /** Default API version when the user doesn't specify one. */
     public static final String DEFAULT_API_VERSION = "v1";
+    public static final String DEFAULT_UPLOAD_MODE = "stream";
+    public static final int DEFAULT_STREAM_MAX_LINES = 5_000;
+    public static final long DEFAULT_STREAM_MAX_BYTES = 1_000_000L;
+
+    private static final String LEGACY_UPLOAD_MODE = "legacy";
+
+    public HttpConfig(String hostUrl, String apiVersion, String authToken, long timeoutSeconds) {
+        this(hostUrl, apiVersion, authToken, timeoutSeconds, DEFAULT_UPLOAD_MODE, 0, 0);
+    }
 
     public HttpConfig {
         // Field-level defaults / normalization. Run inside the compact
-        // constructor so they apply to BOTH the JSON-deserialized path
-        // (via Jackson) and any direct callers (tests, CLI plumbing).
-
+        // constructor so they apply to both JSON deserialization and direct callers.
         if (timeoutSeconds <= 0) {
             timeoutSeconds = 10L;
         }
         if (apiVersion == null || apiVersion.isBlank()) {
             apiVersion = DEFAULT_API_VERSION;
         }
+        if (uploadMode == null || uploadMode.isBlank()) {
+            uploadMode = DEFAULT_UPLOAD_MODE;
+        }
+        uploadMode = uploadMode.trim().toLowerCase(Locale.ROOT);
+        if (!DEFAULT_UPLOAD_MODE.equals(uploadMode) && !LEGACY_UPLOAD_MODE.equals(uploadMode)) {
+            throw new IllegalArgumentException(
+                "HttpConfig.uploadMode must be 'stream' or 'legacy', got: " + uploadMode);
+        }
+        if (DEFAULT_UPLOAD_MODE.equals(uploadMode)) {
+            if (streamMaxLines <= 0) {
+                streamMaxLines = DEFAULT_STREAM_MAX_LINES;
+            }
+            if (streamMaxBytes <= 0) {
+                streamMaxBytes = DEFAULT_STREAM_MAX_BYTES;
+            }
+        } else {
+            streamMaxLines = 0;
+            streamMaxBytes = 0L;
+        }
+
         // Hard reject blank/missing host with a migration-aware error.
         // Jackson silently sets unknown fields to null (we disabled
         // FAIL_ON_UNKNOWN_PROPERTIES in JsonSettings), so a config that
@@ -73,11 +105,19 @@ public record HttpConfig(
         }
     }
 
+    public boolean isStreamMode() {
+        return DEFAULT_UPLOAD_MODE.equals(uploadMode);
+    }
+
+    public String streamEndpoint() {
+        return endpointFor(DEFAULT_UPLOAD_MODE);
+    }
+
     /**
      * Build the full upload URL for a given event type.
      *
      * @param type the event-type segment, e.g. {@code "init"},
-     *             {@code "kernel"}, {@code "metric"} — appended after
+     *             {@code "kernel"}, {@code "metric"}; appended after
      *             the {@code /api/{version}/events/} prefix.
      */
     public String endpointFor(String type) {

src/main/java/com/gpuflight/agent/config/StreamUploadSettings.java

@@ -0,0 +1,24 @@
+package com.gpuflight.agent.config;
+
+/**
+ * Per-tailer batching knobs for stream upload mode. When {@code enabled}, the
+ * tailer collects lines into a batch and hands the whole batch to
+ * {@code Publisher.publishStream} - flushing when {@code maxLines} or
+ * {@code maxBytes} (uncompressed) is reached, or at EOF (no artificial
+ * latency: the moment the file has no more data, whatever is buffered ships).
+ * The cursor advances only after a flush is accepted, so a crash or rejected
+ * batch re-reads from the last accepted offset.
+ *
+ * <p>{@code DISABLED} keeps the original one-publish-per-line flow (legacy
+ * mode, Kafka, and existing tests).
+ */
+public record StreamUploadSettings(boolean enabled, int maxLines, long maxBytes) {
+
+    public static final StreamUploadSettings DISABLED = new StreamUploadSettings(false, 0, 0);
+
+    public static StreamUploadSettings from(HttpConfig http) {
+        return http.isStreamMode()
+            ? new StreamUploadSettings(true, http.streamMaxLines(), http.streamMaxBytes())
+            : DISABLED;
+    }
+}

src/main/java/com/gpuflight/agent/model/LogSourceConfig.java

@@ -20,6 +20,8 @@ public record LogSourceConfig(
 ) {
     public LogSourceConfig {
         if (folder == null) folder = ".";
-        if (logTypes == null || logTypes.isEmpty()) logTypes = List.of("device", "scope", "system");
+        // "sass" carries the SASS-disassembly / source-content artifacts split
+        // out of device.log; tail it by default or they miss live upload.
+        if (logTypes == null || logTypes.isEmpty()) logTypes = List.of("device", "scope", "system", "sass");
     }
 }

src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java

@@ -4,11 +4,16 @@
 import com.gpuflight.agent.config.JsonSettings;
 import com.gpuflight.agent.model.LogWrapper;
 
+import java.io.ByteArrayOutputStream;
+import java.net.InetAddress;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.List;
+import java.util.zip.GZIPOutputStream;
 
 public class HttpPublisher implements Publisher {
     private final HttpConfig config;
@@ -25,11 +30,15 @@ public HttpPublisher(HttpConfig config) {
     @Override
     public boolean publish(String topic, String key, LogWrapper log) {
         try {
-            // Path is structural — assembled inside HttpConfig from
+            // Path is structural - assembled inside HttpConfig from
             // hostUrl + apiVersion + event-type segment. See
             // HttpConfig.endpointFor() for the template.
             String url = config.endpointFor(log.type());
             String message = JsonSettings.MAPPER.writeValueAsString(log);
+            System.out.println("[agent] HTTP event POST starting: url=" + url
+                + " topic=" + topic
+                + " key=" + key
+                + " type=" + log.type());
 
             var requestBuilder = HttpRequest.newBuilder()
                 .uri(URI.create(url))
@@ -38,23 +47,23 @@ public boolean publish(String topic, String key, LogWrapper log) {
                 .header("X-Key", key)
                 .POST(HttpRequest.BodyPublishers.ofString(message));
 
-            if (config.authToken() != null) {
-                requestBuilder.header("Authorization", "Bearer " + config.authToken());
-            }
+            addAuthHeader(requestBuilder);
 
-            // Blocking send is fine on a virtual thread — the carrier thread parks, not blocks.
+            // Blocking send is fine on a virtual thread - the carrier thread parks, not blocks.
             var response = client.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
             if (response.statusCode() >= 200 && response.statusCode() <= 299) {
+                System.out.println("[agent] HTTP event POST accepted: status="
+                    + response.statusCode() + " type=" + log.type());
                 return true;
             }
             if (response.statusCode() == 402) {
-                // GPU limit exceeded — permanent failure, do not retry
+                // GPU limit exceeded - permanent failure, do not retry
                 System.err.println("[GPUFL] ========================================");
                 System.err.println("[GPUFL] GPU limit exceeded for this workspace.");
                 System.err.println("[GPUFL] " + response.body());
                 System.err.println("[GPUFL] Upgrade at https://gpuflight.com/pricing");
                 System.err.println("[GPUFL] ========================================");
-                return true; // advance cursor — retrying won't help
+                return true; // advance cursor - retrying won't help
             }
             System.out.println("HTTP Publish failed [" + url + "]: " + response.statusCode() + " - " + response.body());
             return false;
@@ -67,6 +76,62 @@ public boolean publish(String topic, String key, LogWrapper log) {
         }
     }
 
+    @Override
+    public boolean publishStream(String sessionId, List<String> ndjsonLines) {
+        if (ndjsonLines == null || ndjsonLines.isEmpty()) {
+            return true;
+        }
+        try {
+            String ndjson = String.join("\n", ndjsonLines) + "\n";
+            byte[] body = gzip(ndjson.getBytes(StandardCharsets.UTF_8));
+            String url = config.streamEndpoint();
+            System.out.println("[agent] HTTP stream POST starting: url=" + url
+                + " session=" + sessionId
+                + " lines=" + ndjsonLines.size()
+                + " gzipBytes=" + body.length);
+
+            var requestBuilder = HttpRequest.newBuilder()
+                .uri(URI.create(url))
+                .header("Content-Type", "application/x-ndjson")
+                .header("Content-Encoding", "gzip")
+                .header("X-GpuFlight-Session-Id", sessionId)
+                .header("X-GpuFlight-Hostname", InetAddress.getLocalHost().getHostName())
+                .POST(HttpRequest.BodyPublishers.ofByteArray(body));
+
+            addAuthHeader(requestBuilder);
+
+            var response = client.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
+            if (response.statusCode() >= 200 && response.statusCode() <= 299) {
+                System.out.println("[agent] HTTP stream POST accepted: status="
+                    + response.statusCode() + " session=" + sessionId);
+                return true;
+            }
+            System.out.println("HTTP stream publish failed [" + url + "]: "
+                + response.statusCode() + " - " + response.body());
+            return false;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return false;
+        } catch (Exception e) {
+            System.out.println("HTTP stream connection error: " + e);
+            return false;
+        }
+    }
+
+    private void addAuthHeader(HttpRequest.Builder requestBuilder) {
+        if (config.authToken() != null && !config.authToken().isBlank()) {
+            requestBuilder.header("Authorization", "Bearer " + config.authToken());
+        }
+    }
+
+    private static byte[] gzip(byte[] content) throws Exception {
+        var out = new ByteArrayOutputStream();
+        try (var gzip = new GZIPOutputStream(out)) {
+            gzip.write(content);
+        }
+        return out.toByteArray();
+    }
+
     @Override
     public void close() {
         // Java HttpClient has no explicit close; resources are reclaimed on GC.

src/main/java/com/gpuflight/agent/publisher/KafkaPublisher.java

@@ -27,7 +27,7 @@ public boolean publish(String topic, String key, LogWrapper log) {
         try {
             String message = JsonSettings.MAPPER.writeValueAsString(log.data());
             // producer.send() is async and only blocks when the internal buffer is full.
-            // Blocking on a virtual thread is fine — carrier thread parks, not blocks.
+            // Blocking on a virtual thread is fine - carrier thread parks, not blocks.
             producer.send(new ProducerRecord<>(topic, key, message)).get();
             return true;
         } catch (Exception e) {

src/main/java/com/gpuflight/agent/publisher/Publisher.java

@@ -3,8 +3,14 @@
 import com.gpuflight.agent.model.LogWrapper;
 
 import java.io.Closeable;
+import java.util.List;
 
 public interface Publisher extends Closeable {
     /** Returns true if the event was accepted by the backend, false on failure. */
     boolean publish(String topic, String key, LogWrapper log);
+
+    /** Returns true if the NDJSON batch was accepted by the backend. */
+    default boolean publishStream(String sessionId, List<String> ndjsonLines) {
+        return false;
+    }
 }