fix: cursor issue for the already sent and cursor logic fix

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

@@ -385,9 +385,12 @@ public void tail(Publisher publisher) {
                         continue;
                     }
                     if (!activeFile().exists()) {
-                        idx = 0;
-                        offset = 0;
-                        cursorMgr.update(streamKey, idx, offset);
+                        // Session finished. Leave the cursor at this drained file's
+                        // EOF (drainGz already persisted {idx, endOffset, identity})
+                        // so a restart resumes HERE, skips to EOF and exits. Do NOT
+                        // reset to offset 0 - that keeps the file identity but wipes
+                        // progress, so the next run re-locates this file by content
+                        // signature and re-uploads the whole session (backend 409s it).
                         System.out.println("[" + logType + "] No active file remains; session finished.");
                         return;
                     }
@@ -514,9 +517,10 @@ else if (rotated != null && (isGz(rotated) || rotated.length() >= offset)) {
                                 continue;
                             }
                             if (!activeFile().exists()) {
-                                idx = 0;
-                                offset = 0;
-                                cursorMgr.update(streamKey, idx, offset);
+                                // Session finished. Leave the cursor at this drained
+                                // file's EOF (updated above to {idx, offset, identity})
+                                // so a restart skips to EOF and exits. Do NOT reset to
+                                // offset 0 - that re-uploads the whole session next run.
                                 System.out.println("[" + logType + "] No active file remains; session finished.");
                                 return;
                             }

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

@@ -102,6 +102,15 @@ static void main(String[] args) throws Exception {
             if (!startedSessions.add(key)) return;
             System.out.println("[agent] Tailing session \"" + session.sessionId()
                                + "\" in " + session.folder() + " types=" + session.logTypes());
+            // Count this session's per-channel tailers. When the LAST one finishes
+            // NATURALLY (session drained + every batch accepted), tell the backend the
+            // upload is complete so it can finalize without waiting out its grace
+            // window. A tailer that exits via interruption (agent shutdown) or never
+            // finishes (a channel whose file never appears) does NOT count toward the
+            // signal — so a partial upload is never declared complete; the backend's
+            // grace path covers those.
+            var remaining = new java.util.concurrent.atomic.AtomicInteger(session.logTypes().size());
+            String sid = session.sessionId();
             for (String type : session.logTypes()) {
                 executor.submit(() -> {
                     // Only the "system" channel carries device_metric_batch events.
@@ -110,6 +119,13 @@ static void main(String[] args) throws Exception {
                                                topicPrefix, cursorMgr, consumedFilesQueue, dedup,
                                                streamUploadSettings);
                     tailer.tail(publisher);
+                    // Reached only when tail() returns. Skip on interruption (agent
+                    // shutdown) so an in-flight upload is never declared complete.
+                    if (!Thread.currentThread().isInterrupted()
+                            && remaining.decrementAndGet() == 0
+                            && streamUploadSettings.enabled()) {
+                        signalSessionComplete(publisher, sid);
+                    }
                 });
             }
         };
@@ -176,6 +192,33 @@ static void main(String[] args) throws Exception {
         new CountDownLatch(1).await();
     }
 
+    /**
+     * Tell the backend that every channel of a session has finished uploading.
+     * Best-effort with a few bounded retries: {@code publishSessionComplete}
+     * returns true for terminal outcomes (2xx, or a 4xx like 404 from an older
+     * backend) and false only for transient 5xx/network errors. If it never gets
+     * through, the backend's grace path still finalizes the session, so this is
+     * a pure latency optimization — never a correctness dependency.
+     */
+    private static void signalSessionComplete(Publisher publisher, String sessionId) {
+        for (int attempt = 1; attempt <= 3; attempt++) {
+            try {
+                if (publisher.publishSessionComplete(sessionId)) return;
+            } catch (Exception e) {
+                System.err.println("[agent] session-complete signal error for "
+                                   + sessionId + ": " + e.getMessage());
+            }
+            try {
+                Thread.sleep(2000L);
+            } catch (InterruptedException ie) {
+                Thread.currentThread().interrupt();
+                return;
+            }
+        }
+        System.out.println("[agent] session-complete signal gave up for " + sessionId
+                           + " - backend grace finalize will apply");
+    }
+
     // -------------------------------------------------------------------------
     // Config resolution
     // -------------------------------------------------------------------------

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

@@ -106,6 +106,23 @@ public boolean publishStream(String sessionId, List<String> ndjsonLines) {
                     + response.statusCode() + " session=" + sessionId);
                 return true;
             }
