From 68199d2ad65152fa84d9c9be5008dc7659afdda0 Mon Sep 17 00:00:00 2001 From: pdfinn Date: Fri, 23 Jan 2026 03:04:33 +0700 Subject: [PATCH] feat: Initial implementation of llm9p - LLM as 9P filesystem Exposes Claude as a 9P filesystem, enabling interaction through standard file operations: - ask: write prompt, read response (shim pattern) - model: read/write current model name - temperature: read/write sampling temperature - tokens: read-only token count from last response - new: write to reset conversation - context: read JSON history, write to add system message - _example: usage documentation - stream/chunk: blocking read for streaming responses Includes: - Full 9P2000 protocol implementation (stdlib only) - Anthropic SDK integration with conversation state - Streaming support Co-Authored-By: Claude Opus 4.5 --- .gitignore | 11 + CLAUDE.md | 271 ++++++++++++++++++++++++ README.md | 146 +++++++++++++ cmd/llm9p/main.go | 80 +++++++ go.mod | 12 ++ go.sum | 12 ++ internal/llm/client.go | 335 ++++++++++++++++++++++++++++++ internal/llmfs/ask.go | 79 +++++++ internal/llmfs/context.go | 56 +++++ internal/llmfs/example.go | 62 ++++++ internal/llmfs/new.go | 36 ++++ internal/llmfs/root.go | 28 +++ internal/llmfs/state.go | 91 ++++++++ internal/llmfs/stream.go | 53 +++++ internal/llmfs/tokens.go | 43 ++++ internal/protocol/fs.go | 230 ++++++++++++++++++++ internal/protocol/message.go | 381 ++++++++++++++++++++++++++++++++++ internal/protocol/protocol.go | 327 +++++++++++++++++++++++++++++ internal/protocol/server.go | 346 ++++++++++++++++++++++++++++++ 19 files changed, 2599 insertions(+) create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 README.md create mode 100644 cmd/llm9p/main.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/llm/client.go create mode 100644 internal/llmfs/ask.go create mode 100644 internal/llmfs/context.go create mode 100644 internal/llmfs/example.go create mode 100644 internal/llmfs/new.go create mode 100644 internal/llmfs/root.go create mode 100644 internal/llmfs/state.go create mode 100644 internal/llmfs/stream.go create mode 100644 internal/llmfs/tokens.go create mode 100644 internal/protocol/fs.go create mode 100644 internal/protocol/message.go create mode 100644 internal/protocol/protocol.go create mode 100644 internal/protocol/server.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000000000000000000000000000000000000..5262471d1bce49c767fab45eec9ead3dbf509297 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +# Binary +llm9p + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# OS +.DS_Store diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000000000000000000000000000000000000..39b765d2fcddb614cedd90bb5c161a58216aa247 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,271 @@ +# llm9p - Development Guide + +This guide is for Claude Code and developers working on the llm9p codebase. + +## Quick Reference + +### Build and Run + +```bash +# Build +go build -o llm9p ./cmd/llm9p + +# Run +ANTHROPIC_API_KEY=sk-... ./llm9p -addr :5640 + +# Run with debug logging +ANTHROPIC_API_KEY=sk-... ./llm9p -addr :5640 -debug +``` + +### Testing + +```bash +# Run all tests +go test ./... + +# Run with coverage +go test -cover ./... + +# Test specific package +go test ./internal/protocol/... +go test ./internal/llmfs/... +``` + +### Development Workflow + +```bash +# Install dependencies +go mod tidy + +# Format code +go fmt ./... + +# Vet code +go vet ./... + +# Build +go build -o llm9p ./cmd/llm9p +``` + +## Architecture Overview + +### Project Structure + +``` +llm9p/ +├── cmd/ +│ └── llm9p/ +│ └── main.go # Entry point, CLI flags, server setup +├── internal/ +│ ├── protocol/ # 9P2000 protocol implementation +│ │ ├── protocol.go # Message types, constants, encoding +│ │ ├── message.go # Individual message types +│ │ ├── server.go # Connection handling +│ │ └── fs.go # File/Dir interfaces, base implementations +│ ├── llm/ # LLM client wrapper +│ │ └── client.go # Anthropic API integration +│ └── llmfs/ # LLM filesystem implementation +│ ├── root.go # Root directory construction +│ ├── ask.go # Ask file (shim pattern) +│ ├── state.go # Model, temperature files +│ ├── tokens.go # Read-only token counter +│ ├── new.go # Conversation reset trigger +│ ├── context.go # Conversation history +│ ├── example.go # Usage examples +│ └── stream.go # Streaming interface +├── go.mod +├── go.sum +├── README.md +└── CLAUDE.md +``` + +### Key Components + +1. **Protocol Layer (`internal/protocol/`)** + - Implements 9P2000 protocol + - No external dependencies (stdlib only) + - `File` and `Dir` interfaces define the filesystem abstraction + +2. **LLM Client (`internal/llm/client.go`)** + - Wraps Anthropic SDK + - Manages conversation state + - Supports both sync and streaming responses + - Tracks token usage + +3. **LLM Filesystem (`internal/llmfs/`)** + - Implements each file in the LLM filesystem + - `AskFile` is the core interaction point + - State files (`model`, `temperature`) modify client settings + - `ChunkFile` provides streaming access + +## Adding a New File + +1. Create a new file in `internal/llmfs/`: + +```go +package llmfs + +import ( + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +type MyFile struct { + *protocol.BaseFile + client *llm.Client +} + +func NewMyFile(client *llm.Client) *MyFile { + return &MyFile{ + BaseFile: protocol.NewBaseFile("myfile", 0666), + client: client, + } +} + +func (f *MyFile) Read(p []byte, offset int64) (int, error) { + // Implement read +} + +func (f *MyFile) Write(p []byte, offset int64) (int, error) { + // Implement write +} + +func (f *MyFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + // Update s.Length if dynamic + return s +} +``` + +2. Add to root directory in `internal/llmfs/root.go`: + +```go +root.AddChild(NewMyFile(client)) +``` + +## Protocol Implementation Notes + +### Message Flow + +1. Client sends T-message (request) +2. Server responds with R-message (response) +3. Each message has a tag for matching requests/responses + +### Key 9P Operations + +- `Tversion/Rversion` - Protocol negotiation +- `Tattach/Rattach` - Connect to filesystem +- `Twalk/Rwalk` - Navigate directory tree +- `Topen/Ropen` - Open a file +- `Tread/Rread` - Read from file +- `Twrite/Rwrite` - Write to file +- `Tclunk/Rclunk` - Close a fid + +### File Interfaces + +```go +// File is the interface that files must implement +type File interface { + Stat() Stat + Open(mode uint8) error + Read(p []byte, offset int64) (int, error) + Write(p []byte, offset int64) (int, error) + Close() error +} + +// Dir extends File with directory operations +type Dir interface { + File + Children() []File + Lookup(name string) (File, error) +} +``` + +## Debugging + +### Enable Debug Logging + +```bash +ANTHROPIC_API_KEY=sk-... ./llm9p -debug +``` + +This logs all 9P messages sent and received. + +### Test with 9p Client + +```bash +# Using 9pfuse +9pfuse localhost:5640 /mnt/llm + +# Using Plan 9's 9p tool +9p -a localhost:5640 ls llm +9p -a localhost:5640 read llm/model +9p -a localhost:5640 write llm/ask "Hello" +9p -a localhost:5640 read llm/ask +``` + +### Common Issues + +**"file not found"** +- Check file name spelling +- Ensure file is added to root directory + +**"permission denied"** +- Check file mode (read-only files have mode 0444) +- Write-only files have mode 0222 + +**Connection refused** +- Ensure server is running +- Check address/port + +**API errors** +- Check ANTHROPIC_API_KEY is set +- Check API key is valid +- Check rate limits + +## Code Style + +### Error Handling + +Files should handle errors gracefully and expose them to the user: + +```go +func (f *AskFile) Write(p []byte, offset int64) (int, error) { + response, err := f.client.Ask(ctx, prompt) + if err != nil { + // Store error so it can be read back + f.lastResponse = "Error: " + err.Error() + return len(p), nil + } + f.lastResponse = response + return len(p), nil +} +``` + +### Stat Implementation + +Always implement `Stat()` to return accurate `Length`: + +```go +func (f *MyFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + s.Length = uint64(len(f.content)) + return s +} +``` + +## Future Enhancements + +- [ ] Multiple conversation support (via subdirectories) +- [ ] Prompt templates +- [ ] Response caching +- [ ] Rate limiting +- [ ] Authentication +- [ ] Unix socket support +- [ ] Integration tests + +## Resources + +- [9P Protocol Specification](http://man.cat-v.org/plan_9/5/intro) +- [Anthropic API Documentation](https://docs.anthropic.com/) +- [Plan 9 from User Space](https://9fans.github.io/plan9port/) diff --git a/README.md b/README.md new file mode 100644 index 0000000000000000000000000000000000000000..6a48c7d31e57aca9f52db8cb750ded0fc24c6a79 --- /dev/null +++ b/README.md @@ -0,0 +1,146 @@ +# llm9p + +An LLM (Claude) exposed as a 9P filesystem. + +llm9p enables users, scripts, and AI agents to interact with an LLM through standard filesystem operations. Write a prompt to a file, read the response from the same file. + +## Installation + +```bash +go install github.com/NERVsystems/llm9p/cmd/llm9p@latest +``` + +Or build from source: + +```bash +git clone https://github.com/NERVsystems/llm9p +cd llm9p +go build -o llm9p ./cmd/llm9p +``` + +## Usage + +### Start the Server + +```bash +export ANTHROPIC_API_KEY=sk-ant-... +./llm9p -addr :5640 +``` + +### Mount the Filesystem + +Using 9pfuse (Plan 9 from User Space): + +```bash +mkdir -p /mnt/llm +9pfuse localhost:5640 /mnt/llm +``` + +On macOS with plan9port: + +```bash +9 mount localhost:5640 /mnt/llm +``` + +### Interact with the LLM + +```bash +# Ask a question +echo "What is 2+2?" > /mnt/llm/ask +cat /mnt/llm/ask + +# View token usage +cat /mnt/llm/tokens + +# Change model +echo "claude-3-haiku-20240307" > /mnt/llm/model + +# Adjust temperature +echo "0.5" > /mnt/llm/temperature + +# View conversation history +cat /mnt/llm/context + +# Add a system message +echo "You are a helpful coding assistant." > /mnt/llm/context + +# Reset conversation +echo "" > /mnt/llm/new + +# View help +cat /mnt/llm/_example +``` + +## Filesystem Schema + +``` +/llm/ +├── ask # Write prompt, read response (same file) +├── model # Read/write: current model name +├── temperature # Read/write: temperature float (0.0-2.0) +├── tokens # Read-only: last response token count +├── new # Write anything to start fresh conversation +├── context # Read: conversation history; Write: add system message +├── _example # Read-only: usage examples +└── stream/ # Streaming interface + └── chunk # Read blocks until next chunk, EOF on completion +``` + +### File Behaviors + +| File | Read | Write | +|------|------|-------| +| `ask` | Returns last LLM response | Sends prompt to LLM, stores response | +| `model` | Returns current model name | Sets model for subsequent requests | +| `temperature` | Returns current temperature | Sets temperature (0.0-2.0) | +| `tokens` | Returns last response token count | Permission denied | +| `new` | Permission denied | Any write resets conversation state | +| `context` | Returns JSON conversation history | Appends system message to context | +| `_example` | Returns usage examples | Permission denied | +| `stream/chunk` | Blocks until next chunk, returns it | Permission denied | + +## Shell Scripting + +```bash +#!/bin/sh +# ask.sh - Simple LLM query script + +if [ -z "$1" ]; then + echo "Usage: $0 " + exit 1 +fi + +echo "$1" > /mnt/llm/ask +cat /mnt/llm/ask +``` + +## Configuration + +### Command Line Flags + +| Flag | Default | Description | +|------|---------|-------------| +| `-addr` | `:5640` | Address to listen on | +| `-debug` | `false` | Enable debug logging | + +### Environment Variables + +| Variable | Required | Description | +|----------|----------|-------------| +| `ANTHROPIC_API_KEY` | Yes | Your Anthropic API key | + +## Default Settings + +- **Model**: `claude-sonnet-4-20250514` +- **Temperature**: `0.7` +- **Max Tokens**: `4096` + +## Requirements + +- Go 1.21+ +- Anthropic API key +- 9P client (9pfuse, plan9port, or native Plan 9) + +## License + +MIT diff --git a/cmd/llm9p/main.go b/cmd/llm9p/main.go new file mode 100644 index 0000000000000000000000000000000000000000..62583d54cd8c7034d519524e9a7b859fa512ee7a --- /dev/null +++ b/cmd/llm9p/main.go @@ -0,0 +1,80 @@ +// llm9p exposes an LLM (Claude) as a 9P filesystem. +// +// Usage: +// +// ANTHROPIC_API_KEY=sk-... llm9p -addr :5640 +// +// Mount with: +// +// 9pfuse localhost:5640 /mnt/llm +// +// Interact: +// +// echo "What is 2+2?" > /mnt/llm/ask +// cat /mnt/llm/ask +package main + +import ( + "context" + "flag" + "fmt" + "log" + "net" + "os" + "os/signal" + "syscall" + + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/llmfs" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +func main() { + addr := flag.String("addr", ":5640", "Address to listen on") + debug := flag.Bool("debug", false, "Enable debug logging") + flag.Parse() + + // Get API key from environment + apiKey := os.Getenv("ANTHROPIC_API_KEY") + if apiKey == "" { + fmt.Fprintln(os.Stderr, "Error: ANTHROPIC_API_KEY environment variable not set") + os.Exit(1) + } + + // Create LLM client + client := llm.NewClient(apiKey) + + // Create filesystem + root := llmfs.NewRoot(client) + + // Create 9P server + server := protocol.NewServer(root) + server.SetDebug(*debug) + + // Listen + listener, err := net.Listen("tcp", *addr) + if err != nil { + log.Fatalf("Failed to listen on %s: %v", *addr, err) + } + + log.Printf("llm9p listening on %s", *addr) + log.Printf("Mount with: 9pfuse %s /mnt/llm", listener.Addr()) + + // Handle shutdown gracefully + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go func() { + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + <-sigCh + log.Println("Shutting down...") + cancel() + listener.Close() + }() + + // Serve + if err := server.Serve(ctx, listener); err != nil && ctx.Err() == nil { + log.Fatalf("Server error: %v", err) + } +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000000000000000000000000000000000000..fea183c5aaa44bc74c34037dd783b2ac2e258d99 --- /dev/null +++ b/go.mod @@ -0,0 +1,12 @@ +module github.com/NERVsystems/llm9p + +go 1.21 + +require github.com/anthropics/anthropic-sdk-go v0.2.0-beta.3 + +require ( + github.com/tidwall/gjson v1.14.4 // indirect + github.com/tidwall/match v1.1.1 // indirect + github.com/tidwall/pretty v1.2.1 // indirect + github.com/tidwall/sjson v1.2.5 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000000000000000000000000000000000000..78e0cb45f057862bb959b86a069281921d61f8d3 --- /dev/null +++ b/go.sum @@ -0,0 +1,12 @@ +github.com/anthropics/anthropic-sdk-go v0.2.0-beta.3 h1:b5t1ZJMvV/l99y4jbz7kRFdUp3BSDkI8EhSlHczivtw= +github.com/anthropics/anthropic-sdk-go v0.2.0-beta.3/go.mod h1:AapDW22irxK2PSumZiQXYUFvsdQgkwIWlpESweWZI/c= +github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/gjson v1.14.4 h1:uo0p8EbA09J7RQaflQ1aBRffTR7xedD2bcIVSYxLnkM= +github.com/tidwall/gjson v1.14.4/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA= +github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= +github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4= +github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY= +github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28= diff --git a/internal/llm/client.go b/internal/llm/client.go new file mode 100644 index 0000000000000000000000000000000000000000..3c8342fd7c88c48891f4a22f6aca1a249a1af222 --- /dev/null +++ b/internal/llm/client.go @@ -0,0 +1,335 @@ +// Package llm provides a wrapper around the Anthropic API for use with the 9P filesystem. +package llm + +import ( + "context" + "encoding/json" + "fmt" + "sync" + + "github.com/anthropics/anthropic-sdk-go" + "github.com/anthropics/anthropic-sdk-go/option" +) + +// Message represents a single message in a conversation +type Message struct { + Role string `json:"role"` // "user" or "assistant" + Content string `json:"content"` // message content +} + +// Client wraps the Anthropic API client with conversation state +type Client struct { + client anthropic.Client + mu sync.RWMutex + model string + temperature float64 + messages []Message + lastTokens int + streaming bool + streamChan chan string + streamDone chan struct{} +} + +// NewClient creates a new LLM client +func NewClient(apiKey string) *Client { + client := anthropic.NewClient(option.WithAPIKey(apiKey)) + return &Client{ + client: client, + model: "claude-sonnet-4-20250514", + temperature: 0.7, + messages: make([]Message, 0), + } +} + +// Model returns the current model name +func (c *Client) Model() string { + c.mu.RLock() + defer c.mu.RUnlock() + return c.model +} + +// SetModel sets the model for subsequent requests +func (c *Client) SetModel(model string) { + c.mu.Lock() + defer c.mu.Unlock() + c.model = model +} + +// Temperature returns the current temperature +func (c *Client) Temperature() float64 { + c.mu.RLock() + defer c.mu.RUnlock() + return c.temperature +} + +// SetTemperature sets the temperature for subsequent requests +func (c *Client) SetTemperature(temp float64) error { + if temp < 0.0 || temp > 2.0 { + return fmt.Errorf("temperature must be between 0.0 and 2.0") + } + c.mu.Lock() + defer c.mu.Unlock() + c.temperature = temp + return nil +} + +// LastTokens returns the token count from the last response +func (c *Client) LastTokens() int { + c.mu.RLock() + defer c.mu.RUnlock() + return c.lastTokens +} + +// Messages returns a copy of the conversation history +func (c *Client) Messages() []Message { + c.mu.RLock() + defer c.mu.RUnlock() + result := make([]Message, len(c.messages)) + copy(result, c.messages) + return result +} + +// MessagesJSON returns the conversation history as JSON +func (c *Client) MessagesJSON() ([]byte, error) { + c.mu.RLock() + defer c.mu.RUnlock() + return json.MarshalIndent(c.messages, "", " ") +} + +// AddSystemMessage adds a system message to the context +func (c *Client) AddSystemMessage(content string) { + c.mu.Lock() + defer c.mu.Unlock() + // System messages are prepended to conversations + c.messages = append([]Message{{Role: "system", Content: content}}, c.messages...) +} + +// Reset clears the conversation history +func (c *Client) Reset() { + c.mu.Lock() + defer c.mu.Unlock() + c.messages = make([]Message, 0) + c.lastTokens = 0 +} + +// Ask sends a prompt to the LLM and returns the response +func (c *Client) Ask(ctx context.Context, prompt string) (string, error) { + c.mu.Lock() + // Add user message to history + c.messages = append(c.messages, Message{Role: "user", Content: prompt}) + + // Build the API messages + apiMessages := make([]anthropic.MessageParam, 0, len(c.messages)) + var systemBlocks []anthropic.TextBlockParam + + for _, msg := range c.messages { + switch msg.Role { + case "system": + // Collect system messages + systemBlocks = append(systemBlocks, anthropic.TextBlockParam{ + Text: msg.Content, + }) + case "user": + apiMessages = append(apiMessages, anthropic.NewUserMessage( + anthropic.NewTextBlock(msg.Content), + )) + case "assistant": + apiMessages = append(apiMessages, anthropic.NewAssistantMessage( + anthropic.NewTextBlock(msg.Content), + )) + } + } + + model := c.model + temp := c.temperature + c.mu.Unlock() + + // Build request params + params := anthropic.MessageNewParams{ + Model: anthropic.Model(model), + MaxTokens: 4096, + Messages: apiMessages, + Temperature: anthropic.Float(temp), + } + + // Add system prompt if present + if len(systemBlocks) > 0 { + params.System = systemBlocks + } + + // Make the API call + response, err := c.client.Messages.New(ctx, params) + if err != nil { + // Remove the user message on error + c.mu.Lock() + if len(c.messages) > 0 { + c.messages = c.messages[:len(c.messages)-1] + } + c.mu.Unlock() + return "", fmt.Errorf("API error: %w", err) + } + + // Extract response text + var responseText string + for _, block := range response.Content { + if block.Type == "text" { + responseText += block.Text + } + } + + // Update state + c.mu.Lock() + c.messages = append(c.messages, Message{Role: "assistant", Content: responseText}) + c.lastTokens = int(response.Usage.InputTokens + response.Usage.OutputTokens) + c.mu.Unlock() + + return responseText, nil +} + +// StartStream begins streaming a response for the given prompt +func (c *Client) StartStream(ctx context.Context, prompt string) error { + c.mu.Lock() + if c.streaming { + c.mu.Unlock() + return fmt.Errorf("stream already in progress") + } + + // Add user message to history + c.messages = append(c.messages, Message{Role: "user", Content: prompt}) + + // Build the API messages + apiMessages := make([]anthropic.MessageParam, 0, len(c.messages)) + var systemBlocks []anthropic.TextBlockParam + + for _, msg := range c.messages { + switch msg.Role { + case "system": + systemBlocks = append(systemBlocks, anthropic.TextBlockParam{ + Text: msg.Content, + }) + case "user": + apiMessages = append(apiMessages, anthropic.NewUserMessage( + anthropic.NewTextBlock(msg.Content), + )) + case "assistant": + apiMessages = append(apiMessages, anthropic.NewAssistantMessage( + anthropic.NewTextBlock(msg.Content), + )) + } + } + + model := c.model + temp := c.temperature + + c.streaming = true + c.streamChan = make(chan string, 100) + c.streamDone = make(chan struct{}) + c.mu.Unlock() + + // Start streaming in a goroutine + go func() { + defer func() { + c.mu.Lock() + c.streaming = false + close(c.streamChan) + close(c.streamDone) + c.mu.Unlock() + }() + + // Build request params + params := anthropic.MessageNewParams{ + Model: anthropic.Model(model), + MaxTokens: 4096, + Messages: apiMessages, + Temperature: anthropic.Float(temp), + } + + if len(systemBlocks) > 0 { + params.System = systemBlocks + } + + // Use streaming + stream := c.client.Messages.NewStreaming(ctx, params) + + var fullResponse string + var inputTokens, outputTokens int64 + + for stream.Next() { + event := stream.Current() + + switch event.Type { + case "content_block_delta": + delta := event.Delta + if delta.Type == "text_delta" { + chunk := delta.Text + fullResponse += chunk + select { + case c.streamChan <- chunk: + case <-ctx.Done(): + return + } + } + case "message_delta": + outputTokens = event.Usage.OutputTokens + case "message_start": + inputTokens = event.Message.Usage.InputTokens + } + } + + if err := stream.Err(); err != nil { + // Send error as chunk + select { + case c.streamChan <- fmt.Sprintf("\n[Error: %v]", err): + case <-ctx.Done(): + } + // Remove user message on error + c.mu.Lock() + if len(c.messages) > 0 { + c.messages = c.messages[:len(c.messages)-1] + } + c.mu.Unlock() + return + } + + // Update state with complete response + c.mu.Lock() + c.messages = append(c.messages, Message{Role: "assistant", Content: fullResponse}) + c.lastTokens = int(inputTokens + outputTokens) + c.mu.Unlock() + }() + + return nil +} + +// ReadStreamChunk reads the next chunk from the stream, blocking until available +// Returns empty string and false when stream is complete +func (c *Client) ReadStreamChunk() (string, bool) { + c.mu.RLock() + streamChan := c.streamChan + c.mu.RUnlock() + + if streamChan == nil { + return "", false + } + + chunk, ok := <-streamChan + return chunk, ok +} + +// IsStreaming returns whether a stream is currently in progress +func (c *Client) IsStreaming() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.streaming +} + +// WaitStream waits for the current stream to complete +func (c *Client) WaitStream() { + c.mu.RLock() + done := c.streamDone + c.mu.RUnlock() + + if done != nil { + <-done + } +} diff --git a/internal/llmfs/ask.go b/internal/llmfs/ask.go new file mode 100644 index 0000000000000000000000000000000000000000..00b723f3b3c53ac67fbd7a8cf6d5baba465b8dc2 --- /dev/null +++ b/internal/llmfs/ask.go @@ -0,0 +1,79 @@ +package llmfs + +import ( + "context" + "io" + "strings" + "sync" + + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// AskFile is the main interaction file - write a prompt, read the response +type AskFile struct { + *protocol.BaseFile + client *llm.Client + mu sync.RWMutex + lastResponse string +} + +// NewAskFile creates the ask file +func NewAskFile(client *llm.Client) *AskFile { + return &AskFile{ + BaseFile: protocol.NewBaseFile("ask", 0666), + client: client, + } +} + +func (f *AskFile) Read(p []byte, offset int64) (int, error) { + f.mu.RLock() + content := f.lastResponse + f.mu.RUnlock() + + // Add newline if not present + if content != "" && !strings.HasSuffix(content, "\n") { + content += "\n" + } + + if offset >= int64(len(content)) { + return 0, io.EOF + } + n := copy(p, content[offset:]) + return n, nil +} + +func (f *AskFile) Write(p []byte, offset int64) (int, error) { + prompt := strings.TrimSpace(string(p)) + if prompt == "" { + return len(p), nil // Empty write is a no-op + } + + // Make the API call + response, err := f.client.Ask(context.Background(), prompt) + if err != nil { + // Store error as response so it can be read + f.mu.Lock() + f.lastResponse = "Error: " + err.Error() + f.mu.Unlock() + return len(p), nil // Return success so client knows write completed + } + + f.mu.Lock() + f.lastResponse = response + f.mu.Unlock() + + return len(p), nil +} + +func (f *AskFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + f.mu.RLock() + content := f.lastResponse + f.mu.RUnlock() + if content != "" && !strings.HasSuffix(content, "\n") { + content += "\n" + } + s.Length = uint64(len(content)) + return s +} diff --git a/internal/llmfs/context.go b/internal/llmfs/context.go new file mode 100644 index 0000000000000000000000000000000000000000..9cbf83a9975b27a1bb40bd5ba79f361acd5ac4b2 --- /dev/null +++ b/internal/llmfs/context.go @@ -0,0 +1,56 @@ +package llmfs + +import ( + "io" + "strings" + + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// ContextFile exposes the conversation history +// Read: returns JSON of conversation history +// Write: appends a system message to context +type ContextFile struct { + *protocol.BaseFile + client *llm.Client +} + +// NewContextFile creates the context file +func NewContextFile(client *llm.Client) *ContextFile { + return &ContextFile{ + BaseFile: protocol.NewBaseFile("context", 0666), + client: client, + } +} + +func (f *ContextFile) Read(p []byte, offset int64) (int, error) { + content, err := f.client.MessagesJSON() + if err != nil { + return 0, err + } + // Add newline + content = append(content, '\n') + + if offset >= int64(len(content)) { + return 0, io.EOF + } + n := copy(p, content[offset:]) + return n, nil +} + +func (f *ContextFile) Write(p []byte, offset int64) (int, error) { + // Writing appends a system message to the context + msg := strings.TrimSpace(string(p)) + if msg != "" { + f.client.AddSystemMessage(msg) + } + return len(p), nil +} + +func (f *ContextFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + content, _ := f.client.MessagesJSON() + s.Length = uint64(len(content) + 1) // +1 for newline + return s +} diff --git a/internal/llmfs/example.go b/internal/llmfs/example.go new file mode 100644 index 0000000000000000000000000000000000000000..c99c98618134d8c2b5481279b31962492c415704 --- /dev/null +++ b/internal/llmfs/example.go @@ -0,0 +1,62 @@ +package llmfs + +import ( + "github.com/NERVsystems/llm9p/internal/protocol" +) + +const exampleContent = `LLM 9P Filesystem Usage Examples +================================= + +Basic Interaction: + echo "What is 2+2?" > ask # Send prompt to LLM + cat ask # Read response + +Configuration: + cat model # View current model + echo "claude-3-haiku-20240307" > model # Change model + cat temperature # View current temperature (0.0-2.0) + echo "0.5" > temperature # Set temperature + +Conversation Management: + cat context # View conversation history (JSON) + echo "You are a helpful assistant." > context # Add system message + echo "" > new # Reset conversation + +Token Usage: + cat tokens # View tokens from last response + +Streaming (Advanced): + echo "Tell me a story" > ask # Start generating + cat stream/chunk # Read chunks as they arrive (blocks) + +Shell Scripting: + #!/bin/sh + # Ask the LLM and get response + echo "$1" > /mnt/llm/ask + cat /mnt/llm/ask + +Mounting (Linux/macOS): + # Using 9pfuse (Plan 9 from User Space) + 9pfuse localhost:5640 /mnt/llm + + # Using mount_9p (macOS with plan9port) + mount_9p localhost:5640 /mnt/llm + +Environment: + ANTHROPIC_API_KEY must be set when starting the server + +Files: + ask Read/write: prompt goes in, response comes out + model Read/write: current model name + temperature Read/write: sampling temperature (0.0-2.0) + tokens Read-only: token count from last response + new Write-only: any write resets conversation + context Read: JSON history; Write: add system message + _example Read-only: this help text + stream/chunk Read-only: streaming chunks (blocking) +` + +// NewExampleFile creates the _example file with usage examples +func NewExampleFile() *protocol.StaticFile { + return protocol.NewStaticFile("_example", []byte(exampleContent)) +} diff --git a/internal/llmfs/new.go b/internal/llmfs/new.go new file mode 100644 index 0000000000000000000000000000000000000000..9b5900528b5297fe3637e5a1c9bd333feef52ca0 --- /dev/null +++ b/internal/llmfs/new.go @@ -0,0 +1,36 @@ +package llmfs + +import ( + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// NewFile is a write-only file that resets the conversation when written to +type NewFile struct { + *protocol.BaseFile + client *llm.Client +} + +// NewNewFile creates the new file +func NewNewFile(client *llm.Client) *NewFile { + return &NewFile{ + BaseFile: protocol.NewBaseFile("new", 0222), + client: client, + } +} + +func (f *NewFile) Read(p []byte, offset int64) (int, error) { + return 0, protocol.ErrPermission +} + +func (f *NewFile) Write(p []byte, offset int64) (int, error) { + // Any write resets the conversation + f.client.Reset() + return len(p), nil +} + +func (f *NewFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + s.Length = 0 + return s +} diff --git a/internal/llmfs/root.go b/internal/llmfs/root.go new file mode 100644 index 0000000000000000000000000000000000000000..c59448ac84e8b3a206e7e19a2be495ca2875565d --- /dev/null +++ b/internal/llmfs/root.go @@ -0,0 +1,28 @@ +// Package llmfs implements the LLM filesystem exposed via 9P. +package llmfs + +import ( + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// NewRoot creates the root directory of the LLM filesystem +func NewRoot(client *llm.Client) protocol.Dir { + root := protocol.NewStaticDir("llm") + + // Add all files + root.AddChild(NewAskFile(client)) + root.AddChild(NewModelFile(client)) + root.AddChild(NewTemperatureFile(client)) + root.AddChild(NewTokensFile(client)) + root.AddChild(NewNewFile(client)) + root.AddChild(NewContextFile(client)) + root.AddChild(NewExampleFile()) + + // Add stream directory + streamDir := protocol.NewStaticDir("stream") + streamDir.AddChild(NewChunkFile(client)) + root.AddChild(streamDir) + + return root +} diff --git a/internal/llmfs/state.go b/internal/llmfs/state.go new file mode 100644 index 0000000000000000000000000000000000000000..ddd1c30b480756081ac1e00c743285a5a980c25d --- /dev/null +++ b/internal/llmfs/state.go @@ -0,0 +1,91 @@ +package llmfs + +import ( + "fmt" + "io" + "strconv" + "strings" + + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// ModelFile exposes the current model name (read/write) +type ModelFile struct { + *protocol.BaseFile + client *llm.Client +} + +// NewModelFile creates the model file +func NewModelFile(client *llm.Client) *ModelFile { + return &ModelFile{ + BaseFile: protocol.NewBaseFile("model", 0666), + client: client, + } +} + +func (f *ModelFile) Read(p []byte, offset int64) (int, error) { + content := f.client.Model() + "\n" + if offset >= int64(len(content)) { + return 0, io.EOF + } + n := copy(p, content[offset:]) + return n, nil +} + +func (f *ModelFile) Write(p []byte, offset int64) (int, error) { + model := strings.TrimSpace(string(p)) + if model == "" { + return 0, fmt.Errorf("model name cannot be empty") + } + f.client.SetModel(model) + return len(p), nil +} + +func (f *ModelFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + s.Length = uint64(len(f.client.Model()) + 1) // +1 for newline + return s +} + +// TemperatureFile exposes the current temperature (read/write) +type TemperatureFile struct { + *protocol.BaseFile + client *llm.Client +} + +// NewTemperatureFile creates the temperature file +func NewTemperatureFile(client *llm.Client) *TemperatureFile { + return &TemperatureFile{ + BaseFile: protocol.NewBaseFile("temperature", 0666), + client: client, + } +} + +func (f *TemperatureFile) Read(p []byte, offset int64) (int, error) { + content := fmt.Sprintf("%.2f\n", f.client.Temperature()) + if offset >= int64(len(content)) { + return 0, io.EOF + } + n := copy(p, content[offset:]) + return n, nil +} + +func (f *TemperatureFile) Write(p []byte, offset int64) (int, error) { + tempStr := strings.TrimSpace(string(p)) + temp, err := strconv.ParseFloat(tempStr, 64) + if err != nil { + return 0, fmt.Errorf("invalid temperature: %w", err) + } + if err := f.client.SetTemperature(temp); err != nil { + return 0, err + } + return len(p), nil +} + +func (f *TemperatureFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + content := fmt.Sprintf("%.2f\n", f.client.Temperature()) + s.Length = uint64(len(content)) + return s +} diff --git a/internal/llmfs/stream.go b/internal/llmfs/stream.go new file mode 100644 index 0000000000000000000000000000000000000000..99302e1fd32168f41c7c6f27f90d7e25bba489ed --- /dev/null +++ b/internal/llmfs/stream.go @@ -0,0 +1,53 @@ +package llmfs + +import ( + "io" + + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// ChunkFile provides streaming access to LLM responses +// Reading blocks until the next chunk is available, then returns it +// Returns EOF when the stream is complete +type ChunkFile struct { + *protocol.BaseFile + client *llm.Client +} + +// NewChunkFile creates the stream/chunk file +func NewChunkFile(client *llm.Client) *ChunkFile { + return &ChunkFile{ + BaseFile: protocol.NewBaseFile("chunk", 0444), + client: client, + } +} + +func (f *ChunkFile) Read(p []byte, offset int64) (int, error) { + // If no stream is active, return EOF + if !f.client.IsStreaming() { + return 0, io.EOF + } + + // Block until we get a chunk + chunk, ok := f.client.ReadStreamChunk() + if !ok { + // Stream ended + return 0, io.EOF + } + + // Copy the chunk to the buffer + n := copy(p, chunk) + return n, nil +} + +func (f *ChunkFile) Write(p []byte, offset int64) (int, error) { + return 0, protocol.ErrPermission +} + +func (f *ChunkFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + // Length is unknown for streaming + s.Length = 0 + return s +} diff --git a/internal/llmfs/tokens.go b/internal/llmfs/tokens.go new file mode 100644 index 0000000000000000000000000000000000000000..0e46917cf4e821a950cd8f501ff0afe369e445a5 --- /dev/null +++ b/internal/llmfs/tokens.go @@ -0,0 +1,43 @@ +package llmfs + +import ( + "fmt" + "io" + + "github.com/NERVsystems/llm9p/internal/llm" + "github.com/NERVsystems/llm9p/internal/protocol" +) + +// TokensFile exposes the last response token count (read-only) +type TokensFile struct { + *protocol.BaseFile + client *llm.Client +} + +// NewTokensFile creates the tokens file +func NewTokensFile(client *llm.Client) *TokensFile { + return &TokensFile{ + BaseFile: protocol.NewBaseFile("tokens", 0444), + client: client, + } +} + +func (f *TokensFile) Read(p []byte, offset int64) (int, error) { + content := fmt.Sprintf("%d\n", f.client.LastTokens()) + if offset >= int64(len(content)) { + return 0, io.EOF + } + n := copy(p, content[offset:]) + return n, nil +} + +func (f *TokensFile) Write(p []byte, offset int64) (int, error) { + return 0, protocol.ErrPermission +} + +func (f *TokensFile) Stat() protocol.Stat { + s := f.BaseFile.Stat() + content := fmt.Sprintf("%d\n", f.client.LastTokens()) + s.Length = uint64(len(content)) + return s +} diff --git a/internal/protocol/fs.go b/internal/protocol/fs.go new file mode 100644 index 0000000000000000000000000000000000000000..e137af1e1b5f9feb873a491228890f42138b4494 --- /dev/null +++ b/internal/protocol/fs.go @@ -0,0 +1,230 @@ +package protocol + +import ( + "io" + "sync/atomic" + "time" +) + +// File is the interface that files must implement. +// This is the core abstraction for anything exposed via 9P. +type File interface { + // Stat returns the file's metadata + Stat() Stat + + // Open prepares the file for reading/writing + Open(mode uint8) error + + // Read reads up to len(p) bytes starting at offset + Read(p []byte, offset int64) (n int, err error) + + // Write writes len(p) bytes starting at offset + Write(p []byte, offset int64) (n int, err error) + + // Close releases any resources + Close() error +} + +// Dir is the interface that directories must implement +type Dir interface { + File + + // Children returns the directory's children + Children() []File + + // Lookup finds a child by name + Lookup(name string) (File, error) +} + +// pathCounter generates unique path IDs for qids +var pathCounter uint64 + +func NextPath() uint64 { + return atomic.AddUint64(&pathCounter, 1) +} + +// BaseFile provides a default implementation of common File methods +type BaseFile struct { + Name_ string + Mode_ uint32 + Uid_ string + Gid_ string + Qid_ Qid + Mtime_ time.Time + Length_ uint64 +} + +// NewBaseFile creates a new base file +func NewBaseFile(name string, mode uint32) *BaseFile { + now := time.Now() + qtype := QTFILE + if mode&DMDIR != 0 { + qtype = QTDIR + } + return &BaseFile{ + Name_: name, + Mode_: mode, + Uid_: "llm", + Gid_: "llm", + Mtime_: now, + Qid_: Qid{ + Type: qtype, + Version: 0, + Path: NextPath(), + }, + } +} + +func (f *BaseFile) Stat() Stat { + return Stat{ + Type: 0, + Dev: 0, + Qid: f.Qid_, + Mode: f.Mode_, + Atime: uint32(f.Mtime_.Unix()), + Mtime: uint32(f.Mtime_.Unix()), + Length: f.Length_, + Name: f.Name_, + Uid: f.Uid_, + Gid: f.Gid_, + Muid: f.Uid_, + } +} + +func (f *BaseFile) Open(mode uint8) error { return nil } +func (f *BaseFile) Close() error { return nil } +func (f *BaseFile) Read(p []byte, offset int64) (int, error) { return 0, io.EOF } +func (f *BaseFile) Write(p []byte, offset int64) (int, error) { return 0, ErrPermission } + +// SetLength updates the file length +func (f *BaseFile) SetLength(n uint64) { + f.Length_ = n + f.Mtime_ = time.Now() + f.Qid_.Version++ +} + +// StaticFile is a file with static content +type StaticFile struct { + *BaseFile + Content []byte +} + +// NewStaticFile creates a file with static content +func NewStaticFile(name string, content []byte) *StaticFile { + f := &StaticFile{ + BaseFile: NewBaseFile(name, 0444), + Content: content, + } + f.Length_ = uint64(len(content)) + return f +} + +func (f *StaticFile) Read(p []byte, offset int64) (int, error) { + if offset >= int64(len(f.Content)) { + return 0, io.EOF + } + n := copy(p, f.Content[offset:]) + return n, nil +} + +// StaticDir is a directory with static children +type StaticDir struct { + *BaseFile + children map[string]File + order []string // preserve order for listing +} + +// NewStaticDir creates a new static directory +func NewStaticDir(name string) *StaticDir { + return &StaticDir{ + BaseFile: NewBaseFile(name, DMDIR|0555), + children: make(map[string]File), + order: make([]string, 0), + } +} + +// AddChild adds a child to the directory +func (d *StaticDir) AddChild(f File) { + name := f.Stat().Name + if _, exists := d.children[name]; !exists { + d.order = append(d.order, name) + } + d.children[name] = f +} + +func (d *StaticDir) Children() []File { + result := make([]File, len(d.order)) + for i, name := range d.order { + result[i] = d.children[name] + } + return result +} + +func (d *StaticDir) Lookup(name string) (File, error) { + if f, ok := d.children[name]; ok { + return f, nil + } + return nil, ErrNotFound +} + +func (d *StaticDir) Read(p []byte, offset int64) (int, error) { + // Directory read returns packed stat entries + var buf []byte + for _, f := range d.Children() { + stat := f.Stat() + entry := make([]byte, 256) + n := stat.Encode(entry) + buf = append(buf, entry[:n]...) + } + + if offset >= int64(len(buf)) { + return 0, io.EOF + } + + n := copy(p, buf[offset:]) + return n, nil +} + +// DynamicFile is a file whose content is generated on read +type DynamicFile struct { + *BaseFile + Generator func() []byte +} + +// NewDynamicFile creates a file with dynamic content +func NewDynamicFile(name string, generator func() []byte) *DynamicFile { + return &DynamicFile{ + BaseFile: NewBaseFile(name, 0444), + Generator: generator, + } +} + +func (f *DynamicFile) Read(p []byte, offset int64) (int, error) { + content := f.Generator() + if offset >= int64(len(content)) { + return 0, io.EOF + } + n := copy(p, content[offset:]) + return n, nil +} + +func (f *DynamicFile) Stat() Stat { + s := f.BaseFile.Stat() + s.Length = uint64(len(f.Generator())) + return s +} + +// Errors +type Error string + +func (e Error) Error() string { return string(e) } + +const ( + ErrNotFound Error = "file not found" + ErrPermission Error = "permission denied" + ErrNotDir Error = "not a directory" + ErrIsDir Error = "is a directory" + ErrBadFid Error = "bad fid" + ErrFidInUse Error = "fid already in use" + ErrBadOffset Error = "bad offset" +) diff --git a/internal/protocol/message.go b/internal/protocol/message.go new file mode 100644 index 0000000000000000000000000000000000000000..e51f9dd804a2140e928f9ab3072c516962da4519 --- /dev/null +++ b/internal/protocol/message.go @@ -0,0 +1,381 @@ +package protocol + +import ( + "encoding/binary" + "fmt" +) + +// Message is the interface implemented by all 9P messages +type Message interface { + Type() uint8 + Encode(buf []byte) int +} + +// Request messages (T-messages) + +// TversionMsg negotiates protocol version +type TversionMsg struct { + Msize uint32 // maximum message size + Version string // protocol version string +} + +func (m *TversionMsg) Type() uint8 { return Tversion } + +func (m *TversionMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Msize) + return 4 + EncodeString(buf[4:], m.Version) +} + +func DecodeTversion(buf []byte) (*TversionMsg, error) { + if len(buf) < 6 { + return nil, fmt.Errorf("Tversion too short") + } + m := &TversionMsg{ + Msize: binary.LittleEndian.Uint32(buf[0:4]), + } + m.Version, _ = DecodeString(buf[4:]) + return m, nil +} + +// RversionMsg is the response to Tversion +type RversionMsg struct { + Msize uint32 + Version string +} + +func (m *RversionMsg) Type() uint8 { return Rversion } + +func (m *RversionMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Msize) + return 4 + EncodeString(buf[4:], m.Version) +} + +// TattachMsg attaches to a filesystem +type TattachMsg struct { + Fid uint32 // fid to use for this connection + Afid uint32 // auth fid (NoFid if no auth) + Uname string // user name + Aname string // attach name (filesystem to attach) +} + +func (m *TattachMsg) Type() uint8 { return Tattach } + +func (m *TattachMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Fid) + binary.LittleEndian.PutUint32(buf[4:8], m.Afid) + n := 8 + n += EncodeString(buf[n:], m.Uname) + n += EncodeString(buf[n:], m.Aname) + return n +} + +func DecodeTattach(buf []byte) (*TattachMsg, error) { + if len(buf) < 12 { + return nil, fmt.Errorf("Tattach too short") + } + m := &TattachMsg{ + Fid: binary.LittleEndian.Uint32(buf[0:4]), + Afid: binary.LittleEndian.Uint32(buf[4:8]), + } + n := 8 + var sn int + m.Uname, sn = DecodeString(buf[n:]) + n += sn + m.Aname, _ = DecodeString(buf[n:]) + return m, nil +} + +// RattachMsg is the response to Tattach +type RattachMsg struct { + Qid Qid +} + +func (m *RattachMsg) Type() uint8 { return Rattach } + +func (m *RattachMsg) Encode(buf []byte) int { + return m.Qid.Encode(buf) +} + +// TwalkMsg walks a path +type TwalkMsg struct { + Fid uint32 // starting fid + Newfid uint32 // fid for the result + Names []string // path components to walk +} + +func (m *TwalkMsg) Type() uint8 { return Twalk } + +func (m *TwalkMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Fid) + binary.LittleEndian.PutUint32(buf[4:8], m.Newfid) + binary.LittleEndian.PutUint16(buf[8:10], uint16(len(m.Names))) + n := 10 + for _, name := range m.Names { + n += EncodeString(buf[n:], name) + } + return n +} + +func DecodeTwalk(buf []byte) (*TwalkMsg, error) { + if len(buf) < 10 { + return nil, fmt.Errorf("Twalk too short") + } + m := &TwalkMsg{ + Fid: binary.LittleEndian.Uint32(buf[0:4]), + Newfid: binary.LittleEndian.Uint32(buf[4:8]), + } + nwname := binary.LittleEndian.Uint16(buf[8:10]) + m.Names = make([]string, nwname) + n := 10 + for i := range m.Names { + var sn int + m.Names[i], sn = DecodeString(buf[n:]) + n += sn + } + return m, nil +} + +// RwalkMsg is the response to Twalk +type RwalkMsg struct { + Qids []Qid // qids for each successfully walked element +} + +func (m *RwalkMsg) Type() uint8 { return Rwalk } + +func (m *RwalkMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint16(buf[0:2], uint16(len(m.Qids))) + n := 2 + for i := range m.Qids { + n += m.Qids[i].Encode(buf[n:]) + } + return n +} + +// TopenMsg opens a file +type TopenMsg struct { + Fid uint32 + Mode uint8 +} + +func (m *TopenMsg) Type() uint8 { return Topen } + +func (m *TopenMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Fid) + buf[4] = m.Mode + return 5 +} + +func DecodeTopen(buf []byte) (*TopenMsg, error) { + if len(buf) < 5 { + return nil, fmt.Errorf("Topen too short") + } + return &TopenMsg{ + Fid: binary.LittleEndian.Uint32(buf[0:4]), + Mode: buf[4], + }, nil +} + +// RopenMsg is the response to Topen +type RopenMsg struct { + Qid Qid + Iounit uint32 +} + +func (m *RopenMsg) Type() uint8 { return Ropen } + +func (m *RopenMsg) Encode(buf []byte) int { + n := m.Qid.Encode(buf) + binary.LittleEndian.PutUint32(buf[n:n+4], m.Iounit) + return n + 4 +} + +// TreadMsg reads from a file +type TreadMsg struct { + Fid uint32 + Offset uint64 + Count uint32 +} + +func (m *TreadMsg) Type() uint8 { return Tread } + +func (m *TreadMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Fid) + binary.LittleEndian.PutUint64(buf[4:12], m.Offset) + binary.LittleEndian.PutUint32(buf[12:16], m.Count) + return 16 +} + +func DecodeTread(buf []byte) (*TreadMsg, error) { + if len(buf) < 16 { + return nil, fmt.Errorf("Tread too short") + } + return &TreadMsg{ + Fid: binary.LittleEndian.Uint32(buf[0:4]), + Offset: binary.LittleEndian.Uint64(buf[4:12]), + Count: binary.LittleEndian.Uint32(buf[12:16]), + }, nil +} + +// RreadMsg is the response to Tread +type RreadMsg struct { + Data []byte +} + +func (m *RreadMsg) Type() uint8 { return Rread } + +func (m *RreadMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], uint32(len(m.Data))) + copy(buf[4:], m.Data) + return 4 + len(m.Data) +} + +// TwriteMsg writes to a file +type TwriteMsg struct { + Fid uint32 + Offset uint64 + Data []byte +} + +func (m *TwriteMsg) Type() uint8 { return Twrite } + +func (m *TwriteMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Fid) + binary.LittleEndian.PutUint64(buf[4:12], m.Offset) + binary.LittleEndian.PutUint32(buf[12:16], uint32(len(m.Data))) + copy(buf[16:], m.Data) + return 16 + len(m.Data) +} + +func DecodeTwrite(buf []byte) (*TwriteMsg, error) { + if len(buf) < 16 { + return nil, fmt.Errorf("Twrite too short") + } + count := binary.LittleEndian.Uint32(buf[12:16]) + if len(buf) < int(16+count) { + return nil, fmt.Errorf("Twrite data truncated") + } + return &TwriteMsg{ + Fid: binary.LittleEndian.Uint32(buf[0:4]), + Offset: binary.LittleEndian.Uint64(buf[4:12]), + Data: buf[16 : 16+count], + }, nil +} + +// RwriteMsg is the response to Twrite +type RwriteMsg struct { + Count uint32 +} + +func (m *RwriteMsg) Type() uint8 { return Rwrite } + +func (m *RwriteMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Count) + return 4 +} + +// TclunkMsg closes a fid +type TclunkMsg struct { + Fid uint32 +} + +func (m *TclunkMsg) Type() uint8 { return Tclunk } + +func (m *TclunkMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Fid) + return 4 +} + +func DecodeTclunk(buf []byte) (*TclunkMsg, error) { + if len(buf) < 4 { + return nil, fmt.Errorf("Tclunk too short") + } + return &TclunkMsg{ + Fid: binary.LittleEndian.Uint32(buf[0:4]), + }, nil +} + +// RclunkMsg is the response to Tclunk +type RclunkMsg struct{} + +func (m *RclunkMsg) Type() uint8 { return Rclunk } + +func (m *RclunkMsg) Encode(buf []byte) int { + return 0 +} + +// TstatMsg requests file stats +type TstatMsg struct { + Fid uint32 +} + +func (m *TstatMsg) Type() uint8 { return Tstat } + +func (m *TstatMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint32(buf[0:4], m.Fid) + return 4 +} + +func DecodeTstat(buf []byte) (*TstatMsg, error) { + if len(buf) < 4 { + return nil, fmt.Errorf("Tstat too short") + } + return &TstatMsg{ + Fid: binary.LittleEndian.Uint32(buf[0:4]), + }, nil +} + +// RstatMsg is the response to Tstat +type RstatMsg struct { + Stat Stat +} + +func (m *RstatMsg) Type() uint8 { return Rstat } + +func (m *RstatMsg) Encode(buf []byte) int { + // Rstat has an extra 2-byte length prefix for the stat + statBuf := buf[2:] + n := m.Stat.Encode(statBuf) + binary.LittleEndian.PutUint16(buf[0:2], uint16(n)) + return 2 + n +} + +// RerrorMsg indicates an error +type RerrorMsg struct { + Ename string +} + +func (m *RerrorMsg) Type() uint8 { return Rerror } + +func (m *RerrorMsg) Encode(buf []byte) int { + return EncodeString(buf, m.Ename) +} + +// TflushMsg cancels a pending request +type TflushMsg struct { + Oldtag uint16 +} + +func (m *TflushMsg) Type() uint8 { return Tflush } + +func (m *TflushMsg) Encode(buf []byte) int { + binary.LittleEndian.PutUint16(buf[0:2], m.Oldtag) + return 2 +} + +func DecodeTflush(buf []byte) (*TflushMsg, error) { + if len(buf) < 2 { + return nil, fmt.Errorf("Tflush too short") + } + return &TflushMsg{ + Oldtag: binary.LittleEndian.Uint16(buf[0:2]), + }, nil +} + +// RflushMsg is the response to Tflush +type RflushMsg struct{} + +func (m *RflushMsg) Type() uint8 { return Rflush } + +func (m *RflushMsg) Encode(buf []byte) int { + return 0 +} diff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go new file mode 100644 index 0000000000000000000000000000000000000000..ecad15655498ec1a86b834a9f3c0914d7bf3fee8 --- /dev/null +++ b/internal/protocol/protocol.go @@ -0,0 +1,327 @@ +// Package protocol implements the 9P2000 protocol for the LLM filesystem. +// +// This is a minimal, clean implementation focused on the subset of 9P +// needed for LLM interaction. It is designed to be: +// - Zero external dependencies (stdlib only) +// - LLM-friendly (self-describing, good errors) +// - Simple to understand and maintain +// +// The 9P protocol uses a simple request-response model over a bidirectional +// stream. Each message has a 4-byte size, 1-byte type, and 2-byte tag, +// followed by type-specific payload. +package protocol + +import ( + "encoding/binary" + "fmt" + "io" +) + +// Protocol constants +const ( + // Version is the protocol version we implement + Version = "9P2000" + + // MaxMessageSize is the maximum size of a 9P message + MaxMessageSize = 8192 + + // NoTag is used for Tversion/Rversion which don't use tags + NoTag uint16 = 0xFFFF + + // NoFid represents an invalid fid + NoFid uint32 = 0xFFFFFFFF +) + +// Message types (T = request from client, R = response from server) +const ( + Tversion uint8 = 100 + Rversion uint8 = 101 + Tauth uint8 = 102 + Rauth uint8 = 103 + Tattach uint8 = 104 + Rattach uint8 = 105 + Terror uint8 = 106 // never sent + Rerror uint8 = 107 + Tflush uint8 = 108 + Rflush uint8 = 109 + Twalk uint8 = 110 + Rwalk uint8 = 111 + Topen uint8 = 112 + Ropen uint8 = 113 + Tcreate uint8 = 114 + Rcreate uint8 = 115 + Tread uint8 = 116 + Rread uint8 = 117 + Twrite uint8 = 118 + Rwrite uint8 = 119 + Tclunk uint8 = 120 + Rclunk uint8 = 121 + Tremove uint8 = 122 + Rremove uint8 = 123 + Tstat uint8 = 124 + Rstat uint8 = 125 + Twstat uint8 = 126 + Rwstat uint8 = 127 +) + +// Open modes +const ( + OREAD uint8 = 0 // open for read + OWRITE uint8 = 1 // open for write + ORDWR uint8 = 2 // open for read/write + OEXEC uint8 = 3 // execute (unused in our context) + OTRUNC uint8 = 16 // truncate file first +) + +// File modes (high bits of Stat.Mode) +const ( + DMDIR uint32 = 0x80000000 // directory + DMAPPEND uint32 = 0x40000000 // append only + DMEXCL uint32 = 0x20000000 // exclusive use + DMTMP uint32 = 0x04000000 // temporary file +) + +// Qid represents a unique file identifier +type Qid struct { + Type uint8 // QTDIR, QTFILE, etc. + Version uint32 // version number for cache coherence + Path uint64 // unique path identifier +} + +// Qid types +const ( + QTDIR uint8 = 0x80 // directory + QTAPPEND uint8 = 0x40 // append-only + QTEXCL uint8 = 0x20 // exclusive use + QTTMP uint8 = 0x04 // temporary + QTFILE uint8 = 0x00 // regular file +) + +// Stat represents file metadata +type Stat struct { + Size uint16 // size of this stat structure (for wire format) + Type uint16 // server type + Dev uint32 // server device + Qid Qid // unique id + Mode uint32 // permissions and flags + Atime uint32 // last access time + Mtime uint32 // last modification time + Length uint64 // file length + Name string // file name + Uid string // owner + Gid string // group + Muid string // last modifier +} + +// Encoder handles encoding messages to the wire format +type Encoder struct { + w io.Writer + buf []byte +} + +// NewEncoder creates a new encoder +func NewEncoder(w io.Writer) *Encoder { + return &Encoder{ + w: w, + buf: make([]byte, MaxMessageSize), + } +} + +// Decoder handles decoding messages from the wire format +type Decoder struct { + r io.Reader + buf []byte +} + +// NewDecoder creates a new decoder +func NewDecoder(r io.Reader) *Decoder { + return &Decoder{ + r: r, + buf: make([]byte, MaxMessageSize), + } +} + +// ReadMessage reads a complete 9P message from the stream +func (d *Decoder) ReadMessage() (msgType uint8, tag uint16, payload []byte, err error) { + // Read 4-byte size + if _, err := io.ReadFull(d.r, d.buf[:4]); err != nil { + return 0, 0, nil, fmt.Errorf("reading size: %w", err) + } + size := binary.LittleEndian.Uint32(d.buf[:4]) + + if size < 7 { + return 0, 0, nil, fmt.Errorf("message too small: %d", size) + } + if size > MaxMessageSize { + return 0, 0, nil, fmt.Errorf("message too large: %d", size) + } + + // Read rest of message + remaining := size - 4 + if _, err := io.ReadFull(d.r, d.buf[:remaining]); err != nil { + return 0, 0, nil, fmt.Errorf("reading message: %w", err) + } + + msgType = d.buf[0] + tag = binary.LittleEndian.Uint16(d.buf[1:3]) + payload = d.buf[3:remaining] + + return msgType, tag, payload, nil +} + +// WriteMessage writes a complete 9P message to the stream +func (e *Encoder) WriteMessage(msgType uint8, tag uint16, payload []byte) error { + size := uint32(4 + 1 + 2 + len(payload)) + if size > MaxMessageSize { + return fmt.Errorf("message too large: %d", size) + } + + binary.LittleEndian.PutUint32(e.buf[0:4], size) + e.buf[4] = msgType + binary.LittleEndian.PutUint16(e.buf[5:7], tag) + copy(e.buf[7:], payload) + + _, err := e.w.Write(e.buf[:size]) + return err +} + +// String encoding helpers + +func EncodeString(buf []byte, s string) int { + binary.LittleEndian.PutUint16(buf[0:2], uint16(len(s))) + copy(buf[2:], s) + return 2 + len(s) +} + +func DecodeString(buf []byte) (string, int) { + if len(buf) < 2 { + return "", 0 + } + size := binary.LittleEndian.Uint16(buf[0:2]) + if len(buf) < int(2+size) { + return "", 0 + } + return string(buf[2 : 2+size]), int(2 + size) +} + +// Qid encoding + +func (q *Qid) Encode(buf []byte) int { + buf[0] = q.Type + binary.LittleEndian.PutUint32(buf[1:5], q.Version) + binary.LittleEndian.PutUint64(buf[5:13], q.Path) + return 13 +} + +func DecodeQid(buf []byte) (Qid, int) { + if len(buf) < 13 { + return Qid{}, 0 + } + return Qid{ + Type: buf[0], + Version: binary.LittleEndian.Uint32(buf[1:5]), + Path: binary.LittleEndian.Uint64(buf[5:13]), + }, 13 +} + +// Stat encoding + +func (s *Stat) Encode(buf []byte) int { + // Skip size field, we'll fill it at the end + n := 2 + + // Fixed fields + binary.LittleEndian.PutUint16(buf[n:n+2], s.Type) + n += 2 + binary.LittleEndian.PutUint32(buf[n:n+4], s.Dev) + n += 4 + n += s.Qid.Encode(buf[n:]) + binary.LittleEndian.PutUint32(buf[n:n+4], s.Mode) + n += 4 + binary.LittleEndian.PutUint32(buf[n:n+4], s.Atime) + n += 4 + binary.LittleEndian.PutUint32(buf[n:n+4], s.Mtime) + n += 4 + binary.LittleEndian.PutUint64(buf[n:n+8], s.Length) + n += 8 + + // Variable fields + n += EncodeString(buf[n:], s.Name) + n += EncodeString(buf[n:], s.Uid) + n += EncodeString(buf[n:], s.Gid) + n += EncodeString(buf[n:], s.Muid) + + // Fill in size (total - 2 for size field itself) + s.Size = uint16(n - 2) + binary.LittleEndian.PutUint16(buf[0:2], s.Size) + + return n +} + +func DecodeStat(buf []byte) (Stat, int) { + if len(buf) < 2 { + return Stat{}, 0 + } + + s := Stat{} + s.Size = binary.LittleEndian.Uint16(buf[0:2]) + + if len(buf) < int(s.Size)+2 { + return Stat{}, 0 + } + + n := 2 + s.Type = binary.LittleEndian.Uint16(buf[n : n+2]) + n += 2 + s.Dev = binary.LittleEndian.Uint32(buf[n : n+4]) + n += 4 + + var qn int + s.Qid, qn = DecodeQid(buf[n:]) + n += qn + + s.Mode = binary.LittleEndian.Uint32(buf[n : n+4]) + n += 4 + s.Atime = binary.LittleEndian.Uint32(buf[n : n+4]) + n += 4 + s.Mtime = binary.LittleEndian.Uint32(buf[n : n+4]) + n += 4 + s.Length = binary.LittleEndian.Uint64(buf[n : n+8]) + n += 8 + + var sn int + s.Name, sn = DecodeString(buf[n:]) + n += sn + s.Uid, sn = DecodeString(buf[n:]) + n += sn + s.Gid, sn = DecodeString(buf[n:]) + n += sn + s.Muid, sn = DecodeString(buf[n:]) + n += sn + + return s, int(s.Size) + 2 +} + +// MessageName returns the human-readable name of a message type +func MessageName(t uint8) string { + names := map[uint8]string{ + Tversion: "Tversion", Rversion: "Rversion", + Tauth: "Tauth", Rauth: "Rauth", + Tattach: "Tattach", Rattach: "Rattach", + Rerror: "Rerror", + Tflush: "Tflush", Rflush: "Rflush", + Twalk: "Twalk", Rwalk: "Rwalk", + Topen: "Topen", Ropen: "Ropen", + Tcreate: "Tcreate", Rcreate: "Rcreate", + Tread: "Tread", Rread: "Rread", + Twrite: "Twrite", Rwrite: "Rwrite", + Tclunk: "Tclunk", Rclunk: "Rclunk", + Tremove: "Tremove", Rremove: "Rremove", + Tstat: "Tstat", Rstat: "Rstat", + Twstat: "Twstat", Rwstat: "Rwstat", + } + if name, ok := names[t]; ok { + return name + } + return fmt.Sprintf("unknown(%d)", t) +} diff --git a/internal/protocol/server.go b/internal/protocol/server.go new file mode 100644 index 0000000000000000000000000000000000000000..123c52fb9f692018cd9b490818907379618667db --- /dev/null +++ b/internal/protocol/server.go @@ -0,0 +1,346 @@ +package protocol + +import ( + "context" + "fmt" + "io" + "log" + "net" + "sync" +) + +// Server is a 9P file server +type Server struct { + root Dir + debug bool + mu sync.Mutex + clients map[net.Conn]*clientState +} + +// clientState tracks state for a single client connection +type clientState struct { + fids map[uint32]File + msize uint32 +} + +// NewServer creates a new 9P server with the given root directory +func NewServer(root Dir) *Server { + return &Server{ + root: root, + clients: make(map[net.Conn]*clientState), + } +} + +// SetDebug enables debug logging +func (s *Server) SetDebug(debug bool) { + s.debug = debug +} + +// Serve handles incoming connections on the listener +func (s *Server) Serve(ctx context.Context, listener net.Listener) error { + for { + conn, err := listener.Accept() + if err != nil { + select { + case <-ctx.Done(): + return ctx.Err() + default: + log.Printf("accept error: %v", err) + continue + } + } + + go s.handleConn(conn) + } +} + +// ServeConn handles a single connection (useful for testing) +func (s *Server) ServeConn(conn net.Conn) { + s.handleConn(conn) +} + +func (s *Server) handleConn(conn net.Conn) { + defer conn.Close() + + state := &clientState{ + fids: make(map[uint32]File), + msize: MaxMessageSize, + } + + s.mu.Lock() + s.clients[conn] = state + s.mu.Unlock() + + defer func() { + s.mu.Lock() + delete(s.clients, conn) + s.mu.Unlock() + }() + + dec := NewDecoder(conn) + enc := NewEncoder(conn) + buf := make([]byte, MaxMessageSize) + + for { + msgType, tag, payload, err := dec.ReadMessage() + if err != nil { + if err != io.EOF { + log.Printf("read error: %v", err) + } + return + } + + if s.debug { + log.Printf("< %s tag=%d len=%d", MessageName(msgType), tag, len(payload)) + } + + resp, respType := s.handleMessage(state, msgType, payload, buf) + + if s.debug { + log.Printf("> %s tag=%d len=%d", MessageName(respType), tag, len(resp)) + } + + if err := enc.WriteMessage(respType, tag, resp); err != nil { + log.Printf("write error: %v", err) + return + } + } +} + +func (s *Server) handleMessage(state *clientState, msgType uint8, payload []byte, buf []byte) ([]byte, uint8) { + switch msgType { + case Tversion: + return s.handleVersion(state, payload, buf) + case Tattach: + return s.handleAttach(state, payload, buf) + case Twalk: + return s.handleWalk(state, payload, buf) + case Topen: + return s.handleOpen(state, payload, buf) + case Tread: + return s.handleRead(state, payload, buf) + case Twrite: + return s.handleWrite(state, payload, buf) + case Tclunk: + return s.handleClunk(state, payload, buf) + case Tstat: + return s.handleStat(state, payload, buf) + case Tflush: + return s.handleFlush(state, payload, buf) + default: + return s.errorResponse(buf, fmt.Sprintf("unknown message type: %d", msgType)) + } +} + +func (s *Server) errorResponse(buf []byte, msg string) ([]byte, uint8) { + resp := &RerrorMsg{Ename: msg} + n := resp.Encode(buf) + return buf[:n], Rerror +} + +func (s *Server) handleVersion(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTversion(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + // Negotiate message size + msize := msg.Msize + if msize > MaxMessageSize { + msize = MaxMessageSize + } + state.msize = msize + + // Check version - accept both 9P2000 and Styx (Inferno's name) + version := msg.Version + if msg.Version != Version && msg.Version != "Styx" { + version = "unknown" + } + + if s.debug { + log.Printf("Version negotiation: client=%q responding=%q msize=%d", msg.Version, version, msize) + } + + resp := &RversionMsg{Msize: msize, Version: version} + n := resp.Encode(buf) + return buf[:n], Rversion +} + +func (s *Server) handleAttach(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTattach(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + if _, exists := state.fids[msg.Fid]; exists { + return s.errorResponse(buf, ErrFidInUse.Error()) + } + + state.fids[msg.Fid] = s.root + + resp := &RattachMsg{Qid: s.root.Stat().Qid} + n := resp.Encode(buf) + return buf[:n], Rattach +} + +func (s *Server) handleWalk(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTwalk(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + file, exists := state.fids[msg.Fid] + if !exists { + return s.errorResponse(buf, ErrBadFid.Error()) + } + + if msg.Fid != msg.Newfid { + if _, exists := state.fids[msg.Newfid]; exists { + return s.errorResponse(buf, ErrFidInUse.Error()) + } + } + + // Walk the path + qids := make([]Qid, 0, len(msg.Names)) + current := file + + for _, name := range msg.Names { + dir, ok := current.(Dir) + if !ok { + return s.errorResponse(buf, ErrNotDir.Error()) + } + + next, err := dir.Lookup(name) + if err != nil { + // Return partial walk + break + } + + qids = append(qids, next.Stat().Qid) + current = next + } + + // 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 + } + + resp := &RwalkMsg{Qids: qids} + n := resp.Encode(buf) + return buf[:n], Rwalk +} + +func (s *Server) handleOpen(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTopen(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + file, exists := state.fids[msg.Fid] + if !exists { + return s.errorResponse(buf, ErrBadFid.Error()) + } + + if err := file.Open(msg.Mode); err != nil { + return s.errorResponse(buf, err.Error()) + } + + resp := &RopenMsg{ + Qid: file.Stat().Qid, + Iounit: 0, // 0 means use msize - overhead + } + n := resp.Encode(buf) + return buf[:n], Ropen +} + +func (s *Server) handleRead(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTread(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + file, exists := state.fids[msg.Fid] + if !exists { + return s.errorResponse(buf, ErrBadFid.Error()) + } + + // Limit read size to available buffer + count := msg.Count + maxData := state.msize - 4 - 1 - 2 - 4 // size, type, tag, count + if count > maxData { + count = maxData + } + + data := make([]byte, count) + n, err := file.Read(data, int64(msg.Offset)) + if err != nil && err != io.EOF { + return s.errorResponse(buf, err.Error()) + } + + resp := &RreadMsg{Data: data[:n]} + rn := resp.Encode(buf) + return buf[:rn], Rread +} + +func (s *Server) handleWrite(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTwrite(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + file, exists := state.fids[msg.Fid] + if !exists { + return s.errorResponse(buf, ErrBadFid.Error()) + } + + n, err := file.Write(msg.Data, int64(msg.Offset)) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + resp := &RwriteMsg{Count: uint32(n)} + rn := resp.Encode(buf) + return buf[:rn], Rwrite +} + +func (s *Server) handleClunk(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTclunk(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + file, exists := state.fids[msg.Fid] + if !exists { + return s.errorResponse(buf, ErrBadFid.Error()) + } + + file.Close() + delete(state.fids, msg.Fid) + + resp := &RclunkMsg{} + n := resp.Encode(buf) + return buf[:n], Rclunk +} + +func (s *Server) handleStat(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + msg, err := DecodeTstat(payload) + if err != nil { + return s.errorResponse(buf, err.Error()) + } + + file, exists := state.fids[msg.Fid] + if !exists { + return s.errorResponse(buf, ErrBadFid.Error()) + } + + resp := &RstatMsg{Stat: file.Stat()} + n := resp.Encode(buf) + return buf[:n], Rstat +} + +func (s *Server) handleFlush(state *clientState, payload []byte, buf []byte) ([]byte, uint8) { + // We don't have async operations to cancel, so just respond OK + resp := &RflushMsg{} + n := resp.Encode(buf) + return buf[:n], Rflush +}