package memorywatcher

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
)

// DefaultDebounce is how long the watcher waits after the last filesystem
// event before flushing a commit. Long enough to coalesce a burst of writes
// (e.g. an agent rewriting several kb/* files in the same turn); short
// enough that the user perceives memory commits as near-instant.
const DefaultDebounce = 500 * time.Millisecond

const (
	defaultRetryMin = time.Second
	defaultRetryMax = time.Minute
	pushTimeout     = 60 * time.Second
)

// Committer is the subset of vault.(*Clone) the watcher needs. Declared as
// an interface so tests can substitute a fake; production code passes a
// real *vault.Clone.
type Committer interface {
	// CommitPending stages all changes and creates (or amends) a commit on
	// the clone's bound branch. Returns ErrNothingToCommit when the worktree is
	// clean — the watcher treats that as a successful no-op round.
	CommitPending(ctx context.Context, msg string, opts CommitOpts) (string, error)
	// LastCommitMessage returns the previous HEAD commit message, or empty
	// string when there is no commit yet (fresh clone). Used to derive the
	// amend key for the next commit.
	LastCommitMessage(ctx context.Context) (string, error)
	// LastCommitTime returns the previous HEAD commit's author timestamp.
	// Returns zero when there is no commit yet. Used to enforce the amend
	// deadline so commits don't keep amending indefinitely.
	LastCommitTime(ctx context.Context) (time.Time, error)
}

// CommitOpts mirrors the subset of vault.CommitPendingOpts the watcher cares
// about. Kept as a local type so the watcher does not import vault — the
// dispatcher converts when wiring the real Clone.
type CommitOpts struct {
	// Amend rewrites HEAD instead of stacking a new commit.
	Amend bool
	// AuthorName identifies the actor whose input triggered the wake
	// (human display name, or agent display name for agent-authored wakes).
	// Empty falls back to vault defaults.
	AuthorName string
	// AuthorEmail is the commit author email. Empty falls back to vault defaults.
	AuthorEmail string
}

// ContextResolver returns the conversation context that should be attributed
// to the next commit. Called when the debounce timer fires. May return a
// zero-value CommitContext with empty AgentID when no session is active — in
// that case the watcher still commits, but with the "local edits" fallback
// message.
//
// The resolver is invoked on the watcher's agentteam goroutine, so it must be
// cheap (no blocking I/O). If a slow lookup is required, the resolver should
// cache the latest binding and return the cached snapshot synchronously.
type ContextResolver func(ctx context.Context) CommitContext

// Pusher is the optional push hook the watcher invokes after each successful
// commit. Declared as an interface so tests can substitute a fake; production
// code passes a ClonePusher wrapping *vault.Clone.
type Pusher interface {
	// Push forwards the just-made local commit to the upstream. The watcher
	// swallows ErrNothingToPush (no work to do) and logs other errors — push
	// is best-effort, never blocks commits.
	Push(ctx context.Context, opts PushOpts) (string, error)
}

// PushOpts mirrors the subset of vault.PushOpts the watcher cares about. Kept
// as a local type so the watcher does not import vault; the dispatcher's
// ClonePusher adapter converts.
type PushOpts struct {
	// nothing for now; Force stays at the dispatcher's discretion via the
	// adapter rather than the watcher's hot path.
}

// Syncer checks whether the remote tracking branch has advanced since the
// last push. Called before commit so the watcher can decide whether to amend
// (remote unchanged) or create a new commit (remote advanced). Syncer is
// read-only — it does not modify the worktree or perform any merge.
type Syncer interface {
	// Sync fetches origin and compares local HEAD with origin/<branch>.
	Sync(ctx context.Context) (SyncResult, error)
}

// SyncState mirrors vault.SyncState. Local alias so the watcher package does
// not import vault.
type SyncState int

const (
	SyncUpToDate    SyncState = iota // HEAD == origin/<branch>
	SyncLocalAhead                   // HEAD ahead of origin/<branch>
	SyncRemoteAhead                  // origin/<branch> ahead of HEAD
	SyncDiverged                     // HEAD and origin/<branch> diverged
)

