fix: add backpressure logging, query timeouts, configurable privacy, and response limits

- Add rate-limited warning logging when slow-path queue drops jobs
- Add path canonicalization and directory validation in git watcher
- Return errors from critical hook operations instead of silent logging
- Add configurable FilterLevel (Strict/Moderate/Minimal) for privacy
- Trigger async flush when access tracker buffer full (instead of dropping)
- Add query timeout wrapper (30s default) in SQLite storage
- Add response size limit (5MB) with 413 on overflow in REST API

Author: Patel230

engine/access_tracker.go

@@ -46,10 +46,19 @@ func (at *AccessTracker) Log(ctx context.Context, nodeID string) {
 	at.mu.Lock()
 	if len(at.buffer) < maxAccessBuffer {
 		at.buffer = append(at.buffer, nodeID)
-	} else {
-		slog.Warn("access_tracker: buffer full, dropping access log", "node_id", nodeID)
+		at.mu.Unlock()
+		return
 	}
 	at.mu.Unlock()
+
+	// Buffer full — trigger an async flush instead of dropping
+	slog.Warn("access_tracker: buffer full, triggering forced flush", "node_id", nodeID)
+	go at.Flush(context.Background())
+
+	// After triggering flush, still try to buffer this item
+	at.mu.Lock()
+	at.buffer = append(at.buffer, nodeID)
+	at.mu.Unlock()
 }
 
 // Flush immediately applies all pending access counts to nodes.

git/watcher.go

