package dispatcher

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

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent"
	"github.com/tidbcloud/AgentTeam/desktop/internal/memorywatcher"
	"github.com/tidbcloud/AgentTeam/desktop/internal/nativememory"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
	"github.com/tidbcloud/AgentTeam/desktop/internal/vault"
)

type memoryRestartTarget struct {
	session *interactiveSession
	process *agent.Process
}

// SetMemoryAccessProvider wires the memory binding provider. Optional: when nil
// or never called, the dispatcher runs without memory watcher integration —
// equivalent to pre-Phase-2.3 behavior. Calling this with a nil argument is a
// no-op.
func (d *memoryService) SetMemoryAccessProvider(provider MemoryAccessProvider) {
	if provider == nil {
		return
	}
	d.memoryAccessMu.Lock()
	d.memoryAccess = provider
	d.memoryAccessMu.Unlock()
}

func (d *memoryService) agentMemoryLookupMutex(agentID string) *sync.Mutex {
	v, _ := d.memoryLookupMu.LoadOrStore(strings.TrimSpace(agentID), &sync.Mutex{})
	return v.(*sync.Mutex)
}

// attachMemoryWatcher clones (or reopens) the per-agent memory repo and starts
// a memorywatcher against it. Failures are logged but never propagate: memory
// sync is an enhancement, not a precondition for session start.
//
// The function is called from startSession after the agent workdir exists and
// MEMORY.md has been seeded. On success, sess.watcher and sess.memoryClone are
// populated under sess.mu. On any failure, both are left nil and the session
// proceeds without memory sync.
func (d *memoryService) attachMemoryWatcher(ctx context.Context, sess *interactiveSession, agentID, workDir string) {
	d.attachMemoryWatcherReplacing(ctx, sess, nil, agentID, workDir)
}

type memoryWatcherReplacement struct {
	previous       *interactiveSession
	agentID        string
	workDir        string
	restoreWatcher bool
}

// attachMemoryWatcherReplacing is the session-start replacement transaction.
// The old watcher is fenced and fully drained under the same per-agent lookup
// mutex before EnsureClone/start touches the shared workdir for the new session.
// This prevents both a double watcher and git fetch/checkout racing the old
// watcher's Commit/Sync/Push pipeline.
func (d *memoryService) attachMemoryWatcherReplacing(
	ctx context.Context,
	sess, previous *interactiveSession,
	agentID, workDir string,
) *memoryWatcherReplacement {
	if sess == nil {
		return nil
	}
	owner, err := d.beginMemoryWatcherMutation()
	if err != nil {
		return nil
	}
	defer d.finishMemoryWatcherMutation(owner)
	lookupMu := d.agentMemoryLookupMutex(agentID)
	lookupMu.Lock()
	defer lookupMu.Unlock()
	sess.mu.Lock()
	sessionUnavailable := sess.retired.Load() || sess.resetting
	sess.mu.Unlock()
	if sessionUnavailable {
		return nil
	}
	// A full reset blocks stale status-driven recovery until a real session
	// start establishes a new lifecycle generation for this agent. Reset's
	// final block runs under lookupMu, so it supersedes any concurrent unblock.
	d.allowIdleMemoryRetry(agentID)

	var replacement *memoryWatcherReplacement
	if previous != nil && previous != sess {
		previous.mu.Lock()
		restoreWatcher := previous.watcher != nil || previous.memoryClone != nil || previous.memoryWatcherDrain != nil
		previous.memoryWatcherReplacementFenced = true
		previousWorkDir := previous.workDir
		previous.mu.Unlock()
		if strings.TrimSpace(previousWorkDir) == "" {
			previousWorkDir = workDir
		}
		replacement = &memoryWatcherReplacement{
			previous:       previous,
			agentID:        agentID,
			workDir:        previousWorkDir,
			restoreWatcher: restoreWatcher,
		}
		_ = stopMemoryWatcherWithTimeoutResultAfterDetach(
			previous,
			60*time.Second,
			d.memoryWatcherReplacementAfterOldDetach,
		)
	}
	d.attachMemoryWatcherLocked(ctx, owner, sess, agentID, workDir)
	return replacement
}