// SyncResult mirrors vault.SyncResult for the watcher's pipeline.
type SyncResult struct {
	State SyncState
}

// ChangeListener is the optional downstream hook the watcher invokes after
// every successful commit (and BEFORE push) so owners of derived state can
// reconcile against the canonical clone's new content.
//
// The signal is intentionally content-agnostic: no hash is passed because
// the canonical clone's HEAD is not a trustworthy target revision for a
// derivative — the derivative must derive its own SourceRevision from a
// bounded Scan of canonical content hashes. Passing the HEAD hash here
// would invite callers to lock onto a hint that may be amended away before
// the reconciler runs.
//
// Implementations MUST treat OnMemoryChanged as a fast coalescing enqueue:
// the call happens on the watcher's flushCommit goroutine, inside the same
// commitLock that serializes commits and pushes, so a slow implementation
// would stall the entire memory pipeline. Long reconciliation work must
// hand off to the listener's own goroutine and return immediately. The
// listener must NOT retain the caller's context for async work — derive a
// fresh background context with a bounded timeout if needed.
//
// The watcher fires OnMemoryChanged only when CommitPending returned a
// fresh hash; the ErrNothingToCommit path skips the call. If the canonical
// clone changes but commit fails, no event is delivered here — listeners
// that need to converge on canonical state regardless of commit outcome
// must rely on their own startup/tool-time/periodic reconcile triggers
// (the daemon-scoped reconciler owns those paths; this listener is only
// the fast path for the common case).
//
// Declared as an interface so tests can substitute a fake; production code
// passes the daemon-scoped Memory search reconciler once Task 8 wires it in.
type ChangeListener interface {
	OnMemoryChanged(ctx context.Context)
}

// Options configures New.
type Options struct {
	// MemoryDir is the path to watch. Normally the agent's memory clone at
	// ~/.agentteam/agents/<agentID>/memory. Required.
	MemoryDir string
	// Committer stages and creates the commit. Required.
	Committer Committer
	// Resolve supplies the conversation context for each commit. Required.
	// Pass a no-op resolver (returns zero CommitContext) to commit with the
	// fallback "local edits" message.
	Resolve ContextResolver
	// Push, when non-nil, is called after every successful local commit so
	// changes propagate upstream without waiting for a session-end push.
	// Optional: tests that only exercise commit semantics leave this nil.
	Push Pusher
	// Sync, when non-nil, is called BEFORE commit. It fetches origin and
	// checks whether the remote tracking branch has advanced. If the remote
	// is ahead or diverged, the watcher skips amend and creates a new commit
	// instead. Optional: tests that only exercise commit semantics leave
	// this nil.
	Sync Syncer
	// Notify, when non-nil, is called after every successful commit and
	// BEFORE push, so downstream derivative owners can reconcile against
	// the canonical clone's new content. Push may block for up to 60s on a
	// slow network, so Notify runs first to keep the signal prompt.
	// Implementations must be a fast coalescing enqueue (see ChangeListener
	// docs); the watcher never blocks on the listener's reconciliation work.
	// Optional: tests that only exercise commit/push semantics leave this nil.
	Notify ChangeListener
	// Debounce overrides DefaultDebounce. Zero falls back to the default.
	// Tests set this to a small value to keep assertions fast.
	Debounce time.Duration
	// Logger receives lifecycle events. Nil falls back to slog.Default().
	Logger *slog.Logger
	// Status, when non-nil, receives runtime-only synchronization transitions.
	// The callback must return promptly. It may request a non-blocking RetrySync,
	// but must not synchronously close the watcher from its own pipeline callback.
	Status StatusListener
	// RetryMin and RetryMax bound exponential push retry delays. Zero values use
	// production defaults. Tests may shorten both without changing semantics.
	RetryMin time.Duration
	RetryMax time.Duration
	// RetryOnStart probes for commits left ahead by a prior watcher or daemon
	// run. Production enables it; isolated commit tests may leave it false.
	RetryOnStart bool
}

