From bad7a7b752122022f1b03df6cc9e3deedf49a835 Mon Sep 17 00:00:00 2001 From: pdfinn Date: Wed, 25 Feb 2026 09:53:37 +0700 Subject: [PATCH] feat(llmfs): async Write + per-session streaming for live token delivery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enable lucibridge to stream LLM tokens into the Lucifer conversation zone as they are generated, without waiting for the full response. Changes: - session.go: add BeginGeneration/EndGeneration/SendChunk/GetStreamCh/ WaitDone methods on Session; streamCh (cap 256) carries raw text chunks during generation; doneCh signals completion; EndGeneration closes channels but does NOT nil streamCh (late readers still see closed chan) - session_ask.go: Write() is now async — calls BeginGeneration(), spawns goroutine, returns immediately; Read() calls WaitDone() before accessing LastResponse so pread blocks until generation completes - session_stream.go (new): /n/llm/N/stream file; Read() blocks on <-ch returning each text chunk as it arrives; returns EOF when generation is done or no generation is active (channel nil or closed) - session_dir.go: register stream file in Children() and Lookup() - client.go: AskWithRequest() branches on req.StreamFunc != nil to use SSE Messages.NewStreaming() path; text_delta events forwarded to StreamFunc; session.Ask() sets StreamFunc=session.SendChunk when GetStreamCh() is non-nil - server.go: add detailed walk debug logging (names, types, failures) controlled by existing -debug flag Co-Authored-By: Claude Sonnet 4.6 --- internal/llm/client.go | 34 ++++++++-- internal/llm/session.go | 109 ++++++++++++++++++++++++++++++- internal/llmfs/session_ask.go | 66 ++++++++++++------- internal/llmfs/session_dir.go | 5 +- internal/llmfs/session_stream.go | 51 +++++++++++++++ internal/protocol/server.go | 11 ++++ 6 files changed, 243 insertions(+), 33 deletions(-) create mode 100644 internal/llmfs/session_stream.go diff --git a/internal/llm/client.go b/internal/llm/client.go index 66f2dd523209a2949eb07382ecc3490bcb064f0a..3122c300e2877aa4258cb6d239c79d2677f2a166 100644 --- a/internal/llm/client.go +++ b/internal/llm/client.go @@ -686,11 +686,37 @@ func (c *Client) AskWithRequest(ctx context.Context, req AskRequest) (AskRespons } } + var ( + resp *anthropic.Message + latencyMs int64 + ) startTime := time.Now() - resp, err := c.client.Messages.New(ctx, params) - latencyMs := time.Since(startTime).Milliseconds() - if err != nil { - return AskResponse{}, fmt.Errorf("API error: %w", err) + + if req.StreamFunc != nil { + // Streaming path: use SSE, call StreamFunc for each text_delta chunk. + // Accumulate the full message for STOP:/TOOL: formatting after streaming ends. + stream := c.client.Messages.NewStreaming(ctx, params) + var acc anthropic.Message + for stream.Next() { + event := stream.Current() + if event.Type == "content_block_delta" && event.Delta.Type == "text_delta" { + req.StreamFunc(event.Delta.Text) + } + _ = acc.Accumulate(event) + } + latencyMs = time.Since(startTime).Milliseconds() + if err := stream.Err(); err != nil { + return AskResponse{}, fmt.Errorf("API streaming error: %w", err) + } + resp = &acc + } else { + // Blocking path: single HTTP request, no streaming. + r, err := c.client.Messages.New(ctx, params) + latencyMs = time.Since(startTime).Milliseconds() + if err != nil { + return AskResponse{}, fmt.Errorf("API error: %w", err) + } + resp = r } tokens := int(resp.Usage.InputTokens + resp.Usage.OutputTokens) diff --git a/internal/llm/session.go b/internal/llm/session.go index ecfb1298bcfd6a88f0962eb3e6ebed14ec143168..1588a23efc652b276dd9aee0a0209d833c24f64b 100644 --- a/internal/llm/session.go +++ b/internal/llm/session.go @@ -48,6 +48,11 @@ type Session struct { mu sync.RWMutex closed bool + + // Async generation + streaming support + streamCh chan string // raw text chunks during generation; nil when idle + doneCh chan struct{} // closed when generation completes; nil when idle + streamMu sync.Mutex // guards streamCh and doneCh } // NewSession creates a new session with the given ID and defaults. @@ -246,6 +251,68 @@ func (s *Session) IsClosed() bool { return s.closed } +// BeginGeneration allocates the stream channel and completion signal. +// Must be called before starting an async LLM generation. +func (s *Session) BeginGeneration() { + s.streamMu.Lock() + defer s.streamMu.Unlock() + s.streamCh = make(chan string, 256) + s.doneCh = make(chan struct{}) +} + +// EndGeneration closes the stream channel and completion signal. +// Called (via defer) when the async generation goroutine finishes. +// +// streamCh is closed but NOT set to nil: any buffered chunks remain readable +// from the closed channel even if the reader hasn't opened the file yet. +// BeginGeneration() will overwrite streamCh with a fresh channel next time. +func (s *Session) EndGeneration() { + s.streamMu.Lock() + ch := s.streamCh + done := s.doneCh + // Do NOT nil streamCh — leave closed channel readable for late-opening readers. + s.doneCh = nil + s.streamMu.Unlock() + if ch != nil { + close(ch) + } + if done != nil { + close(done) + } +} + +// SendChunk sends a text chunk to the stream channel (non-blocking; drops if buffer full). +func (s *Session) SendChunk(text string) { + s.streamMu.Lock() + ch := s.streamCh + s.streamMu.Unlock() + if ch == nil { + return + } + select { + case ch <- text: + default: // drop if buffer full + } +} + +// GetStreamCh returns the current stream channel, or nil if no generation is active. +func (s *Session) GetStreamCh() chan string { + s.streamMu.Lock() + defer s.streamMu.Unlock() + return s.streamCh +} + +// WaitDone blocks until the current generation completes, or returns immediately +// if no generation is in progress. +func (s *Session) WaitDone() { + s.streamMu.Lock() + done := s.doneCh + s.streamMu.Unlock() + if done != nil { + <-done + } +} + // SessionManager manages sessions and provides API access. // The APIClient is stateless - all conversation state is in sessions. type SessionManager struct { @@ -343,7 +410,8 @@ func (sm *SessionManager) Ask(ctx context.Context, id int, prompt string) (strin prefill := session.prefill session.mu.RUnlock() - // Build request with session's settings + // Build request with session's settings. + // Enable streaming when a generation is already in progress (BeginGeneration was called). req := AskRequest{ Messages: history, Prompt: prompt, @@ -354,12 +422,27 @@ func (sm *SessionManager) Ask(ctx context.Context, id int, prompt string) (strin Prefill: prefill, ToolDefs: session.Tools(), } + if session.GetStreamCh() != nil { + req.StreamFunc = session.SendChunk + } // Make API call (stateless) ar, err := sm.apiClient.AskWithRequest(ctx, req) if err != nil { - session.SetLastResponse("Error: " + err.Error()) - return "", err + if isContentFilterError(err) { + // Content filtering: reset history and retry once with a clean slate. + session.Reset() + req.Messages = nil + ar2, err2 := sm.apiClient.AskWithRequest(ctx, req) + if err2 != nil { + session.SetLastResponse("Error: " + err2.Error()) + return "", err2 + } + ar = ar2 + } else { + session.SetLastResponse("Error: " + err.Error()) + return "", err + } } // Update session state — store structured content for tool turns @@ -403,9 +486,21 @@ func (sm *SessionManager) AskWithToolResults(ctx context.Context, id int, result // Prompt is intentionally empty — tool results ARE the new user turn. // Prefill is intentionally empty — prefill is inappropriate mid-tool-loop. } + if session.GetStreamCh() != nil { + req.StreamFunc = session.SendChunk + } ar, err := sm.apiClient.AskWithRequest(ctx, req) if err != nil { + if isContentFilterError(err) { + // Content filtering on a tool-result turn: the offending content is + // in the tool results. Reset history and return a synthetic end_turn + // so the agent can surface a message rather than hard-crashing. + session.Reset() + synthetic := "STOP:end_turn\nContent filtering policy blocked a tool result. The session history has been reset. Please try a different approach or rephrase your request." + session.SetLastResponse(synthetic) + return synthetic, nil + } session.SetLastResponse("Error: " + err.Error()) return "", err } @@ -550,6 +645,14 @@ type AskRequest struct { Prefill string ToolDefs []ToolDef // non-nil enables native tool_use protocol ToolResults []ToolResult // non-nil: submit tool results as a new user turn + StreamFunc func(string) // optional; called for each text_delta chunk during streaming +} + +// isContentFilterError returns true when the API rejected the request due to +// Anthropic's content filtering policy. The session history should be reset +// before retrying in this case. +func isContentFilterError(err error) bool { + return strings.Contains(err.Error(), "content filtering policy") } // Errors diff --git a/internal/llmfs/session_ask.go b/internal/llmfs/session_ask.go index a00b575cc2e4ae4f7b4fd8627dae6ea2aaa509a3..0a6ae3ea901bd14730616e4f3656597035f0cb0a 100644 --- a/internal/llmfs/session_ask.go +++ b/internal/llmfs/session_ask.go @@ -29,12 +29,16 @@ func NewSessionAskFile(sm *llm.SessionManager, id int) *SessionAskFile { } // Read returns the last response from this session. +// Blocks until any in-progress generation completes before returning content. func (f *SessionAskFile) Read(p []byte, offset int64) (int, error) { session := f.sm.Get(f.id) if session == nil { return 0, protocol.ErrNotFound } + // Wait for the background generation goroutine (if any) to finish. + session.WaitDone() + content := session.LastResponse() if content != "" && !strings.HasSuffix(content, "\n") { content += "\n" @@ -49,6 +53,9 @@ func (f *SessionAskFile) Read(p []byte, offset int64) (int, error) { } // Write sends a prompt to the LLM using this session's settings. +// Returns immediately after starting the generation in a background goroutine. +// The response is available via Read (pread) once generation completes. +// // If the write begins with "TOOL_RESULTS\n", it is parsed as tool execution // results and submitted via AskWithToolResults instead of a plain Ask. // @@ -69,37 +76,46 @@ func (f *SessionAskFile) Write(p []byte, offset int64) (int, error) { return len(p), nil // Empty write is a no-op } - ctx := context.Background() + session := f.sm.Get(f.id) + if session == nil { + return 0, protocol.ErrNotFound + } + if session.IsClosed() { + return 0, protocol.ErrPermission + } - // Detect TOOL_RESULTS prefix → submit tool results, not a plain prompt - if strings.HasPrefix(prompt, "TOOL_RESULTS\n") { - results, err := parseToolResults(prompt) - if err != nil { - log.Printf("llm9p: SessionAskFile.Write TOOL_RESULTS parse error: %v", err) - // Store error in session so client can read it back - session := f.sm.Get(f.id) - if session != nil { + // Allocate streaming channel and done signal before launching goroutine. + // This ensures the stream file can observe a non-nil channel immediately + // after the write returns (no race between Write and stream Read). + session.BeginGeneration() + + // Launch generation goroutine — Write returns to the caller immediately. + sm := f.sm + id := f.id + go func() { + defer session.EndGeneration() + ctx := context.Background() + + if strings.HasPrefix(prompt, "TOOL_RESULTS\n") { + results, err := parseToolResults(prompt) + if err != nil { + log.Printf("llm9p: async gen TOOL_RESULTS parse error: %v", err) session.SetLastResponse("Error: " + err.Error()) + return } - return len(p), nil - } - log.Printf("llm9p: SessionAskFile.Write submitting %d tool results", len(results)) - _, err = f.sm.AskWithToolResults(ctx, f.id, results) - if err != nil { - log.Printf("llm9p: SessionAskFile.Write tool results error: %v", err) + log.Printf("llm9p: async gen submitting %d tool results", len(results)) + if _, err = sm.AskWithToolResults(ctx, id, results); err != nil { + log.Printf("llm9p: async gen tool results error: %v", err) + } + return } - return len(p), nil - } - // Regular text prompt - log.Printf("llm9p: SessionAskFile.Write prompt: %s", prompt[:min(len(prompt), 50)]) - response, err := f.sm.Ask(ctx, f.id, prompt) - if err != nil { - log.Printf("llm9p: SessionAskFile.Write error: %v", err) - return len(p), nil - } + log.Printf("llm9p: async gen prompt: %s", prompt[:min(len(prompt), 50)]) + if _, err := sm.Ask(ctx, id, prompt); err != nil { + log.Printf("llm9p: async gen error: %v", err) + } + }() - log.Printf("llm9p: SessionAskFile.Write success, response len=%d", len(response)) return len(p), nil } diff --git a/internal/llmfs/session_dir.go b/internal/llmfs/session_dir.go index b84e8a92a6289c2a0eeafa094ca51e719cf752e5..6eb9c6c0efd6f4e0d0df01f9ee87d6ffc959c556 100644 --- a/internal/llmfs/session_dir.go +++ b/internal/llmfs/session_dir.go @@ -10,7 +10,7 @@ import ( ) // SessionDir represents a single session directory: /n/llm/N/ -// Contains: ask, compact, context, ctl, model, temperature, system, thinking, prefill, usage +// Contains: ask, compact, context, ctl, model, temperature, system, thinking, prefill, tools, usage, stream type SessionDir struct { *protocol.BaseFile sm *llm.SessionManager @@ -43,6 +43,7 @@ func (d *SessionDir) Children() []protocol.File { NewSessionSystemFile(d.sm, d.id), NewSessionThinkingFile(d.sm, d.id), NewSessionPrefillFile(d.sm, d.id), + NewSessionStreamFile(d.sm, d.id), NewSessionToolsFile(d.sm, d.id), NewSessionUsageFile(d.sm, d.id), } @@ -74,6 +75,8 @@ func (d *SessionDir) Lookup(name string) (protocol.File, error) { return NewSessionThinkingFile(d.sm, d.id), nil case "prefill": return NewSessionPrefillFile(d.sm, d.id), nil + case "stream": + return NewSessionStreamFile(d.sm, d.id), nil case "tools": return NewSessionToolsFile(d.sm, d.id), nil case "usage": diff --git a/internal/llmfs/session_stream.go b/internal/llmfs/session_stream.go new file mode 100644 index 0000000000000000000000000000000000000000..b000abf31cf0d8c5649a53b96d3f6b321fafc54f --- /dev/null +++ b/internal/llmfs/session_stream.go @@ -0,0 +1,51 @@ +package llmfs + +import ( + "io" + + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// SessionStreamFile is the stream file for a specific session: /n/llm/N/stream +// Read blocks until the next chunk is available, returning chunks as they arrive +// and io.EOF when generation completes (or no generation is active). +type SessionStreamFile struct { + *protocol.BaseFile + sm *llm.SessionManager + id int +} + +// NewSessionStreamFile creates a stream file for the given session. +func NewSessionStreamFile(sm *llm.SessionManager, id int) *SessionStreamFile { + return &SessionStreamFile{ + BaseFile: protocol.NewBaseFile("stream", 0444), + sm: sm, + id: id, + } +} + +// Read blocks until the next text chunk is available and copies it into p. +// Returns io.EOF when generation completes or no generation is active. +// The offset parameter is ignored — this is a streaming file, not seekable. +func (f *SessionStreamFile) Read(p []byte, offset int64) (int, error) { + session := f.sm.Get(f.id) + if session == nil { + return 0, protocol.ErrNotFound + } + + ch := session.GetStreamCh() + if ch == nil { + // No active generation. + return 0, io.EOF + } + + chunk, ok := <-ch + if !ok { + // Channel closed: generation complete. + return 0, io.EOF + } + + n := copy(p, chunk) + return n, nil +} diff --git a/internal/protocol/server.go b/internal/protocol/server.go index 123c52fb9f692018cd9b490818907379618667db..6eb5bab9ddc18879a2bf9871dff48828b72f0415 100644 --- a/internal/protocol/server.go +++ b/internal/protocol/server.go @@ -207,15 +207,24 @@ func (s *Server) handleWalk(state *clientState, payload []byte, buf []byte) ([]b for _, name := range msg.Names { dir, ok := current.(Dir) if !ok { + if s.debug { + log.Printf(" walk: %q is not a directory", name) + } return s.errorResponse(buf, ErrNotDir.Error()) } next, err := dir.Lookup(name) if err != nil { + if s.debug { + log.Printf(" walk: lookup %q failed: %v (walked %d/%d)", name, err, len(qids), len(msg.Names)) + } // Return partial walk break } + if s.debug { + log.Printf(" walk: %q -> %T", name, next) + } qids = append(qids, next.Stat().Qid) current = next } @@ -223,6 +232,8 @@ func (s *Server) handleWalk(state *clientState, payload []byte, buf []byte) ([]b // Only update fid if we walked at least one element (or no elements requested) if len(qids) == len(msg.Names) { state.fids[msg.Newfid] = current + } else if s.debug { + log.Printf(" walk: partial walk %d/%d names succeeded; newfid %d NOT registered", len(qids), len(msg.Names), msg.Newfid) } resp := &RwalkMsg{Qids: qids}