// restoreReplacedMemoryWatcher best-effort restores the old session after a
// replacement launch fails before publication. Restoration is itself an
// admitted mutation and runs under the same per-agent lookup mutex, so shutdown
// cannot revive a watcher and a concurrent Refresh cannot race the clone.
func (d *memoryService) restoreReplacedMemoryWatcher(ctx context.Context, replacement *memoryWatcherReplacement) {
	if replacement == nil || replacement.previous == nil {
		return
	}
	owner, err := d.beginMemoryWatcherMutation()
	if err != nil {
		// Shutdown owns the old session's final state; do not clear its fence or
		// start a watcher after mutation admission has closed.
		return
	}
	defer d.finishMemoryWatcherMutation(owner)

	lookupMu := d.agentMemoryLookupMutex(replacement.agentID)
	lookupMu.Lock()
	defer lookupMu.Unlock()

	previous := replacement.previous
	key := strings.TrimSpace(previous.sessionKey)
	if key == "" {
		key = strings.TrimSpace(previous.agentID)
	}
	current, ok := d.sessionManager.Get(key)
	if !ok || current != previous {
		return
	}
	previous.mu.Lock()
	process := previous.process
	eligible := previous.memoryWatcherReplacementFenced &&
		!previous.retired.Load() && !previous.resetting && !previous.memoryWatcherTerminalFenced &&
		process != nil
	previous.mu.Unlock()
	if !eligible || !process.AcceptingDeliveries() {
		return
	}
	previous.mu.Lock()
	eligible = previous.process == process && previous.memoryWatcherReplacementFenced &&
		!previous.retired.Load() && !previous.resetting && !previous.memoryWatcherTerminalFenced
	if eligible {
		previous.memoryWatcherReplacementFenced = false
	}
	previous.mu.Unlock()
	if !eligible {
		return
	}
	if !process.AcceptingDeliveries() {
		previous.mu.Lock()
		if previous.process == process && !previous.retired.Load() {
			previous.memoryWatcherReplacementFenced = true
		}
		previous.mu.Unlock()
		return
	}
	if replacement.restoreWatcher {
		d.attachMemoryWatcherLocked(ctx, owner, previous, replacement.agentID, replacement.workDir)
	}
}

// attachMemoryWatcherLocked performs lookup+authoritative gate+attach while
// the caller holds agentMemoryLookupMutex(agentID). refresh/rebind use this
// directly to avoid recursively locking the per-agent serialization mutex.
func (d *memoryService) attachMemoryWatcherLocked(ctx context.Context, owner *memoryWatcherMutationOwner, sess *interactiveSession, agentID, workDir string) {
	// A concurrent teardown clears sess.watcher before its FlushAndClose/final
	// push finishes. Do not open/reconfigure the same clone during that detached
	// drain window. Existing ownership and terminal/reset/replacement fences also
	// make the operation a no-op before provider, git, or Watcher.Start work: the
	// publication fence is the final authority, not the first duplicate check.
	if !waitForSessionMemoryWatcherDrain(ctx, sess) {
		return
	}
	d.memoryAccessMu.RLock()
	provider := d.memoryAccess
	d.memoryAccessMu.RUnlock()
	if provider == nil {
		return
	}

	access, state, err := provider.Lookup(ctx, agentID)
	if err != nil {
		d.setMemoryLookupStatus(agentID, degradedMemorySyncStatus(memorywatcher.FailureReason(err, memorywatcher.StatusReasonLookupFailed)))
		d.scheduleIdleMemoryRetry(agentID)
		if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
			reg.setSourceTrusted(agentID, false)
		}
		slog.WarnContext(ctx, "memory watcher: binding lookup failed; continuing without watcher",
			"agent_id", agentID, "error", err)
		return
	}
	if state != MemoryAccessBound {
		if state == MemoryAccessUnknown {
			d.setMemoryLookupStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonLookupFailed))
			d.scheduleIdleMemoryRetry(agentID)
		} else {
			d.finishIdleMemoryRetry(agentID, ctx)
			d.clearMemoryLookupStatus(agentID)
			d.clearMemorySyncStatus(agentID)
		}
		if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
			reg.setSourceTrusted(agentID, false)
			if state == MemoryAccessUnbound {
				reg.closeAgent(agentID)
			}
		}
		return
	}
	d.clearMemoryLookupStatus(agentID)
	if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
		reg.closeAgentIfBindingChanged(agentID, access)
		reg.setSourceTrusted(agentID, true)
	}

	clone, migration, nativeSync, ok := d.ensureMemoryCloneForAccess(ctx, agentID, workDir, access)
	if !ok {
		d.scheduleIdleMemoryRetry(agentID)
		return
	}
	// Best-effort wire the search reconciler as the watcher's Notify listener.
	// The reconciler is owned by d.memorySearch (daemon scope); the session
	// watcher holds only a ChangeListener reference, never a Close handle.
	notify := d.lookupSearchReconciler(ctx, agentID, access)
	if d.startMemoryWatcherForAccess(ctx, owner, sess, agentID, access, clone, nativeSync, notify) && migration != nil {
		sess.mu.Lock()
		sess.pendingLegacyPreferencesMigration = migration
		sess.mu.Unlock()
	}
}

func waitForSessionMemoryWatcherDrain(ctx context.Context, sess *interactiveSession) bool {
	if sess == nil {
		return false
	}
	if ctx == nil {
		ctx = context.Background()
	}
	for {
		sess.mu.Lock()
		blocked := sess.retired.Load() || sess.resetting || sess.memoryWatcherReplacementFenced || sess.memoryWatcherTerminalFenced
		occupied := sess.watcher != nil || sess.memoryClone != nil || sess.memoryPushAuth != nil
		var done <-chan struct{}
		if sess.memoryWatcherDrain != nil {
			done = sess.memoryWatcherDrain.done
		}
		sess.mu.Unlock()
		if blocked || occupied {
			return false
		}
		if done == nil {
			return true
		}
		select {
		case <-done:
		case <-ctx.Done():
			return false
		}
	}
}

