From 3bc3d58a9d4462b458dd7216e1ef675aedb1968a Mon Sep 17 00:00:00 2001
From: sun-970 <3843544764@qq.com>
Date: Sun, 20 Sep 2026 09:19:03 +0000
Subject: [PATCH] feat(cli): add one-way mem put --watch (#110)

Foreground poll watcher consumes the shared ingest core. New files ingest
after one quiet interval; changed files are reported not re-ingested;
local deletes never propagate.
---
 CHANGELOG.md                              |   4 +
 GOAL.md                                   |   2 +-
 README.md                                 |   1 +
 SPEC.md                                   |   2 +-
 server/cmd/mem/cmds_file.go               |  24 +
 server/cmd/mem/cmds_watch.go              | 510 ++++++++++++++++++++++
 server/cmd/mem/cmds_watch_lock.go         |  38 ++
 server/cmd/mem/cmds_watch_lock_other.go   |  20 +
 server/cmd/mem/cmds_watch_lock_unix.go    |  22 +
 server/cmd/mem/cmds_watch_lock_windows.go |  29 ++
 server/cmd/mem/cmds_watch_report.go       |  87 ++++
 server/cmd/mem/cmds_watch_test.go         | 405 +++++++++++++++++
 server/internal/ingest/ingest.go          |   6 +
 13 files changed, 1148 insertions(+), 2 deletions(-)
 create mode 100644 server/cmd/mem/cmds_watch.go
 create mode 100644 server/cmd/mem/cmds_watch_lock.go
 create mode 100644 server/cmd/mem/cmds_watch_lock_other.go
 create mode 100644 server/cmd/mem/cmds_watch_lock_unix.go
 create mode 100644 server/cmd/mem/cmds_watch_lock_windows.go
 create mode 100644 server/cmd/mem/cmds_watch_report.go
 create mode 100644 server/cmd/mem/cmds_watch_test.go

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 563e24c..d4e0a13 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,10 @@ The project publishes 0.x prerelease versions; a stable release line is not yet
 
 ## [Unreleased]
 
+### Added
+
+- `mem put <path> --watch` one-way foreground directory watch: poll `--interval` (default 30s), ingest new files after one quiet interval, report changed files without re-ingesting, and never propagate local deletes. Refs #110.
+
 ### Changed
 
 - Ingest cursor locks try non-blocking exclusive locks and give up after 5s so a wedged peer becomes a warning instead of a silent hang. Refs #139.
diff --git a/GOAL.md b/GOAL.md
index 8515d46..c2a9804 100644
--- a/GOAL.md
+++ b/GOAL.md
@@ -121,7 +121,7 @@ mem 只有一份数据和能力内核，但面向两类使用者：
 - [x] 实现 API / CLI / Web 的 workspace export 与空目标 `fresh` import。
 - [x] 实现导入前校验、幂等 ledger、冲突明细、失败补偿和导入后重新索引。
 - [x] 实现 `merge_conservative`。
-- [ ] 实现增量包、断点上传与完整本地同步盘体验。
+- [ ] 实现增量包、断点上传与完整本地同步盘体验。 One-way `mem put --watch` is #110 and is not this item.
 
 ### P2 — 让全部 Agent 数据可见可控
 
diff --git a/README.md b/README.md
index a9b7b2e..a63ed2e 100644
--- a/README.md
+++ b/README.md
@@ -260,6 +260,7 @@ curl --fail http://127.0.0.1:8080/healthz
 export MEM_SERVER=http://localhost:8080
 mem auth login
 mem put ~/Photos --recursive
+mem put ~/Photos --watch          # one-way watch of new files (#110); not a sync drive
 # 可选：同步端附带可信的拍摄时间、位置和来源；AI 建议稍后在 Web 中确认
 mem put ~/Photos/IMG_0001.jpg \
   --captured-at 2026-07-29T08:00:00+08:00 \
diff --git a/SPEC.md b/SPEC.md
index 8755146..228a86c 100644
--- a/SPEC.md
+++ b/SPEC.md
@@ -607,7 +607,7 @@ mem put <path> --tag <tag>...
 mem put <path> --captured-at <rfc3339> --lat <lat> --lon <lon> \
   [--location-accuracy <meters>] [--place <label>] \
   [--source-kind mobile|ai_device|other] [--source-name <description>]
-mem put <path> --watch                    # 守护，新文件自动入
+mem put <path> --watch                    # 守护，新文件自动入（#110 单向监视：变更只上报不重摄，本地删除不传播）
 
 # 取
 mem get <file_id> -o <path>
diff --git a/server/cmd/mem/cmds_file.go b/server/cmd/mem/cmds_file.go
index fc2a8a5..fd7dbc8 100644
--- a/server/cmd/mem/cmds_file.go
+++ b/server/cmd/mem/cmds_file.go
@@ -19,6 +19,10 @@ import (
 func newPutCmd() *cobra.Command {
 	var (
 		recursive  bool
+		watch      bool
+		interval   time.Duration
+		dryRun     bool
+		limit      int
 		tag        []string
 		name       string
 		mimeFlag   string
@@ -68,6 +72,22 @@ func newPutCmd() *cobra.Command {
 			}
 
 			target := args[0]
+			if watch {
+				if target == "-" {
+					return errors.New("--watch does not read stdin")
+				}
+				return runPutWatch(cmd, watchOptions{
+					root:           target,
+					dest:           toFolder,
+					interval:       interval,
+					format:         format,
+					dryRun:         dryRun,
+					limit:          limit,
+					tags:           tag,
+					sourceMetadata: sourceMetadata,
+					client:         c,
+				})
+			}
 			if target == "-" {
 				if name == "" {
 					return errors.New("--name required when reading from stdin")
@@ -89,6 +109,10 @@ func newPutCmd() *cobra.Command {
 		},
 	}
 	cmd.Flags().BoolVarP(&recursive, "recursive", "r", false, "recurse into directories")
+	cmd.Flags().BoolVar(&watch, "watch", false, "one-way foreground watch; ingest new files after one quiet interval")
+	cmd.Flags().DurationVar(&interval, "interval", 30*time.Second, "watch poll interval")
+	cmd.Flags().BoolVar(&dryRun, "dry-run", false, "watch: plan only; do not upload or advance cursors")
+	cmd.Flags().IntVar(&limit, "limit", 0, "watch: stop after this many uploads per cycle (0 = no limit)")
 	cmd.Flags().StringArrayVar(&tag, "tag", nil, "tag(s) to attach (repeatable)")
 	cmd.Flags().StringVar(&name, "name", "", "override file name (required for stdin)")
 	cmd.Flags().StringVar(&mimeFlag, "mime", "", "override MIME type")
diff --git a/server/cmd/mem/cmds_watch.go b/server/cmd/mem/cmds_watch.go
new file mode 100644
index 0000000..46b69bb
--- /dev/null
+++ b/server/cmd/mem/cmds_watch.go
@@ -0,0 +1,510 @@
+package main
+
+import (
+	"context"
+	"crypto/sha1"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"io"
+	"mime"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+	"time"
+
+	"github.com/PeterGuy326/mem/server/internal/apiclient"
+	"github.com/PeterGuy326/mem/server/internal/ingest"
+	"github.com/spf13/cobra"
+)
+
+const (
+	watchGiveUpCycles = 10
+	watchReportCap    = 200
+)
+
+var errWatchLocked = errors.New("another watcher already holds this root")
+
+type watchOptions struct {
+	root           string
+	dest           string
+	interval       time.Duration
+	format         string
+	dryRun         bool
+	limit          int
+	tags           []string
+	sourceMetadata *apiclient.FileSourceMetadata
+	client         *httpClient
+}
+
+type fileStat struct {
+	size  int64
+	mtime time.Time
+}
+
+type watchItem struct {
+	LocalPath    string `json:"local_path"`
+	FileID       string `json:"file_id,omitempty"`
+	Path         string `json:"path,omitempty"`
+	SHA256Prefix string `json:"sha256_prefix,omitempty"`
+	Status       string `json:"status"`
+	Code         string `json:"code,omitempty"`
+}
+
+type watchReport struct {
+	Scanned   int         `json:"scanned"`
+	Ingested  int         `json:"ingested"`
+	Deduped   int         `json:"deduped"`
+	Unchanged int         `json:"unchanged"`
+	Changed   int         `json:"changed"`
+	LocalGone int         `json:"local_gone"`
+	Failed    int         `json:"failed"`
+	Items     []watchItem `json:"items"`
+}
+
+type watchSession struct {
+	opts        watchOptions
+	root        string
+	cursorDir   string
+	reportPath  string
+	stdout      io.Writer
+	pending     map[string]fileStat
+	consecutive int
+	giveUpCode  ingest.Code
+	sleep       func(context.Context, time.Duration) error
+}
+
+func runPutWatch(cmd *cobra.Command, opts watchOptions) error {
+	if opts.interval <= 0 {
+		return errors.New("--interval must be positive")
+	}
+	root, err := ingest.CanonicalRoot(opts.root)
+	if err != nil {
+		return newCliError(1, err.Error(), "set put --watch to an accessible path")
+	}
+	if _, err := os.Stat(root); err != nil {
+		if os.IsNotExist(err) {
+			cmd.SilenceUsage = true
+			return newCliError(2, "watched path not found", "pass an existing directory")
+		}
+		return newCliError(1, err.Error(), "")
+	}
+
+	state := cliStateRoot()
+	key := watchRootKey(root)
+	lockPath := filepath.Join(state, "watch", "locks", key)
+	lock, err := acquireWatchLock(lockPath)
+	if err != nil {
+		if errors.Is(err, errWatchLocked) {
+			return newCliError(1, "another watcher is already running for this directory",
+				"wait for the other process to exit")
+		}
+		return err
+	}
+	defer lock.release()
+
+	sess := &watchSession{
+		opts:       opts,
+		root:       root,
+		cursorDir:  filepath.Join(state, "watch", "cursors", key),
+		reportPath: filepath.Join(state, "watch", "reports", key+".jsonl"),
+		stdout:     cmd.OutOrStdout(),
+		pending:    map[string]fileStat{},
+		sleep: func(ctx context.Context, d time.Duration) error {
+			timer := time.NewTimer(d)
+			defer timer.Stop()
+			select {
+			case <-ctx.Done():
+				return nil
+			case <-timer.C:
+				return nil
+			}
+		},
+	}
+
+	ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	return sess.loop(ctx)
+}
+
+func (s *watchSession) loop(ctx context.Context) error {
+	sleep := s.sleep
+	if sleep == nil {
+		sleep = func(ctx context.Context, d time.Duration) error {
+			timer := time.NewTimer(d)
+			defer timer.Stop()
+			select {
+			case <-ctx.Done():
+				return nil
+			case <-timer.C:
+				return nil
+			}
+		}
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+		if err := s.cycle(context.Background()); err != nil {
+			return err
+		}
+		if err := s.giveUpError(); err != nil {
+			return err
+		}
+		if err := sleep(ctx, s.opts.interval); err != nil {
+			return err
+		}
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+	}
+}
+
+func (s *watchSession) giveUpError() error {
+	if s.consecutive < watchGiveUpCycles {
+		return nil
+	}
+	switch s.giveUpCode {
+	case ingest.CodeAuth:
+		return newCliError(3, "watch stopped after repeated auth failures", "run `mem auth login`")
+	case ingest.CodePlanQuota:
+		return newCliError(4, "watch stopped after repeated plan/quota failures", "")
+	case ingest.CodeProviderTimeout:
+		return newCliError(5, "watch stopped after repeated provider/timeout failures", "")
+	default:
+		return newCliError(1, "watch stopped after repeated failures", "")
+	}
+}
+
+func (s *watchSession) cycle(ctx context.Context) error {
+	report := watchReport{Items: []watchItem{}}
+	paths, err := ingest.Walk(s.root, nil)
+	if err != nil {
+		return err
+	}
+
+	live := make(map[string]struct{}, len(paths))
+	for _, p := range paths {
+		live[p] = struct{}{}
+	}
+
+	cursors, err := loadWatchCursors(s.cursorDir)
+	if err != nil {
+		return err
+	}
+	for _, cp := range cursors {
+		if cp.LastLine < 1 {
+			continue
+		}
+		if _, ok := live[cp.Abs]; ok {
+			continue
+		}
+		report.LocalGone++
+		report.Items = append(report.Items, watchItem{
+			LocalPath:    cp.Abs,
+			FileID:       cp.FileID,
+			Path:         cp.RemotePath,
+			SHA256Prefix: shaPrefix(cp.SHA256),
+			Status:       "local_gone",
+		})
+	}
+
+	var toIngest []string
+	for _, abs := range paths {
+		report.Scanned++
+		fi, err := os.Stat(abs)
+		if err != nil {
+			report.Failed++
+			code := ingest.Classify(err)
+			report.Items = append(report.Items, watchItem{
+				LocalPath: abs,
+				Status:    "failed",
+				Code:      string(code),
+			})
+			continue
+		}
+		st := fileStat{size: fi.Size(), mtime: fi.ModTime()}
+		cp := ingest.LoadCursor(s.cursorDir, abs)
+		if cp.LastLine >= 1 {
+			if sameStat(cp, st) {
+				report.Unchanged++
+				delete(s.pending, abs)
+				continue
+			}
+			sum, err := hashFile(abs)
+			if err != nil {
+				report.Failed++
+				report.Items = append(report.Items, watchItem{
+					LocalPath: abs,
+					Status:    "failed",
+					Code:      string(ingest.Classify(err)),
+				})
+				continue
+			}
+			if cp.SHA256 != "" && sum == cp.SHA256 {
+				report.Unchanged++
+				_ = ingest.SaveCursor(s.cursorDir, ingest.Cursor{
+					Abs:        abs,
+					Size:       st.size,
+					ModTime:    st.mtime.UTC().Format("2006-01-02T15:04:05Z07:00"),
+					LastLine:   cp.LastLine,
+					SHA256:     cp.SHA256,
+					FileID:     cp.FileID,
+					RemotePath: cp.RemotePath,
+				})
+				delete(s.pending, abs)
+				continue
+			}
+			report.Changed++
+			report.Items = append(report.Items, watchItem{
+				LocalPath:    abs,
+				FileID:       cp.FileID,
+				Path:         cp.RemotePath,
+				SHA256Prefix: shaPrefix(cp.SHA256),
+				Status:       "changed",
+			})
+			delete(s.pending, abs)
+			continue
+		}
+
+		prev, seen := s.pending[abs]
+		if !seen || prev.size != st.size || !prev.mtime.Equal(st.mtime) {
+			s.pending[abs] = st
+			continue
+		}
+		toIngest = append(toIngest, abs)
+	}
+	for pending := range s.pending {
+		if _, ok := live[pending]; !ok {
+			delete(s.pending, pending)
+		}
+	}
+
+	records := map[string]uploadRecord{}
+	fatalAttempts := 0
+	fatalCode := ingest.Code("")
+	nonFatalOnly := true
+	attempts := 0
+
+	for _, abs := range toIngest {
+		attempts++
+		rec := uploadRecord{}
+		core, err := ingest.Run(ctx, []string{abs}, ingest.Options{
+			StateDir: s.cursorDir,
+			DryRun:   s.opts.dryRun,
+			Limit:    s.opts.limit,
+		}, parseWatchFile, s.uploadWatchFile(&rec))
+		if rec.local != "" {
+			records[abs] = rec
+		}
+		if err != nil {
+			code := ingest.Classify(err)
+			if core.Failures != nil {
+				for c := range core.Failures {
+					code = c
+					break
+				}
+			}
+			report.Failed++
+			report.Items = append(report.Items, watchItem{
+				LocalPath: abs,
+				Status:    "failed",
+				Code:      string(code),
+			})
+			if isGiveUpCode(code) {
+				fatalAttempts++
+				fatalCode = code
+			} else {
+				nonFatalOnly = false
+			}
+			continue
+		}
+		if s.opts.dryRun {
+			report.Ingested += core.Ingested
+			report.Items = append(report.Items, watchItem{
+				LocalPath: abs,
+				Status:    "ingested",
+			})
+			delete(s.pending, abs)
+			continue
+		}
+		nonFatalOnly = false
+		item := watchItem{
+			LocalPath:    abs,
+			FileID:       rec.fileID,
+			Path:         rec.remotePath,
+			SHA256Prefix: shaPrefix(rec.sha256),
+			Status:       "ingested",
+		}
+		if rec.deduped {
+			report.Deduped++
+			item.Status = "deduped"
+		} else {
+			report.Ingested++
+		}
+		if rec.sha256 != "" {
+			size, mtime, _ := ingest.FileState(abs)
+			_ = ingest.SaveCursor(s.cursorDir, ingest.Cursor{
+				Abs:        abs,
+				Size:       size,
+				ModTime:    mtime,
+				LastLine:   1,
+				SHA256:     rec.sha256,
+				FileID:     rec.fileID,
+				RemotePath: rec.remotePath,
+			})
+		}
+		report.Items = append(report.Items, item)
+		delete(s.pending, abs)
+	}
+
+	if attempts > 0 && fatalAttempts == attempts && nonFatalOnly {
+		s.consecutive++
+		s.giveUpCode = fatalCode
+	} else if attempts > 0 {
+		s.consecutive = 0
+		s.giveUpCode = ""
+	}
+
+	if err := writeWatchReport(s.stdout, s.opts.format, report); err != nil {
+		return err
+	}
+	return appendWatchReport(s.reportPath, report)
+}
+
+type uploadRecord struct {
+	local      string
+	fileID     string
+	remotePath string
+	sha256     string
+	deduped    bool
+}
+
+func parseWatchFile(abs string, skipBefore int) ([]ingest.Unit, int, error) {
+	if skipBefore >= 1 {
+		return nil, 0, nil
+	}
+	return []ingest.Unit{{Line: 1, Body: abs}}, 0, nil
+}
+
+func (s *watchSession) uploadWatchFile(rec *uploadRecord) ingest.UploadFunc {
+	return func(ctx context.Context, abs string, u ingest.Unit) (ingest.Outcome, error) {
+		if s.opts.client == nil {
+			return ingest.Outcome{}, errors.New("watch upload client is not configured")
+		}
+		f, err := os.Open(abs)
+		if err != nil {
+			return ingest.Outcome{}, err
+		}
+		defer f.Close()
+		sum := sha256.New()
+		tee := io.TeeReader(f, sum)
+		rel, err := filepath.Rel(s.root, abs)
+		if err != nil {
+			rel = filepath.Base(abs)
+		}
+		folder := s.opts.dest
+		if d := filepath.Dir(rel); d != "" && d != "." {
+			folder = joinFolder(s.opts.dest, filepath.ToSlash(d))
+		}
+		name := filepath.Base(abs)
+		mimeType := mime.TypeByExtension(filepath.Ext(name))
+		var resp map[string]any
+		if err := s.opts.client.api.UploadMultipartWithSourceMetadata(
+			ctx, name, mimeType, folder, tee, s.opts.tags, s.opts.sourceMetadata, &resp,
+		); err != nil {
+			return ingest.Outcome{}, err
+		}
+		fileObj, _ := resp["file"].(map[string]any)
+		rec.local = abs
+		rec.sha256 = hex.EncodeToString(sum.Sum(nil))
+		if fileObj != nil {
+			rec.fileID, _ = fileObj["id"].(string)
+			rec.remotePath, _ = fileObj["path"].(string)
+			if rec.remotePath == "" {
+				if p, ok := fileObj["path"].(string); ok {
+					rec.remotePath = p
+				}
+			}
+		}
+		deduped, _ := resp["deduped"].(bool)
+		rec.deduped = deduped
+		return ingest.Outcome{Deduplicated: deduped}, nil
+	}
+}
+
+func sameStat(cp ingest.Cursor, st fileStat) bool {
+	if cp.Size != st.size {
+		return false
+	}
+	if cp.ModTime == "" {
+		return false
+	}
+	return cp.ModTime == st.mtime.UTC().Format("2006-01-02T15:04:05Z07:00")
+}
+
+func hashFile(path string) (string, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return "", err
+	}
+	defer f.Close()
+	sum := sha256.New()
+	if _, err := io.Copy(sum, f); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(sum.Sum(nil)), nil
+}
+
+func shaPrefix(sum string) string {
+	if len(sum) < 12 {
+		return sum
+	}
+	return sum[:12]
+}
+
+func watchRootKey(root string) string {
+	sum := sha1.Sum([]byte(root))
+	return hex.EncodeToString(sum[:])
+}
+
+func isGiveUpCode(code ingest.Code) bool {
+	switch code {
+	case ingest.CodeAuth, ingest.CodePlanQuota, ingest.CodeProviderTimeout:
+		return true
+	default:
+		return false
+	}
+}
+
+func loadWatchCursors(stateDir string) ([]ingest.Cursor, error) {
+	entries, err := os.ReadDir(stateDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	var out []ingest.Cursor
+	for _, e := range entries {
+		if e.IsDir() || filepath.Ext(e.Name()) != ".json" {
+			continue
+		}
+		b, err := os.ReadFile(filepath.Join(stateDir, e.Name()))
+		if err != nil {
+			continue
+		}
+		var cp ingest.Cursor
+		if json.Unmarshal(b, &cp) != nil || cp.Abs == "" {
+			continue
+		}
+		out = append(out, cp)
+	}
+	return out, nil
+}
diff --git a/server/cmd/mem/cmds_watch_lock.go b/server/cmd/mem/cmds_watch_lock.go
new file mode 100644
index 0000000..1937915
--- /dev/null
+++ b/server/cmd/mem/cmds_watch_lock.go
@@ -0,0 +1,38 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+)
+
+type watchLock struct {
+	file *os.File
+}
+
+func acquireWatchLock(path string) (*watchLock, error) {
+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+		return nil, fmt.Errorf("create watch lock dir: %w", err)
+	}
+	file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return nil, fmt.Errorf("open watch lock: %w", err)
+	}
+	if err := tryLockWatchFile(file); err != nil {
+		_ = file.Close()
+		if isWatchLockBusy(err) {
+			return nil, errWatchLocked
+		}
+		return nil, fmt.Errorf("acquire watch lock: %w", err)
+	}
+	return &watchLock{file: file}, nil
+}
+
+func (l *watchLock) release() {
+	if l == nil || l.file == nil {
+		return
+	}
+	_ = unlockWatchFile(l.file)
+	_ = l.file.Close()
+	l.file = nil
+}
diff --git a/server/cmd/mem/cmds_watch_lock_other.go b/server/cmd/mem/cmds_watch_lock_other.go
new file mode 100644
index 0000000..bff3048
--- /dev/null
+++ b/server/cmd/mem/cmds_watch_lock_other.go
@@ -0,0 +1,20 @@
+//go:build !(darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris || windows)
+
+package main
+
+import (
+	"fmt"
+	"os"
+)
+
+func tryLockWatchFile(_ *os.File) error {
+	return fmt.Errorf("watch locks are not supported on this operating system")
+}
+
+func isWatchLockBusy(_ error) bool {
+	return false
+}
+
+func unlockWatchFile(_ *os.File) error {
+	return nil
+}
diff --git a/server/cmd/mem/cmds_watch_lock_unix.go b/server/cmd/mem/cmds_watch_lock_unix.go
new file mode 100644
index 0000000..9d169f8
--- /dev/null
+++ b/server/cmd/mem/cmds_watch_lock_unix.go
@@ -0,0 +1,22 @@
+//go:build darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
+
+package main
+
+import (
+	"errors"
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+func tryLockWatchFile(file *os.File) error {
+	return unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB)
+}
+
+func isWatchLockBusy(err error) bool {
+	return errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EWOULDBLOCK)
+}
+
+func unlockWatchFile(file *os.File) error {
+	return unix.Flock(int(file.Fd()), unix.LOCK_UN)
+}
diff --git a/server/cmd/mem/cmds_watch_lock_windows.go b/server/cmd/mem/cmds_watch_lock_windows.go
new file mode 100644
index 0000000..f6691a2
--- /dev/null
+++ b/server/cmd/mem/cmds_watch_lock_windows.go
@@ -0,0 +1,29 @@
+//go:build windows
+
+package main
+
+import (
+	"errors"
+	"os"
+
+	"golang.org/x/sys/windows"
+)
+
+func tryLockWatchFile(file *os.File) error {
+	return windows.LockFileEx(
+		windows.Handle(file.Fd()),
+		windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY,
+		0,
+		1,
+		0,
+		&windows.Overlapped{},
+	)
+}
+
+func isWatchLockBusy(err error) bool {
+	return errors.Is(err, windows.ERROR_LOCK_VIOLATION)
+}
+
+func unlockWatchFile(file *os.File) error {
+	return windows.UnlockFileEx(windows.Handle(file.Fd()), 0, 1, 0, &windows.Overlapped{})
+}
diff --git a/server/cmd/mem/cmds_watch_report.go b/server/cmd/mem/cmds_watch_report.go
new file mode 100644
index 0000000..1f29e2f
--- /dev/null
+++ b/server/cmd/mem/cmds_watch_report.go
@@ -0,0 +1,87 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+)
+
+func writeWatchReport(w io.Writer, format string, report watchReport) error {
+	if format == "json" {
+		enc := json.NewEncoder(w)
+		enc.SetEscapeHTML(false)
+		return enc.Encode(report)
+	}
+	_, err := fmt.Fprintf(w,
+		"watch scanned=%d ingested=%d deduped=%d unchanged=%d changed=%d local_gone=%d failed=%d\n",
+		report.Scanned, report.Ingested, report.Deduped, report.Unchanged,
+		report.Changed, report.LocalGone, report.Failed)
+	return err
+}
+
+func appendWatchReport(path string, report watchReport) error {
+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+		return fmt.Errorf("create watch report dir: %w", err)
+	}
+	line, err := json.Marshal(report)
+	if err != nil {
+		return err
+	}
+
+	existing, err := readJSONL(path)
+	if err != nil {
+		return err
+	}
+	row := append(append([]byte(nil), line...), '\n')
+	existing = append(existing, row)
+	if len(existing) > watchReportCap {
+		existing = existing[len(existing)-watchReportCap:]
+	}
+
+	tmp, err := os.CreateTemp(filepath.Dir(path), ".watch-report-*.tmp")
+	if err != nil {
+		return err
+	}
+	tmpName := tmp.Name()
+	defer func() {
+		_ = tmp.Close()
+		_ = os.Remove(tmpName)
+	}()
+	if err := tmp.Chmod(0o600); err != nil {
+		return err
+	}
+	for _, row := range existing {
+		if _, err := tmp.Write(row); err != nil {
+			return err
+		}
+	}
+	if err := tmp.Sync(); err != nil {
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	return os.Rename(tmpName, path)
+}
+
+func readJSONL(path string) ([][]byte, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	defer f.Close()
+	var lines [][]byte
+	sc := bufio.NewScanner(f)
+	sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
+	for sc.Scan() {
+		row := append([]byte(nil), sc.Bytes()...)
+		lines = append(lines, append(row, '\n'))
+	}
+	return lines, sc.Err()
+}
diff --git a/server/cmd/mem/cmds_watch_test.go b/server/cmd/mem/cmds_watch_test.go
new file mode 100644
index 0000000..c0f1416
--- /dev/null
+++ b/server/cmd/mem/cmds_watch_test.go
@@ -0,0 +1,405 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/PeterGuy326/mem/server/internal/ingest"
+)
+
+type watchUploadLog struct {
+	mu      sync.Mutex
+	methods []string
+	posts   int
+	deletes int
+	dedupe  bool
+	path    string
+}
+
+func (l *watchUploadLog) record(method, path string) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.methods = append(l.methods, method+" "+path)
+	if method == http.MethodPost {
+		l.posts++
+	}
+	if method == http.MethodDelete {
+		l.deletes++
+	}
+}
+
+func newWatchFileServer(t *testing.T, log *watchUploadLog) *httptest.Server {
+	t.Helper()
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		log.record(r.Method, r.URL.Path)
+		if r.Method == http.MethodDelete {
+			t.Errorf("watch must not issue DELETE %s", r.URL.Path)
+			http.Error(w, "delete forbidden", http.StatusMethodNotAllowed)
+			return
+		}
+		if r.Method != http.MethodPost || r.URL.Path != "/v1/files" {
+			http.NotFound(w, r)
+			return
+		}
+		name := "file.bin"
+		if _, hdr, err := r.FormFile("file"); err == nil {
+			name = hdr.Filename
+		}
+		remote := log.path
+		if remote == "" {
+			remote = "/vault/" + name
+		}
+		status := http.StatusCreated
+		if log.dedupe {
+			status = http.StatusOK
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(status)
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"file": map[string]any{
+				"id":     "11111111-1111-1111-1111-111111111111",
+				"name":   name,
+				"path":   remote,
+				"sha256": "abc",
+			},
+			"deduped": log.dedupe,
+		})
+	}))
+}
+
+func newTestWatch(t *testing.T, root string, client *httpClient) *watchSession {
+	t.Helper()
+	canonical, err := ingest.CanonicalRoot(root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	state := t.TempDir()
+	t.Setenv("MEM_STATE_DIR", state)
+	return &watchSession{
+		opts: watchOptions{
+			root:     canonical,
+			dest:     "/Inbox",
+			interval: time.Second,
+			format:   "json",
+			client:   client,
+		},
+		root:       canonical,
+		cursorDir:  filepath.Join(state, "watch", "cursors", "test"),
+		reportPath: filepath.Join(state, "watch", "reports", "test.jsonl"),
+		stdout:     io.Discard,
+		pending:    map[string]fileStat{},
+	}
+}
+
+func decodeWatchJSON(t *testing.T, raw string) watchReport {
+	t.Helper()
+	var report watchReport
+	if err := json.Unmarshal([]byte(strings.TrimSpace(raw)), &report); err != nil {
+		t.Fatalf("report json: %v (%s)", err, raw)
+	}
+	return report
+}
+
+func TestWatchIngestsNewFileAfterQuietInterval(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "note.txt")
+	if err := os.WriteFile(path, []byte("hello"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	log := &watchUploadLog{}
+	srv := newWatchFileServer(t, log)
+	defer srv.Close()
+	client := newHTTPClient(&cliConfig{Server: srv.URL, Token: "tok"})
+	sess := newTestWatch(t, dir, client)
+	var buf bytes.Buffer
+	sess.stdout = &buf
+
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	first := decodeWatchJSON(t, buf.String())
+	if first.Scanned != 1 || first.Ingested != 0 || log.posts != 0 {
+		t.Fatalf("first cycle = %+v posts=%d, want pending", first, log.posts)
+	}
+
+	buf.Reset()
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	second := decodeWatchJSON(t, buf.String())
+	if second.Ingested != 1 || log.posts != 1 {
+		t.Fatalf("second cycle = %+v posts=%d, want ingest", second, log.posts)
+	}
+	if len(second.Items) != 1 || second.Items[0].Status != "ingested" {
+		t.Fatalf("items = %+v", second.Items)
+	}
+
+	buf.Reset()
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	third := decodeWatchJSON(t, buf.String())
+	if third.Unchanged != 1 || third.Ingested != 0 || log.posts != 1 {
+		t.Fatalf("third cycle = %+v posts=%d, want unchanged", third, log.posts)
+	}
+}
+
+func TestWatchFailureContinuesWithStableCode(t *testing.T) {
+	dir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("a"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, "b.txt"), []byte("b"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	var calls int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		if calls == 1 {
+			w.WriteHeader(http.StatusUnauthorized)
+			_, _ = w.Write([]byte(`{"error":"unauthorized"}`))
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusCreated)
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"file":    map[string]any{"id": "id-ok", "path": "/Inbox/b.txt", "name": "b.txt"},
+			"deduped": false,
+		})
+	}))
+	defer srv.Close()
+	sess := newTestWatch(t, dir, newHTTPClient(&cliConfig{Server: srv.URL, Token: "tok"}))
+	var buf bytes.Buffer
+	sess.stdout = &buf
+	_ = sess.cycle(context.Background())
+	buf.Reset()
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	report := decodeWatchJSON(t, buf.String())
+	if report.Failed != 1 || report.Ingested != 1 {
+		t.Fatalf("report = %+v, want 1 failed and 1 ingested", report)
+	}
+	var sawAuth bool
+	for _, item := range report.Items {
+		if item.Status == "failed" && item.Code == string(ingest.CodeAuth) {
+			sawAuth = true
+		}
+	}
+	if !sawAuth {
+		t.Fatalf("items = %+v, want auth failure code", report.Items)
+	}
+}
+
+func TestWatchLocalGoneDoesNotDelete(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "gone.txt")
+	if err := os.WriteFile(path, []byte("x"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	log := &watchUploadLog{}
+	srv := newWatchFileServer(t, log)
+	defer srv.Close()
+	sess := newTestWatch(t, dir, newHTTPClient(&cliConfig{Server: srv.URL, Token: "tok"}))
+	_ = sess.cycle(context.Background())
+	_ = sess.cycle(context.Background())
+	if err := os.Remove(path); err != nil {
+		t.Fatal(err)
+	}
+	var buf bytes.Buffer
+	sess.stdout = &buf
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	report := decodeWatchJSON(t, buf.String())
+	if report.LocalGone != 1 {
+		t.Fatalf("report = %+v, want local_gone", report)
+	}
+	if log.deletes != 0 {
+		t.Fatalf("deletes = %d", log.deletes)
+	}
+	for _, m := range log.methods {
+		if strings.HasPrefix(m, http.MethodDelete) {
+			t.Fatalf("unexpected method %s", m)
+		}
+	}
+}
+
+func TestWatchChangedFileIsReportedNotReingested(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "edit.txt")
+	if err := os.WriteFile(path, []byte("one"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	log := &watchUploadLog{}
+	srv := newWatchFileServer(t, log)
+	defer srv.Close()
+	sess := newTestWatch(t, dir, newHTTPClient(&cliConfig{Server: srv.URL, Token: "tok"}))
+	_ = sess.cycle(context.Background())
+	_ = sess.cycle(context.Background())
+	posts := log.posts
+	if err := os.WriteFile(path, []byte("two-different"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	var buf bytes.Buffer
+	sess.stdout = &buf
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	report := decodeWatchJSON(t, buf.String())
+	if report.Changed != 1 {
+		t.Fatalf("report = %+v, want changed", report)
+	}
+	if log.posts != posts {
+		t.Fatalf("posts = %d after change, want %d", log.posts, posts)
+	}
+	buf.Reset()
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	again := decodeWatchJSON(t, buf.String())
+	if again.Changed != 1 || log.posts != posts {
+		t.Fatalf("second changed cycle = %+v posts=%d", again, log.posts)
+	}
+}
+
+func TestWatchDedupedItemUsesServerPath(t *testing.T) {
+	dir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(dir, "dup.txt"), []byte("same"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	log := &watchUploadLog{dedupe: true, path: "/already/elsewhere/dup.txt"}
+	srv := newWatchFileServer(t, log)
+	defer srv.Close()
+	sess := newTestWatch(t, dir, newHTTPClient(&cliConfig{Server: srv.URL, Token: "tok"}))
+	var buf bytes.Buffer
+	sess.stdout = &buf
+	_ = sess.cycle(context.Background())
+	buf.Reset()
+	if err := sess.cycle(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	report := decodeWatchJSON(t, buf.String())
+	if report.Deduped != 1 || report.Ingested != 0 {
+		t.Fatalf("report = %+v", report)
+	}
+	if len(report.Items) != 1 || report.Items[0].Path != "/already/elsewhere/dup.txt" {
+		t.Fatalf("item = %+v", report.Items)
+	}
+}
+
+func TestWatchMissingRootExitsNotFound(t *testing.T) {
+	setWorkspaceTestConfig(t, "http://127.0.0.1:1", "tok", "ws")
+	root := newRootCmd()
+	root.SetArgs([]string{"put", filepath.Join(t.TempDir(), "missing"), "--watch"})
+	err := root.Execute()
+	var ce *cliError
+	if err == nil || !errors.As(err, &ce) || ce.code != 2 {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestWatchLockFailsFast(t *testing.T) {
+	dir := t.TempDir()
+	lockPath := filepath.Join(dir, "lock")
+	first, err := acquireWatchLock(lockPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer first.release()
+	_, err = acquireWatchLock(lockPath)
+	if !errors.Is(err, errWatchLocked) {
+		t.Fatalf("err = %v, want errWatchLocked", err)
+	}
+}
+
+func TestWatchLoopExitsZeroOnCancel(t *testing.T) {
+	dir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("a"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	log := &watchUploadLog{}
+	srv := newWatchFileServer(t, log)
+	defer srv.Close()
+	sess := newTestWatch(t, dir, newHTTPClient(&cliConfig{Server: srv.URL, Token: "tok"}))
+	sess.opts.interval = time.Hour
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- sess.loop(ctx) }()
+	time.Sleep(20 * time.Millisecond)
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("loop err = %v, want nil (exit 0)", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("watch loop did not exit")
+	}
+}
+
+func TestWatchReportJSONLIsCapped(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "r.jsonl")
+	for i := 0; i < 205; i++ {
+		if err := appendWatchReport(path, watchReport{Scanned: i, Items: []watchItem{}}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	b, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	lines := strings.Split(strings.TrimSpace(string(b)), "\n")
+	if len(lines) != watchReportCap {
+		t.Fatalf("lines = %d, want %d", len(lines), watchReportCap)
+	}
+	var last watchReport
+	if err := json.Unmarshal([]byte(lines[len(lines)-1]), &last); err != nil {
+		t.Fatal(err)
+	}
+	if last.Scanned != 204 {
+		t.Fatalf("last scanned = %d, want 204", last.Scanned)
+	}
+}
+
+func TestWatchJSONHasClosedVocabulary(t *testing.T) {
+	report := watchReport{
+		Scanned: 1, Ingested: 1, Items: []watchItem{{
+			LocalPath: "/tmp/a", FileID: "id", Path: "/Inbox/a", Status: "ingested",
+		}},
+	}
+	var buf bytes.Buffer
+	if err := writeWatchReport(&buf, "json", report); err != nil {
+		t.Fatal(err)
+	}
+	var raw map[string]any
+	if err := json.Unmarshal(buf.Bytes(), &raw); err != nil {
+		t.Fatal(err)
+	}
+	for _, key := range []string{"scanned", "ingested", "deduped", "unchanged", "changed", "local_gone", "failed", "items"} {
+		if _, ok := raw[key]; !ok {
+			t.Fatalf("missing %s", key)
+		}
+	}
+}
+
+func TestWatchFlagIsRegistered(t *testing.T) {
+	cmd, _, err := newRootCmd().Find([]string{"put"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cmd.Flags().Lookup("watch") == nil || cmd.Flags().Lookup("interval") == nil {
+		t.Fatal("put --watch/--interval flags missing")
+	}
+}
diff --git a/server/internal/ingest/ingest.go b/server/internal/ingest/ingest.go
index e852629..7a7941f 100644
--- a/server/internal/ingest/ingest.go
+++ b/server/internal/ingest/ingest.go
@@ -121,6 +121,12 @@ type Cursor struct {
 	ModTime  string `json:"mtime"`
 	LastLine int    `json:"last_line"`
 
+	// Optional file-plane locators used by put --watch. Transcript connectors
+	// leave them empty; omitempty keeps existing qoder cursors byte-compatible.
+	SHA256     string `json:"sha256,omitempty"`
+	FileID     string `json:"file_id,omitempty"`
+	RemotePath string `json:"remote_path,omitempty"`
+
 	// Corrupt is set in memory when a stored cursor failed to decode. It is
 	// never persisted.
 	Corrupt bool `json:"-"`