// Watcher observes MemoryDir and flushes commits through Committer. A single
// goroutine runs the fsnotify agentteam; Close is idempotent and safe to call
// from any goroutine.
type Watcher struct {
	opts   Options
	logger *slog.Logger

	mu         sync.Mutex
	watcher    *fsnotify.Watcher
	cancel     chan struct{}
	stopped    bool
	started    bool
	debounce   *time.Timer
	hasPending bool // events arrived since last successful flush
	// commitFailed keeps a later retry push from reporting ready while file
	// changes are still present but not durably represented in Git.
	commitFailed bool
	commitLock   sync.Mutex

	statusMu sync.RWMutex
	status   Status

	retryTimer   *time.Timer
	retryAttempt int
	retryGen     uint64
	retryRunning bool
}

// New constructs a watcher. The fsnotify.Watcher is not created until Start
// is called, so a New call is safe even when MemoryDir does not exist yet.
func New(opts Options) (*Watcher, error) {
	if strings.TrimSpace(opts.MemoryDir) == "" {
		return nil, fmt.Errorf("memorywatcher: memory dir is required")
	}
	if opts.Committer == nil {
		return nil, fmt.Errorf("memorywatcher: committer is required")
	}
	if opts.Resolve == nil {
		return nil, fmt.Errorf("memorywatcher: context resolver is required")
	}
	debounce := opts.Debounce
	if debounce <= 0 {
		debounce = DefaultDebounce
	}
	logger := opts.Logger
	if logger == nil {
		logger = slog.Default()
	}
	return &Watcher{
		opts:   opts,
		logger: logger,
		cancel: make(chan struct{}),
		status: Status{State: StatusReady},
		// Use a placeholder timer so reset logic is uniform; it never fires
		// because the goroutine is not running until Start.
		debounce: time.AfterFunc(debounce, func() {}),
	}, nil
}

// Start begins watching MemoryDir. Returns an error if MemoryDir does not
// exist, is not a directory, or the fsnotify watcher cannot be created. The
// call is idempotent: calling Start on an already-started watcher is a
// no-op. Start on a closed watcher returns ErrClosed.
func (w *Watcher) Start(ctx context.Context) error {
	w.mu.Lock()
	if w.stopped {
		w.mu.Unlock()
		return fmt.Errorf("memorywatcher: %w", ErrClosed)
	}
	if w.started {
		w.mu.Unlock()
		return nil
	}
	w.mu.Unlock()

	info, err := os.Stat(w.opts.MemoryDir)
	if err != nil {
		return fmt.Errorf("memorywatcher: stat memory dir: %w", err)
	}
	if !info.IsDir() {
		return fmt.Errorf("memorywatcher: %s is not a directory", w.opts.MemoryDir)
	}

	fsw, err := fsnotify.NewWatcher()
	if err != nil {
		return fmt.Errorf("memorywatcher: create fsnotify watcher: %w", err)
	}

	w.mu.Lock()
	w.watcher = fsw
	w.started = true
	// Replace placeholder timer with the real one. The timer starts armed
	// but the callback re-arms itself inside the agentteam goroutine, so this
	// initial timer is effectively a no-op until the first event.
	w.debounce.Stop()
	w.mu.Unlock()

	if err := w.addTree(w.opts.MemoryDir); err != nil {
		// Best-effort cleanup if initial walk fails.
		_ = fsw.Close()
		return fmt.Errorf("memorywatcher: walk memory dir: %w", err)
	}

	go w.agentteam()
	if w.opts.RetryOnStart && w.opts.Push != nil {
		// Recover local work synchronously before Start returns so callers cannot
		// begin another Git mutation while an uncommitted startup round is still
		// pending. The potentially slow network push remains asynchronous.
		w.recoverLocalChangesOnStart(ctx)
	} else {
		w.publishStatus(Status{State: StatusReady})
	}
	w.logger.InfoContext(ctx, "memorywatcher started",
		"memory_dir", w.opts.MemoryDir,
		"debounce", w.opts.Debounce,
	)
	return nil
}