// lookupSearchReconcilers returns the ChangeListener to wire into the watcher's
// Notify. Returns nil when no registry is wired or the binding fails retention
// validation, so the watcher still runs without search-derivative sync.
func (d *memoryService) lookupSearchReconciler(ctx context.Context, agentID string, access MemoryAccess) memorywatcher.ChangeListener {
	reg, waitErr := d.snapshotMemorySearch(ctx)
	if reg == nil {
		if waitErr != nil {
			slog.WarnContext(ctx, "memory search: registry snapshot wait cancelled; watcher runs without search sync",
				"agent_id", agentID, "error", waitErr)
		}
		return nil
	}
	rec, _, err := reg.getOrCreate(ctx, agentID, access)
	if err != nil {
		slog.WarnContext(ctx, "memory search: reconciler not started; watcher runs without search sync",
			"agent_id", agentID, "memory_id", access.MemoryID, "error", err)
		return nil
	}
	return rec
}

// publishMemoryWatcher is the single publication boundary for all attach and
// rebind paths. Shutdown closes mutation admission under the same mutex before
// SessionManager.Close retires sessions. Consequently the admission generation
// and retired-session check below are one provable boundary: publication either
// wins before shutdown closes admission and is included in the final drain, or
// it is rejected and synchronously cleaned up by its mutation owner.
func (d *memoryService) publishMemoryWatcher(
	owner *memoryWatcherMutationOwner,
	sess *interactiveSession,
	clone *vault.Clone,
	access MemoryAccess,
	nativeSync *nativememory.Sync,
	watcher *memorywatcher.Watcher,
) bool {
	if d == nil || sess == nil || clone == nil || watcher == nil {
		return false
	}
	d.memoryWatcherMutationMu.Lock()
	defer d.memoryWatcherMutationMu.Unlock()
	if !d.memoryWatcherMutationOwnerCurrentLocked(owner) || sess.retired.Load() {
		return false
	}
	sess.mu.Lock()
	defer sess.mu.Unlock()
	if sess.retired.Load() || sess.resetting || sess.memoryWatcherReplacementFenced || sess.memoryWatcherTerminalFenced || sess.watcher != nil || sess.memoryClone != nil || sess.memoryPushAuth != nil || sess.memoryWatcherDrain != nil {
		return false
	}
	sess.memoryClone = clone
	sess.memoryPushAuth = access.Auth
	sess.memoryProtocol = access.Protocol
	sess.memoryGeneration = access.ContentGeneration
	sess.memoryContentVersionID = access.ContentVersionID
	sess.memoryNativeSync = nativeSync
	sess.watcher = watcher
	return true
}

// stopMemoryWatcher closes the session's memory watcher and flushes any
// in-flight local commits to upstream. Idempotent and safe to call from any
// goroutine. The watcher + clone + auth are copied out under sess.mu and all
// downstream work (Close + final push) runs outside the lock to avoid
// blocking other session readers.
//
// Final push ordering:
//  1. Close the watcher (blocks until any in-flight commit finishes, so the
//     local clone is fully flushed before we push).
//  2. Call vault.Clone.Push synchronously with a 60s timeout. Failures are
//     logged — the next session's Pull-equivalent flow can still catch up.
//
// Called from every session teardown path (Dispatcher.Close, StopByAgent,
// StopByServerAgentID, stopAll, resetSession) so the watcher never outlives
// its owning session.
func stopMemoryWatcher(sess *interactiveSession) {
	_ = stopMemoryWatcherWithTimeoutResult(sess, 60*time.Second)
}

func (d *memoryService) stopMemoryWatcherAndRetry(sess *interactiveSession) {
	result := stopMemoryWatcherWithTimeoutResult(sess, 60*time.Second)
	if result.pushErr == nil || sess == nil {
		return
	}
	agentID := strings.TrimSpace(sess.agentID)
	if agentID == "" {
		return
	}
	d.setMemorySyncStatus(agentID, memorywatcher.Status{
		State:  memorywatcher.StatusPendingSync,
		Reason: memorywatcher.FailureReason(result.pushErr, memorywatcher.StatusReasonPushFailed),
	})
	d.scheduleIdleMemoryRetry(agentID)
}

type stopMemoryWatcherResult struct {
	clone   *vault.Clone
	pushErr error
}

// memoryWatcherDrain is the per-session completion owner for the interval
// after the published watcher fields have been cleared but before
// FlushAndClose/final push has completed. stopMemoryWatcher callers that arrive
// during this interval join done and observe the same result. This keeps
// repeated or concurrent shutdown paths from treating detached as drained.
type memoryWatcherDrain struct {
	done   chan struct{}
	result stopMemoryWatcherResult
}

func stopMemoryWatcherWithTimeoutResult(sess *interactiveSession, pushTimeout time.Duration) stopMemoryWatcherResult {
	return stopMemoryWatcherWithTimeoutResultAfterDetach(sess, pushTimeout, nil)
}

