From adb08fc51d259028308acb63c14a0e46e0f6e8a7 Mon Sep 17 00:00:00 2001
From: Levi Neely <lneely@pascom.net>
Date: Tue, 6 Oct 2026 12:01:26 +0200
Subject: [PATCH] server: fix eventwait to use Stream instead of BlockOnce

EventValue was storing only the latest event and using hash comparison,
which caused events to be overwritten if they arrived faster than the
client could read them.

Now eventwait uses Stream mode with EventStream which delivers each
event as it arrives. Events won't be lost due to rapid arrival.
---
 cmd/olliesrv/internal/fs/spec.go       | 10 +++------
 cmd/olliesrv/internal/session/event.go | 28 +++++++++++++-------------
 kde/gui/ollie9pclient.cpp              |  3 ---
 3 files changed, 17 insertions(+), 24 deletions(-)

diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go
index d86eb71..e103ff9 100644
--- a/cmd/olliesrv/internal/fs/spec.go
+++ b/cmd/olliesrv/internal/fs/spec.go
@@ -117,7 +117,7 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
 	}
 
 	// Event subscription for eventwait (lives for server lifetime).
-	eventRead, eventSignal := session.EventValue(session.SubscribeEvents(cfg.Ctx, "*"))
+	eventRead, eventSignal := session.EventStream(session.SubscribeEvents(cfg.Ctx, "*"))
 
 	return virtfs.DirNode("/",
 		virtfs.UID(treeSpecUID),
@@ -240,12 +240,8 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
 			}),
 		),
 		virtfs.FileNode("eventwait", 0444,
-			virtfs.Doc("Blocking read for server events"),
-			virtfs.Read(func() ([]byte, error) {
-				data, _, _ := eventRead()
-				return data, nil
-			}),
-			virtfs.BlockOnce(eventRead, eventSignal),
+			virtfs.Doc("Streaming read for server events"),
+			virtfs.Stream(eventRead, eventSignal),
 		),
 		virtfs.FileNode("generate", 0666,
 			virtfs.Doc("One-shot LLM generation"),
diff --git a/cmd/olliesrv/internal/session/event.go b/cmd/olliesrv/internal/session/event.go
index 1a2d5f7..39855e8 100644
--- a/cmd/olliesrv/internal/session/event.go
+++ b/cmd/olliesrv/internal/session/event.go
@@ -2,8 +2,6 @@ package session
 
 import (
 	"context"
-	"crypto/sha256"
-	"encoding/hex"
 	"strings"
 	"sync"
 
@@ -44,12 +42,12 @@ func SubscribeEvents(ctx context.Context, topic string) <-chan Event {
 	return ch
 }
 
-// EventValue adapts an event channel into BlockOnce-compatible (readFn, signalFn).
-// A goroutine drains the channel; each arrival updates the value and fires the signal.
-func EventValue(ch <-chan Event) (readFn func() ([]byte, string, error), signalFn func() <-chan struct{}) {
+// EventStream returns functions suitable for virtfs.Stream - delivers events as they arrive.
+// Each read returns the next event; blocks until one is available.
+func EventStream(ch <-chan Event) (readFn func(base string) ([]byte, string, error), signalFn func() <-chan struct{}) {
 	var mu sync.Mutex
-	var data []byte
-	var hash string
+	var pending []byte
+	var counter int64
 	sig := make(chan struct{})
 
 	go func() {
@@ -59,22 +57,24 @@ func EventValue(ch <-chan Event) (readFn func() ([]byte, string, error), signalF
 				line += " " + ev.Payload
 			}
 			d := []byte(line + "\n")
-			h := sha256.Sum256(d)
 			mu.Lock()
-			data = d
-			hash = hex.EncodeToString(h[:8])
+			pending = d
+			counter++
 			close(sig)
 			sig = make(chan struct{})
 			mu.Unlock()
 		}
 	}()
 
-	readFn = func() ([]byte, string, error) {
+	readFn = func(base string) ([]byte, string, error) {
 		mu.Lock()
 		defer mu.Unlock()
-		out := make([]byte, len(data))
-		copy(out, data)
-		return out, hash, nil
+		if len(pending) == 0 {
+			return nil, base, nil // no data, will block on signal
+		}
+		out := pending
+		pending = nil
+		return out, "", nil
 	}
 	signalFn = func() <-chan struct{} {
 		mu.Lock()
diff --git a/kde/gui/ollie9pclient.cpp b/kde/gui/ollie9pclient.cpp
index cba23f6..50f2818 100644
--- a/kde/gui/ollie9pclient.cpp
+++ b/kde/gui/ollie9pclient.cpp
@@ -252,7 +252,6 @@ void Ollie9pClient::handleEvent(const QString &eventLine)
             refreshSessions();
         } else if (rest.startsWith(QLatin1String("bypass."))) {
             // Bypass event: payload is "id\tcmd\tcwd"
-            qDebug() << "handleEvent: bypass event" << eventLine;
             if (payload.isEmpty()) return;
             const int tab1 = payload.indexOf(QLatin1Char('\t'));
             if (tab1 < 0) return;
@@ -262,7 +261,6 @@ void Ollie9pClient::handleEvent(const QString &eventLine)
             const QString id = payload.left(tab1).toString();
             const QString cmd = payload.mid(tab1 + 1, tab2 - tab1 - 1).toString();
             const QString cwd = payload.mid(tab2 + 1).toString();
-            qDebug() << "handleEvent: emitting bypassRequested" << sessionId << id << cmd.left(50);
             emit bypassRequested(sessionId.toString(), id, cmd, cwd);
         }
     }
@@ -856,7 +854,6 @@ void Ollie9pClient::checkPendingBypass()
     QString cwd = obj["cwd"].toString();
     
     if (!id.isEmpty() && !cmd.isEmpty()) {
-        qDebug() << "checkPendingBypass: found pending" << id << cmd.left(50);
         emit bypassRequested(m_activeSessionId, id, cmd, cwd);
     }
 }