// recoverLocalChangesOnStart durably commits files left by an earlier watcher
// or daemon run before Start returns, then hands cloud delivery to the normal
// asynchronous retry pipeline. Startup recovery always creates a new commit:
// deciding whether an amend is safe requires a network fetch and must not add a
// push-timeout-sized delay to session startup.
func (w *Watcher) recoverLocalChangesOnStart(parent context.Context) {
	w.commitLock.Lock()
	defer w.commitLock.Unlock()

	if parent == nil {
		parent = context.Background()
	}
	ctx, cancel := context.WithTimeout(parent, 30*time.Second)
	defer cancel()

	commitCtx := w.opts.Resolve(ctx)
	hash, err := w.opts.Committer.CommitPending(ctx, BuildCommitMessage(commitCtx), CommitOpts{
		AuthorName: commitCtx.ActorName,
	})
	if err != nil && !errors.Is(err, ErrNothingToCommit) {
		w.mu.Lock()
		w.commitFailed = true
		w.mu.Unlock()
		w.logger.WarnContext(ctx, "memorywatcher: startup commit failed", "error", err)
		w.publishStatus(Status{State: StatusDegraded, Reason: StatusReasonCommitFailed})
		return
	}

	w.mu.Lock()
	w.commitFailed = false
	w.mu.Unlock()
	w.publishStatus(Status{State: StatusPendingSync})
	if err == nil {
		w.logger.InfoContext(ctx, "memorywatcher startup recovery committed", "hash", hash)
		if w.opts.Notify != nil {
			w.opts.Notify.OnMemoryChanged(ctx)
		}
	}
	w.requestPushRetry(0)
}

// ErrClosed is returned (or wrapped) when Start is called on a watcher that
// has already been closed.
var ErrClosed = errors.New("watcher closed")

// addTree walks dir recursively and adds every subdirectory to the fsnotify
// watcher, skipping any directory whose name is ".git" (case-sensitive —
// AGS-backed repos use lowercase). Symbolic links are not followed.
func (w *Watcher) addTree(dir string) error {
	return filepath.WalkDir(dir, func(path string, d os.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if !d.IsDir() {
			return nil
		}
		if d.Name() == ".git" {
			return filepath.SkipDir
		}
		if err := w.watcher.Add(path); err != nil {
			// A directory disappearing between WalkDir and Add is
			// recoverable — log and continue. Other errors abort Start.
			if !errors.Is(err, os.ErrNotExist) {
				return fmt.Errorf("add watch %s: %w", path, err)
			}
			w.logger.Debug("memorywatcher: dir vanished during walk", "path", path)
		}
		return nil
	})
}

// shouldIgnore reports whether an event on path should be dropped. The
// watcher subscribes to directories and filters per-event so a stray write
// inside memory/.git/ (e.g. .git/index changing during a CommitPending) is
// not re-processed.
func (w *Watcher) shouldIgnore(path string) bool {
	rel, err := filepath.Rel(w.opts.MemoryDir, path)
	if err != nil {
		return true
	}
	// Anything under .git/ is ignored. We also ignore the .git directory
	// itself (Rel returns ".git" for the dir).
	first := rel
	if i := strings.Index(rel, string(os.PathSeparator)); i >= 0 {
		first = rel[:i]
	}
	return first == ".git"
}

// agentteam consumes fsnotify events until Close cancels it. Create events on
// directories trigger an addTree on the new subdir so newly-created kb/
// subdirectories stay watched.
func (w *Watcher) agentteam() {
	debounce := w.opts.Debounce
	if debounce <= 0 {
		debounce = DefaultDebounce
	}
	for {
		select {
		case <-w.cancel:
			return
		case event, ok := <-w.watcher.Events:
			if !ok {
				return
			}
			w.handleEvent(event, debounce)
		case err, ok := <-w.watcher.Errors:
			if !ok {
				return
			}
			w.logger.Debug("memorywatcher: fsnotify error", "error", err)
		}
	}
}