// stopMemoryWatcherForShutdown is the shutdown variant: it uses a shorter
// push timeout so daemon shutdown is not blocked by slow git pushes. The
// optional test seam runs after the watcher is detached from sess but while
// this authoritative drain still owns its FlushAndClose/final push work.
func (d *Dispatcher) stopMemoryWatcherForShutdown(sess *interactiveSession) {
	var afterDetach func(*interactiveSession)
	if d != nil {
		afterDetach = d.memoryWatcherShutdownAfterDetach
	}
	_ = stopMemoryWatcherWithTimeoutResultAfterDetach(sess, 3*time.Second, afterDetach)
}

func stopMemoryWatcherWithTimeoutResultAfterDetach(
	sess *interactiveSession,
	pushTimeout time.Duration,
	afterDetach func(*interactiveSession),
) stopMemoryWatcherResult {
	if sess == nil {
		return stopMemoryWatcherResult{}
	}
	sess.mu.Lock()
	if active := sess.memoryWatcherDrain; active != nil {
		sess.mu.Unlock()
		<-active.done
		return active.result
	}
	watcher := sess.watcher
	clone := sess.memoryClone
	auth := sess.memoryPushAuth
	if watcher == nil && clone == nil {
		sess.memoryPushAuth = nil
		sess.memoryProtocol = ""
		sess.memoryGeneration = 0
		sess.memoryContentVersionID = ""
		sess.memoryNativeSync = nil
		sess.mu.Unlock()
		return stopMemoryWatcherResult{}
	}
	drain := &memoryWatcherDrain{done: make(chan struct{})}
	sess.memoryWatcherDrain = drain
	sess.watcher = nil
	sess.memoryClone = nil
	sess.memoryPushAuth = nil
	sess.memoryProtocol = ""
	sess.memoryGeneration = 0
	sess.memoryContentVersionID = ""
	sess.memoryNativeSync = nil
	sess.mu.Unlock()

	result := stopMemoryWatcherResult{clone: clone}

	if watcher != nil {
		if afterDetach != nil {
			afterDetach(sess)
		}
		if err := watcher.FlushAndClose(); err != nil {
			slog.Warn("memory watcher: close failed", "agent_id", sess.agentID, "error", err)
		}
	}
	if clone != nil && clone.UsesRemote() {
		pushCtx, cancel := context.WithTimeout(context.Background(), pushTimeout)
		if _, err := clone.Push(pushCtx, vault.PushOpts{Auth: auth}); err != nil {
			if !errors.Is(err, vault.ErrNothingToPush) {
				slog.Warn("memory watcher: final push failed", "agent_id", sess.agentID, "error", err)
				result.pushErr = err
			}
		}
		cancel()
		if watcher != nil {
			watcher.RecordFinalPush(result.pushErr)
		}
	}

	// Publish the result before closing done. Channel close synchronizes every
	// joining teardown's result read. Clear the active pointer under sess.mu so
	// a later independent teardown cannot enter until this drain is complete.
	sess.mu.Lock()
	drain.result = result
	if sess.memoryWatcherDrain == drain {
		sess.memoryWatcherDrain = nil
	}
	close(drain.done)
	sess.mu.Unlock()
	return result
}

func retainSessionMemoryOfflineAfterRevoke(ctx context.Context, sess *interactiveSession, pushTimeout time.Duration) error {
	result := stopMemoryWatcherWithTimeoutResult(sess, pushTimeout)
	clone := result.clone
	if clone == nil {
		return nil
	}
	if !clone.UsesRemote() {
		return nil
	}
	if result.pushErr != nil {
		return fmt.Errorf("final push before retain offline: %w", result.pushErr)
	}
	if err := clone.RetainOffline(ctx, time.Time{}); err != nil {
		slog.WarnContext(ctx, "memory access refresh: failed to retain offline memory clone",
			"agent_id", sess.agentID,
			"memory_id", clone.MemoryID(),
			"error", err)
		return err
	}
	return nil
}

func retainMemoryDirOfflineAfterRevoke(ctx context.Context, memoryDir, agentID string) error {
	local, err := vault.IsLocalJournal(memoryDir)
	if err != nil {
		return err
	}
	if local {
		return nil
	}
	state, err := vault.ReadRetentionState(memoryDir)
	if err != nil {
		return err
	}
	return vault.RetainOffline(ctx, vault.RetainOfflineOpts{
		TargetPath: memoryDir,
		MemoryID:   state.MemoryID,
		AgentID:    agentID,
		Branch:     state.Branch,
	})
}

