6 files changed, 243 insertions(+), 33 deletions(-)

M internal/llm/client.go
M internal/llm/session.go
M internal/llmfs/session_ask.go
M internal/llmfs/session_dir.go
A internal/llmfs/session_stream.go
M internal/protocol/server.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)

@@ 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

@@ 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
 }
 

@@ 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":

@@ 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
+}

@@ 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}