func (w *Watcher) handleEvent(event fsnotify.Event, debounce time.Duration) {
	if w.shouldIgnore(event.Name) {
		return
	}
	// Only file/dir mutations count. fsnotify emits Chmod on macOS that we
	// don't care about; Rename is followed by Create on the destination, so
	// we let Create carry the signal.
	if !event.Has(fsnotify.Create) && !event.Has(fsnotify.Write) && !event.Has(fsnotify.Remove) {
		return
	}

	// New directories must be added to the watcher so files created inside
	// them later are observed. Walk shallow — addTree is recursive.
	if event.Has(fsnotify.Create) {
		if info, err := os.Stat(event.Name); err == nil && info.IsDir() {
			if err := w.addTree(event.Name); err != nil {
				w.logger.Debug("memorywatcher: add new subdir", "path", event.Name, "error", err)
			}
		}
	}

	w.mu.Lock()
	if w.stopped {
		w.mu.Unlock()
		return
	}
	w.hasPending = true
	if w.debounce != nil {
		w.debounce.Stop()
	}
	w.debounce = time.AfterFunc(debounce, w.flushCommit)
	w.mu.Unlock()
}

// flushCommit is the debounced handler. It runs on its own goroutine spawned
// by time.AfterFunc. We hold commitLock for the duration so a late trigger
// from the previous debounce window cannot race with a new commit.
//
// Pipeline: Sync → Commit → Push. Sync runs first so the watcher can
// decide whether to amend (remote unchanged) or create a new commit (remote
// advanced). All three steps run inside commitLock so subsequent flushes
// cannot interleave with an in-flight pipeline.
func (w *Watcher) flushCommit() {
	w.commitLock.Lock()
	defer w.commitLock.Unlock()
	w.flushCommitLocked()
}

func (w *Watcher) flushCommitLocked() {
	w.mu.Lock()
	if w.stopped {
		w.mu.Unlock()
		return
	}
	if !w.hasPending {
		w.mu.Unlock()
		return
	}
	w.hasPending = false
	w.mu.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	commitCtx := w.opts.Resolve(ctx)
	msg := BuildCommitMessage(commitCtx)

	prevMsg, err := w.opts.Committer.LastCommitMessage(ctx)
	if err != nil {
		w.logger.WarnContext(ctx, "memorywatcher: cannot read last commit; creating new commit",
			"error", err)
	}
	prevTime, _ := w.opts.Committer.LastCommitTime(ctx)
	amend := shouldAmend(prevMsg, prevTime, commitCtx)

	// If we intend to amend, first check whether the remote branch has
	// advanced. If it has, we must not amend — rewriting history that
	// already diverged from the remote would cause push conflicts.
	if amend && w.opts.Sync != nil {
		syncCtx, syncCancel := context.WithTimeout(context.Background(), 60*time.Second)
		result, err := w.opts.Sync.Sync(syncCtx)
		syncCancel()
		if err != nil {
			w.logger.WarnContext(ctx, "memorywatcher: pre-commit sync failed; creating new commit",
				"error", err)
			amend = false
		} else {
			w.logger.InfoContext(ctx, "memorywatcher synced",
				"state", syncStateName(result.State),
			)
			if result.State == SyncRemoteAhead || result.State == SyncDiverged {
				w.logger.InfoContext(ctx, "memorywatcher: remote advanced; creating new commit instead of amend",
					"state", syncStateName(result.State))
				amend = false
			}
		}
	}

	hash, err := w.opts.Committer.CommitPending(ctx, msg, CommitOpts{
		Amend:      amend,
		AuthorName: commitCtx.ActorName,
	})
	if err != nil {
		if errors.Is(err, ErrNothingToCommit) {
			w.mu.Lock()
			w.commitFailed = false
			w.mu.Unlock()
			w.logger.Debug("memorywatcher: nothing to commit after debounce window")
			w.pushNow("")
			return
		}
		w.mu.Lock()
		w.commitFailed = true
		w.mu.Unlock()
		w.logger.WarnContext(ctx, "memorywatcher: commit failed", "error", err, "amend", amend)
		w.publishStatus(Status{State: StatusDegraded, Reason: StatusReasonCommitFailed})
		return
	}
	w.mu.Lock()
	w.commitFailed = false
	w.mu.Unlock()
	w.logger.InfoContext(ctx, "memorywatcher committed",
		"hash", hash,
		"amend", amend,
		"channel", commitCtx.ChannelID,
		"thread", commitCtx.ThreadTarget,
		"task", commitCtx.TaskScope,
	)
	// The local commit is durable before any remote I/O begins. Publish pending
	// now so a slow push is observable without delaying or misreporting the local
	// save as cloud-synced.
	w.publishStatus(Status{State: StatusPendingSync})

	// Notify downstream derivative owners BEFORE push. The canonical clone
	// HEAD already advanced; push may block up to 60s on a slow network and
	// has no bearing on canonical content. The listener is responsible for
	// returning quickly (fast coalescing enqueue) — see ChangeListener docs.
	// Push outcome does not affect whether the event was delivered.
	if w.opts.Notify != nil {
		w.opts.Notify.OnMemoryChanged(ctx)
	}

	w.pushNow(hash)
}