func (d *memoryService) refreshAgentMemoryAccessWithSnapshot(ctx context.Context, agentID string, snapshot *MemoryAccessLookup) (restartTarget *memoryRestartTarget) {
	if ctx == nil {
		ctx = context.Background()
	}
	// Wait before taking mutation admission or lookupMu: startSession needs the
	// same lookup mutex to attach its watcher before it can finish this start.
	if _, ok := d.sessionManager.FindByLookupID(agentID); !ok {
		if startDone, inFlight := d.sessionManager.StartInFlight(agentID); inFlight {
			if d.memoryRefreshBeforeWaitForStart != nil {
				d.memoryRefreshBeforeWaitForStart()
			}
			select {
			case <-startDone:
			case <-ctx.Done():
				return nil
			}
		}
	}
	owner, err := d.beginMemoryWatcherMutation()
	if err != nil {
		return
	}
	defer d.finishMemoryWatcherMutation(owner)

	lookupMu := d.agentMemoryLookupMutex(agentID)
	lookupMu.Lock()
	defer lookupMu.Unlock()
	if ctx != nil && ctx.Err() != nil {
		return
	}

	d.memoryAccessMu.RLock()
	provider := d.memoryAccess
	d.memoryAccessMu.RUnlock()
	if provider == nil {
		slog.InfoContext(ctx, "memory access refresh: no provider wired; skipping",
			"agent_id", agentID)
		return
	}

	var access MemoryAccess
	var state MemoryAccessState
	if snapshot != nil {
		access, state, err = snapshot.Access, snapshot.State, snapshot.Err
	} else {
		access, state, err = provider.Lookup(ctx, agentID)
	}
	if err != nil {
		if ctx != nil && ctx.Err() != nil {
			return
		}
		d.setMemoryLookupStatus(agentID, degradedMemorySyncStatus(memorywatcher.FailureReason(err, memorywatcher.StatusReasonLookupFailed)))
		d.scheduleIdleMemoryRetry(agentID)
		if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
			reg.setSourceTrusted(agentID, false)
		}
		slog.WarnContext(ctx, "memory access refresh: lookup failed",
			"agent_id", agentID, "error", err)
		return
	}
	if state == MemoryAccessUnknown {
		d.setMemoryLookupStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonLookupFailed))
		d.scheduleIdleMemoryRetry(agentID)
		// Preserve the last trusted clone/generation, but freeze derivative
		// writes. Watcher callbacks and tool-time freshness hints may continue
		// to arrive; the Reconciler gate drops them and marks searches stale
		// until a later authoritative Bound refresh resumes reconciliation.
		if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
			reg.setSourceTrusted(agentID, false)
		}
		slog.WarnContext(ctx, "memory access refresh: lookup state unknown; freezing search derivative writes",
			"agent_id", agentID)
		return
	}
	d.clearMemoryLookupStatus(agentID)
	bound := state == MemoryAccessBound
	if !bound {
		d.finishIdleMemoryRetry(agentID, ctx)
		d.clearMemorySyncStatus(agentID)
		// Detaching a watcher performs one final serialized push, whose status
		// callback may republish a snapshot while the drain is in progress. An
		// authoritative unbind owns the final state, so clear that callback's
		// transient projection again before releasing the per-agent mutation lock.
		defer d.clearMemorySyncStatus(agentID)
	}

	// Registry serving-generation gates. These run BEFORE any session/clone
	// branching so every downstream path observes a consistent registry state:
	//
	//   - Unbound (authoritative revoke): pull the serving generation unconditionally.
	//     The downstream retainOffline branches also call closeAgent (idempotent),
	//     but the early pull guarantees we never serve hits against a clone whose
	//     binding was just revoked, regardless of session/clone state.
	//   - Bound identity change: pull any active reconciler whose (memoryID, branch)
	//     disagrees with the authoritative access, so the new binding's Start can
	//     acquire the per-derivative-root owner lock (shared across fingerprints
	//     for the same agent). Covers no-session A→B, branch-only switches, and
	//     the watcher=nil case (registry may still hold A from a prior binding).
	//     Same-identity is a no-op.
	//
	// snapshotMemorySearch bounded-waits on the swap chan if a registry
	// replacement is in progress (Enable mid-transaction). This ensures the
	// same Refresh call converges on the new registry rather than degrading
	// silently and waiting up to 30min for the next periodic trigger.
	reg, waitErr := d.snapshotMemorySearch(ctx)
	if waitErr != nil {
		slog.WarnContext(ctx, "memory access refresh: registry snapshot wait cancelled",
			"agent_id", agentID, "error", waitErr)
		// Fall through: reg may still be non-nil if the wait raced. If nil,
		// the branches below observe no registry and degrade gracefully.
	}
	if reg != nil {
		if !bound {
			// Authoritative Unbound is a persistent deny gate, not merely an
			// entry close. Set it before fencing so a late stale-access
			// getOrCreate cannot capture the new epoch and resurrect a writer.
			reg.setSourceTrusted(agentID, false)
			reg.closeAgent(agentID)
		} else {
			// Fence any old identity first, then reopen the authoritative gate
			// for this Bound snapshot. This lets Unknown(A)->Bound(B) converge
			// in one refresh while never resuming A before it is closed.
			reg.closeAgentIfBindingChanged(agentID, access)
			reg.setSourceTrusted(agentID, true)
		}
	}

	sess, ok := d.sessionManager.FindByLookupID(agentID)
	if !ok || sess == nil {
		workDir := sessionWorkDir(agentID)
		memoryDir := filepath.Join(workDir, "memory")
		if bound {
			slog.InfoContext(ctx, "memory access refresh: no session, pre-cloning memory repo",
				"agent_id", agentID, "memory_id", access.MemoryID)
			d.preCloneMemory(ctx, access, agentID, workDir)
		} else if _, err := os.Stat(memoryDir); err == nil {
			slog.InfoContext(ctx, "memory access refresh: no session, access revoked, retaining memory clone offline",
				"agent_id", agentID)
			// Serving generation was already pulled by the unconditional closeAgent
			// above (blocker 1). Re-close defensively in case a concurrent
			// refresh populated an entry between the gate and this branch;
			// closeAgent is idempotent.
			if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
				reg.closeAgent(agentID)
			}
			if err := retainMemoryDirOfflineAfterRevoke(ctx, memoryDir, agentID); err != nil {
				slog.WarnContext(ctx, "memory access refresh: failed to retain memory clone offline",
					"agent_id", agentID, "error", err)
			}
		}
		return
	}
	d.finishIdleMemoryRetry(agentID, ctx)

	sess.mu.Lock()
	sessionUnavailable := sess.retired.Load() || sess.resetting || sess.memoryWatcherReplacementFenced || sess.memoryWatcherTerminalFenced
	hasWatcher := sess.watcher != nil
	currentMemoryID := ""
	currentBranch := ""
	currentProtocol := MemorySyncProtocol("")
	currentGeneration := uint64(0)
	currentContentVersionID := ""
	process := sess.process
	processUsesWorkspaceRootMemory := sess.processUsesWorkspaceRootMemory
	if sess.memoryClone != nil {
		currentMemoryID = sess.memoryClone.MemoryID()
		currentBranch = sess.memoryClone.Branch()
	}
	currentProtocol = sess.memoryProtocol
	currentGeneration = sess.memoryGeneration
	currentContentVersionID = sess.memoryContentVersionID
	if sess.memoryNativeSync != nil {
		currentContentVersionID = sess.memoryNativeSync.RemoteVersionID()
	}
	workDir := sess.workDir
	sess.mu.Unlock()
	if sessionUnavailable {
		slog.InfoContext(ctx, "memory access refresh: session is retiring; skipping watcher mutation",
			"agent_id", agentID)
		return
	}

	// Binding identity is (memoryID, branch). Both feed the reconciler's
	// binding fingerprint, so a branch-only switch must trigger rebind too —
	// otherwise the prior fingerprint's reconciler keeps serving a generation
	// indexed against a branch the clone no longer tracks.
	branchChanged := (currentProtocol == MemorySyncProtocolLegacyGitV1 || access.Protocol == MemorySyncProtocolLegacyGitV1) &&
		strings.TrimSpace(currentBranch) != strings.TrimSpace(access.Branch)
	bindingChanged := hasWatcher && currentMemoryID != "" &&
		(currentMemoryID != access.MemoryID || branchChanged || currentProtocol != access.Protocol || currentGeneration != access.ContentGeneration ||
			(access.Protocol == MemorySyncProtocolNativeContentV1 && currentContentVersionID != access.ContentVersionID))
	nativeHeadChanged := hasWatcher && currentProtocol == MemorySyncProtocolNativeContentV1 &&
		access.Protocol == MemorySyncProtocolNativeContentV1 && currentMemoryID == access.MemoryID &&
		currentGeneration == access.ContentGeneration && currentContentVersionID != access.ContentVersionID

	switch {
	case bound && !hasWatcher:
		slog.InfoContext(ctx, "memory access refresh: access bound, attaching watcher",
			"agent_id", agentID, "memory_id", access.MemoryID)
		d.attachMemoryWatcherLocked(ctx, owner, sess, agentID, workDir)
		sess.mu.Lock()
		if sess.memoryClone != nil && process != nil && processUsesWorkspaceRootMemory {
			restartTarget = &memoryRestartTarget{session: sess, process: process}
		}
		sess.mu.Unlock()
		if restartTarget == nil {
			d.finalizePendingLegacyPreferencesMigration(ctx, sess, false)
		}

	case bound && hasWatcher && nativeHeadChanged && process != nil:
		// A live runtime can write the Memory root without taking the watcher/git
		// lock. Fence this session and retire its exact Process generation before
		// a later session start stages and publishes the new cloud tree.
		// The search generation is exact-tree derived as well, so retire it now;
		// the replacement session must build a new generation from B rather than
		// reusing the same-binding index created for A.
		if reg != nil {
			reg.closeAgent(agentID)
		}
		sess.mu.Lock()
		sess.memoryWatcherReplacementFenced = true
		sess.mu.Unlock()
		d.stopMemoryWatcherAndRetry(sess)
		restartTarget = &memoryRestartTarget{session: sess, process: process}

	case bound && hasWatcher && bindingChanged:
		slog.InfoContext(ctx, "memory access refresh: memory binding switched, rebinding local clone",
			"agent_id", agentID,
			"memory_id_from", currentMemoryID,
			"memory_id_to", access.MemoryID,
			"branch_from", currentBranch,
			"branch_to", access.Branch,
			"replace_local", access.ReplaceLocalContentOnSwitch)
		d.rebindAgentMemoryLocked(ctx, owner, sess, agentID, workDir, access)

	case !bound && hasWatcher:
		slog.InfoContext(ctx, "memory access refresh: access revoked, retaining memory clone offline",
			"agent_id", agentID)
		// Serving generation was already pulled by the unconditional closeAgent
		// above (blocker 1). Re-close defensively against a concurrent refresh
		// repopulating the entry; closeAgent is idempotent.
		if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
			reg.closeAgent(agentID)
		}
		if err := retainSessionMemoryOfflineAfterRevoke(ctx, sess, 60*time.Second); err != nil {
			slog.WarnContext(ctx, "memory access refresh: failed to retain session memory clone offline",
				"agent_id", agentID, "error", err)
		}

	case bound && hasWatcher:
		slog.InfoContext(ctx, "memory access refresh: access unchanged, watcher already attached",
			"agent_id", agentID)

	default:
		slog.InfoContext(ctx, "memory access refresh: no access and no watcher; nothing to do",
			"agent_id", agentID)
	}
	return restartTarget
}