+            if (response.statusCode() == 409) {
+                // Session already finalized on the backend - typically this agent
+                // re-tailing a finished session after a restart. The data is
+                // already stored and re-profiling is required to replace it, so
+                // retrying NEVER succeeds. Advance like a 2xx (mirrors the 402
+                // handling in publish()) so the tailer drains to EOF and exits
+                // instead of re-POSTing the same batch every 5s forever.
+                System.out.println("[agent] HTTP stream POST: session already uploaded (409) - "
+                    + "skipping, session=" + sessionId);
+                return true;
+            }
+            if (response.statusCode() == 402) {
+                // GPU/workspace limit exceeded - permanent; retrying won't help.
+                System.err.println("[GPUFL] GPU limit exceeded for this workspace - "
+                    + "skipping session=" + sessionId + ". " + response.body());
+                return true;
+            }
             System.out.println("HTTP stream publish failed [" + url + "]: "
                 + response.statusCode() + " - " + response.body());
             return false;
@@ -118,6 +135,45 @@ public boolean publishStream(String sessionId, List<String> ndjsonLines) {
         }
     }
 
+    @Override
+    public boolean publishSessionComplete(String sessionId) {
+        try {
+            String url = config.endpointFor("session-complete");
+            System.out.println("[agent] HTTP session-complete POST starting: url=" + url
+                + " session=" + sessionId);
+
+            var requestBuilder = HttpRequest.newBuilder()
+                .uri(URI.create(url))
+                .header("X-GpuFlight-Session-Id", sessionId)
+                .POST(HttpRequest.BodyPublishers.noBody());
+
+            addAuthHeader(requestBuilder);
+
+            var response = client.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
+            int sc = response.statusCode();
+            if (sc >= 200 && sc <= 299) {
+                System.out.println("[agent] HTTP session-complete accepted: status=" + sc
+                    + " session=" + sessionId);
+                return true;
+            }
+            if (sc >= 400 && sc < 500) {
+                // Old backend without the endpoint (404), or a client error: nothing
+                // to retry — the backend's grace path still finalizes the session.
+                System.out.println("[agent] HTTP session-complete not applicable (status=" + sc
+                    + ") session=" + sessionId + " - backend grace finalize will apply");
+                return true;
+            }
+            System.out.println("HTTP session-complete failed [" + url + "]: " + sc + " - " + response.body());
+            return false;   // 5xx: transient, worth a bounded retry
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return false;
+        } catch (Exception e) {
+            System.out.println("HTTP session-complete connection error: " + e);
+            return false;
+        }
+    }
+
     private void addAuthHeader(HttpRequest.Builder requestBuilder) {
         if (config.authToken() != null && !config.authToken().isBlank()) {
             requestBuilder.header("Authorization", "Bearer " + config.authToken());

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

@@ -13,4 +13,19 @@ public interface Publisher extends Closeable {
     default boolean publishStream(String sessionId, List<String> ndjsonLines) {
         return false;
     }
+
+    /**
+     * Signal the backend that EVERY channel of {@code sessionId} has finished
+     * uploading — the agent has drained all per-channel tailers and every batch
+     * was accepted. Lets the backend finalize the session immediately instead of
+     * waiting out its conservative grace/settle window.
+     *
+     * <p>Returns true when the signal was delivered OR is pointless to retry
+     * (2xx, or a 4xx like 404 from an older backend without the endpoint); false
+     * only on a transient failure (5xx / network) worth retrying. Default no-op
+     * (e.g. Kafka, where there is no such finalize step) returns true.
+     */
+    default boolean publishSessionComplete(String sessionId) {
+        return true;
+    }
 }

src/test/java/com/gpuflight/agent/LogTailerTest.java

@@ -632,4 +632,53 @@ void tail_compressedActive_resumesFromOffset(@TempDir Path tempDir) throws Excep
         assertEquals(1, publisher.events.size(), "Only the tail after the offset should be read");
         assertTrue(publisher.events.get(0).data().contains("k2"));
     }
+
+    // ---- A finished session is not re-uploaded when the agent restarts ----
+
+    @Test
+    void tail_finishedRotatedGzSession_notReuploadedOnRestart(@TempDir Path tempDir) throws Exception {
+        // A finished session whose channel rotated: only <channel>.1.log.gz
+        // remains (no active <channel>.log). Run 1 drains it fully and tail()
+        // returns. Run 2 over the SAME files + cursor must NOT re-publish: the
+        // cursor left at the gz's EOF makes the restart skip-to-EOF and exit.
+        //
+        // Regression: the terminal "session finished" branch used to reset the
+        // cursor to offset 0 while keeping the file identity, so run 2 re-located
+        // the gz by content signature and re-uploaded the whole session - which
+        // the backend then rejected with 409 "session already uploaded", looping
+        // forever.
+        Files.createDirectories(tempDir.resolve("app"));
+        String content =
+            "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k1\"}\n" +
+            "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k2\"}\n";
+        Path plain = tempDir.resolve("app/device.1.log");
+        Files.writeString(plain, content);
+        Path gz = tempDir.resolve("app/device.1.log.gz");
+        gzip(plain, gz);
+        Files.delete(plain); // finished: only the rotated .gz remains, no active .log
+
+        File cursorFile = tempDir.resolve("cursor.json").toFile();
+
+        // Run 1: drains the gz fully; tail() returns on "session finished".
+        CapturingStreamPublisher pub1 = new CapturingStreamPublisher();
+        LogTailer tailer1 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace",
+                new CursorManager(cursorFile), new LinkedBlockingQueue<>(), null,
+                new StreamUploadSettings(true, 10, 1_000_000L));
+        Thread t1 = startTailer(tailer1, pub1);
+        awaitEvents(pub1.batches, 1, 3000);
+        t1.join(3000); // tail() exits by itself once the finished session is drained
+        assertFalse(t1.isAlive(), "tailer should exit on its own after draining a finished session");
+        assertEquals(1, pub1.batches.size(), "run 1 publishes the finished session once");
+
+        // Run 2: same files, same persisted cursor. Must NOT re-publish anything.
+        CapturingStreamPublisher pub2 = new CapturingStreamPublisher();
+        LogTailer tailer2 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace",
+                new CursorManager(cursorFile), new LinkedBlockingQueue<>(), null,
+                new StreamUploadSettings(true, 10, 1_000_000L));
+        Thread t2 = startTailer(tailer2, pub2);
+        t2.join(3000); // should skip-to-EOF and exit immediately
+        assertFalse(t2.isAlive(), "restart of a finished session should exit immediately");
+        assertTrue(pub2.batches.isEmpty(),
+                "a finished session must NOT be re-uploaded on restart (got " + pub2.batches + ")");
+    }
 }