@@ -3,7 +3,9 @@ package git
 import (
 	"context"
 	"fmt"
+	"os"
 	"os/exec"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -26,8 +28,22 @@ type Watcher struct {
 }
 
 // New creates a git Watcher for the given project directory.
-func New(store storage.Storage, g graph.Graph, dir string) *Watcher {
-	return &Watcher{store: store, graph: g, dir: dir}
+// Returns an error if the directory path is invalid or does not exist.
+func New(store storage.Storage, g graph.Graph, dir string) (*Watcher, error) {
+	absDir, err := filepath.Abs(dir)
+	if err != nil {
+		return nil, fmt.Errorf("git watcher: invalid path %q: %w", dir, err)
+	}
+	absDir = filepath.Clean(absDir)
+
+	info, err := os.Stat(absDir)
+	if err != nil {
+		return nil, fmt.Errorf("git watcher: path %q does not exist: %w", absDir, err)
+	}
+	if !info.IsDir() {
+		return nil, fmt.Errorf("git watcher: path %q is not a directory", absDir)
+	}
+	return &Watcher{store: store, graph: g, dir: absDir}, nil
 }
 
 // StalesSince returns stale reports for files changed since the given time.

hooks/runner.go

@@ -74,9 +74,14 @@ func (r *Runner) SessionStart(ctx context.Context, in *HookInput) error {
 	}
 
 	// Write session ID to a temp file for other hooks to pick up
-	if err := os.WriteFile(sessionFile(r.project), []byte(sessionID), 0600); err != nil {
-		// Best-effort: log but don't fail the hook
+	sf := sessionFile(r.project)
+	if err := os.MkdirAll(filepath.Dir(sf), 0755); err != nil {
+		fmt.Fprintf(os.Stderr, "yaad: warning: could not create session dir: %v\n", err)
+		return fmt.Errorf("create session dir: %w", err)
+	}
+	if err := os.WriteFile(sf, []byte(sessionID), 0600); err != nil {
 		fmt.Fprintf(os.Stderr, "yaad: warning: could not write session file: %v\n", err)
+		return fmt.Errorf("write session file: %w", err)
 	}
 
 	// Auto-decay: keep graph lean without manual intervention
@@ -143,6 +148,7 @@ func (r *Runner) SessionEnd(ctx context.Context, in *HookInput) error {
 			Agent:   in.Agent,
 		}); err != nil {
 			fmt.Fprintf(os.Stderr, "yaad: warning: could not store session summary: %v\n", err)
+			return fmt.Errorf("store session summary: %w", err)
 		}
 	}
 
@@ -152,6 +158,9 @@ func (r *Runner) SessionEnd(ctx context.Context, in *HookInput) error {
 	// Clean up session file (best-effort)
 	if rmErr := os.Remove(sessionFile(r.project)); rmErr != nil {
 		fmt.Fprintf(os.Stderr, "yaad: warning: could not remove session file: %v\n", rmErr)
+		if err == nil {
+			return fmt.Errorf("remove session file: %w", rmErr)
+		}
 	}
 	return err
 }

ingest/dual_stream.go

@@ -12,6 +12,7 @@ import (
 	"context"
 	"log/slog"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
@@ -39,6 +40,9 @@ type DualStream struct {
 	once     sync.Once
 	lastNode map[string]string // project → last node ID (temporal backbone)
 	mu       sync.Mutex
+
+	dropped     atomic.Int64
+	lastDropLog atomic.Int64 // unix timestamp of last drop warning
 }
 
 // New creates a DualStream ingestion manager.
@@ -90,6 +94,15 @@ func (ds *DualStream) Remember(ctx context.Context, in engine.RememberInput) (*s
 	}:
 	default:
 		// Queue full — skip slow path for this node (graceful degradation)
+		total := ds.dropped.Add(1)
+		now := time.Now().Unix()
+		last := ds.lastDropLog.Load()
+		if now-last >= 10 {
+			if ds.lastDropLog.CompareAndSwap(last, now) {
+				slog.Warn("slow-path queue full, dropping jobs",
+					"dropped_total", total, "node_id", node.ID)
+			}
+		}
 	}
 
 	return node, nil

internal/server/mcp.go

@@ -364,7 +364,10 @@ func (s *MCPServer) handleStale(ctx context.Context, req mcp.CallToolRequest) (*
 	if project == "" {
 		project, _ = os.Getwd()
 	}
-	watcher := gitwatch.New(s.eng.Store(), s.eng.Graph(), project)
+	watcher, err := gitwatch.New(s.eng.Store(), s.eng.Graph(), project)
+	if err != nil {
+		return nil, err
+	}
 	since := time.Now().Add(-7 * 24 * time.Hour)
 	reports, err := watcher.StalesSince(ctx, since)
 	if err != nil {

internal/server/rest.go

@@ -25,7 +25,8 @@ import (
 )
 
 const (
-	maxRequestBodySize = 1 << 20 // 1 MB
+	maxRequestBodySize = 1 << 20      // 1 MB
+	maxResponseSize    = 5 * (1 << 20) // 5 MB
 	maxGraphDepth      = 5
 )
 
@@ -201,7 +202,7 @@ func (s *RESTServer) handleRecall(w http.ResponseWriter, r *http.Request) {
 		httpErr(w, err, 500)
 		return
 	}
-	httpJSON(w, result, 200)
+	httpJSONCapped(w, result, 200)
 }
 
 func (s *RESTServer) handleContext(w http.ResponseWriter, r *http.Request) {
@@ -211,7 +212,7 @@ func (s *RESTServer) handleContext(w http.ResponseWriter, r *http.Request) {
 		httpErr(w, err, 500)
 		return
 	}
-	httpJSON(w, result, 200)
+	httpJSONCapped(w, result, 200)
 }
 
 func (s *RESTServer) handleLink(w http.ResponseWriter, r *http.Request) {
@@ -275,7 +276,7 @@ func (s *RESTServer) handleSubgraph(w http.ResponseWriter, r *http.Request) {
 		httpErr(w, err, 500)
 		return
 	}
-	httpJSON(w, sg, 200)
+	httpJSONCapped(w, sg, 200)
 }
 
 func (s *RESTServer) handleImpact(w http.ResponseWriter, r *http.Request) {
@@ -296,7 +297,7 @@ func (s *RESTServer) handleImpact(w http.ResponseWriter, r *http.Request) {
 			nodes = append(nodes, n)
 		}
 	}
-	httpJSON(w, nodes, 200)
+	httpJSONCapped(w, nodes, 200)
 }
 
 func (s *RESTServer) handleForget(w http.ResponseWriter, r *http.Request) {
@@ -475,7 +476,11 @@ func (s *RESTServer) handleStale(w http.ResponseWriter, r *http.Request) {
 		httpJSON(w, map[string]string{"status": "no project directory configured"}, 200)
 		return
 	}
-	watcher := gitwatch.New(s.eng.Store(), s.eng.Graph(), s.projectDir)
+	watcher, err := gitwatch.New(s.eng.Store(), s.eng.Graph(), s.projectDir)
+	if err != nil {
+		httpErr(w, err, 500)
+		return
+	}
 	since := time.Now().Add(-7 * 24 * time.Hour) // last 7 days
 	reports, err := watcher.StalesSince(r.Context(), since)
 	if err != nil {
@@ -537,7 +542,7 @@ func (s *RESTServer) handleHybridRecall(w http.ResponseWriter, r *http.Request)
 		return
 	}
 	reranked := engine.Rerank(r.Context(), scored, s.eng.Store())
-	httpJSON(w, reranked, 200)
+	httpJSONCapped(w, reranked, 200)
 }
 
 func (s *RESTServer) handleProactive(w http.ResponseWriter, r *http.Request) {
@@ -759,6 +764,23 @@ func httpJSON(w http.ResponseWriter, v any, code int) {
 	json.NewEncoder(w).Encode(v)
 }
 
+// httpJSONCapped encodes the response as JSON but returns a 413 error if the
+// serialized payload exceeds maxResponseSize.
+func httpJSONCapped(w http.ResponseWriter, v any, code int) {
+	data, err := json.Marshal(v)
+	if err != nil {
+		httpErr(w, fmt.Errorf("marshal response: %w", err), 500)
+		return
+	}
+	if len(data) > maxResponseSize {
+		httpErr(w, fmt.Errorf("response size %d bytes exceeds limit of %d bytes", len(data), maxResponseSize), 413)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(code)
+	w.Write(data)
+}
+
 func httpErr(w http.ResponseWriter, err error, code int) {
 	httpJSON(w, map[string]string{"error": err.Error()}, code)
 }

privacy/filter.go

@@ -6,7 +6,23 @@ import (
 	"strings"
 )
 
-var patterns = []*regexp.Regexp{
+// FilterLevel controls how aggressively content is filtered.
+type FilterLevel int
+
+const (
+	// Strict strips everything (emails, IPs, phones, secrets).
+	Strict FilterLevel = iota
+	// Moderate keeps infrastructure IPs (10.x, 192.168.x) and work-related emails.
+	Moderate
+	// Minimal only strips high-entropy secrets and explicit API keys.
+	Minimal
+)
+
+// DefaultFilterLevel is used by Filter() when no level is specified.
+var DefaultFilterLevel = Moderate
+
+// secretPatterns are always stripped (all levels including Minimal).
+var secretPatterns = []*regexp.Regexp{
 	// API keys
 	regexp.MustCompile(`sk-[a-zA-Z0-9]{20,}`),                    // OpenAI
 	regexp.MustCompile(`AKIA[A-Z0-9]{16}`),                        // AWS Access Key
@@ -26,23 +42,55 @@ var patterns = []*regexp.Regexp{
 	regexp.MustCompile(`-----BEGIN\s+\w+\s+PRIVATE\s+KEY-----[\s\S]*?-----END\s+\w+\s+PRIVATE\s+KEY-----`),
 	// Connection strings with passwords
 	regexp.MustCompile(`(?i)(postgres|mysql|mongodb)://[^\s"]+:[^\s"]+@[^\s"]+`),
-	// PII: email addresses
-	regexp.MustCompile(`[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}`),
+}
+
+// piiPatterns are stripped by Strict and Moderate (but not Minimal).
+var piiPatterns = []*regexp.Regexp{
 	// PII: US phone numbers
 	regexp.MustCompile(`\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b`),
 	// PII: US Social Security Numbers
 	regexp.MustCompile(`\b\d{3}-\d{2}-\d{4}\b`),
-	// PII: IPv4 addresses (non-localhost, non-private documentation ranges)
-	regexp.MustCompile(`\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b`),
 	// PII: credit card numbers (basic Luhn-eligible patterns)
 	regexp.MustCompile(`\b(?:\d[ -]*?){13,19}\b`),
 }
 
-// Filter replaces secrets in content with [REDACTED].
+// strictOnlyPatterns are only stripped in Strict mode.
+var strictOnlyPatterns = []*regexp.Regexp{
+	// PII: email addresses (Moderate keeps work emails)
+	regexp.MustCompile(`[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}`),
+	// PII: IPv4 addresses (Moderate keeps infrastructure IPs like 10.x, 192.168.x)
+	regexp.MustCompile(`\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b`),
+}
+
+// patterns is kept for backward compatibility — contains all patterns (Strict behavior).
+var patterns = append(append(append([]*regexp.Regexp{}, secretPatterns...), piiPatterns...), strictOnlyPatterns...)
+
+// Filter replaces secrets in content with [REDACTED] using DefaultFilterLevel.
 func Filter(content string) string {
-	for _, p := range patterns {
+	return FilterWithLevel(content, DefaultFilterLevel)
+}
+
+// FilterWithLevel replaces secrets in content with [REDACTED] at the specified level.
+func FilterWithLevel(content string, level FilterLevel) string {
+	// All levels strip explicit secrets
+	for _, p := range secretPatterns {
 		content = p.ReplaceAllString(content, "[REDACTED]")
 	}
+
+	// Moderate and Strict also strip PII patterns (phones, SSNs, credit cards)
+	if level <= Moderate {
+		for _, p := range piiPatterns {
+			content = p.ReplaceAllString(content, "[REDACTED]")
+		}
+	}
+
+	// Strict strips everything including emails and all IPs
+	if level == Strict {
+		for _, p := range strictOnlyPatterns {
+			content = p.ReplaceAllString(content, "[REDACTED]")
+		}
+	}
+
 	// Catch high-entropy tokens that regexes might miss.
 	// Only target tokens that look like standalone secrets (no JSON, no code).
 	for _, word := range strings.Fields(content) {

privacy/filter_test.go

@@ -49,7 +49,6 @@ func TestFilter_PII(t *testing.T) {
 		name  string
 		input string
 	}{
-		{"email", "contact user@example.com for help"},
 		{"SSN", "ssn is 123-45-6789"},
 		{"phone", "call 555-123-4567"},
 	}
@@ -63,6 +62,48 @@ func TestFilter_PII(t *testing.T) {
 	}
 }
 
+func TestFilter_Strict(t *testing.T) {
+	// Strict mode strips emails and all IPs
+	result := FilterWithLevel("contact user@example.com for help", Strict)
+	if !strings.Contains(result, "[REDACTED]") {
+		t.Errorf("expected email redaction in Strict mode, got: %s", result)
+	}
+	result = FilterWithLevel("server at 203.0.113.5", Strict)
+	if !strings.Contains(result, "[REDACTED]") {
+		t.Errorf("expected IP redaction in Strict mode, got: %s", result)
+	}
+}
+
+func TestFilter_Moderate(t *testing.T) {
+	// Moderate keeps work-related emails and infrastructure IPs
+	result := FilterWithLevel("contact user@example.com for help", Moderate)
+	if strings.Contains(result, "[REDACTED]") {
+		t.Errorf("Moderate should keep emails, got: %s", result)
+	}
+	result = FilterWithLevel("server at 10.0.0.1", Moderate)
+	if strings.Contains(result, "[REDACTED]") {
+		t.Errorf("Moderate should keep infrastructure IPs, got: %s", result)
+	}
+	// But Moderate still strips SSN and secrets
+	result = FilterWithLevel("ssn is 123-45-6789", Moderate)
+	if !strings.Contains(result, "[REDACTED]") {
+		t.Errorf("Moderate should redact SSN, got: %s", result)
+	}
+}
+
+func TestFilter_Minimal(t *testing.T) {
+	// Minimal only strips high-entropy secrets and explicit API keys
+	result := FilterWithLevel("my key is sk-abcdefghijklmnopqrstuvwxyz", Minimal)
+	if !strings.Contains(result, "[REDACTED]") {
+		t.Errorf("Minimal should redact API keys, got: %s", result)
+	}
+	// Minimal does NOT strip SSN or phone
+	result = FilterWithLevel("ssn is 123-45-6789", Minimal)
+	if strings.Contains(result, "[REDACTED]") {
+		t.Errorf("Minimal should not redact SSN, got: %s", result)
+	}
+}
+
 func TestFilter_ConnectionStrings(t *testing.T) {
 	input := "db url: postgres://admin:s3cret@db.example.com:5432/mydb"
 	result := Filter(input)