// rebindAgentMemory switches the session's memory clone to a new binding. The
// caller must have confirmed that the binding has changed (memoryID OR branch
// differs from the prior clone's identity, since both feed the reconciler's
// binding fingerprint).
//
// Rebind ordering (per the Task-5c search-integration contract):
//  1. Close the search reconciler for this agent. This pulls the serving
//     generation (Current/Submit) before the watcher stops, so no in-flight
//     Notify event arrives against a closing reconciler. The per-binding
//     owner lock is released as part of Close, unblocking the new binding's
//     reconciler for the same resolved root.
//  2. Stop the existing watcher, flushing pending commits and pushing to the
//     OLD origin.
//  3. Rebind the clone:
//     - ReplaceLocalContentOnSwitch=true: backs up the local clone and creates
//     a fresh clone from the new binding.
//     - ReplaceLocalContentOnSwitch=false: keeps the local clone, repoints its
//     origin remote to the new binding, force-pushes the local history to
//     the new origin.
//  4. Attach a new watcher with a freshly looked-up reconciler as Notify.
//
// On failure the session is left without a watcher; the next refresh retry
// will attempt rebind again.
func (d *memoryService) rebindAgentMemoryLocked(ctx context.Context, owner *memoryWatcherMutationOwner, sess *interactiveSession, agentID, workDir string, access MemoryAccess) {
	// Identity gate already ran in refreshAgentMemoryAccess and closed any
	// stale reconciler for this agent. Re-close defensively in case rebind is
	// entered from a path that skipped the gate; closeAgent is idempotent.
	if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
		reg.closeAgent(agentID)
	}
	result := stopMemoryWatcherWithTimeoutResult(sess, 60*time.Second)
	clone := result.clone
	if clone == nil {
		slog.WarnContext(ctx, "memory access refresh: rebind has no prior clone; attaching fresh watcher",
			"agent_id", agentID)
		d.attachMemoryWatcherLocked(ctx, owner, sess, agentID, workDir)
		return
	}
	if result.pushErr != nil {
		slog.WarnContext(ctx, "memory access refresh: rebind preceded by push failure to prior origin; attempting rebind anyway",
			"agent_id", agentID, "error", result.pushErr)
	}
	if access.Protocol == MemorySyncProtocolNativeContentV1 {
		d.attachMemoryWatcherLocked(ctx, owner, sess, agentID, workDir)
		return
	}
	if !clone.UsesRemote() {
		if err := clone.RetainOffline(ctx, time.Time{}); err != nil {
			d.setMemorySyncStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonCloneFailed))
			return
		}
		d.attachMemoryWatcherLocked(ctx, owner, sess, agentID, workDir)
		return
	}

	rebindResult, err := clone.Rebind(ctx, vault.RebindOpts{
		NewMemoryID:  access.MemoryID,
		NewRepoURL:   access.OriginURL,
		NewBranch:    access.Branch,
		ReplaceLocal: access.ReplaceLocalContentOnSwitch,
		Auth:         access.Auth,
	})
	if err != nil {
		d.setMemorySyncStatus(agentID, degradedMemorySyncStatus(memorywatcher.FailureReason(err, memorywatcher.StatusReasonCloneFailed)))
		slog.WarnContext(ctx, "memory access refresh: rebind failed; watcher remains off",
			"agent_id", agentID, "memory_id", access.MemoryID, "error", err)
		return
	}
	notify := d.lookupSearchReconciler(ctx, agentID, access)
	d.startMemoryWatcherForAccess(ctx, owner, sess, agentID, access, rebindResult.Clone, nil, notify)
}