src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java

@@ -143,6 +143,48 @@ void publishStream_sendsGzippedNdjsonBatch() {
         assertTrue(lastBody.get().endsWith("\n"));
     }
 
+    @Test
+    void publishStream_alreadyUploaded409_advancesAsAccepted() {
+        // 409 "session already uploaded" is terminal - retrying never succeeds.
+        // publishStream must return true (advance) so the tailer drains to EOF and
+        // exits instead of re-POSTing the same batch every 5s forever.
+        statusToReturn.set(409);
+        HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L);
+        HttpPublisher pub = new HttpPublisher(config);
+
+        boolean ok = pub.publishStream("session-1",
+            List.of("{\"type\":\"kernel_event\",\"session_id\":\"session-1\"}"));
+
+        assertTrue(ok, "409 already-uploaded must advance the cursor, not retry forever");
+    }
+
+    @Test
+    void publishStream_limitExceeded402_advancesAsAccepted() {
+        // 402 (workspace/GPU limit) is permanent; retrying won't help.
+        statusToReturn.set(402);
+        HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L);
+        HttpPublisher pub = new HttpPublisher(config);
+
+        boolean ok = pub.publishStream("session-1",
+            List.of("{\"type\":\"kernel_event\",\"session_id\":\"session-1\"}"));
+
+        assertTrue(ok, "402 limit-exceeded must advance, not retry forever");
+    }
+
+    @Test
+    void publishStream_serverError500_signalsRetry() {
+        // A transient 5xx must remain retriable (return false) so the tailer backs
+        // off and retries from the same offset.
+        statusToReturn.set(500);
+        HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L);
+        HttpPublisher pub = new HttpPublisher(config);
+
+        boolean ok = pub.publishStream("session-1",
+            List.of("{\"type\":\"kernel_event\",\"session_id\":\"session-1\"}"));
+
+        assertFalse(ok, "5xx is transient and must be retried");
+    }
+
     @Test
     void publish_nonSuccessResponse_doesNotThrow() throws InterruptedException {
         statusToReturn.set(500);
@@ -169,4 +211,37 @@ void close_doesNotThrow() {
         HttpPublisher pub = new HttpPublisher(config);
         assertDoesNotThrow(pub::close);
     }
+
+    // ── publishSessionComplete (agent "all channels uploaded" signal) ──
+
+    @Test
+    void publishSessionComplete_postsToEndpointWithSessionHeader_andReturnsTrueOn200() {
+        statusToReturn.set(200);
+        HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5);
+        HttpPublisher pub = new HttpPublisher(config);
+
+        boolean ok = pub.publishSessionComplete("session-xyz");
+
+        assertTrue(ok);
+        assertEquals("/api/v1/events/session-complete", lastPath.get());
+        assertEquals("session-xyz", lastSessionId.get());
+    }
+
+    @Test
+    void publishSessionComplete_oldBackend404_returnsTrueSoAgentDoesNotRetry() {
+        // An older backend without the endpoint 404s; the agent must treat it as
+        // terminal (the grace path still finalizes the session) and not loop.
+        statusToReturn.set(404);
+        HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5);
+        boolean ok = new HttpPublisher(config).publishSessionComplete("session-xyz");
+        assertTrue(ok);
+    }
+
+    @Test
+    void publishSessionComplete_serverError500_returnsFalseForRetry() {
+        statusToReturn.set(500);
+        HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5);
+        boolean ok = new HttpPublisher(config).publishSessionComplete("session-xyz");
+        assertFalse(ok);
+    }
 }