// ApplyAndFlush serializes one caller-owned filesystem mutation with the
// watcher's Commit -> Sync -> Push pipeline. It is intended for cutover work
// that must update the watched clone only after an external writer has stopped
// and must not race a debounce or retry Git operation.
//
// The watcher remains active after the call. Success means the mutation and
// every earlier pending filesystem event reached StatusReady; commit or push
// failures remain represented by the watcher's ordinary retry state.
func (w *Watcher) ApplyAndFlush(apply func() error) error {
	if w == nil || apply == nil {
		return errors.New("memorywatcher: apply callback is required")
	}
	w.commitLock.Lock()
	defer w.commitLock.Unlock()

	w.mu.Lock()
	if w.stopped {
		w.mu.Unlock()
		return fmt.Errorf("memorywatcher: %w", ErrClosed)
	}
	if w.debounce != nil {
		w.debounce.Stop()
	}
	w.mu.Unlock()

	if err := apply(); err != nil {
		w.mu.Lock()
		if !w.stopped {
			// The callback may have partially mutated the worktree before
			// returning its error, so preserve it for the ordinary pipeline.
			w.hasPending = true
			if w.debounce != nil {
				w.debounce.Stop()
			}
			debounce := w.opts.Debounce
			if debounce <= 0 {
				debounce = DefaultDebounce
			}
			w.debounce = time.AfterFunc(debounce, w.flushCommit)
		}
		w.mu.Unlock()
		return err
	}

	// Force a round even if fsnotify has not delivered the mutation yet. The
	// commit lock keeps timer/retry Git work out of the mutation interval.
	w.mu.Lock()
	if w.stopped {
		w.mu.Unlock()
		return fmt.Errorf("memorywatcher: %w", ErrClosed)
	}
	if w.debounce != nil {
		w.debounce.Stop()
	}
	w.hasPending = true
	w.mu.Unlock()
	w.flushCommitLocked()

	status := w.Status()
	if status.State != StatusReady {
		return fmt.Errorf("memorywatcher: mutation sync incomplete: state=%s reason=%s", status.State, status.Reason)
	}
	return nil
}

// pushNow runs push synchronously on the caller's goroutine. It is invoked
// from flushCommit after Commit+Sync so push carries the post-sync HEAD.
// Synchronous push (rather than the previous async goroutine) is what
// serializes git ops on the underlying Clone — the next flushCommit cannot
// commit/push while this push is in flight because both hold commitLock.
func (w *Watcher) pushNow(hash string) {
	pusher := w.opts.Push
	if pusher == nil {
		w.publishStatus(Status{State: StatusReady})
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), pushTimeout)
	defer cancel()
	if _, err := pusher.Push(ctx, PushOpts{}); err != nil {
		if errors.Is(err, ErrNothingToPush) {
			w.logger.Debug("memorywatcher: nothing to push", "hash", hash)
			w.markPushReady()
			return
		}
		w.logger.WarnContext(ctx, "memorywatcher: push failed",
			"hash", hash, "error", err)
		w.markPushPending(FailureReason(err, StatusReasonPushFailed))
		return
	}
	w.markPushReady()
	w.logger.InfoContext(ctx, "memorywatcher pushed", "hash", hash)
}

func syncStateName(s SyncState) string {
	switch s {
	case SyncUpToDate:
		return "up-to-date"
	case SyncLocalAhead:
		return "local-ahead"
	case SyncRemoteAhead:
		return "remote-ahead"
	case SyncDiverged:
		return "diverged"
	default:
		return "unknown"
	}
}