// preCloneMemory ensures the memory repo is cloned to the agent workdir even
// when no session is active, so the next session start can reopen it without
// a full download. Failures are logged but not propagated.
func (d *memoryService) preCloneMemory(ctx context.Context, access MemoryAccess, agentID, workDir string) {
	if access.ImportExistingLocalContent && workDir == sessionWorkDir(agentID) {
		seedAgentRootFromLatestSession(agentID, workDir)
	}
	clone, migration, nativeSync, ok := d.ensureMemoryCloneForAccess(ctx, agentID, workDir, access)
	if !ok {
		d.scheduleIdleMemoryRetry(agentID)
		return
	}
	if migration != nil {
		_, err := migration.finalize(ctx, clone, nil, true)
		if err != nil {
			slog.WarnContext(ctx, "memory import: close legacy user preference migration failed",
				"agent_id", agentID, "error", err)
		}
	}
	status, err := syncIdleMemoryAccess(ctx, clone, nativeSync, agentID, access.Auth)
	d.setMemorySyncStatus(agentID, status)
	if err != nil {
		d.scheduleIdleMemoryRetry(agentID)
		slog.WarnContext(ctx, "memory access refresh: pre-clone sync incomplete; local content remains safe",
			"agent_id", agentID, "memory_id", access.MemoryID, "reason", status.Reason, "error", err)
	} else {
		d.finishIdleMemoryRetry(agentID, ctx)
	}
	slog.InfoContext(ctx, "memory access refresh: pre-clone available",
		"agent_id", agentID, "memory_id", access.MemoryID, "path", clone.Path())

	// Best-effort: pull the search reconciler for the binding so an agent with
	// no active session still produces a generation. Identity gate already ran
	// in refreshAgentMemoryAccess so any stale reconciler for a prior binding
	// has been closed and its per-derivative-root owner lock released; this
	// getOrCreate is the one that publishes the new fingerprint.
	if reg, _ := d.snapshotMemorySearch(ctx); reg != nil {
		if _, _, err := reg.getOrCreate(ctx, agentID, access); err != nil {
			slog.WarnContext(ctx, "memory search: pre-clone reconciler start failed",
				"agent_id", agentID, "memory_id", access.MemoryID, "error", err)
		}
	}
}

