feat: persist passive browser activity batches

internal/core/browserreceiver/receiver.go

@@ -12,18 +12,20 @@ import (
 	"log"
 	"net"
 	"net/http"
-	"net/url"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/verstak/verstak-desktop/internal/core/events"
+	"github.com/verstak/verstak-desktop/internal/core/hostname"
 )
 
 const (
 	capturePath         = "/api/browser-inbox/v1/captures"
+	activityBatchPath   = "/api/browser-activity/v1/batches"
 	DefaultAddr         = "127.0.0.1:47731"
 	DefaultCaptureURL   = "http://" + DefaultAddr + capturePath
+	DefaultActivityURL  = "http://" + DefaultAddr + activityBatchPath
 	receiverTokenHeader = "X-Verstak-Receiver-Token"
 )
 
@@ -43,6 +45,9 @@ const (
 	maxFileTextBytes       = 2 * 1024 * 1024
 	maxFileBytes           = 8 * 1024 * 1024
 	maxFileDataBase64Bytes = 4 * ((maxFileBytes + 2) / 3)
+	maxActivityBodyBytes   = 256 * 1024
+	maxActivityEntries     = 100
+	maxActivityDuration    = 10 * time.Minute
 )
 
 type Receiver struct {
@@ -55,10 +60,12 @@ type Receiver struct {
 type WorkspaceProvider func() string
 
 type Options struct {
-	RequireToken  bool
-	ReceiverToken string
-	Available     func() bool
-	Persist       func(events.Event) error
+	RequireToken      bool
+	ReceiverToken     string
+	Available         func() bool
+	Persist           func(events.Event) error
+	ActivityAvailable func() bool
+	PersistActivity   func(events.Event) error
 }
 
 type Server struct {
@@ -107,6 +114,23 @@ type CaptureBrowser struct {
 	Name string `json:"name"`
 }
 
+// ActivityBatchPayload contains only domain-level time accounting. It never
+// accepts or emits page URLs, titles, content, or navigation history.
+type ActivityBatchPayload struct {
+	SchemaVersion int             `json:"schemaVersion"`
+	BatchID       string          `json:"batchId"`
+	CreatedAt     string          `json:"createdAt"`
+	Source        string          `json:"source"`
+	Entries       []ActivityEntry `json:"entries"`
+}
+
+type ActivityEntry struct {
+	Hostname        string `json:"hostname"`
+	StartedAt       string `json:"startedAt"`
+	EndedAt         string `json:"endedAt"`
+	DurationSeconds int64  `json:"durationSeconds"`
+}
+
 func New(bus *events.Bus, providers ...WorkspaceProvider) *Receiver {
 	return NewWithOptions(bus, Options{}, providers...)
 }
@@ -141,6 +165,18 @@ func (r *Receiver) SetPersistence(available func() bool, persist func(events.Eve
 	r.options.Persist = persist
 }
 
+// SetActivityPersistence configures durable passive browser activity storage.
+// An acknowledgement is sent only after persist returns successfully.
+func (r *Receiver) SetActivityPersistence(available func() bool, persist func(events.Event) error) {
+	if r == nil {
+		return
+	}
+	r.optionsMu.Lock()
+	defer r.optionsMu.Unlock()
+	r.options.ActivityAvailable = available
+	r.options.PersistActivity = persist
+}
+
 func Start(addr string, receiver *Receiver) (*Server, error) {
 	if receiver == nil {
 		return nil, fmt.Errorf("receiver is required")
@@ -181,6 +217,10 @@ func (s *Server) Close() error {
 func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
 
+	if req.URL.Path == activityBatchPath {
+		r.serveActivityBatch(w, req)
+		return
+	}
 	if req.URL.Path != capturePath {
 		http.NotFound(w, req)
 		return
@@ -256,6 +296,59 @@ func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	})
 }
 
+func (r *Receiver) serveActivityBatch(w http.ResponseWriter, req *http.Request) {
+	if req.Method != http.MethodPost {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		_ = json.NewEncoder(w).Encode(map[string]string{"error": "method not allowed"})
+		return
+	}
+	if err := r.validateReceiverToken(req); err != nil {
+		writeError(w, http.StatusUnauthorized, err.Error())
+		return
+	}
+	options := r.currentOptions()
+	if options.ActivityAvailable == nil || !options.ActivityAvailable() || options.PersistActivity == nil {
+		writeError(w, http.StatusServiceUnavailable, "activity storage unavailable")
+		return
+	}
+
+	defer req.Body.Close()
+	decoder := json.NewDecoder(http.MaxBytesReader(w, req.Body, maxActivityBodyBytes))
+	var batch ActivityBatchPayload
+	if err := decoder.Decode(&batch); err != nil {
+		writeActivityDecodeError(w, err)
+		return
+	}
+	if err := decoder.Decode(&struct{}{}); err != io.EOF {
+		writeActivityDecodeError(w, err)
+		return
+	}
+	if err := batch.NormalizeAndValidate(); err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	event := events.Event{
+		Name:      "browser.activity.batch",
+		Timestamp: time.Now().UTC().Format(time.RFC3339Nano),
+		Payload:   batch.EventPayload(),
+	}
+	if err := options.PersistActivity(event); err != nil {
+		log.Printf("[browserreceiver] persist activity %s: %v", batch.BatchID, err)
+		writeError(w, http.StatusServiceUnavailable, "activity storage unavailable")
+		return
+	}
+	if r.bus != nil {
+		r.bus.Publish(event)
+	}
+
+	w.WriteHeader(http.StatusAccepted)
+	_ = json.NewEncoder(w).Encode(map[string]string{
+		"status":  "accepted",
+		"batchId": batch.BatchID,
+	})
+}
+
 func (r *Receiver) currentOptions() Options {
 	if r == nil {
 		return Options{}
@@ -370,6 +463,80 @@ func (p CapturePayload) Validate() error {
 	return nil
 }
 
+// NormalizeAndValidate validates a batch before it is made durable and replaces
+// each submitted hostname with the shared canonical A-label form.
+func (p *ActivityBatchPayload) NormalizeAndValidate() error {
+	if p == nil || p.SchemaVersion != 1 {
+		return fmt.Errorf("unsupported schemaVersion")
+	}
+	if strings.TrimSpace(p.BatchID) == "" {
+		return fmt.Errorf("batchId is required")
+	}
+	if err := validateCaptureText(p.BatchID, "batchId", maxCaptureIDBytes); err != nil {
+		return err
+	}
+	if strings.TrimSpace(p.CreatedAt) == "" {
+		return fmt.Errorf("createdAt is required")
+	}
+	if _, err := time.Parse(time.RFC3339, p.CreatedAt); err != nil {
+		return fmt.Errorf("createdAt is invalid")
+	}
+	if err := validateCaptureText(p.CreatedAt, "createdAt", maxCapturedAtBytes); err != nil {
+		return err
+	}
+	if strings.TrimSpace(p.Source) == "" {
+		return fmt.Errorf("source is required")
+	}
+	if err := validateCaptureText(p.Source, "source", maxCaptureSourceBytes); err != nil {
+		return err
+	}
+	if len(p.Entries) == 0 || len(p.Entries) > maxActivityEntries {
+		return fmt.Errorf("entries must contain between 1 and %d items", maxActivityEntries)
+	}
+	for index := range p.Entries {
+		entry := &p.Entries[index]
+		canonical := hostname.NormalizeHostnameV1(entry.Hostname)
+		if canonical == "" {
+			return fmt.Errorf("entries[%d].hostname is invalid", index)
+		}
+		startedAt, err := time.Parse(time.RFC3339, entry.StartedAt)
+		if err != nil {
+			return fmt.Errorf("entries[%d].startedAt is invalid", index)
+		}
+		endedAt, err := time.Parse(time.RFC3339, entry.EndedAt)
+		if err != nil {
+			return fmt.Errorf("entries[%d].endedAt is invalid", index)
+		}
+		interval := endedAt.Sub(startedAt)
+		if interval <= 0 || interval > maxActivityDuration {
+			return fmt.Errorf("entries[%d] interval must be between 1 second and %s", index, maxActivityDuration)
+		}
+		if entry.DurationSeconds <= 0 || entry.DurationSeconds > int64(maxActivityDuration/time.Second) || time.Duration(entry.DurationSeconds)*time.Second > interval {
+			return fmt.Errorf("entries[%d].durationSeconds is invalid", index)
+		}
+		entry.Hostname = canonical
+	}
+	return nil
+}
+
+func (p ActivityBatchPayload) EventPayload() map[string]interface{} {
+	entries := make([]map[string]interface{}, 0, len(p.Entries))
+	for _, entry := range p.Entries {
+		entries = append(entries, map[string]interface{}{
+			"hostname":        entry.Hostname,
+			"startedAt":       entry.StartedAt,
+			"endedAt":         entry.EndedAt,
+			"durationSeconds": entry.DurationSeconds,
+		})
+	}
+	return map[string]interface{}{
+		"batchId":   strings.TrimSpace(p.BatchID),
+		"createdAt": strings.TrimSpace(p.CreatedAt),
+		"source":    strings.TrimSpace(p.Source),
+		"entries":   entries,
+	}
+}
+
 func (p CapturePayload) validateFile() error {
 	if p.File == nil || strings.TrimSpace(p.File.Name) == "" {
 		return fmt.Errorf("file.name is required")
@@ -450,10 +617,19 @@ func (p CapturePayload) EventPayload() map[string]interface{} {
 }
 
 func captureDomain(rawURL, fallback string) string {
-	if u, err := url.Parse(strings.TrimSpace(rawURL)); err == nil && u.Hostname() != "" {
-		return u.Hostname()
+	if normalized := hostname.NormalizeURLHostnameV1(rawURL); normalized != "" {
+		return normalized
 	}
-	return strings.TrimSpace(fallback)
+	return hostname.NormalizeHostnameV1(fallback)
+}
+
+func writeActivityDecodeError(w http.ResponseWriter, err error) {
+	var maxBytesErr *http.MaxBytesError
+	if errors.As(err, &maxBytesErr) {
+		writeError(w, http.StatusRequestEntityTooLarge, "activity payload exceeds limit")
+		return
+	}
+	writeError(w, http.StatusBadRequest, "invalid JSON")
 }
 
 func writeError(w http.ResponseWriter, status int, message string) {

internal/core/browserreceiver/receiver_test.go

@@ -238,6 +238,101 @@ func TestReceiverLeavesCaptureUnassignedWithoutCurrentWorkspace(t *testing.T) {
 	}
 }
 
+func TestReceiverAcceptsDomainActivityBatchAfterDurablePersistence(t *testing.T) {
+	bus := events.NewBus()
+	received := make(chan events.Event, 1)
+	bus.Subscribe("browser.activity.batch", func(event events.Event) {
+		received <- event
+	})
+	persisted := 0
+	receiver := NewWithOptions(bus, Options{
+		ActivityAvailable: func() bool { return true },
+		PersistActivity: func(event events.Event) error {
+			persisted++
+			return nil
+		},
+	})
+	body := `{
+		"schemaVersion": 1,
+		"batchId": "batch-123",
+		"createdAt": "2026-07-12T10:05:00.000Z",
+		"source": "verstak-browser-extension",
+		"entries": [{
+			"hostname": "пример.рф",
+			"startedAt": "2026-07-12T10:00:00.000Z",
+			"endedAt": "2026-07-12T10:05:00.000Z",
+			"durationSeconds": 300
+		}]
+	}`
+	req := httptest.NewRequest(http.MethodPost, "/api/browser-activity/v1/batches", bytes.NewBufferString(body))
+	rec := httptest.NewRecorder()
+
+	receiver.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusAccepted, rec.Body.String())
+	}
+	if persisted != 1 {
+		t.Fatalf("persisted = %d, want 1 before acknowledgement", persisted)
+	}
+	var response map[string]string
+	if err := json.Unmarshal(rec.Body.Bytes(), &response); err != nil {
+		t.Fatalf("response json: %v", err)
+	}
+	if response["batchId"] != "batch-123" || response["status"] != "accepted" {
+		t.Fatalf("response = %+v, want accepted batch-123", response)
+	}
+	event := <-received
+	payload, ok := event.Payload.(map[string]interface{})
+	if !ok {
+		t.Fatalf("event payload type = %T, want map[string]interface{}", event.Payload)
+	}
+	if payload["batchId"] != "batch-123" {
+		t.Fatalf("event payload = %+v, want batch id", payload)
+	}
+	entries, ok := payload["entries"].([]map[string]interface{})
+	if !ok || len(entries) != 1 || entries[0]["hostname"] != "xn--e1afmkfd.xn--p1ai" {
+		t.Fatalf("event entries = %+v, want one canonical hostname", payload["entries"])
+	}
+	if _, ok := payload["url"]; ok {
+		t.Fatalf("activity payload must not contain URL: %+v", payload)
+	}
+}
+
+func TestReceiverRejectsDomainActivityWithoutConsumerOrValidInterval(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		receiver *Receiver
+		body     string
+		wantCode int
+	}{
+		{
+			name:     "consumer unavailable",
+			receiver: New(events.NewBus()),
+			body:     `{"schemaVersion":1,"batchId":"batch-a","createdAt":"2026-07-12T10:05:00Z","source":"verstak-browser-extension","entries":[{"hostname":"example.com","startedAt":"2026-07-12T10:00:00Z","endedAt":"2026-07-12T10:05:00Z","durationSeconds":300}]}`,
+			wantCode: http.StatusServiceUnavailable,
+		},
+		{
+			name: "duration too long",
+			receiver: NewWithOptions(events.NewBus(), Options{
+				ActivityAvailable: func() bool { return true },
+				PersistActivity:   func(events.Event) error { return nil },
+			}),
+			body:     `{"schemaVersion":1,"batchId":"batch-b","createdAt":"2026-07-12T10:05:00Z","source":"verstak-browser-extension","entries":[{"hostname":"example.com","startedAt":"2026-07-12T10:00:00Z","endedAt":"2026-07-12T10:20:01Z","durationSeconds":1201}]}`,
+			wantCode: http.StatusBadRequest,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			req := httptest.NewRequest(http.MethodPost, "/api/browser-activity/v1/batches", strings.NewReader(tc.body))
+			rec := httptest.NewRecorder()
+			tc.receiver.ServeHTTP(rec, req)
+			if rec.Code != tc.wantCode {
+				t.Fatalf("status = %d, want %d; body=%s", rec.Code, tc.wantCode, rec.Body.String())
+			}
+		})
+	}
+}
+
 func TestReceiverRequiresTokenWhenPaired(t *testing.T) {
 	bus := events.NewBus()
 	received := make(chan events.Event, 1)

internal/core/storage/api.go

@@ -20,6 +20,17 @@ type Storage struct {
 	vault *vault.Vault
 }
 
+// NDJSONRetention bounds append-only plugin data. Records are compacted after
+// a successful append so settings.json never becomes an event log.
+type NDJSONRetention struct {
+	TimestampField   string
+	MaxAge           time.Duration
+	MaxEntries       int
+	MaxBytes         int64
+	DeduplicateField string
+	DeduplicateValue string
+}
+
 // New creates a new Storage instance backed by the given vault.
 func New(v *vault.Vault) *Storage {
 	return &Storage{vault: v}
@@ -234,6 +245,204 @@ func (s *Storage) WritePluginDataJSON(pluginID, name string, data map[string]int
 	return atomicWrite(path, encoded)
 }
 
+// ReadPluginDataNDJSON reads an append-only named data file. A missing file is
+// represented by an empty slice.
+func (s *Storage) ReadPluginDataNDJSON(pluginID, name string) ([]map[string]interface{}, error) {
+	if err := validatePluginID(pluginID); err != nil {
+		return nil, err
+	}
+	if err := validateStorageName("data", name); err != nil {
+		return nil, err
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	var records []map[string]interface{}
+	err := s.withOpenVault(func(vaultPath string) error {
+		var err error
+		records, err = readPluginDataNDJSONAt(vaultPath, pluginID, name)
+		return err
+	})
+	return records, err
+}
+
+// AppendPluginDataNDJSON appends records durably and then applies bounded
+// retention. It returns false without writing when the supplied idempotency
+// value is already present in the retained log.
+func (s *Storage) AppendPluginDataNDJSON(pluginID, name string, records []map[string]interface{}, retention NDJSONRetention) (bool, error) {
+	if err := validatePluginID(pluginID); err != nil {
+		return false, err
+	}
+	if err := validateStorageName("data", name); err != nil {
+		return false, err
+	}
+	if len(records) == 0 {
+		return false, fmt.Errorf("NDJSON records are empty")
+	}
+	if retention.MaxEntries < 0 || retention.MaxBytes < 0 || retention.MaxAge < 0 {
+		return false, fmt.Errorf("NDJSON retention values must not be negative")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	stored := false
+	err := s.withOpenVault(func(vaultPath string) error {
+		existing, err := readPluginDataNDJSONAt(vaultPath, pluginID, name)
+		if err != nil {
+			return err
+		}
+		if retention.DeduplicateField != "" && retention.DeduplicateValue != "" {
+			for _, record := range existing {
+				if fmt.Sprint(record[retention.DeduplicateField]) == retention.DeduplicateValue {
+					return nil
+				}
+			}
+		}
+		path := pluginDataNDJSONPath(vaultPath, pluginID, name)
+		if err := appendNDJSON(path, records); err != nil {
+			return err
+		}
+		stored = true
+		compacted := compactNDJSONRecords(append(existing, records...), retention, time.Now().UTC())
+		if !sameNDJSONRecords(append(existing, records...), compacted) {
+			return writeNDJSON(path, compacted)
+		}
+		return nil
+	})
+	return stored, err
+}
+
+// WritePluginDataNDJSON replaces a named append-only data file. It is reserved
+// for explicit user actions such as clearing Activity, never normal event
+// ingestion.
+func (s *Storage) WritePluginDataNDJSON(pluginID, name string, records []map[string]interface{}) error {
+	if err := validatePluginID(pluginID); err != nil {
+		return err
+	}
+	if err := validateStorageName("data", name); err != nil {
+		return err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.withOpenVault(func(vaultPath string) error {
+		return writeNDJSON(pluginDataNDJSONPath(vaultPath, pluginID, name), records)
+	})
+}
+
+func pluginDataNDJSONPath(vaultPath, pluginID, name string) string {
+	return filepath.Join(vaultPath, ".verstak", "plugin-data", pluginID, name+".ndjson")
+}
+
+func readPluginDataNDJSONAt(vaultPath, pluginID, name string) ([]map[string]interface{}, error) {
+	data, err := os.ReadFile(pluginDataNDJSONPath(vaultPath, pluginID, name))
+	if err != nil {
+		if os.IsNotExist(err) {
+			return []map[string]interface{}{}, nil
+		}
+		return nil, fmt.Errorf("failed to read NDJSON data %s for plugin %s: %w", name, pluginID, err)
+	}
+	if len(data) == 0 {
+		return []map[string]interface{}{}, nil
+	}
+	lines := strings.Split(strings.TrimSuffix(string(data), "\n"), "\n")
+	records := make([]map[string]interface{}, 0, len(lines))
+	for index, line := range lines {
+		if strings.TrimSpace(line) == "" {
+			continue
+		}
+		var record map[string]interface{}
+		if err := json.Unmarshal([]byte(line), &record); err != nil {
+			return nil, fmt.Errorf("corrupt NDJSON data %s line %d for plugin %s: %w", name, index+1, pluginID, err)
+		}
+		records = append(records, record)
+	}
+	return records, nil
+}
+
+func appendNDJSON(path string, records []map[string]interface{}) error {
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		return fmt.Errorf("failed to create NDJSON data directory: %w", err)
+	}
+	file, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
+	if err != nil {
+		return fmt.Errorf("failed to open NDJSON data file: %w", err)
+	}
+	defer file.Close()
+	for _, record := range records {
+		encoded, err := json.Marshal(record)
+		if err != nil {
+			return fmt.Errorf("failed to marshal NDJSON record: %w", err)
+		}
+		if _, err := file.Write(append(encoded, '\n')); err != nil {
+			return fmt.Errorf("failed to append NDJSON record: %w", err)
+		}
+	}
+	if err := file.Sync(); err != nil {
+		return fmt.Errorf("failed to sync NDJSON data: %w", err)
+	}
+	return nil
+}
+
+func writeNDJSON(path string, records []map[string]interface{}) error {
+	data := make([]byte, 0)
+	for _, record := range records {
+		encoded, err := json.Marshal(record)
+		if err != nil {
+			return fmt.Errorf("failed to marshal NDJSON record: %w", err)
+		}
+		data = append(data, encoded...)
+		data = append(data, '\n')
+	}
+	return atomicWrite(path, data)
+}
+
+func compactNDJSONRecords(records []map[string]interface{}, retention NDJSONRetention, now time.Time) []map[string]interface{} {
+	kept := make([]map[string]interface{}, 0, len(records))
+	cutoff := now.Add(-retention.MaxAge)
+	for _, record := range records {
+		if retention.MaxAge > 0 && retention.TimestampField != "" {
+			if raw, ok := record[retention.TimestampField].(string); ok {
+				if timestamp, err := time.Parse(time.RFC3339, raw); err == nil && timestamp.Before(cutoff) {
+					continue
+				}
+			}
+		}
+		kept = append(kept, record)
+	}
+	if retention.MaxEntries > 0 && len(kept) > retention.MaxEntries {
+		kept = kept[len(kept)-retention.MaxEntries:]
+	}
+	if retention.MaxBytes > 0 {
+		for len(kept) > 0 && ndjsonSize(kept) > retention.MaxBytes {
+			kept = kept[1:]
+		}
+	}
+	return kept
+}
+
+func ndjsonSize(records []map[string]interface{}) int64 {
+	var total int64
+	for _, record := range records {
+		encoded, err := json.Marshal(record)
+		if err == nil {
+			total += int64(len(encoded) + 1)
+		}
+	}
+	return total
+}
+
+func sameNDJSONRecords(left, right []map[string]interface{}) bool {
+	if len(left) != len(right) {
+		return false
+	}
+	for index := range left {
+		leftJSON, leftErr := json.Marshal(left[index])
+		rightJSON, rightErr := json.Marshal(right[index])
+		if leftErr != nil || rightErr != nil || string(leftJSON) != string(rightJSON) {
+			return false
+		}
+	}
+	return true
+}
+
 // ─── Cache JSON API ───────────────────────────────────────
 
 // ReadPluginCacheJSON reads a named JSON cache file for a plugin.

internal/core/storage/ndjson_test.go

@@ -0,0 +1,91 @@
+package storage
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestAppendPluginDataNDJSONCompactsAndDeduplicates(t *testing.T) {
+	s, vaultDir := newTestStorage(t)
+	retention := NDJSONRetention{
+		TimestampField:   "occurredAt",
+		MaxAge:           60 * 24 * time.Hour,
+		MaxEntries:       2,
+		MaxBytes:         16 * 1024,
+		DeduplicateField: "sourceBatchId",
+		DeduplicateValue: "batch-1",
+	}
+	old := time.Now().UTC().Add(-61 * 24 * time.Hour).Format(time.RFC3339)
+	now := time.Now().UTC().Format(time.RFC3339)
+
+	stored, err := s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{
+		{"activityId": "old", "sourceBatchId": "old-batch", "occurredAt": old},
+	}, retention)
+	if err != nil || !stored {
+		t.Fatalf("old append = (%v, %v), want (true, nil)", stored, err)
+	}
+
+	stored, err = s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{
+		{"activityId": "current-1", "sourceBatchId": "batch-1", "occurredAt": now},
+	}, retention)
+	if err != nil || !stored {
+		t.Fatalf("first current append = (%v, %v), want (true, nil)", stored, err)
+	}
+
+	stored, err = s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{
+		{"activityId": "duplicate", "sourceBatchId": "batch-1", "occurredAt": now},
+	}, retention)
+	if err != nil || stored {
+		t.Fatalf("duplicate append = (%v, %v), want (false, nil)", stored, err)
+	}
+
+	for _, item := range []struct {
+		id    string
+		batch string
+	}{
+		{id: "current-2", batch: "batch-3"},
+		{id: "current-3", batch: "batch-4"},
+	} {
+		retention.DeduplicateValue = item.batch
+		if stored, err := s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{
+			{"activityId": item.id, "sourceBatchId": item.batch, "occurredAt": now},
+		}, retention); err != nil || !stored {
+			t.Fatalf("append %s = (%v, %v), want (true, nil)", item.id, stored, err)
+		}
+	}
+
+	records, err := s.ReadPluginDataNDJSON("verstak.activity", "activity-events")
+	if err != nil {
+		t.Fatalf("ReadPluginDataNDJSON: %v", err)
+	}
+	if len(records) != 2 || records[0]["activityId"] != "current-2" || records[1]["activityId"] != "current-3" {
+		t.Fatalf("records = %+v, want current-2 and current-3", records)
+	}
+	if _, err := os.Stat(filepath.Join(vaultDir, "VerstakVault", ".verstak", "plugin-data", "verstak.activity", "activity-events.ndjson")); err != nil {
+		t.Fatalf("activity data file missing: %v", err)
+	}
+}
+
+func TestWritePluginDataNDJSONReplacesRecordsForExplicitUserClear(t *testing.T) {
+	s, _ := newTestStorage(t)
+	if _, err := s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{
+		{"activityId": "one", "occurredAt": time.Now().UTC().Format(time.RFC3339)},
+		{"activityId": "two", "occurredAt": time.Now().UTC().Format(time.RFC3339)},
+	}, NDJSONRetention{}); err != nil {
+		t.Fatalf("AppendPluginDataNDJSON: %v", err)
+	}
+	if err := s.WritePluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{
+		{"activityId": "two", "occurredAt": time.Now().UTC().Format(time.RFC3339)},
+	}); err != nil {
+		t.Fatalf("WritePluginDataNDJSON: %v", err)
+	}
+	records, err := s.ReadPluginDataNDJSON("verstak.activity", "activity-events")
+	if err != nil {
+		t.Fatalf("ReadPluginDataNDJSON: %v", err)
+	}
+	if len(records) != 1 || records[0]["activityId"] != "two" {
+		t.Fatalf("records = %+v, want only two", records)
+	}
+}