// amendMaxAge is the maximum age of the previous commit after which a new
// commit is created instead of amending, even when the amend key matches.
const amendMaxAge = 6 * time.Hour

// shouldAmend reports whether the next commit should amend the previous one
// instead of stacking. The rules are:
//  1. The previous commit must have been a memory watcher commit (non-empty
//     amend key parsed from trailers).
//  2. The amend key must equal the next context's key.
//  3. The previous commit must not be older than amendMaxAge (6h). This
//     prevents indefinite amend that could accumulate too many changes in a
//     single commit.
func shouldAmend(prevMessage string, prevTime time.Time, next CommitContext) bool {
	prevKey := AmendKeyFromTrailers(parseTrailers(prevMessage))
	if prevKey == "" {
		return false
	}
	if prevKey != next.AmendKey() {
		return false
	}
	if !prevTime.IsZero() && time.Since(prevTime) > amendMaxAge {
		return false
	}
	return true
}

// Close stops the watcher WITHOUT flushing pending events. Idempotent and
// safe to call from any goroutine. After Close, Start returns ErrClosed and
// pending debounce timers are cancelled without firing — events that arrived
// inside the debounce window but had not flushed yet are dropped.
//
// Use FlushAndClose at session teardown when you need every fsnotify event
// observed up to that point to be committed and pushed. Plain Close is fine
// for explicit "stop watching but don't commit what I just typed" use cases.
//
// Close blocks until any in-flight commit finishes so a caller that pushes
// after Close returns sees every already-debounced write in the local clone.
func (w *Watcher) Close() error {
	w.mu.Lock()
	if w.stopped {
		w.mu.Unlock()
		return nil
	}
	w.stopped = true
	w.retryGen++
	close(w.cancel)
	if w.debounce != nil {
		w.debounce.Stop()
	}
	if w.retryTimer != nil {
		w.retryTimer.Stop()
		w.retryTimer = nil
	}
	watcher := w.watcher
	w.mu.Unlock()

	if watcher != nil {
		if err := watcher.Close(); err != nil {
			// Still drain the in-flight commit so the caller's subsequent
			// final push is not racing with a half-finished local write.
			w.commitLock.Lock()
			w.commitLock.Unlock()
			return err
		}
	}
	w.commitLock.Lock()
	w.commitLock.Unlock()
	return nil
}

// FlushAndClose drains any pending debounce window, then closes the watcher.
// After FlushAndClose returns, every fsnotify event observed before the call
// has been flushed through Commit → Sync → Push. Use this at
// session teardown so a "user typed then immediately stopped the session"
// race does not lose the final write.
//
// The flush runs synchronously on the caller's goroutine. If another
// debounce fire is in flight, FlushAndClose waits for it to finish via
// commitLock before running its own.
func (w *Watcher) FlushAndClose() error {
	w.mu.Lock()
	if w.stopped {
		w.mu.Unlock()
		return nil
	}
	// Disarm the timer so it cannot fire concurrently with our manual flush.
	// hasPending is the source of truth: even if the timer already fired but
	// is waiting on commitLock, our flush will pick up its work via the flag.
	if w.debounce != nil {
		w.debounce.Stop()
	}
	pending := w.hasPending
	w.mu.Unlock()

	if pending {
		flushCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer cancel()
		// flushCommit acquires commitLock so we serialize against any in-flight
		// timer-driven flush.
		w.flushCommitWithCtx(flushCtx)
	}

	return w.Close()
}

// flushCommitWithCtx is the manual-flush entry point used by FlushAndClose.
// It is identical to flushCommit except it allows the caller to bound the
// total work via ctx; the per-step timeouts inside flushCommit still apply.
// Implemented as a thin wrapper so the timer-driven path stays unchanged.
func (w *Watcher) flushCommitWithCtx(_ context.Context) {
	// hasPending guarantees a real round; commitLock serializes against any
	// concurrent timer fire. We do not propagate ctx into the inner steps
	// because they already establish their own timeouts.
	w.flushCommit()
}