// buildMemoryContextResolver returns the callback the watcher uses to label
// each commit. The closure reads the live session binding under sess.mu so
// the watcher always attributes commits to the current conversation scope
// (channel, thread, task) even when the active target changes mid-session.
//
// Per the agreed contract (thread #agentteam-in-agentteam:57d48e57), commit author is
// the agent's own identity — never the human actor. Channel/Thread/Task live
// in the commit body and trailers.
func (d *memoryService) buildMemoryContextResolver(sess *interactiveSession) memorywatcher.ContextResolver {
	return func(ctx context.Context) memorywatcher.CommitContext {
		sess.mu.Lock()
		defer sess.mu.Unlock()

		out := memorywatcher.CommitContext{
			AgentID:      sess.agentID,
			ActorName:    agentAuthorName(sess.agent.Name, sess.agentID),
			ChannelID:    sess.binding.channelID,
			ChannelType:  sess.binding.targetScopeType,
			ThreadTarget: threadTargetFromBinding(sess.binding),
			ThreadTitle:  threadTitleFromBinding(sess.binding),
		}

		if t := sess.activeTaskTarget.Normalize(); t.Valid() {
			if strings.TrimSpace(t.TaskScope) != "" {
				out.TaskScope = t.TaskScope
			}
			if strings.TrimSpace(t.Source) != "" {
				out.TriggerSource = t.Source
			}
			if t.TaskNumber > 0 {
				out.TaskNumber = t.TaskNumber
			}
			if strings.TrimSpace(t.TaskTitle) != "" {
				out.TaskTitle = t.TaskTitle
			}
		}
		return out
	}
}

func threadTitleFromBinding(binding conversationBinding) string {
	agentteamMeta, ok := binding.meta.(*types.AgentTeamMeta)
	if !ok || agentteamMeta == nil {
		return ""
	}
	root, ok := types.ReplyThreadRootFromMetadata(agentteamMeta.Metadata)
	if !ok {
		return ""
	}
	return memoryContextTitle(root.Content)
}

func memoryContextTitle(value string) string {
	trimmed := strings.Join(strings.Fields(value), " ")
	if trimmed == "" {
		return ""
	}
	const maxRunes = 120
	runes := []rune(trimmed)
	if len(runes) <= maxRunes {
		return trimmed
	}
	if maxRunes <= 1 {
		return string(runes[:maxRunes])
	}
	return string(runes[:maxRunes-1]) + "…"
}

// agentAuthorName picks a non-empty display name for commit attribution. The
// agent config name wins; falling back to the agent ID keeps commits
// attributable even when the config is malformed.
func agentAuthorName(name, agentID string) string {
	if trimmed := strings.TrimSpace(name); trimmed != "" {
		return trimmed
	}
	return agentID
}