package dispatcher import ( "archive/zip" "bytes" "context" "errors" "fmt" "io" "os" "path/filepath" "strings" "sync" "testing" "time" "github.com/go-git/go-git/v5" "github.com/go-git/go-git/v5/config" "github.com/go-git/go-git/v5/plumbing" "github.com/stretchr/testify/require" "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/profile" "github.com/tidbcloud/AgentTeam/desktop/internal/types" "github.com/tidbcloud/AgentTeam/desktop/internal/vault" ) // fakeMemoryAccessProvider is a stub MemoryAccessProvider for unit tests. It // returns whatever the test registered via set(), guarded by a mutex because // Lookup runs on the watcher goroutine while tests assert on the main goroutine. type fakeMemoryAccessProvider struct { mu sync.Mutex bindings map[string]MemoryAccess err error unknown bool calls int } type fakeMemoryAccessBatchProvider struct { batches []map[string]MemoryAccessLookup } func nativeMemoryStatePath(workDir string, access MemoryAccess) string { return filepath.Join(workDir, ".memory-sync", fmt.Sprintf("native-%s-g%d.json", safeVersionSegment(access.MemoryID), access.ContentGeneration)) } func nativeMemoryReplaceTxnRoot(root string) string { return filepath.Join(filepath.Dir(root), "."+filepath.Base(root)+".native-refresh") } func (p *fakeMemoryAccessBatchProvider) Lookup(context.Context, string) (MemoryAccess, MemoryAccessState, error) { return MemoryAccess{}, MemoryAccessUnknown, errors.New("single lookup is not expected") } func (p *fakeMemoryAccessBatchProvider) LookupBatch(ctx context.Context, _ []string) <-chan map[string]MemoryAccessLookup { out := make(chan map[string]MemoryAccessLookup, len(p.batches)) go func() { defer close(out) for _, batch := range p.batches { select { case out <- batch: case <-ctx.Done(): return } } }() return out } type fakeNativeMemoryRemote struct { mu sync.Mutex calls []nativememory.Checkpoint archiveCalls []string archiveStarted chan<- struct{} releaseArchive <-chan struct{} err error expectedGeneration uint64 archiveFiles map[string]string archiveFilesByVersion map[string]map[string]string } func (r *fakeNativeMemoryRemote) PublishCheckpoint(_ context.Context, _ string, input nativememory.Checkpoint) (nativememory.Result, error) { r.mu.Lock() defer r.mu.Unlock() r.calls = append(r.calls, input) if r.err != nil { return nativememory.Result{}, r.err } if r.expectedGeneration != 0 && input.Generation != r.expectedGeneration { return nativememory.Result{}, errors.New("stale native memory generation") } return nativememory.Result{ VersionID: "ctv_next", OperationID: input.OperationID, ReplicaID: input.ReplicaID, Generation: input.Generation, CoveredFromSequence: input.CoveredFromSequence, CoveredThroughSequence: input.CoveredThroughSequence, }, nil } func (r *fakeNativeMemoryRemote) OpenVersionArchive(ctx context.Context, _, versionID string) (io.ReadCloser, error) { r.mu.Lock() r.archiveCalls = append(r.archiveCalls, versionID) r.mu.Unlock() if r.archiveStarted != nil { select { case r.archiveStarted <- struct{}{}: default: } } if r.releaseArchive != nil { select { case <-r.releaseArchive: case <-ctx.Done(): return nil, ctx.Err() } } var buffer bytes.Buffer writer := zip.NewWriter(&buffer) files := r.archiveFiles if selected := r.archiveFilesByVersion[versionID]; selected != nil { files = selected } if len(files) == 0 { files = map[string]string{"cloud.md": "cloud\n"} } for name, content := range files { file, err := writer.Create(name) if err != nil { return nil, err } if _, err := file.Write([]byte(content)); err != nil { return nil, err } } if err := writer.Close(); err != nil { return nil, err } return io.NopCloser(bytes.NewReader(buffer.Bytes())), nil } func (r *fakeNativeMemoryRemote) archiveCallCount(versionID string) int { r.mu.Lock() defer r.mu.Unlock() count := 0 for _, called := range r.archiveCalls { if called == versionID { count++ } } return count } func (r *fakeNativeMemoryRemote) setError(err error) { r.mu.Lock() r.err = err r.mu.Unlock() } func (r *fakeNativeMemoryRemote) callCount() int { r.mu.Lock() defer r.mu.Unlock() return len(r.calls) } func nativeCheckpointFileContent(t *testing.T, files []nativememory.File, name string) string { t.Helper() for _, file := range files { if file.Path == name { return string(file.Content) } } t.Fatalf("native checkpoint file %q not found", name) return "" } type blockingMemoryAccessProvider struct { entered chan struct{} exited chan struct{} } func (p *blockingMemoryAccessProvider) Lookup(ctx context.Context, _ string) (MemoryAccess, MemoryAccessState, error) { close(p.entered) <-ctx.Done() close(p.exited) return MemoryAccess{}, MemoryAccessUnknown, ctx.Err() } func (f *fakeMemoryAccessProvider) Lookup(_ context.Context, agentID string) (MemoryAccess, MemoryAccessState, error) { f.mu.Lock() defer f.mu.Unlock() f.calls++ if f.err != nil { return MemoryAccess{}, MemoryAccessUnknown, f.err } if f.unknown { return MemoryAccess{}, MemoryAccessUnknown, nil } g, ok := f.bindings[agentID] if !ok { return MemoryAccess{}, MemoryAccessUnbound, nil } return g, MemoryAccessBound, nil } func (f *fakeMemoryAccessProvider) set(agentID string, access MemoryAccess) { f.mu.Lock() defer f.mu.Unlock() if access.Protocol == "" { access.Protocol = MemorySyncProtocolLegacyGitV1 } if f.bindings == nil { f.bindings = map[string]MemoryAccess{} } f.bindings[agentID] = access } func (f *fakeMemoryAccessProvider) remove(agentID string) { f.mu.Lock() defer f.mu.Unlock() delete(f.bindings, agentID) } func (f *fakeMemoryAccessProvider) setErr(err error) { f.mu.Lock() defer f.mu.Unlock() f.err = err } func (f *fakeMemoryAccessProvider) setUnknown(unknown bool) { f.mu.Lock() defer f.mu.Unlock() f.unknown = unknown } func (f *fakeMemoryAccessProvider) callCount() int { f.mu.Lock() defer f.mu.Unlock() return f.calls } // makeOriginRepoForMemory seeds a normal source repo with one commit on main // and returns its file:// URL. Mirrors the helper in memorywatcher/e2e_test.go // but lives here so dispatcher tests stay self-contained. func makeOriginRepoForMemory(t *testing.T, originDir string) string { t.Helper() repo, err := git.PlainInit(originDir, false) require.NoError(t, err) require.NoError(t, repo.Storer.SetReference(plumbing.NewSymbolicReference(plumbing.HEAD, plumbing.NewBranchReferenceName("main")))) wt, err := repo.Worktree() require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(originDir, "MEMORY.md"), []byte("# Origin memory\n\nInitial content.\n"), 0o644)) _, err = wt.Add("MEMORY.md") require.NoError(t, err) _, err = wt.Commit("initial commit", &git.CommitOptions{}) require.NoError(t, err) return "file://" + originDir } // makeBareOriginRepoForMemory seeds a bare origin with one commit on main and // returns its file:// URL. Bare origins accept pushes to any ref without the // "branch is currently checked out" error non-bare repos return — necessary // for tests that verify the agent's commit round-trips back to origin via the // watcher's push path. Mirrors the helper in memorywatcher/e2e_test.go but // lives here so dispatcher tests stay self-contained. func makeBareOriginRepoForMemory(t *testing.T, originDir string) string { t.Helper() bareRepo, err := git.PlainInit(originDir, true) require.NoError(t, err) require.NoError(t, bareRepo.Storer.SetReference(plumbing.NewSymbolicReference(plumbing.HEAD, plumbing.NewBranchReferenceName("main")))) staging := t.TempDir() stagingRepo, err := git.PlainInit(staging, false) require.NoError(t, err) require.NoError(t, stagingRepo.Storer.SetReference(plumbing.NewSymbolicReference(plumbing.HEAD, plumbing.NewBranchReferenceName("main")))) wt, err := stagingRepo.Worktree() require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(staging, "MEMORY.md"), []byte("# Origin memory\n\nInitial content.\n"), 0o644)) _, err = wt.Add("MEMORY.md") require.NoError(t, err) _, err = wt.Commit("initial commit", &git.CommitOptions{}) require.NoError(t, err) _, err = stagingRepo.CreateRemote(&config.RemoteConfig{ Name: "origin", URLs: []string{"file://" + originDir}, }) require.NoError(t, err) require.NoError(t, stagingRepo.PushContext(context.Background(), &git.PushOptions{ RemoteName: "origin", RefSpecs: []config.RefSpec{"+refs/heads/main:refs/heads/main"}, })) return "file://" + originDir } // refMessageOnOrigin reads the commit message at the given ref name on a bare // origin. Empty string if the ref does not exist (used to assert "not pushed"). func refMessageOnOrigin(t *testing.T, originDir, refName string) string { t.Helper() repo, err := git.PlainOpen(originDir) require.NoError(t, err) ref, err := repo.Reference(plumbing.NewBranchReferenceName(refName), true) if err != nil { return "" } commit, err := repo.CommitObject(ref.Hash()) require.NoError(t, err) return commit.Message } func newDispatcherWithMemoryProvider(t *testing.T, provider MemoryAccessProvider) *Dispatcher { t.Helper() d := mustNewDispatcher("") d.SetMemoryAccessProvider(provider) t.Cleanup(d.closeIdleMemoryRetries) return d } // headMessageAt reads the HEAD commit message at clonePath. func headMessageAt(t *testing.T, clonePath string) string { t.Helper() repo, err := git.PlainOpen(clonePath) require.NoError(t, err) head, err := repo.Head() require.NoError(t, err) commit, err := repo.CommitObject(head.Hash()) require.NoError(t, err) return commit.Message } // headAuthorNameAt reads the HEAD commit author name at clonePath. func headAuthorNameAt(t *testing.T, clonePath string) string { t.Helper() repo, err := git.PlainOpen(clonePath) require.NoError(t, err) head, err := repo.Head() require.NoError(t, err) commit, err := repo.CommitObject(head.Hash()) require.NoError(t, err) return commit.Author.Name } // TestAttachMemoryWatcherClonesAndStartsWatcher drives attachMemoryWatcher // directly with a real vault clone + fake provider and verifies: // - The clone is created at workDir/memory // - The watcher is recorded on sess and observes file writes // - The commit message carries Channel/Thread/Task context and the author // carries agent identity func TestAttachMemoryWatcherClonesAndStartsWatcher(t *testing.T) { t.Parallel() origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_alice" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_alice", OriginURL: origin, Branch: "alice-memory", }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, binding: conversationBinding{ channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam", }, } d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.watcher, "watcher should be attached when binding exists") require.NotNil(t, sess.memoryClone, "memory clone should be created") require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) return ok && status.State == memorywatcher.StatusReady }, time.Second, 5*time.Millisecond, "startup push probe should converge to ready") // Write a file inside the clone path. The watcher should commit within a // short debounce window. require.NoError(t, os.WriteFile( filepath.Join(sess.memoryClone.Path(), "MEMORY.md"), []byte("# alice's memory\n\nwatcher-driven commit.\n"), 0o644, )) require.Eventually(t, func() bool { return strings.TrimSpace(headMessageAt(t, sess.memoryClone.Path())) != "initial commit" }, 3*time.Second, 15*time.Millisecond) msg := headMessageAt(t, sess.memoryClone.Path()) require.Contains(t, msg, "Update memory from channel #agentteam") require.Contains(t, msg, "AgentTeam-Channel: #agentteam") require.NotContains(t, msg, "AgentTeam-Agent:") require.Equal(t, "agt_alice", headAuthorNameAt(t, sess.memoryClone.Path())) } func TestAttachNativeMemoryWatcherPublishesCheckpointWithoutAGSRemote(t *testing.T) { oldInterval := nativeMemoryCheckpointMinInterval nativeMemoryCheckpointMinInterval = time.Nanosecond t.Cleanup(func() { nativeMemoryCheckpointMinInterval = oldInterval }) workDir := t.TempDir() agentID := "agt_native" remote := &fakeNativeMemoryRemote{} provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_base", ContentGeneration: 1, NativeRemote: remote, }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, binding: conversationBinding{channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam"}} d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) require.NotNil(t, sess.watcher) require.NotNil(t, sess.memoryClone) require.False(t, sess.memoryClone.UsesRemote()) local, err := vault.IsLocalJournal(sess.memoryClone.Path()) require.NoError(t, err) require.True(t, local) require.Equal(t, "cloud\n", readFileString(t, filepath.Join(sess.memoryClone.Path(), "cloud.md"))) require.NoError(t, os.WriteFile(filepath.Join(sess.memoryClone.Path(), "MEMORY.md"), []byte("# native local\n"), 0o644)) require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) return ok && status.State == memorywatcher.StatusReady && remote.callCount() == 1 }, 3*time.Second, 15*time.Millisecond) stopMemoryWatcher(sess) require.Equal(t, 1, remote.callCount(), "teardown must not call Git push or duplicate an acknowledged native checkpoint") require.FileExists(t, nativeMemoryStatePath(workDir, MemoryAccess{MemoryID: "mem_native", ContentGeneration: 1})) } func TestOpenNativeMemoryJournalPreservesFallbackAlongsideCloudCanonical(t *testing.T) { oldInterval := nativeMemoryCheckpointMinInterval nativeMemoryCheckpointMinInterval = time.Nanosecond t.Cleanup(func() { nativeMemoryCheckpointMinInterval = oldInterval }) workDir := t.TempDir() require.NoError(t, os.WriteFile(filepath.Join(workDir, "MEMORY.md"), []byte("# local fallback\n"), 0o644)) remote := &fakeNativeMemoryRemote{archiveFiles: map[string]string{ "MEMORY.md": "# cloud canonical\n", "cloud.md": "cloud-only\n", }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_base", ContentGeneration: 1, ImportExistingLocalContent: true, NativeRemote: remote, } clone, syncer, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.Equal(t, "# cloud canonical\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) require.Equal(t, "# local fallback\n", readFileString(t, filepath.Join(workDir, "MEMORY.md")), "the source fallback must remain available offline") require.Equal(t, "# local fallback\n", readFileString(t, filepath.Join(clone.Path(), ".conflicts", "local-import", "MEMORY.md"))) _, err = syncer.Publish(t.Context()) require.NoError(t, err) require.Len(t, remote.calls, 1) require.Equal(t, "# cloud canonical\n", nativeCheckpointFileContent(t, remote.calls[0].Files, "MEMORY.md")) require.Equal(t, "# local fallback\n", nativeCheckpointFileContent(t, remote.calls[0].Files, ".conflicts/local-import/MEMORY.md")) } func TestOpenNativeMemoryJournalFencesStaleGenerationAndKeepsBothOutboxes(t *testing.T) { oldInterval := nativeMemoryCheckpointMinInterval nativeMemoryCheckpointMinInterval = time.Nanosecond t.Cleanup(func() { nativeMemoryCheckpointMinInterval = oldInterval }) workDir := t.TempDir() remote := &fakeNativeMemoryRemote{expectedGeneration: 3} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_base", ContentGeneration: 3, NativeRemote: remote, } clone, staleSync, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "local.md"), []byte("offline edit\n"), 0o644)) revision, err := clone.CommitPending(t.Context(), "offline edit", vault.CommitPendingOpts{}) require.NoError(t, err) require.NoError(t, staleSync.RecordLocalRevision(revision)) remote.expectedGeneration = 4 _, err = staleSync.Publish(t.Context()) require.ErrorContains(t, err, "stale native memory generation") require.FileExists(t, nativeMemoryStatePath(workDir, access)) currentAccess := access currentAccess.ContentGeneration = 4 _, currentSync, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, currentAccess) require.NoError(t, err) _, err = currentSync.Publish(t.Context()) require.NoError(t, err) require.FileExists(t, nativeMemoryStatePath(workDir, currentAccess)) require.Len(t, remote.calls, 2) require.EqualValues(t, 3, remote.calls[0].Generation) require.EqualValues(t, 4, remote.calls[1].Generation) require.Equal(t, "offline edit\n", nativeCheckpointFileContent(t, remote.calls[1].Files, "local.md")) } func TestOpenNativeMemoryJournalMaterializesRemoteRestoreAsLocalVersion(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_old": {"MEMORY.md": "old cloud\n", "removed.md": "old only\n"}, "ctv_restored": {"MEMORY.md": "restored cloud\n", "restored.md": "new only\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_old", ContentGeneration: 3, NativeRemote: remote, } clone, syncer, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.False(t, syncer.Pending()) oldHead, err := clone.HeadRevision(t.Context()) require.NoError(t, err) access.ContentVersionID = "ctv_restored" refreshed, refreshedSync, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.False(t, refreshedSync.Pending(), "an exact cloud restore is a local baseline, not an outbound edit") require.Equal(t, "restored cloud\n", readFileString(t, filepath.Join(refreshed.Path(), "MEMORY.md"))) require.Equal(t, "new only\n", readFileString(t, filepath.Join(refreshed.Path(), "restored.md"))) require.NoFileExists(t, filepath.Join(refreshed.Path(), "removed.md")) newHead, err := refreshed.HeadRevision(t.Context()) require.NoError(t, err) require.NotEqual(t, oldHead, newHead, "restoring exact cloud content must create a new immutable local version") commit, err := refreshed.Repository().CommitObject(plumbing.NewHash(newHead)) require.NoError(t, err) require.Len(t, commit.ParentHashes, 1) require.Equal(t, oldHead, commit.ParentHashes[0].String(), "the prior local lineage remains recoverable") } func TestOpenNativeMemoryJournalMaterializesSameTreeRestoreAsDistinctLocalVersion(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_old": {"MEMORY.md": "same tree\n"}, "ctv_restored": {"MEMORY.md": "same tree\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_old", ContentGeneration: 3, NativeRemote: remote, } clone, _, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) oldHead, err := clone.HeadRevision(t.Context()) require.NoError(t, err) oldCommit, err := clone.Repository().CommitObject(plumbing.NewHash(oldHead)) require.NoError(t, err) access.ContentVersionID = "ctv_restored" refreshed, syncer, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.False(t, syncer.Pending()) newHead, err := refreshed.HeadRevision(t.Context()) require.NoError(t, err) require.NotEqual(t, oldHead, newHead, "a distinct ContentVersion must have a distinct local journal version") newCommit, err := refreshed.Repository().CommitObject(plumbing.NewHash(newHead)) require.NoError(t, err) require.Equal(t, oldCommit.TreeHash, newCommit.TreeHash) require.Equal(t, []plumbing.Hash{plumbing.NewHash(oldHead)}, newCommit.ParentHashes) } func TestOpenNativeMemoryJournalRecoversCleanSidecarCrashBeforeRemoteRead(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_old": {"MEMORY.md": "old cloud\n"}, "ctv_concurrent": {"MEMORY.md": "must not be read\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_old", ContentGeneration: 3, NativeRemote: remote, } clone, _, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "MEMORY.md"), []byte("clean unseen local commit\n"), 0o644)) crashHead, err := clone.CommitPending(t.Context(), "commit before sidecar crash", vault.CommitPendingOpts{}) require.NoError(t, err) access.ContentVersionID = "ctv_concurrent" _, _, err = openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.ErrorIs(t, err, nativememory.ErrRemoteHeadChanged) require.Zero(t, remote.archiveCallCount("ctv_concurrent"), "admission must fail before opening the concurrent archive") head, err := clone.HeadRevision(t.Context()) require.NoError(t, err) require.Equal(t, crashHead, head) require.Equal(t, "clean unseen local commit\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) access.ContentVersionID = "ctv_old" _, recovered, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.True(t, recovered.Pending(), "the unseen clean commit must remain durable pending work") } func TestOpenNativeMemoryJournalLostResponseAckStillMaterializesLaterHead(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_old": {"MEMORY.md": "old cloud\n"}, "ctv_restored": {"MEMORY.md": "restored after ack\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_old", ContentGeneration: 3, ActiveReplicaID: "replica-1", NativeRemote: remote, } clone, syncer, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "MEMORY.md"), []byte("published V1\n"), 0o644)) publishedHead, err := clone.CommitPending(t.Context(), "publish V1", vault.CommitPendingOpts{}) require.NoError(t, err) require.NoError(t, syncer.RecordLocalRevision(publishedHead)) remote.setError(errors.New("response lost")) _, err = syncer.Publish(t.Context()) require.EqualError(t, err, "response lost") remote.setError(nil) access.AcknowledgedSequence = 1 access.ContentVersionID = "ctv_restored" refreshed, refreshedSync, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.False(t, refreshedSync.Pending()) require.Equal(t, "restored after ack\n", readFileString(t, filepath.Join(refreshed.Path(), "MEMORY.md"))) newHead, err := refreshed.HeadRevision(t.Context()) require.NoError(t, err) require.NotEqual(t, publishedHead, newHead) require.Equal(t, 1, remote.archiveCallCount("ctv_restored")) } func TestOpenNativeMemoryJournalLostResponseAckRejectsLaterHeadWithTail(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_old": {"MEMORY.md": "old cloud\n"}, "ctv_restored": {"MEMORY.md": "must not be read\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_old", ContentGeneration: 3, ActiveReplicaID: "replica-1", NativeRemote: remote, } clone, syncer, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "MEMORY.md"), []byte("published V1\n"), 0o644)) publishedHead, err := clone.CommitPending(t.Context(), "publish V1", vault.CommitPendingOpts{}) require.NoError(t, err) require.NoError(t, syncer.RecordLocalRevision(publishedHead)) remote.setError(errors.New("response lost")) _, err = syncer.Publish(t.Context()) require.EqualError(t, err, "response lost") require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "MEMORY.md"), []byte("local tail\n"), 0o644)) tailHead, err := clone.CommitPending(t.Context(), "local tail", vault.CommitPendingOpts{}) require.NoError(t, err) require.NoError(t, syncer.RecordLocalRevision(tailHead)) access.AcknowledgedSequence = 1 access.ContentVersionID = "ctv_restored" _, _, err = openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.ErrorIs(t, err, nativememory.ErrRemoteHeadChanged) require.Zero(t, remote.archiveCallCount("ctv_restored")) require.Equal(t, "local tail\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) head, err := clone.HeadRevision(t.Context()) require.NoError(t, err) require.Equal(t, tailHead, head) } func TestOpenNativeMemoryJournalFailsClosedOnConcurrentHeadWithLocalEdit(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_old": {"MEMORY.md": "old cloud\n"}, "ctv_concurrent": {"MEMORY.md": "other writer\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_old", ContentGeneration: 3, NativeRemote: remote, } clone, _, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "MEMORY.md"), []byte("local unsynced\n"), 0o644)) access.ContentVersionID = "ctv_concurrent" _, _, err = openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.ErrorIs(t, err, nativememory.ErrRemoteHeadChanged) require.Equal(t, "local unsynced\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) require.NoFileExists(t, filepath.Join(clone.Path(), ".conflicts", "cloud-ctv_concurrent")) } func TestNativeMemoryStatePathDoesNotAliasLongIdentifiers(t *testing.T) { workDir := t.TempDir() left := nativeMemoryStatePath(workDir, MemoryAccess{MemoryID: "mem_shared_prefix_aaaaaaaa", ContentGeneration: 1}) right := nativeMemoryStatePath(workDir, MemoryAccess{MemoryID: "mem_shared_prefix_bbbbbbbb", ContentGeneration: 1}) require.NotEqual(t, left, right) require.Equal(t, left, nativeMemoryStatePath(workDir, MemoryAccess{MemoryID: "mem_shared_prefix_aaaaaaaa", ContentGeneration: 1})) } func TestNativeMemoryArchivePathsRejectMetadataAndSymlinkTraversal(t *testing.T) { for _, value := range []string{"../outside", "/absolute", " leading.md", `windows\\path`, ".git/config", "nested/.derived/index", "nested/.agentteam-state"} { _, err := safeNativeArchivePath(value) require.Error(t, err, value) } root := t.TempDir() outside := t.TempDir() require.NoError(t, os.Symlink(outside, filepath.Join(root, "linked"))) _, err := safeNativeMemoryTarget(root, "linked/escaped.md") require.ErrorContains(t, err, "traverses a symlink") } func TestOpenNativeMemoryJournalRejectsTargetFilesystemPathConflictsBeforeReplace(t *testing.T) { for _, tc := range []struct { name string files map[string]string }{ {name: "case equivalent", files: map[string]string{"Notes.md": "one", "notes.md": "two"}}, {name: "unicode equivalent", files: map[string]string{"\u00e9.md": "one", "e\u0301.md": "two"}}, {name: "file directory prefix", files: map[string]string{"notes": "file", "notes/item.md": "child"}}, } { t.Run(tc.name, func(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_a": {"MEMORY.md": "stable\n"}, "ctv_b": tc.files, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_a", ContentGeneration: 1, NativeRemote: remote, } clone, _, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) stableHead, err := clone.HeadRevision(t.Context()) require.NoError(t, err) access.ContentVersionID = "ctv_b" _, _, err = openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.Error(t, err) head, headErr := clone.HeadRevision(t.Context()) require.NoError(t, headErr) require.Equal(t, stableHead, head) require.Equal(t, "stable\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) }) } } func TestOpenNativeMemoryJournalRefreshFailureNeverExposesPartialTreeOrDeletesLaterEdits(t *testing.T) { tests := []struct { name string hooks *nativeMemoryRefreshHooks }{ {name: "replace", hooks: &nativeMemoryRefreshHooks{beforeSwitch: func() error { return errors.New("replace failed") }}}, {name: "commit", hooks: &nativeMemoryRefreshHooks{commit: func(context.Context, *vault.Clone) (string, error) { return "", errors.New("commit failed") }}}, {name: "complete persist", hooks: &nativeMemoryRefreshHooks{complete: func(*nativememory.Sync, string, string) error { return errors.New("complete persist failed") }}}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_a": {"MEMORY.md": "stable\n", "stable.md": "keep\n"}, "ctv_b": {"MEMORY.md": "remote\n", "remote.md": "new\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_a", ContentGeneration: 1, NativeRemote: remote, } clone, _, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) stableHead, err := clone.HeadRevision(t.Context()) require.NoError(t, err) access.ContentVersionID = "ctv_b" access.nativeRefreshHooks = tc.hooks _, _, err = openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.Error(t, err) head, headErr := clone.HeadRevision(t.Context()) require.NoError(t, headErr) require.Equal(t, stableHead, head) require.Equal(t, "stable\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) require.Equal(t, "keep\n", readFileString(t, filepath.Join(clone.Path(), "stable.md"))) require.NoFileExists(t, filepath.Join(clone.Path(), "remote.md")) // The runtime may continue locally after attach fails. Retry must turn // this edit into pending work before another remote read or replace. require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "after-failure.md"), []byte("local survives\n"), 0o644)) access.nativeRefreshHooks = nil _, _, err = openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.ErrorIs(t, err, nativememory.ErrRemoteHeadChanged) require.Equal(t, 1, remote.archiveCallCount("ctv_b")) require.Equal(t, "stable\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) require.Equal(t, "local survives\n", readFileString(t, filepath.Join(clone.Path(), "after-failure.md"))) require.NoFileExists(t, filepath.Join(clone.Path(), "remote.md")) }) } } func TestNativeMemoryRefreshTransactionRecoversEveryCrashPhase(t *testing.T) { for _, phase := range []string{"pre-manifest", "backing-up", "installing", "switched", "committed", "adopted"} { t.Run(phase, func(t *testing.T) { workDir := t.TempDir() remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_a": {"MEMORY.md": "old memory\n", "old.md": "old only\n"}, "ctv_b": {"MEMORY.md": "new memory\n", "new.md": "new only\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_a", ContentGeneration: 1, NativeRemote: remote, } clone, _, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) oldHead, err := clone.HeadRevision(t.Context()) require.NoError(t, err) access.ContentVersionID = "ctv_b" syncer, err := nativememory.Open(nativememory.Options{ Root: clone.Path(), StatePath: nativeMemoryStatePath(workDir, access), Remote: remote, Binding: nativememory.Binding{ MemoryID: access.MemoryID, ContentID: access.ContentID, ContentVersionID: access.ContentVersionID, Generation: access.ContentGeneration, ReplicaID: access.ActiveReplicaID, }, }) require.NoError(t, err) require.NoError(t, syncer.BeginRemoteRefresh("ctv_b")) files, err := readNativeMemoryArchive(t.Context(), access) require.NoError(t, err) var txn *nativeMemoryReplaceTxn if phase == "pre-manifest" { txnRoot := nativeMemoryReplaceTxnRoot(clone.Path()) require.NoError(t, os.Mkdir(txnRoot, 0o700)) require.NoError(t, syncSessionResumeStateDir(filepath.Dir(clone.Path()))) stageRoot := filepath.Join(txnRoot, "stage") require.NoError(t, os.Mkdir(stageRoot, 0o700)) require.NoError(t, os.WriteFile(filepath.Join(stageRoot, "MEMORY.md"), []byte("new memory\n"), 0o644)) require.NoError(t, syncNativeMemoryStage(stageRoot)) // persistManifest writes and fsyncs a temporary file before its // atomic rename. Simulate a kill in that exact pre-rename window. require.NoError(t, os.WriteFile(filepath.Join(txnRoot, ".manifest-crash.tmp"), []byte(`{"partial":true}`), 0o600)) } else { txn, err = prepareNativeMemoryFiles(t.Context(), clone, "ctv_b", files) require.NoError(t, err) } switch phase { case "pre-manifest": case "backing-up": require.NoError(t, os.MkdirAll(txn.backupRoot, 0o700)) require.NoError(t, os.Rename(filepath.Join(txn.root, txn.manifest.OldEntries[0]), filepath.Join(txn.backupRoot, txn.manifest.OldEntries[0]))) case "installing": require.NoError(t, os.MkdirAll(txn.backupRoot, 0o700)) for _, name := range txn.manifest.OldEntries { require.NoError(t, os.Rename(filepath.Join(txn.root, name), filepath.Join(txn.backupRoot, name))) } txn.manifest.Phase = nativeMemoryReplaceInstalling require.NoError(t, txn.persistManifest()) require.NoError(t, os.Rename(filepath.Join(txn.stageRoot, txn.manifest.NewEntries[0]), filepath.Join(txn.root, txn.manifest.NewEntries[0]))) case "switched", "committed", "adopted": require.NoError(t, txn.switchRoots()) if phase == "committed" || phase == "adopted" { revision, commitErr := clone.CommitPending(t.Context(), "Adopt crash test cloud version", vault.CommitPendingOpts{AllowEmpty: true}) require.NoError(t, commitErr) require.NoError(t, txn.recordCommittedRevision(revision)) if phase == "adopted" { require.NoError(t, syncer.CompleteRemoteRefresh("ctv_b", revision)) } } } // Simulate a new process: only the durable sidecar and manifest are // available. Incomplete phases must restore the entire old tree before // any remote retry; a durably adopted phase must retain the entire new tree. restarted, err := nativememory.Open(nativememory.Options{ Root: clone.Path(), StatePath: nativeMemoryStatePath(workDir, access), Remote: remote, Binding: nativememory.Binding{ MemoryID: access.MemoryID, ContentID: access.ContentID, ContentVersionID: access.ContentVersionID, Generation: access.ContentGeneration, ReplicaID: access.ActiveReplicaID, }, }) require.NoError(t, err) require.NoError(t, recoverNativeMemoryReplaceTxn(t.Context(), clone, restarted)) require.NoDirExists(t, nativeMemoryReplaceTxnRoot(clone.Path())) if phase == "adopted" { require.Equal(t, "new memory\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) require.Equal(t, "new only\n", readFileString(t, filepath.Join(clone.Path(), "new.md"))) require.NoFileExists(t, filepath.Join(clone.Path(), "old.md")) return } head, headErr := clone.HeadRevision(t.Context()) require.NoError(t, headErr) require.Equal(t, oldHead, head) require.Equal(t, "old memory\n", readFileString(t, filepath.Join(clone.Path(), "MEMORY.md"))) require.Equal(t, "old only\n", readFileString(t, filepath.Join(clone.Path(), "old.md"))) require.NoFileExists(t, filepath.Join(clone.Path(), "new.md")) // The same restart may then retry the exact cloud head. It can expose // only the complete new tree after the transaction and sidecar finish. refreshed, _, err := openNativeMemoryJournal(t.Context(), "agt_native", workDir, access) require.NoError(t, err) require.Equal(t, "new memory\n", readFileString(t, filepath.Join(refreshed.Path(), "MEMORY.md"))) require.Equal(t, "new only\n", readFileString(t, filepath.Join(refreshed.Path(), "new.md"))) require.NoFileExists(t, filepath.Join(refreshed.Path(), "old.md")) }) } } func TestRefreshMemoryAccessDoesNotLetSlowMutationStarveHealthyGroup(t *testing.T) { t.Setenv(profile.ConfigDirEnvKey, t.TempDir()) slowRelease := make(chan struct{}) slowStarted := make(chan struct{}, 1) slowRemote := &fakeNativeMemoryRemote{ archiveFilesByVersion: map[string]map[string]string{"ctv_slow": {"MEMORY.md": "slow\n"}}, archiveStarted: slowStarted, releaseArchive: slowRelease, } fastRemote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{"ctv_fast": {"MEMORY.md": "fast\n"}}} provider := &fakeMemoryAccessBatchProvider{batches: []map[string]MemoryAccessLookup{ {"agt_slow_mutation": {State: MemoryAccessBound, Access: MemoryAccess{ MemoryID: "mem_slow", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_slow", ContentVersionID: "ctv_slow", ContentGeneration: 1, NativeRemote: slowRemote, }}}, {"agt_fast_mutation": {State: MemoryAccessBound, Access: MemoryAccess{ MemoryID: "mem_fast", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_fast", ContentVersionID: "ctv_fast", ContentGeneration: 1, NativeRemote: fastRemote, }}}, }} d := newDispatcherWithMemoryProvider(t, provider) done := make(chan struct{}) go func() { d.RefreshMemoryAccess(t.Context(), []string{"agt_slow_mutation", "agt_fast_mutation"}) close(done) }() select { case <-slowStarted: case <-time.After(time.Second): t.Fatal("slow archive mutation did not start") } fastMemory := filepath.Join(sessionWorkDir("agt_fast_mutation"), "memory") require.Eventually(t, func() bool { content, err := os.ReadFile(filepath.Join(fastMemory, "MEMORY.md")) return err == nil && string(content) == "fast\n" }, time.Second, 10*time.Millisecond, "healthy group materialization must finish while another archive read is blocked") close(slowRelease) select { case <-done: case <-time.After(3 * time.Second): t.Fatal("batch mutation refresh did not finish after slow group recovered") } require.Equal(t, "slow\n", readFileString(t, filepath.Join(sessionWorkDir("agt_slow_mutation"), "memory", "MEMORY.md"))) } func TestAttachMemoryWatcherRecoversPendingResolvedRoot(t *testing.T) { t.Parallel() tmp := t.TempDir() workDir := filepath.Join(tmp, "agent") require.NoError(t, os.MkdirAll(workDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "MEMORY.md"), []byte("# Workspace fallback\n"), 0o644)) agentID := "agt_pending" access := MemoryAccess{ MemoryID: "mem_pending", OriginURL: "file://" + filepath.Join(tmp, "missing-origin"), Branch: "main", } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID} d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) require.Nil(t, sess.watcher) require.Nil(t, sess.memoryClone) memoryDir := filepath.Join(workDir, "memory") require.FileExists(t, filepath.Join(memoryDir, "MEMORY.md")) require.FileExists(t, filepath.Join(memoryDir, pendingLocalMemoryMarker)) require.Equal(t, "# Workspace fallback\n", readFileString(t, filepath.Join(workDir, "MEMORY.md"))) require.NoError(t, os.WriteFile(filepath.Join(memoryDir, "MEMORY.md"), []byte("# Memory\n\nRemember pending fact.\n"), 0o644)) require.NoError(t, os.MkdirAll(filepath.Join(memoryDir, "knowledge", "concepts"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(memoryDir, "knowledge", "concepts", "pending.md"), []byte("# Pending fact\n"), 0o644)) access.OriginURL = makeBareOriginRepoForMemory(t, filepath.Join(tmp, "origin")) provider.set(agentID, access) d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.watcher) require.NotNil(t, sess.memoryClone) require.Equal(t, memoryDir, sess.memoryClone.Path()) require.NoFileExists(t, filepath.Join(memoryDir, pendingLocalMemoryMarker)) require.Equal(t, "# Origin memory\n\nInitial content.\n", readFileString(t, filepath.Join(memoryDir, "MEMORY.md"))) require.Equal(t, "# Pending fact\n", readFileString(t, filepath.Join(memoryDir, "knowledge", "concepts", "pending.md"))) require.Equal(t, "# Workspace fallback\n", readFileString(t, filepath.Join(workDir, "MEMORY.md")), "pending recovery must not import or remove the workspace fallback when legacy import is disabled") conflicts, err := filepath.Glob(filepath.Join(memoryDir, ".conflicts", "MEMORY.md.conflict-*")) require.NoError(t, err) require.Len(t, conflicts, 1) require.Equal(t, "# Memory\n\nRemember pending fact.\n", readFileString(t, conflicts[0])) } func TestEnsurePendingLocalMemoryRootClaimsOneBindingConcurrently(t *testing.T) { for i := 0; i < 20; i++ { workDir := filepath.Join(t.TempDir(), "agent") accesses := []MemoryAccess{ {MemoryID: "mem_alpha", Branch: "main"}, {MemoryID: "mem_beta", Branch: "agents/beta"}, } errs := make([]error, len(accesses)) start := make(chan struct{}) var wg sync.WaitGroup for idx, access := range accesses { wg.Add(1) go func() { defer wg.Done() <-start errs[idx] = ensurePendingLocalMemoryRoot(workDir, "agt_shared", access) }() } close(start) wg.Wait() winner := -1 for idx, err := range errs { if err == nil { require.Equal(t, -1, winner, "only one binding may claim the pending root") winner = idx } else { require.Contains(t, err.Error(), "another binding") } } require.NotEqual(t, -1, winner) memoryDir := filepath.Join(workDir, "memory") matching, err := matchingPendingLocalMemoryRoot(memoryDir, "agt_shared", accesses[winner]) require.NoError(t, err) require.True(t, matching) require.FileExists(t, filepath.Join(memoryDir, "MEMORY.md")) } } func TestPrepareExistingLocalMemoryRejectsCorruptPendingMarker(t *testing.T) { workDir := t.TempDir() memoryDir := filepath.Join(workDir, "memory") require.NoError(t, os.MkdirAll(memoryDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(memoryDir, pendingLocalMemoryMarker), []byte("not json\n"), 0o600)) require.NoError(t, os.WriteFile(filepath.Join(memoryDir, "MEMORY.md"), []byte("# Local memory\n"), 0o644)) prepared, err := prepareExistingLocalMemoryDirForImport(context.Background(), workDir, "agt_alpha", MemoryAccess{ MemoryID: "mem_alpha", Branch: "main", }) require.Error(t, err) require.Nil(t, prepared) require.Contains(t, err.Error(), "decode state") require.FileExists(t, filepath.Join(memoryDir, "MEMORY.md")) } func TestAttachMemoryWatcherIsolatesPendingRootFromOldBinding(t *testing.T) { for _, importExisting := range []bool{false, true} { t.Run(fmt.Sprintf("import_existing_%t", importExisting), func(t *testing.T) { tmp := t.TempDir() workDir := filepath.Join(tmp, "agent") agentID := "agt_rebound" require.NoError(t, ensurePendingLocalMemoryRoot(workDir, agentID, MemoryAccess{ MemoryID: "mem_a", Branch: "main", })) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "private-a.md"), []byte("A only\n"), 0o600)) provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_b", OriginURL: makeBareOriginRepoForMemory(t, filepath.Join(tmp, "origin-b")), Branch: "main", ImportExistingLocalContent: importExisting, }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.memoryClone) require.NoFileExists(t, filepath.Join(workDir, "memory", "private-a.md")) backups, err := filepath.Glob(filepath.Join(workDir, ".memory-pending-backups", "memory.pending-*")) require.NoError(t, err) require.Len(t, backups, 1) require.Equal(t, "A only\n", readFileString(t, filepath.Join(backups[0], "private-a.md"))) require.FileExists(t, filepath.Join(backups[0], pendingLocalMemoryMarker)) }) } } func TestAttachMemoryWatcherPublishesNewPendingRootAfterRebindCloneFailure(t *testing.T) { for _, importExisting := range []bool{false, true} { t.Run(fmt.Sprintf("import_existing_%t", importExisting), func(t *testing.T) { tmp := t.TempDir() workDir := filepath.Join(tmp, "agent") agentID := "agt_rebound_offline" require.NoError(t, ensurePendingLocalMemoryRoot(workDir, agentID, MemoryAccess{ MemoryID: "mem_a", Branch: "main", })) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "private-a.md"), []byte("A only\n"), 0o600)) accessB := MemoryAccess{ MemoryID: "mem_b", OriginURL: "file://" + filepath.Join(tmp, "missing-origin-b"), Branch: "main", ImportExistingLocalContent: importExisting, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, accessB) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) require.Nil(t, sess.memoryClone) memoryDir := filepath.Join(workDir, "memory") require.NoFileExists(t, filepath.Join(memoryDir, "private-a.md")) require.FileExists(t, filepath.Join(memoryDir, "MEMORY.md")) matching, err := matchingPendingLocalMemoryRoot(memoryDir, agentID, accessB) require.NoError(t, err) require.True(t, matching) backups, err := filepath.Glob(filepath.Join(workDir, ".memory-pending-backups", "memory.pending-*")) require.NoError(t, err) require.Len(t, backups, 1) require.Equal(t, "A only\n", readFileString(t, filepath.Join(backups[0], "private-a.md"))) }) } } func TestAttachMemoryWatcherRetriesImportAfterTwoCloneFailures(t *testing.T) { tmp := t.TempDir() workDir := filepath.Join(tmp, "agent") agentID := "agt_clone_retry" require.NoError(t, os.MkdirAll(filepath.Join(workDir, "memory"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "local.md"), []byte("survives retries\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "MEMORY.md"), []byte("# Local memory\n"), 0o644)) access := MemoryAccess{ MemoryID: "mem_clone_retry", OriginURL: "file://" + filepath.Join(tmp, "missing-origin"), Branch: "main", ImportExistingLocalContent: true, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) for attempt := 1; attempt <= 2; attempt++ { sess := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) require.Nil(t, sess.memoryClone) require.Equal(t, "survives retries\n", readFileString(t, filepath.Join(workDir, "memory", "local.md"))) state, err := readExistingLocalMemoryImportState(filepath.Join(workDir, "memory")) require.NoError(t, err) require.NotNil(t, state) require.True(t, existingLocalMemoryImportStateMatches(*state, agentID, access)) } access.OriginURL = makeBareOriginRepoForMemory(t, filepath.Join(tmp, "origin")) provider.set(agentID, access) sess := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.memoryClone) require.Equal(t, "survives retries\n", readFileString(t, filepath.Join(workDir, "memory", "local.md"))) require.NoFileExists(t, filepath.Join(workDir, "memory", existingLocalMemoryImportMarker)) require.NoFileExists(t, filepath.Join(workDir, "memory", pendingLocalMemoryMarker)) require.NoDirExists(t, filepath.Join(workDir, ".memory-imports")) } func TestCollectExistingLocalMemoryKeepsOrdinaryPlaceholderContent(t *testing.T) { workDir := t.TempDir() stagedPath := filepath.Join(workDir, "staged") require.NoError(t, os.MkdirAll(stagedPath, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(stagedPath, "MEMORY.md"), []byte(pendingLocalMemoryIndex), 0o644)) sources, _, _, err := collectExistingLocalMemoryImportSources(workDir, &preparedExistingLocalMemoryDir{ stagedPath: stagedPath, pending: false, }) require.NoError(t, err) require.Len(t, sources, 1) require.Equal(t, "MEMORY.md", sources[0].rel) sources, _, _, err = collectExistingLocalMemoryImportSources(workDir, &preparedExistingLocalMemoryDir{ stagedPath: stagedPath, pending: true, }) require.NoError(t, err) require.Empty(t, sources) } func TestAttachMemoryWatcherKeepsCachedCloneWritableOffline(t *testing.T) { t.Parallel() tmp := t.TempDir() originPath := filepath.Join(tmp, "origin") origin := makeBareOriginRepoForMemory(t, originPath) workDir := filepath.Join(tmp, "agent") agentID := "agt_cached_offline" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{MemoryID: "mem_cached", OriginURL: origin, Branch: "main"}) d := newDispatcherWithMemoryProvider(t, provider) online := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(context.Background(), online, agentID, workDir) require.NotNil(t, online.watcher) stopMemoryWatcher(online) require.NoError(t, os.Rename(originPath, originPath+"-offline")) offline := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(context.Background(), offline, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(offline) }) require.NotNil(t, offline.watcher) require.NotNil(t, offline.memoryClone) require.Equal(t, filepath.Join(workDir, "memory"), offline.memoryClone.Path()) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "offline.md"), []byte("# Offline write\n"), 0o644)) require.Eventually(t, func() bool { return strings.Contains(headMessageAt(t, filepath.Join(workDir, "memory")), "Update memory") }, 3*time.Second, 15*time.Millisecond) require.FileExists(t, filepath.Join(workDir, "memory", "offline.md")) } // TestAttachMemoryWatcherSkipsWhenNoBinding verifies that a missing binding // leaves the watcher nil and the session usable. This is the "degrade" path: // unbound agents without memory repos must continue to work. func TestAttachMemoryWatcherSkipsWhenNoBinding(t *testing.T) { t.Parallel() provider := &fakeMemoryAccessProvider{} // no bindings set d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: "agt_unprovisioned"} d.attachMemoryWatcher(context.Background(), sess, "agt_unprovisioned", t.TempDir()) require.Nil(t, sess.watcher) require.Nil(t, sess.memoryClone) _, ok := d.MemorySyncStatus("agt_unprovisioned") require.False(t, ok, "authoritatively unbound agents must not expose a sync failure") } // TestAttachMemoryWatcherSkipsWhenProviderNil confirms that without a provider // wired (the default Dispatcher state), nothing happens. func TestAttachMemoryWatcherSkipsWhenProviderNil(t *testing.T) { t.Parallel() d := mustNewDispatcher("") // no SetMemoryAccessProvider call sess := &interactiveSession{agentID: "agt_alice"} d.attachMemoryWatcher(context.Background(), sess, "agt_alice", t.TempDir()) require.Nil(t, sess.watcher) } // TestAttachMemoryWatcherDegradesOnLookupError verifies that a provider error // does not propagate — the session simply runs without a watcher. func TestAttachMemoryWatcherDegradesOnLookupError(t *testing.T) { t.Parallel() provider := &fakeMemoryAccessProvider{} provider.setErr(errFakeProviderFailure) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: "agt_alice"} d.attachMemoryWatcher(context.Background(), sess, "agt_alice", t.TempDir()) require.Nil(t, sess.watcher, "lookup error must not produce a watcher") status, ok := d.MemorySyncStatus("agt_alice") require.True(t, ok) require.Equal(t, memorywatcher.StatusDegraded, status.State) require.Equal(t, memorywatcher.StatusReasonLookupFailed, status.Reason) provider.setErr(nil) d.RetryMemorySync(context.Background(), "agt_alice") require.Eventually(t, func() bool { _, ok = d.MemorySyncStatus("agt_alice") return !ok }, time.Second, 10*time.Millisecond, "explicit retry must reuse RefreshMemoryAccess and clear an authoritative unbound result") } func TestRetryMemorySyncRecoversIdleCloneWithoutActiveSession(t *testing.T) { configDir := t.TempDir() t.Setenv(profile.ConfigDirEnvKey, configDir) originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) agentID := "agt_pending_no_session" access := MemoryAccess{MemoryID: "mem_pending_no_session", Protocol: MemorySyncProtocolLegacyGitV1, OriginURL: origin, Branch: "main"} provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) d.RefreshMemoryAccess(t.Context(), []string{agentID}) clone, _, _, ok := d.ensureMemoryCloneForAccess(t.Context(), agentID, sessionWorkDir(agentID), access) require.True(t, ok) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "pending.md"), []byte("pending\n"), 0o644)) require.Equal(t, "initial commit", strings.TrimSpace(refMessageOnOrigin(t, originDir, "main"))) d.setMemorySyncStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonCommitFailed)) offlineOrigin := originDir + "-offline" require.NoError(t, os.Rename(originDir, offlineOrigin)) d.RetryMemorySync(t.Context(), agentID) require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) return ok && status.State == memorywatcher.StatusPendingSync && status.Reason == memorywatcher.StatusReasonPushFailed }, time.Second, 10*time.Millisecond) require.NoError(t, os.Rename(offlineOrigin, originDir)) d.RetryMemorySync(t.Context(), agentID) require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) return ok && status.State == memorywatcher.StatusReady && strings.Contains(refMessageOnOrigin(t, originDir, "main"), "Update memory from local edits") }, time.Second, 10*time.Millisecond) } func TestNativeMemoryIdleSyncPreservesOfflineJournalAndNeverUsesGitRemote(t *testing.T) { t.Setenv(profile.ConfigDirEnvKey, t.TempDir()) oldInterval := nativeMemoryCheckpointMinInterval nativeMemoryCheckpointMinInterval = time.Nanosecond t.Cleanup(func() { nativeMemoryCheckpointMinInterval = oldInterval }) agentID := "agt_native_idle" workDir := sessionWorkDir(agentID) memoryDir := filepath.Join(workDir, "memory") require.NoError(t, os.MkdirAll(memoryDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(memoryDir, "MEMORY.md"), []byte("# offline\n"), 0o644)) remote := &fakeNativeMemoryRemote{err: errors.New("quota exceeded")} provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_base", ContentGeneration: 3, NativeRemote: remote, // A stale legacy remote must be ignored rather than selected as fallback. OriginURL: "file:///must-not-be-used", Branch: "legacy-must-not-be-used", }) d := newDispatcherWithMemoryProvider(t, provider) t.Cleanup(d.closeIdleMemoryRetries) d.RefreshMemoryAccess(t.Context(), []string{agentID}) status, ok := d.MemorySyncStatus(agentID) require.True(t, ok) require.Equal(t, memorywatcher.StatusPendingSync, status.State) require.Equal(t, memorywatcher.StatusReasonPushFailed, status.Reason) require.Equal(t, 1, remote.callCount()) require.FileExists(t, filepath.Join(memoryDir, ".git", "HEAD")) local, err := vault.IsLocalJournal(memoryDir) require.NoError(t, err) require.True(t, local) require.FileExists(t, nativeMemoryStatePath(workDir, MemoryAccess{MemoryID: "mem_native", ContentGeneration: 3})) require.Equal(t, "# offline\n", readFileString(t, filepath.Join(memoryDir, "MEMORY.md"))) require.NoError(t, os.WriteFile(filepath.Join(memoryDir, "later.md"), []byte("kept\n"), 0o644)) remote.setError(nil) d.RetryMemorySync(t.Context(), agentID) require.Eventually(t, func() bool { status, exists := d.MemorySyncStatus(agentID) return exists && status.State == memorywatcher.StatusReady && remote.callCount() >= 2 }, 3*time.Second, 20*time.Millisecond) require.Equal(t, "kept\n", readFileString(t, filepath.Join(memoryDir, "later.md"))) callsBeforeRevoke := remote.callCount() provider.remove(agentID) d.RefreshMemoryAccess(t.Context(), []string{agentID}) require.FileExists(t, filepath.Join(memoryDir, ".git", "HEAD"), "native revoke preserves the local version journal") require.FileExists(t, nativeMemoryStatePath(workDir, MemoryAccess{MemoryID: "mem_native", ContentGeneration: 3}), "native revoke preserves the pending outbox sidecar") require.Equal(t, callsBeforeRevoke, remote.callCount(), "revoke must not publish or fall back to a Git remote") } func TestRefreshMemoryAccessRebindsActiveNativeSessionOnExactHeadChange(t *testing.T) { workDir := t.TempDir() agentID := "agt_native_active_restore" remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_a": {"MEMORY.md": "same tree\n"}, "ctv_b": {"MEMORY.md": "same tree\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_a", ContentGeneration: 3, NativeRemote: remote, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, workDir: workDir, agent: agent.Config{Name: "Native Restore"}} d.sessionManager.sessions[agentID] = sess t.Cleanup(func() { stopMemoryWatcher(sess) d.closeIdleMemoryRetries() }) d.attachMemoryWatcher(t.Context(), sess, agentID, workDir) require.NotNil(t, sess.watcher) oldWatcher := sess.watcher oldHead, err := sess.memoryClone.HeadRevision(t.Context()) require.NoError(t, err) access.ContentVersionID = "ctv_b" provider.set(agentID, access) d.RefreshMemoryAccess(t.Context(), []string{agentID}) sess.mu.Lock() watcher := sess.watcher clone := sess.memoryClone admittedVersion := sess.memoryContentVersionID sess.mu.Unlock() require.NotNil(t, watcher, "clean same-generation head change must replace and reattach the watcher") require.NotSame(t, oldWatcher, watcher) require.Equal(t, "ctv_b", admittedVersion) require.Equal(t, 1, remote.archiveCallCount("ctv_b")) newHead, err := clone.HeadRevision(t.Context()) require.NoError(t, err) require.NotEqual(t, oldHead, newHead) commit, err := clone.Repository().CommitObject(plumbing.NewHash(newHead)) require.NoError(t, err) require.Equal(t, []plumbing.Hash{plumbing.NewHash(oldHead)}, commit.ParentHashes) } func TestRefreshMemoryAccessDoesNotRebindAfterOwnDurableNativePublish(t *testing.T) { workDir := t.TempDir() agentID := "agt_native_own_publish" remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{"ctv_a": {"MEMORY.md": "cloud\n"}}} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_a", ContentGeneration: 3, NativeRemote: remote, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, workDir: workDir, agent: agent.Config{Name: "Native Publish"}} d.sessionManager.sessions[agentID] = sess t.Cleanup(func() { stopMemoryWatcher(sess) }) d.attachMemoryWatcher(t.Context(), sess, agentID, workDir) require.NotNil(t, sess.watcher) originalWatcher := sess.watcher require.NoError(t, os.WriteFile(filepath.Join(sess.memoryClone.Path(), "local.md"), []byte("publish\n"), 0o644)) revision, err := sess.memoryClone.CommitPending(t.Context(), "local publish", vault.CommitPendingOpts{}) require.NoError(t, err) require.NoError(t, sess.memoryNativeSync.RecordLocalRevision(revision)) result, err := sess.memoryNativeSync.Publish(t.Context()) require.NoError(t, err) require.Equal(t, "ctv_next", result.VersionID) access.ContentVersionID = "ctv_next" access.AcknowledgedSequence = 1 provider.set(agentID, access) archiveCalls := remote.archiveCallCount("ctv_next") d.RefreshMemoryAccess(t.Context(), []string{agentID}) sess.mu.Lock() currentWatcher := sess.watcher currentVersion := sess.memoryNativeSync.RemoteVersionID() sess.mu.Unlock() require.Same(t, originalWatcher, currentWatcher) require.Equal(t, "ctv_next", currentVersion) require.Equal(t, archiveCalls, remote.archiveCallCount("ctv_next")) } func TestRefreshMemoryAccessKeepsActiveNativePendingOnExactHeadChange(t *testing.T) { workDir := t.TempDir() agentID := "agt_native_active_conflict" remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_a": {"MEMORY.md": "old cloud\n"}, "ctv_b": {"MEMORY.md": "must not be read\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_a", ContentGeneration: 3, NativeRemote: remote, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, workDir: workDir, agent: agent.Config{Name: "Native Conflict"}} d.sessionManager.sessions[agentID] = sess t.Cleanup(func() { stopMemoryWatcher(sess) d.closeIdleMemoryRetries() }) d.attachMemoryWatcher(t.Context(), sess, agentID, workDir) require.NotNil(t, sess.watcher) memoryDir := sess.memoryClone.Path() remote.setError(errors.New("cloud unavailable")) require.NoError(t, os.WriteFile(filepath.Join(memoryDir, "MEMORY.md"), []byte("local pending\n"), 0o644)) access.ContentVersionID = "ctv_b" provider.set(agentID, access) d.RefreshMemoryAccess(t.Context(), []string{agentID}) sess.mu.Lock() watcher := sess.watcher clone := sess.memoryClone sess.mu.Unlock() require.Nil(t, watcher, "a conflicting exact head must leave the watcher fenced") require.Nil(t, clone) require.Zero(t, remote.archiveCallCount("ctv_b"), "pending admission must fail before reading B") require.Equal(t, "local pending\n", readFileString(t, filepath.Join(memoryDir, "MEMORY.md"))) status, ok := d.MemorySyncStatus(agentID) require.True(t, ok) require.Equal(t, memorywatcher.StatusPendingSync, status.State) require.Equal(t, memorywatcher.StatusReasonPushFailed, status.Reason) } func TestRefreshMemoryAccessFencesActiveRuntimeBeforeExternalNativeHeadMaterialization(t *testing.T) { workDir := t.TempDir() agentID := "agt_native_live_fence" remote := &fakeNativeMemoryRemote{archiveFilesByVersion: map[string]map[string]string{ "ctv_a": {"MEMORY.md": "old\n"}, "ctv_b": {"MEMORY.md": "new\n"}, }} access := MemoryAccess{ MemoryID: "mem_native", Protocol: MemorySyncProtocolNativeContentV1, ContentID: "ctn_native", ContentVersionID: "ctv_a", ContentGeneration: 3, NativeRemote: remote, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, workDir: workDir, process: &agent.Process{}, agent: agent.Config{Name: "Live Fence"}} d.sessionManager.sessions[agentID] = sess d.attachMemoryWatcher(t.Context(), sess, agentID, workDir) require.NotNil(t, sess.watcher) access.ContentVersionID = "ctv_b" provider.set(agentID, access) d.RefreshMemoryAccess(t.Context(), []string{agentID}) _, live := d.sessionManager.FindByLookupID(agentID) require.False(t, live, "the exact process generation must retire before materialization") require.Zero(t, remote.archiveCallCount("ctv_b")) require.Equal(t, "old\n", readFileString(t, filepath.Join(workDir, "memory", "MEMORY.md"))) replacement := &interactiveSession{agentID: agentID, workDir: workDir, agent: agent.Config{Name: "Live Fence"}} d.sessionManager.sessions[agentID] = replacement d.attachMemoryWatcher(t.Context(), replacement, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(replacement) }) require.NotNil(t, replacement.watcher) require.Equal(t, 1, remote.archiveCallCount("ctv_b")) require.Equal(t, "new\n", readFileString(t, filepath.Join(workDir, "memory", "MEMORY.md"))) } func TestIdleMemoryCloneRetriesAutomaticallyWithoutActiveSession(t *testing.T) { configDir := t.TempDir() t.Setenv(profile.ConfigDirEnvKey, configDir) originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) agentID := "agt_automatic_idle_retry" access := MemoryAccess{MemoryID: "mem_automatic_idle_retry", Protocol: MemorySyncProtocolLegacyGitV1, OriginURL: origin, Branch: "main"} provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) t.Cleanup(d.closeIdleMemoryRetries) d.RefreshMemoryAccess(t.Context(), []string{agentID}) clone, _, _, ok := d.ensureMemoryCloneForAccess(t.Context(), agentID, sessionWorkDir(agentID), access) require.True(t, ok) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "automatic.md"), []byte("automatic retry\n"), 0o644)) offlineOrigin := originDir + "-offline" require.NoError(t, os.Rename(originDir, offlineOrigin)) lookupsBeforeFailure := provider.callCount() d.RefreshMemoryAccess(t.Context(), []string{agentID}) status, ok := d.MemorySyncStatus(agentID) require.True(t, ok) require.Equal(t, memorywatcher.StatusPendingSync, status.State) require.Equal(t, memorywatcher.StatusReasonPushFailed, status.Reason) // The first retry uses a one-second floor, so a failure cannot create a hot // provider/Git agentteam while the remote remains unavailable. time.Sleep(100 * time.Millisecond) require.Equal(t, lookupsBeforeFailure+1, provider.callCount()) require.NoError(t, os.Rename(offlineOrigin, originDir)) require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) return ok && status.State == memorywatcher.StatusReady && strings.Contains(refMessageOnOrigin(t, originDir, "main"), "Update memory from local edits") }, 4*time.Second, 25*time.Millisecond) status, ok = d.MemorySyncStatus(agentID) require.True(t, ok) require.Equal(t, memorywatcher.StatusReady, status.State) } func TestIdleMemoryRecoveryRetriesColdLookupFailure(t *testing.T) { t.Setenv(profile.ConfigDirEnvKey, t.TempDir()) origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) agentID := "agt_cold_lookup_retry" provider := &fakeMemoryAccessProvider{} provider.setErr(errFakeProviderFailure) d := newDispatcherWithMemoryProvider(t, provider) t.Cleanup(d.closeIdleMemoryRetries) d.RefreshMemoryAccess(t.Context(), []string{agentID}) status, ok := d.MemorySyncStatus(agentID) require.True(t, ok) require.Equal(t, memorywatcher.StatusReasonLookupFailed, status.Reason) provider.set(agentID, MemoryAccess{MemoryID: "mem_cold_lookup_retry", OriginURL: origin, Branch: "main"}) provider.setErr(nil) require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) _, statErr := os.Stat(filepath.Join(sessionWorkDir(agentID), "memory", ".git")) return ok && status.State == memorywatcher.StatusReady && statErr == nil }, 4*time.Second, 25*time.Millisecond) } func TestIdleMemoryRecoveryRetriesColdCloneFailure(t *testing.T) { t.Setenv(profile.ConfigDirEnvKey, t.TempDir()) originPath := filepath.Join(t.TempDir(), "origin") agentID := "agt_cold_clone_retry" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_cold_clone_retry", OriginURL: "file://" + originPath, Branch: "main", }) d := newDispatcherWithMemoryProvider(t, provider) t.Cleanup(d.closeIdleMemoryRetries) d.RefreshMemoryAccess(t.Context(), []string{agentID}) status, ok := d.MemorySyncStatus(agentID) require.True(t, ok) require.Equal(t, memorywatcher.StatusReasonCloneFailed, status.Reason) makeBareOriginRepoForMemory(t, originPath) require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) _, statErr := os.Stat(filepath.Join(sessionWorkDir(agentID), "memory", ".git")) return ok && status.State == memorywatcher.StatusReady && statErr == nil }, 4*time.Second, 25*time.Millisecond) } func TestActiveSessionRecoversInitialLookupFailureAndRestarts(t *testing.T) { origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) agentID := "agt_active_lookup_recovery" provider := &fakeMemoryAccessProvider{} provider.setErr(errFakeProviderFailure) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, agent: agent.Config{Name: "Active Recovery"}, workDir: t.TempDir(), process: &agent.Process{}, processUsesWorkspaceRootMemory: true, } d.sessionManager.sessions[agentID] = sess d.attachMemoryWatcher(t.Context(), sess, agentID, sess.workDir) require.Nil(t, sess.watcher) provider.set(agentID, MemoryAccess{MemoryID: "mem_active_lookup_recovery", OriginURL: origin, Branch: "main"}) provider.setErr(nil) require.Eventually(t, func() bool { _, live := d.sessionManager.FindByLookupID(agentID) _, err := os.Stat(filepath.Join(sess.workDir, "memory", ".git")) return !live && err == nil }, 4*time.Second, 25*time.Millisecond) } func TestActiveSessionLookupFailureAutomaticallyAppliesUnbind(t *testing.T) { origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) agentID := "agt_active_unbind_recovery" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{MemoryID: "mem_active_unbind_recovery", OriginURL: origin, Branch: "main"}) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, agent: agent.Config{Name: "Active Unbind Recovery"}, workDir: t.TempDir(), } d.sessionManager.sessions[agentID] = sess t.Cleanup(func() { stopMemoryWatcher(sess) }) d.attachMemoryWatcher(t.Context(), sess, agentID, sess.workDir) require.NotNil(t, sess.watcher) provider.setErr(errFakeProviderFailure) d.RefreshMemoryAccess(t.Context(), []string{agentID}) provider.remove(agentID) provider.setErr(nil) require.Eventually(t, func() bool { sess.mu.Lock() watcher := sess.watcher sess.mu.Unlock() _, ok := d.MemorySyncStatus(agentID) return watcher == nil && !ok }, 4*time.Second, 25*time.Millisecond) } func TestFailedTeardownPushRecoversWithoutAnotherSession(t *testing.T) { configDir := t.TempDir() t.Setenv(profile.ConfigDirEnvKey, configDir) originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) agentID := "agt_teardown_retry" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{MemoryID: "mem_teardown_retry", OriginURL: origin, Branch: "main"}) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, agent: agent.Config{Name: "Teardown Retry"}, workDir: sessionWorkDir(agentID), } d.sessionManager.sessions[agentID] = sess d.attachMemoryWatcher(t.Context(), sess, agentID, sess.workDir) require.NotNil(t, sess.watcher) offlineOrigin := originDir + "-offline" require.NoError(t, os.Rename(originDir, offlineOrigin)) require.NoError(t, os.WriteFile(filepath.Join(sess.workDir, "memory", "teardown.md"), []byte("recover after teardown\n"), 0o644)) require.Eventually(t, func() bool { status, ok := d.MemorySyncStatus(agentID) return ok && status.State == memorywatcher.StatusPendingSync }, 3*time.Second, 25*time.Millisecond) d.stopMemoryWatcherAndRetry(sess) d.sessionManager.RemoveIfCurrent(agentID, sess) require.NoError(t, os.Rename(offlineOrigin, originDir)) require.Eventually(t, func() bool { return strings.Contains(refMessageOnOrigin(t, originDir, "main"), "Update memory") }, 4*time.Second, 25*time.Millisecond) owners := d.closeMemoryWatcherMutationAdmission() require.NoError(t, waitForMemoryWatcherMutationOwners(t.Context(), owners)) } func TestIdleMemoryRetryCoalescesWhileRecoveryIsRunning(t *testing.T) { d := mustNewDispatcher("") agentID := "agt_retry_coalesce" d.scheduleIdleMemoryRetry(agentID) d.memoryIdleRetryMu.Lock() retry := d.memoryIdleRetries[agentID] require.NotNil(t, retry) require.True(t, retry.timer.Stop()) retry.timer = nil retry.running = true attempt := retry.attempt d.memoryIdleRetryMu.Unlock() d.scheduleIdleMemoryRetry(agentID) d.memoryIdleRetryMu.Lock() require.Nil(t, retry.timer, "an in-flight recovery must own the only retry slot for its agent") require.Equal(t, attempt, retry.attempt, "coalesced requests must not advance backoff") d.memoryIdleRetryMu.Unlock() d.closeIdleMemoryRetries() } func TestRetryMemorySyncQueuesRecoveryWithoutBlockingCaller(t *testing.T) { provider := &blockingMemoryAccessProvider{ entered: make(chan struct{}), exited: make(chan struct{}), } d := newDispatcherWithMemoryProvider(t, provider) agentID := "agt_retry_nonblocking" d.setMemoryLookupStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonLookupFailed)) started := time.Now() d.RetryMemorySync(t.Context(), agentID) require.Less(t, time.Since(started), 100*time.Millisecond, "explicit retry must enqueue recovery instead of holding the API request") select { case <-provider.entered: case <-time.After(time.Second): t.Fatal("queued recovery did not enter access lookup") } d.cancelIdleMemoryRetry(agentID) select { case <-provider.exited: case <-time.After(time.Second): t.Fatal("canceling queued recovery did not cancel the in-flight lookup") } } func TestRetryMemorySyncDoesNotWaitForMemoryLookupLock(t *testing.T) { d := mustNewDispatcher("") agentID := "agt_retry_lookup_busy" d.setMemoryLookupStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonLookupFailed)) lookupMu := d.agentMemoryLookupMutex(agentID) lookupMu.Lock() defer lookupMu.Unlock() started := time.Now() d.RetryMemorySync(t.Context(), agentID) require.Less(t, time.Since(started), 100*time.Millisecond, "explicit retry must not wait for an in-flight lookup, clone, or sync") d.memoryIdleRetryMu.Lock() _, queued := d.memoryIdleRetries[agentID] d.memoryIdleRetryMu.Unlock() require.True(t, queued, "explicit retry must enqueue recovery while lookup is busy") d.cancelIdleMemoryRetry(agentID) } func TestClosingMemoryRetryAdmissionCancelsInFlightLookup(t *testing.T) { provider := &blockingMemoryAccessProvider{ entered: make(chan struct{}), exited: make(chan struct{}), } d := newDispatcherWithMemoryProvider(t, provider) agentID := "agt_retry_shutdown" d.scheduleIdleMemoryRetry(agentID) d.memoryIdleRetryMu.Lock() retry := d.memoryIdleRetries[agentID] require.NotNil(t, retry) require.True(t, retry.timer.Stop()) retry.timer = nil d.memoryIdleRetryMu.Unlock() go d.runIdleMemoryRetry(agentID, retry) select { case <-provider.entered: case <-time.After(time.Second): t.Fatal("idle retry did not enter access lookup") } owners := d.closeMemoryWatcherMutationAdmission() ctx, cancel := context.WithTimeout(t.Context(), time.Second) defer cancel() require.NoError(t, waitForMemoryWatcherMutationOwners(ctx, owners)) select { case <-provider.exited: case <-ctx.Done(): t.Fatal("dispatcher shutdown did not cancel in-flight idle lookup") } } func TestIdleMemoryRetryDelayIsBoundedAndStaggered(t *testing.T) { first := idleMemoryRetryDelay("agt_first", 0) second := idleMemoryRetryDelay("agt_second", 0) require.GreaterOrEqual(t, first, 3*idleMemoryRetryMin/4) require.LessOrEqual(t, first, idleMemoryRetryMin) require.NotEqual(t, first, second, "different agents should not retry in lockstep") require.LessOrEqual(t, idleMemoryRetryDelay("agt_first", 100), idleMemoryRetryMax) } // TestStopMemoryWatcherIsIdempotentAndClearsFields exercises the helper // directly: calling it twice must not panic, and after the first call the // watcher/memoryClone fields are nil. func TestStopMemoryWatcherIsIdempotentAndClearsFields(t *testing.T) { t.Parallel() origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() provider := &fakeMemoryAccessProvider{} provider.set("agt_alice", MemoryAccess{MemoryID: "mem_alice", OriginURL: origin, Branch: "alice-memory"}) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: "agt_alice"} d.attachMemoryWatcher(context.Background(), sess, "agt_alice", workDir) require.NotNil(t, sess.watcher) stopMemoryWatcher(sess) require.Nil(t, sess.watcher, "stopMemoryWatcher must clear the watcher field") require.Nil(t, sess.memoryClone, "stopMemoryWatcher must clear the clone field") require.NotPanics(t, func() { stopMemoryWatcher(sess) }, "second stop must be a no-op") } // TestBuildMemoryContextResolverDerivesIdentityFromAgent checks the resolver // uses the agent name (not human actor) per the agreed contract, and threads // channel/thread/task through from binding + activeTaskTarget. func TestBuildMemoryContextResolverDerivesIdentityFromAgent(t *testing.T) { t.Parallel() d := mustNewDispatcher("") sess := &interactiveSession{ agentID: "agt_alice", agent: agent.Config{Name: "Alice Agent"}, binding: conversationBinding{ channelID: "#engineering", targetScopeType: "thread", conversationKey: "#engineering:abc123", meta: &types.AgentTeamMeta{Metadata: map[string]any{ "replyThreadRoot": map[string]any{ "content": "Memory integration plan\nwith extra whitespace", }, }}, }, activeTaskTarget: types.ActiveTaskTarget{ Channel: "#engineering", TaskNumber: 42, Source: "task_tool_call", TaskScope: "#tsk_xyz", TaskTitle: "Fix memory commit messages", ThreadTarget: "#engineering:abc123", }, } resolver := d.buildMemoryContextResolver(sess) ctx := resolver(context.Background()) require.Equal(t, "agt_alice", ctx.AgentID) require.Equal(t, "Alice Agent", ctx.ActorName, "author must be agent identity, not human actor") require.Equal(t, "#engineering", ctx.ChannelID) require.Equal(t, "#engineering:abc123", ctx.ThreadTarget) require.Equal(t, "Memory integration plan with extra whitespace", ctx.ThreadTitle) require.Equal(t, "#tsk_xyz", ctx.TaskScope) require.Equal(t, int64(42), ctx.TaskNumber) require.Equal(t, "Fix memory commit messages", ctx.TaskTitle) require.Equal(t, "task_tool_call", ctx.TriggerSource) } func TestBuildMemoryContextResolverDoesNotTreatTaskToolSourceAsTaskScope(t *testing.T) { t.Parallel() d := mustNewDispatcher("") sess := &interactiveSession{ agentID: "agt_alice", binding: conversationBinding{ channelID: "#engineering", conversationKey: "#engineering:abc123", }, activeTaskTarget: types.ActiveTaskTarget{ Channel: "#engineering", TaskNumber: 42, Source: "task_tool_call", ThreadTarget: "#engineering:abc123", }, } ctx := d.buildMemoryContextResolver(sess)(context.Background()) require.Empty(t, ctx.TaskScope) require.Equal(t, int64(42), ctx.TaskNumber) require.Equal(t, "task_tool_call", ctx.TriggerSource) require.NotContains(t, memorywatcher.BuildCommitMessage(ctx), "Update memory from task_tool_call") } // TestBuildMemoryContextResolverHandlesTaskScopeConversation verifies the // disambiguation rule: when the binding's conversation key IS a task scope // (no thread separator), ThreadTarget stays empty so the commit subject falls // back to the task scope. The task scope must come from the explicit // ActiveTaskTarget.TaskScope field, never backfilled from Source. func TestBuildMemoryContextResolverHandlesTaskScopeConversation(t *testing.T) { t.Parallel() d := mustNewDispatcher("") sess := &interactiveSession{ agentID: "agt_alice", binding: conversationBinding{conversationKey: "#tsk_only"}, activeTaskTarget: types.ActiveTaskTarget{ Channel: "#eng", TaskNumber: 7, TaskScope: "#tsk_only", Source: "task_dispatch", }, } resolver := d.buildMemoryContextResolver(sess) ctx := resolver(context.Background()) require.Empty(t, ctx.ThreadTarget, "task scope key must not be treated as a thread") require.Equal(t, "#tsk_only", ctx.TaskScope, "explicit task scope flows to the commit context") require.Equal(t, "task_dispatch", ctx.TriggerSource, "Source is provenance, never backfilled into TaskScope") } // TestThreadTargetFromBinding covers the thread-vs-task-scope-vs-channel // classification used by the resolver. func TestThreadTargetFromBinding(t *testing.T) { t.Parallel() cases := []struct { name string bind conversationBinding want string }{ {"empty", conversationBinding{}, ""}, {"channel-only", conversationBinding{conversationKey: "#eng"}, ""}, {"task-scope", conversationBinding{conversationKey: "#tsk_x"}, ""}, {"thread", conversationBinding{conversationKey: "#eng:AbC123"}, "#eng:abc123"}, {"dm", conversationBinding{conversationKey: "dm:@alice"}, ""}, {"dm-thread", conversationBinding{conversationKey: "dm:@alice:DeF456"}, "dm:@alice:def456"}, {"team", conversationBinding{conversationKey: "#team_abcdefghijklmn"}, ""}, {"team-thread", conversationBinding{conversationKey: "#team_abcdefghijklmn:AbCd1234"}, "#team_abcdefghijklmn:abcd1234"}, {"malformed-task-thread", conversationBinding{conversationKey: "#tsk_x:deadbeef"}, ""}, {"malformed-team", conversationBinding{conversationKey: "#team_short:deadbeef"}, ""}, {"invalid-short-id", conversationBinding{conversationKey: "#eng:abc"}, ""}, } for _, c := range cases { c := c t.Run(c.name, func(t *testing.T) { t.Parallel() require.Equal(t, c.want, threadTargetFromBinding(c.bind)) }) } } // TestAgentAuthorName covers the fallback chain. func TestAgentAuthorName(t *testing.T) { t.Parallel() require.Equal(t, "Alice", agentAuthorName("Alice", "agt_x")) require.Equal(t, "agt_x", agentAuthorName("", "agt_x")) require.Equal(t, "agt_x", agentAuthorName(" ", "agt_x")) } var errFakeProviderFailure = errString("fake provider failure") type errString string func (e errString) Error() string { return string(e) } // TestStopMemoryWatcherFinalPushLandsOnOrigin covers the only dispatcher- // level code path not exercised by the package-scoped e2e tests: the final // push that stopMemoryWatcher issues after the watcher is closed. The whole // pipeline (sync → commit → push) is covered in // memorywatcher/e2e_test.go, but the FINAL push that runs in stopMemoryWatcher // after watcher.FlushAndClose() returns is wired only here. Verifying it at // the dispatcher level guards against regressions in: // // - sess.memoryPushAuth retention across the watcher's lifetime // - clone.Push being called after FlushAndClose drains pending debounce // - the 60s timeout context wrapping the final push // // Without this test, a regression that turned stopMemoryWatcher into a no-op // (e.g. nil memoryPushAuth, missing Push call) would not appear until manual // end-to-end testing. func TestStopMemoryWatcherFinalPushLandsOnOrigin(t *testing.T) { t.Parallel() // Use a bare origin so the agent's push to refs/heads/ is // accepted without "branch is currently checked out" errors. originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) workDir := t.TempDir() agentID := "agt_alice" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_alice", OriginURL: origin, Branch: "alice-memory", }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, binding: conversationBinding{ channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam", }, } d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) require.NotNil(t, sess.watcher) require.NotNil(t, sess.memoryClone) // Sanity: before any agent write, origin has no bound branch ref. require.Empty(t, refMessageOnOrigin(t, originDir, "alice-memory"), "bound branch ref must not exist on origin before any push") require.NoError(t, os.WriteFile( filepath.Join(sess.memoryClone.Path(), "MEMORY.md"), []byte("# alice's memory\n\nfinal-push-on-stop path.\n"), 0o644, )) // Wait for the watcher to commit locally. We don't push mid-session here // (the pipeline does, but we want to verify the FINAL push specifically); // we wait for the commit hash to advance past "initial commit". require.Eventually(t, func() bool { return strings.TrimSpace(headMessageAt(t, sess.memoryClone.Path())) != "initial commit" }, 3*time.Second, 15*time.Millisecond) require.NotEqual(t, "initial commit", strings.TrimSpace(headMessageAt(t, sess.memoryClone.Path())), "watcher must commit before stop") // Sanity: even after commit, origin has no bound branch ref yet. The // pipeline normally pushes after each commit, but a tiny debounce or // ordering could in principle race this assertion. We accept either // state and only require the FINAL assertion to hold. // Then call stopMemoryWatcher — this must flush the pending debounce and // issue a final push. stopMemoryWatcher(sess) // Final push must have landed bound branch on origin with our commit. pushed := refMessageOnOrigin(t, originDir, "alice-memory") require.NotEmpty(t, pushed, "stopMemoryWatcher must push bound branch to origin") require.Contains(t, pushed, "Update memory from channel #agentteam") require.NotContains(t, pushed, "AgentTeam-Agent:") } func TestRefreshMemoryAccessRestartsRuntimeAfterFirstAttach(t *testing.T) { t.Parallel() origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_alice" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_alice", OriginURL: origin, Branch: "alice-memory", ImportExistingLocalContent: true, }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, process: &agent.Process{}, processUsesWorkspaceRootMemory: true, agent: agent.Config{ Name: "Alice", }, binding: conversationBinding{ channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam", }, workDir: workDir, } d.sessionManager.sessions["agt_alice"] = sess require.Nil(t, sess.watcher) // Refresh attaches managed Memory, then retires the stale runtime contract. d.RefreshMemoryAccess(context.Background(), []string{agentID}) _, ok := d.sessionManager.FindByLookupID(agentID) require.False(t, ok, "first managed attach must retire the stale runtime") require.FileExists(t, filepath.Join(workDir, "memory", "MEMORY.md")) markers, err := filepath.Glob(filepath.Join(workDir, legacyUserPreferencesMigrationStateDir, legacyUserPreferencesMigrationMarkerPrefix+"*.done")) require.NoError(t, err) require.Len(t, markers, 1, "the migration window closes only after the fallback runtime is retired") } func TestFallbackRetirementImportsFinalLegacyPreferenceWrite(t *testing.T) { originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) workDir := t.TempDir() agentID := "agt_final_legacy_cutover_write" access := MemoryAccess{ MemoryID: "mem_final_legacy_cutover_write", OriginURL: origin, Branch: "main", ImportExistingLocalContent: true, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) process := &agent.Process{} sess := &interactiveSession{ sessionKey: agentID, agentID: agentID, workDir: workDir, process: process, processUsesWorkspaceRootMemory: true, } d.sessionManager.sessions[agentID] = sess t.Cleanup(func() { stopMemoryWatcher(sess) }) target := d.memoryService.refreshAgentMemoryAccessWithSnapshot(t.Context(), agentID, nil) require.NotNil(t, target) require.Same(t, process, target.process) require.FileExists(t, legacyUserPreferencesMigrationStatePath( workDir, agentID, access.MemoryID, access.Branch, legacyUserPreferencesMigrationReadySuffix, )) require.NoFileExists(t, filepath.Join(workDir, "memory", "profiles", "user.md"), "legacy preferences remain unstable until the fallback Process stops") legacyPath := filepath.Join(workDir, "kb", "user-preferences.md") require.NoError(t, os.MkdirAll(filepath.Dir(legacyPath), 0o755)) require.NoError(t, os.WriteFile(legacyPath, []byte("# User Preferences\n\n- Preserve the final fallback write.\n"), 0o644)) require.True(t, d.interruptSessionProcessForRestart(target.session, target.process)) profile := readFileString(t, filepath.Join(workDir, "memory", "profiles", "user.md")) require.Contains(t, profile, "Preserve the final fallback write.") require.FileExists(t, legacyUserPreferencesMigrationStatePath( workDir, agentID, access.MemoryID, access.Branch, legacyUserPreferencesMigrationDoneSuffix, )) require.NoFileExists(t, legacyUserPreferencesMigrationStatePath( workDir, agentID, access.MemoryID, access.Branch, legacyUserPreferencesMigrationReadySuffix, )) require.Contains(t, refMessageOnOrigin(t, originDir, "main"), "Update memory") } func TestRefreshMemoryAccessDoesNotRestartRuntimeCompiledWithPendingManagedMemory(t *testing.T) { origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_pending_managed_runtime" access := MemoryAccess{ MemoryID: "mem_pending_managed_runtime", OriginURL: origin, Branch: "main", } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) require.NoError(t, ensurePendingLocalMemoryRoot(workDir, agentID, access)) process := &agent.Process{} sess := &interactiveSession{agentID: agentID, workDir: workDir, process: process} d.sessionManager.sessions[agentID] = sess t.Cleanup(func() { sess.mu.Lock() sess.process = nil sess.mu.Unlock() stopMemoryWatcher(sess) }) restartTarget := d.memoryService.refreshAgentMemoryAccessWithSnapshot(t.Context(), agentID, nil) require.Nil(t, restartTarget, "a runtime compiled with pending memory/MEMORY.md already has the managed contract") require.Same(t, process, sess.process) require.NotNil(t, sess.watcher) markers, err := filepath.Glob(filepath.Join(workDir, legacyUserPreferencesMigrationStateDir, legacyUserPreferencesMigrationMarkerPrefix+"*.done")) require.NoError(t, err) require.Len(t, markers, 1, "a managed pending runtime can close the absent legacy migration without restart") } func TestRefreshMemoryAccessWaitsForInFlightFallbackProcess(t *testing.T) { origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_inflight_fallback_runtime" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_inflight_fallback_runtime", OriginURL: origin, Branch: "main", ImportExistingLocalContent: true, }) d := newDispatcherWithMemoryProvider(t, provider) reservation, _, owns := d.sessionManager.BeginStart(agentID, nil) require.True(t, owns) process := &agent.Process{} sess := &interactiveSession{ sessionKey: agentID, agentID: agentID, workDir: workDir, processUsesWorkspaceRootMemory: true, } waiting := make(chan struct{}) d.memoryRefreshBeforeWaitForStart = func() { close(waiting) } refreshed := make(chan struct{}) go func() { d.RefreshMemoryAccess(t.Context(), []string{agentID}) close(refreshed) }() <-waiting select { case <-refreshed: t.Fatal("refresh must not pre-clone or finalize while a Process start is unpublished") default: } _, published := d.sessionManager.PublishStart(reservation, sess, process, "fallback-generation", true) require.True(t, published) d.sessionManager.FinishStart(reservation) <-refreshed _, live := d.sessionManager.FindByLookupID(agentID) require.False(t, live, "the exact fallback Process published by the in-flight start must be retired") markers, err := filepath.Glob(filepath.Join(workDir, legacyUserPreferencesMigrationStateDir, legacyUserPreferencesMigrationMarkerPrefix+"*"+legacyUserPreferencesMigrationDoneSuffix)) require.NoError(t, err) require.Len(t, markers, 1) } func TestManagedMemoryRestartDoesNotRetireNewProcessGeneration(t *testing.T) { origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_replaced_after_memory_attach" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_replaced_after_memory_attach", OriginURL: origin, Branch: "main", }) d := newDispatcherWithMemoryProvider(t, provider) staleProcess := &agent.Process{} stale := &interactiveSession{ agentID: agentID, workDir: workDir, process: staleProcess, processUsesWorkspaceRootMemory: true, } d.sessionManager.sessions[agentID] = stale restartTarget := d.memoryService.refreshAgentMemoryAccessWithSnapshot(t.Context(), agentID, nil) require.Same(t, stale, restartTarget.session) require.Same(t, staleProcess, restartTarget.process) t.Cleanup(func() { stale.mu.Lock() stale.process = nil stale.mu.Unlock() stopMemoryWatcher(stale) }) stale.mu.Lock() stale.process = &agent.Process{} stale.mu.Unlock() require.False(t, d.interruptSessionProcessForRestart(restartTarget.session, restartTarget.process)) current, ok := d.sessionManager.FindByLookupID(agentID) require.True(t, ok) require.Same(t, stale, current) } // TestRefreshMemoryAccessDetachesWatcherWhenRevoked verifies that when access // is revoked for an agent with an active watcher, the refresh detaches it. func TestRefreshMemoryAccessDetachesWatcherWhenRevoked(t *testing.T) { t.Parallel() origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_bob" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_bob", OriginURL: origin, Branch: "bob-memory", }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, agent: agent.Config{ Name: "Bob", }, binding: conversationBinding{ channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam", }, workDir: workDir, } d.sessionManager.sessions[agentID] = sess // Attach watcher first d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) require.NotNil(t, sess.watcher, "precondition: watcher should be attached") memoryDir := sess.memoryClone.Path() // Revoke access provider.remove(agentID) // Refresh should detach d.RefreshMemoryAccess(context.Background(), []string{agentID}) require.Nil(t, sess.watcher, "watcher should be detached after refresh revokes access") require.Nil(t, sess.memoryClone, "clone handle should be cleared after refresh revokes access") _, ok := d.MemorySyncStatus(agentID) require.False(t, ok, "authoritatively unbound agents must not retain teardown sync status") require.NoDirExists(t, filepath.Join(memoryDir, ".git"), "revoked memory clone should be retained offline without .git") require.FileExists(t, filepath.Join(memoryDir, "MEMORY.md"), "revoked memory content should stay readable offline") state, err := vault.ReadRetentionState(memoryDir) require.NoError(t, err) require.Equal(t, "mem_bob", state.MemoryID) require.Equal(t, agentID, state.AgentID) require.Equal(t, "bob-memory", state.Branch) require.NotEmpty(t, state.RevokedAt) _, err = time.Parse(time.RFC3339Nano, state.RevokedAt) require.NoError(t, err) } func TestRefreshMemoryAccessPreservesWatcherWhenLookupUnknown(t *testing.T) { t.Parallel() origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_unknown_refresh" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_unknown_refresh", OriginURL: origin, Branch: "unknown-refresh-memory", }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, agent: agent.Config{Name: "Unknown Refresh"}, binding: conversationBinding{ channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam", }, workDir: workDir, } d.sessionManager.sessions[agentID] = sess d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.watcher, "precondition: watcher should be attached") memoryDir := sess.memoryClone.Path() provider.setErr(errFakeProviderFailure) d.RefreshMemoryAccess(context.Background(), []string{agentID}) require.NotNil(t, sess.watcher, "unknown lookup must preserve active watcher") require.NotNil(t, sess.memoryClone, "unknown lookup must preserve clone handle") require.DirExists(t, filepath.Join(memoryDir, ".git"), "unknown lookup must not retain/detach clone") status, ok := d.MemorySyncStatus(agentID) require.True(t, ok) require.Equal(t, memorywatcher.StatusDegraded, status.State) require.Equal(t, memorywatcher.StatusReasonLookupFailed, status.Reason, "lookup degradation must take precedence over the preserved watcher's status") lookupsBeforeRetry := provider.callCount() provider.setErr(nil) d.RetryMemorySync(t.Context(), agentID) require.Eventually(t, func() bool { status, ok = d.MemorySyncStatus(agentID) return provider.callCount() == lookupsBeforeRetry+1 && ok && status.Reason != memorywatcher.StatusReasonLookupFailed }, time.Second, 10*time.Millisecond, "explicit retry must refresh authoritative access instead of retrying only the active watcher") } // TestRefreshMemoryAccessRetainsOfflineCloneWhenRevokedWithoutSession verifies // that no-session revokes reuse the same retention metadata written during the // pre-clone path. func TestRefreshMemoryAccessRetainsOfflineCloneWhenRevokedWithoutSession(t *testing.T) { t.Setenv(profile.ConfigDirEnvKey, t.TempDir()) origin := makeOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) agentID := "agt_nobody" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_no_session", OriginURL: origin, Branch: "no-session-memory", }) d := newDispatcherWithMemoryProvider(t, provider) d.RefreshMemoryAccess(context.Background(), []string{agentID}) memoryDir := filepath.Join(sessionWorkDir(agentID), "memory") require.DirExists(t, filepath.Join(memoryDir, ".git"), "pre-clone should create a managed memory clone") provider.remove(agentID) d.RefreshMemoryAccess(context.Background(), []string{agentID}) require.NoDirExists(t, filepath.Join(memoryDir, ".git"), "revoked no-session clone should be detached") require.FileExists(t, filepath.Join(memoryDir, "MEMORY.md"), "revoked no-session clone should preserve content") state, err := vault.ReadRetentionState(memoryDir) require.NoError(t, err) require.Equal(t, "mem_no_session", state.MemoryID) require.Equal(t, agentID, state.AgentID) require.Equal(t, "no-session-memory", state.Branch) require.NotEmpty(t, state.RevokedAt) _, err = time.Parse(time.RFC3339Nano, state.RevokedAt) require.NoError(t, err) } func TestRefreshMemoryAccessPreCloneImportsLatestLegacySessionMemory(t *testing.T) { t.Setenv(profile.ConfigDirEnvKey, t.TempDir()) ctx := context.Background() originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) agentID := "agt_legacy" agentRoot := sessionWorkDir(agentID) legacySession := filepath.Join(agentRoot, "sessions", "legacy-1") require.NoError(t, os.MkdirAll(legacySession, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(legacySession, "MEMORY.md"), []byte("# legacy local memory\n"), 0o644)) legacyPreferences := filepath.Join(agentRoot, "kb", "user-preferences.md") require.NoError(t, os.MkdirAll(filepath.Dir(legacyPreferences), 0o755)) require.NoError(t, os.WriteFile(legacyPreferences, []byte("# User Preferences\n\n- Prefer focused tests.\n"), 0o644)) provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_legacy", OriginURL: origin, Branch: "legacy-memory", ImportExistingLocalContent: true, }) d := newDispatcherWithMemoryProvider(t, provider) d.RefreshMemoryAccess(ctx, []string{agentID}) memoryDir := filepath.Join(agentRoot, "memory") require.DirExists(t, filepath.Join(memoryDir, ".git")) conflicts, err := filepath.Glob(filepath.Join(memoryDir, ".conflicts", "MEMORY.md.conflict-*")) require.NoError(t, err) require.Len(t, conflicts, 1) require.Equal(t, "# legacy local memory\n", readFileString(t, conflicts[0])) require.Contains(t, readFileString(t, filepath.Join(memoryDir, "profiles", "user.md")), "Prefer focused tests") require.FileExists(t, legacyPreferences, "Local Knowledge Index source must remain independent") require.Contains(t, refMessageOnOrigin(t, originDir, "legacy-memory"), "import existing local memory") } // TestRefreshMemoryAccessNoopWhenNoProvider verifies that refresh does nothing // when no provider is wired. func TestRefreshMemoryAccessNoopWhenNoProvider(t *testing.T) { t.Parallel() d := mustNewDispatcher("") // no provider d.RefreshMemoryAccess(context.Background(), []string{"agt_anyone"}) } func TestAttachMemoryWatcherImportsExistingLocalMemoryDirectory(t *testing.T) { originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) workDir := t.TempDir() agentID := "agt_alice" require.NoError(t, os.MkdirAll(filepath.Join(workDir, "memory", "kb"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "kb", "local.md"), []byte("# local\n"), 0o644)) provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_alice", OriginURL: origin, Branch: "alice-memory", ImportExistingLocalContent: true, }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.watcher) require.FileExists(t, filepath.Join(workDir, "memory", "kb", "local.md")) require.NoDirExists(t, filepath.Join(workDir, ".memory-imports")) require.Contains(t, refMessageOnOrigin(t, originDir, "alice-memory"), "import existing local memory") } func TestImportExistingLocalMemoryIntoCloneConflictsAndPushesOnce(t *testing.T) { ctx := context.Background() originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) workDir := t.TempDir() agentID := "agt_alice" clonePath := filepath.Join(workDir, "memory") clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_alice", AgentID: agentID, Branch: "alice-memory", RepoURL: origin, TargetPath: clonePath, }) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(workDir, "MEMORY.md"), []byte("# local root\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(clonePath, "MEMORY.md"), []byte("clone root\n"), 0o644)) require.NoError(t, os.MkdirAll(filepath.Join(clonePath, "kb"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(clonePath, "kb", "same.md"), []byte("clone side\n"), 0o644)) require.NoError(t, os.MkdirAll(filepath.Join(clonePath, "dir-conflict"), 0o755)) staged := filepath.Join(workDir, ".memory-imports", "memory.import-test") require.NoError(t, os.MkdirAll(filepath.Join(staged, "kb"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(staged, "MEMORY.md"), []byte("staged root\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(staged, "dir-conflict"), []byte("local file where clone has dir\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(staged, "kb", "local.md"), []byte("local only\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(staged, "kb", "same.md"), []byte("local conflict\n"), 0o644)) importExistingLocalMemoryIntoClone(ctx, workDir, &preparedExistingLocalMemoryDir{ originalPath: filepath.Join(workDir, "memory"), stagedPath: staged, }, clone, agentID, nil) require.NoFileExists(t, filepath.Join(workDir, "MEMORY.md")) require.NoDirExists(t, staged) require.Equal(t, "clone root\n", readFileString(t, filepath.Join(clonePath, "MEMORY.md"))) require.Equal(t, "clone side\n", readFileString(t, filepath.Join(clonePath, "kb", "same.md"))) require.Equal(t, "local only\n", readFileString(t, filepath.Join(clonePath, "kb", "local.md"))) require.DirExists(t, filepath.Join(clonePath, "dir-conflict")) rootConflicts, err := filepath.Glob(filepath.Join(clonePath, ".conflicts", "MEMORY.md.conflict-*")) require.NoError(t, err) require.Len(t, rootConflicts, 2) rootConflictContents := make([]string, 0, len(rootConflicts)) for _, conflict := range rootConflicts { rootConflictContents = append(rootConflictContents, readFileString(t, conflict)) } require.ElementsMatch(t, []string{"# local root\n", "staged root\n"}, rootConflictContents) kbConflicts, err := filepath.Glob(filepath.Join(clonePath, ".conflicts", "kb", "same.md.conflict-*")) require.NoError(t, err) require.Len(t, kbConflicts, 1) require.Equal(t, "local conflict\n", readFileString(t, kbConflicts[0])) dirConflicts, err := filepath.Glob(filepath.Join(clonePath, ".conflicts", "dir-conflict.conflict-*")) require.NoError(t, err) require.Len(t, dirConflicts, 1) require.Equal(t, "local file where clone has dir\n", readFileString(t, dirConflicts[0])) require.Contains(t, refMessageOnOrigin(t, originDir, "alice-memory"), "import existing local memory") } func TestImportExistingLocalMemoryIntoCloneKeepsLocalSourcesWhenPushFails(t *testing.T) { ctx := context.Background() originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) workDir := t.TempDir() agentID := "agt_alice" clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_alice", AgentID: agentID, Branch: "alice-memory", RepoURL: origin, TargetPath: filepath.Join(workDir, "memory"), }) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(workDir, "MEMORY.md"), []byte("# local root\n"), 0o644)) staged := filepath.Join(workDir, ".memory-imports", "memory.import-test") require.NoError(t, os.MkdirAll(staged, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(staged, "local.md"), []byte("local only\n"), 0o644)) require.NoError(t, os.RemoveAll(originDir)) importExistingLocalMemoryIntoClone(ctx, workDir, &preparedExistingLocalMemoryDir{ originalPath: filepath.Join(workDir, "memory"), stagedPath: staged, }, clone, agentID, nil) require.FileExists(t, filepath.Join(workDir, "MEMORY.md")) require.FileExists(t, filepath.Join(staged, "local.md")) } func TestAttachMemoryWatcherRetriesRetainedPartialImport(t *testing.T) { ctx := context.Background() tmp := t.TempDir() originDir := filepath.Join(tmp, "origin") origin := makeBareOriginRepoForMemory(t, originDir) agentID := "agt_retry_import" seed, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_retry", AgentID: "agt_seed", Branch: "main", RepoURL: origin, TargetPath: filepath.Join(tmp, "seed"), }) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(seed.Path(), ".conflicts"), []byte("blocks conflict directory\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(seed.Path(), "z-conflict.md"), []byte("remote wins\n"), 0o644)) _, err = seed.CommitPending(ctx, "seed import conflict", vault.CommitPendingOpts{}) require.NoError(t, err) _, err = seed.Push(ctx, vault.PushOpts{}) require.NoError(t, err) workDir := filepath.Join(tmp, "agent") require.NoError(t, os.MkdirAll(filepath.Join(workDir, "memory"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "a-success.md"), []byte("import once\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "memory", "z-conflict.md"), []byte("local preserved\n"), 0o644)) access := MemoryAccess{ MemoryID: "mem_retry", OriginURL: origin, Branch: "main", ImportExistingLocalContent: true, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) first := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(ctx, first, agentID, workDir) require.NotNil(t, first.memoryClone) firstClone := first.memoryClone stopMemoryWatcher(first) staged, err := filepath.Glob(filepath.Join(workDir, ".memory-imports", "memory.import-*")) require.NoError(t, err) require.Len(t, staged, 1) require.FileExists(t, filepath.Join(staged[0], existingLocalMemoryImportMarker)) require.Equal(t, "import once\n", readFileString(t, filepath.Join(staged[0], "a-success.md"))) require.Equal(t, "local preserved\n", readFileString(t, filepath.Join(staged[0], "z-conflict.md"))) require.Equal(t, "import once\n", readFileString(t, filepath.Join(workDir, "memory", "a-success.md"))) require.Equal(t, "remote wins\n", readFileString(t, filepath.Join(workDir, "memory", "z-conflict.md"))) require.NoError(t, os.Remove(filepath.Join(workDir, "memory", ".conflicts"))) _, err = firstClone.CommitPending(ctx, "remove conflict blocker", vault.CommitPendingOpts{}) require.NoError(t, err) _, err = firstClone.Push(ctx, vault.PushOpts{}) require.NoError(t, err) second := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(ctx, second, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(second) }) require.NotNil(t, second.memoryClone) require.NoDirExists(t, filepath.Join(workDir, ".memory-imports")) require.Equal(t, "import once\n", readFileString(t, filepath.Join(workDir, "memory", "a-success.md"))) conflicts, err := filepath.Glob(filepath.Join(workDir, "memory", ".conflicts", "z-conflict.md.conflict-*")) require.NoError(t, err) require.Len(t, conflicts, 1) require.Equal(t, "local preserved\n", readFileString(t, conflicts[0])) } func TestImportExistingLocalMemoryIntoCloneMigratesRootIdentityToCloudMemory(t *testing.T) { ctx := context.Background() originDir := filepath.Join(t.TempDir(), "origin") origin := makeBareOriginRepoForMemory(t, originDir) workDir := t.TempDir() agentID := "agt_alice" clonePath := filepath.Join(workDir, "memory") clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_alice", AgentID: agentID, Branch: "alice-memory", RepoURL: origin, TargetPath: clonePath, }) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(workDir, "IDENTITY.md"), []byte("# Local Role\n"), 0o644)) importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, agentID, nil) require.Equal(t, "# Local Role\n", readFileString(t, filepath.Join(clonePath, "system", "identity.md"))) require.NoFileExists(t, filepath.Join(workDir, "IDENTITY.md"), "remote Memory owns the authoritative identity after import") require.Contains(t, refMessageOnOrigin(t, originDir, "alice-memory"), "import existing local memory") } func TestImportExistingLocalMemoryIntoCloneDoesNotOverwriteCloudMemoryIdentity(t *testing.T) { ctx := context.Background() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_alice" clonePath := filepath.Join(workDir, "memory") clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_alice", AgentID: agentID, Branch: "alice-memory", RepoURL: origin, TargetPath: clonePath, }) require.NoError(t, err) require.NoError(t, os.MkdirAll(filepath.Join(clonePath, "system"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(clonePath, "system", "identity.md"), []byte("# Remote Identity\n"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "IDENTITY.md"), []byte("# Local Role\n"), 0o644)) importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, agentID, nil) require.Equal(t, "# Remote Identity\n", readFileString(t, filepath.Join(clonePath, "system", "identity.md"))) conflicts, err := filepath.Glob(filepath.Join(clonePath, ".conflicts", "system", "identity.md.conflict-*")) require.NoError(t, err) require.Empty(t, conflicts, "remote identity should win without writing duplicate conflict files") } func TestMergeLegacyUserPreferencesUsesActiveProfileOnce(t *testing.T) { destPath := filepath.Join(t.TempDir(), "profiles", "user.md") legacy := []byte("# User Preferences\n\n- Prefer focused tests.\n") changed, err := mergeLegacyUserPreferences(destPath, legacy) require.NoError(t, err) require.True(t, changed) require.Equal(t, "# User Profile\n\n## Imported User Preferences\n\n- Prefer focused tests.\n", readFileString(t, destPath)) changed, err = mergeLegacyUserPreferences(destPath, legacy) require.NoError(t, err) require.False(t, changed, "a retained Local Knowledge source must not duplicate the one-time import") require.Equal(t, 1, strings.Count(readFileString(t, destPath), "Prefer focused tests")) existingPath := filepath.Join(t.TempDir(), "profiles", "user.md") require.NoError(t, os.MkdirAll(filepath.Dir(existingPath), 0o755)) require.NoError(t, os.WriteFile(existingPath, []byte("# User Profile\n\n- Existing preference.\n"), 0o644)) changed, err = mergeLegacyUserPreferences(existingPath, legacy) require.NoError(t, err) require.True(t, changed) merged := readFileString(t, existingPath) require.Contains(t, merged, "Existing preference") require.Contains(t, merged, "## Imported User Preferences") require.Contains(t, merged, "Prefer focused tests") } func TestLegacyUserPreferencesMigrationDoesNotRestoreEditedProfile(t *testing.T) { ctx := t.Context() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_legacy_profile_once" clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_legacy_profile_once", AgentID: agentID, Branch: "main", RepoURL: origin, TargetPath: filepath.Join(workDir, "memory"), }) require.NoError(t, err) legacyPath := filepath.Join(workDir, "kb", "user-preferences.md") require.NoError(t, os.MkdirAll(filepath.Dir(legacyPath), 0o755)) require.NoError(t, os.WriteFile(legacyPath, []byte("# User Preferences\n\n- Prefer focused tests.\n"), 0o644)) migration := importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, agentID, nil) require.NotNil(t, migration) completed, err := migration.finalize(ctx, clone, nil, true) require.NoError(t, err) require.True(t, completed) complete, err := legacyUserPreferencesMigrationComplete(workDir, agentID, clone) require.NoError(t, err) require.True(t, complete) profilePath := filepath.Join(clone.Path(), "profiles", "user.md") require.NoError(t, os.WriteFile(profilePath, []byte("# User Profile\n\n- Updated active preference.\n"), 0o644)) require.NoError(t, os.WriteFile(legacyPath, []byte("# User Preferences\n\n- Changed stale preference.\n"), 0o644)) importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, agentID, nil) require.Equal(t, "# User Profile\n\n- Updated active preference.\n", readFileString(t, profilePath)) } func TestLegacyUserPreferencesMigrationClosesWhenLegacyFileIsAbsent(t *testing.T) { ctx := t.Context() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_no_legacy_profile" access := MemoryAccess{ MemoryID: "mem_no_legacy_profile", OriginURL: origin, Branch: "main", ImportExistingLocalContent: true, } d := newDispatcherWithMemoryProvider(t, nil) d.memoryService.preCloneMemory(ctx, access, agentID, workDir) profilePath := filepath.Join(workDir, "memory", "profiles", "user.md") require.NoFileExists(t, profilePath) legacyPath := filepath.Join(workDir, "kb", "user-preferences.md") require.NoError(t, os.MkdirAll(filepath.Dir(legacyPath), 0o755)) require.NoError(t, os.WriteFile(legacyPath, []byte("# Local Knowledge\n\nThis is not Memory.\n"), 0o644)) d.memoryService.preCloneMemory(ctx, access, agentID, workDir) require.NoFileExists(t, profilePath, "a later Local Knowledge file must not reopen the one-time Memory migration") } func TestLegacyUserPreferencesCutoverStateSurvivesLaunchFailure(t *testing.T) { ctx := t.Context() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_legacy_probe_launch_failure" access := MemoryAccess{ MemoryID: "mem_legacy_probe_launch_failure", OriginURL: origin, Branch: "main", ImportExistingLocalContent: true, } provider := &fakeMemoryAccessProvider{} provider.set(agentID, access) d := newDispatcherWithMemoryProvider(t, provider) failedLaunch := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(ctx, failedLaunch, agentID, workDir) require.NotNil(t, failedLaunch.watcher) require.FileExists(t, legacyUserPreferencesMigrationStatePath( workDir, agentID, access.MemoryID, access.Branch, legacyUserPreferencesMigrationReadySuffix, )) stopMemoryWatcher(failedLaunch) legacyPath := filepath.Join(workDir, "kb", "user-preferences.md") require.NoError(t, os.MkdirAll(filepath.Dir(legacyPath), 0o755)) require.NoError(t, os.WriteFile(legacyPath, []byte("# Local Knowledge\n\nCreated after the failed launch.\n"), 0o644)) retry := &interactiveSession{agentID: agentID, workDir: workDir} d.attachMemoryWatcher(ctx, retry, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(retry) }) require.NotNil(t, retry.watcher) require.NoFileExists(t, filepath.Join(workDir, "memory", "profiles", "user.md"), "ready state must await authoritative runtime cutover before its final legacy-source probe") require.FileExists(t, legacyUserPreferencesMigrationStatePath( workDir, agentID, access.MemoryID, access.Branch, legacyUserPreferencesMigrationReadySuffix, )) } func TestImportExistingLocalMemoryIntoCloneBacksUpWhenOverLimit(t *testing.T) { ctx := context.Background() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_alice", AgentID: "agt_alice", Branch: "alice-memory", RepoURL: origin, TargetPath: filepath.Join(workDir, "memory"), }) require.NoError(t, err) oldLimit := existingLocalMemoryImportMaxBytes existingLocalMemoryImportMaxBytes = 1 << 10 t.Cleanup(func() { existingLocalMemoryImportMaxBytes = oldLimit }) staged := filepath.Join(workDir, ".memory-imports", "memory.import-test") require.NoError(t, os.MkdirAll(staged, 0o755)) largePath := filepath.Join(staged, "large.md") file, err := os.Create(largePath) require.NoError(t, err) require.NoError(t, file.Truncate(1<<30), "sparse file proves rejection does not require a full-file allocation") require.NoError(t, file.Close()) importExistingLocalMemoryIntoClone(ctx, workDir, &preparedExistingLocalMemoryDir{ originalPath: filepath.Join(workDir, "memory"), stagedPath: staged, }, clone, "agt_alice", nil) require.NoFileExists(t, filepath.Join(clone.Path(), "large.md")) require.NoDirExists(t, staged) backups, err := filepath.Glob(filepath.Join(workDir, ".memory-backups", "memory.backup-*")) require.NoError(t, err) require.Len(t, backups, 1) info, err := os.Stat(filepath.Join(backups[0], "large.md")) require.NoError(t, err) require.Equal(t, int64(1<<30), info.Size()) } func TestImportExistingLocalMemoryIntoCloneBacksUpWhenCumulativeLimitExceeded(t *testing.T) { for _, tc := range []struct { name string contents []string }{ {name: "importable", contents: []string{"four", "more"}}, {name: "whitespace", contents: []string{" ", "\t\t\t\t"}}, } { t.Run(tc.name, func(t *testing.T) { testImportExistingLocalMemoryCumulativeLimit(t, tc.contents) }) } } func TestLegacyPreferenceFinalizerSharesGeneralImportBudget(t *testing.T) { ctx := t.Context() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_shared_import_budget", AgentID: "agt_shared_import_budget", Branch: "main", RepoURL: origin, TargetPath: filepath.Join(workDir, "memory"), }) require.NoError(t, err) oldLimit := existingLocalMemoryImportMaxBytes existingLocalMemoryImportMaxBytes = 64 t.Cleanup(func() { existingLocalMemoryImportMaxBytes = oldLimit }) general := strings.Repeat("g", 40) legacy := strings.Repeat("l", 40) require.NoError(t, os.WriteFile(filepath.Join(workDir, "MEMORY.md"), []byte(general), 0o644)) require.NoError(t, os.MkdirAll(filepath.Join(workDir, "kb"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "kb", "user-preferences.md"), []byte(legacy), 0o644)) migration := importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, "agt_shared_import_budget", nil) require.NotNil(t, migration) require.Equal(t, int64(len(general)), migration.generalImportedBytes) ready, err := legacyUserPreferencesMigrationReady(workDir, "agt_shared_import_budget", clone) require.NoError(t, err) require.NotNil(t, ready) require.Equal(t, int64(len(general)), ready.GeneralImportedBytes) migration = importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, "agt_shared_import_budget", nil) require.NotNil(t, migration) require.Equal(t, int64(len(general)), migration.generalImportedBytes, "a retry must recover the cumulative budget from durable ready state") completed, err := migration.finalize(ctx, clone, nil, true) require.NoError(t, err) require.True(t, completed) require.NoFileExists(t, filepath.Join(clone.Path(), "profiles", "user.md"), "legacy source must not exceed the remaining cumulative import budget") complete, err := legacyUserPreferencesMigrationComplete(workDir, "agt_shared_import_budget", clone) require.NoError(t, err) require.True(t, complete) } func TestLegacyPreferenceBudgetSurvivesCrashAfterGeneralSourceRemoval(t *testing.T) { ctx := t.Context() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() agentID := "agt_crash_safe_import_budget" clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_crash_safe_import_budget", AgentID: agentID, Branch: "main", RepoURL: origin, TargetPath: filepath.Join(workDir, "memory"), }) require.NoError(t, err) oldLimit := existingLocalMemoryImportMaxBytes existingLocalMemoryImportMaxBytes = 64 t.Cleanup(func() { existingLocalMemoryImportMaxBytes = oldLimit }) general := strings.Repeat("g", 40) legacy := strings.Repeat("l", 40) generalPath := filepath.Join(workDir, "MEMORY.md") require.NoError(t, os.WriteFile(generalPath, []byte(general), 0o644)) require.NoError(t, os.MkdirAll(filepath.Join(workDir, "kb"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "kb", "user-preferences.md"), []byte(legacy), 0o644)) // Simulate a crash after the general commit/push and source removal but // before advancing the durable phase to cutover-ready. The pending phase is // written before any import mutation and must reserve the consumed bytes // across restart even when the original source has disappeared. require.NoError(t, markLegacyUserPreferencesGeneralImportPending( workDir, agentID, clone.MemoryID(), clone.Branch(), int64(len(general)), )) require.NoError(t, os.WriteFile(filepath.Join(clone.Path(), "MEMORY.md"), []byte(general), 0o644)) _, err = clone.CommitPending(ctx, "chore: simulate completed general import", vault.CommitPendingOpts{}) require.NoError(t, err) _, err = clone.Push(ctx, vault.PushOpts{}) require.NoError(t, err) require.NoError(t, os.Remove(generalPath)) migration := importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, agentID, nil) require.NotNil(t, migration) require.Equal(t, int64(len(general)), migration.generalImportedBytes) ready, err := legacyUserPreferencesMigrationReady(workDir, agentID, clone) require.NoError(t, err) require.Equal(t, legacyUserPreferencesMigrationPhaseCutoverReady, ready.Phase) require.Equal(t, int64(len(general)), ready.GeneralImportedBytes) completed, err := migration.finalize(ctx, clone, nil, true) require.NoError(t, err) require.True(t, completed) require.NoFileExists(t, filepath.Join(clone.Path(), "profiles", "user.md"), "restart must not restore the budget consumed before the crash") } func testImportExistingLocalMemoryCumulativeLimit(t *testing.T, contents []string) { t.Helper() ctx := context.Background() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_alice", AgentID: "agt_alice", Branch: "alice-memory", RepoURL: origin, TargetPath: filepath.Join(workDir, "memory"), }) require.NoError(t, err) oldLimit := existingLocalMemoryImportMaxBytes existingLocalMemoryImportMaxBytes = 7 t.Cleanup(func() { existingLocalMemoryImportMaxBytes = oldLimit }) staged := filepath.Join(workDir, ".memory-imports", "memory.import-test") require.NoError(t, os.MkdirAll(staged, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(staged, "a.md"), []byte(contents[0]), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(staged, "b.md"), []byte(contents[1]), 0o644)) importExistingLocalMemoryIntoClone(ctx, workDir, &preparedExistingLocalMemoryDir{ originalPath: filepath.Join(workDir, "memory"), stagedPath: staged, }, clone, "agt_alice", nil) require.NoFileExists(t, filepath.Join(clone.Path(), "a.md"), "collection must finish before any partial import") require.NoFileExists(t, filepath.Join(clone.Path(), "b.md")) require.NoDirExists(t, staged) backups, err := filepath.Glob(filepath.Join(workDir, ".memory-backups", "memory.backup-*")) require.NoError(t, err) require.Len(t, backups, 1) require.Equal(t, contents[0], readFileString(t, filepath.Join(backups[0], "a.md"))) require.Equal(t, contents[1], readFileString(t, filepath.Join(backups[0], "b.md"))) } func TestImportExistingLocalMemoryIntoCloneRemovesEmptyRootMemory(t *testing.T) { ctx := context.Background() origin := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin")) workDir := t.TempDir() clone, err := vault.EnsureClone(ctx, vault.EnsureCloneOpts{ MemoryID: "mem_alice", AgentID: "agt_alice", Branch: "alice-memory", RepoURL: origin, TargetPath: filepath.Join(workDir, "memory"), }) require.NoError(t, err) require.NoError(t, os.WriteFile(filepath.Join(workDir, "MEMORY.md"), []byte(" \n\t"), 0o644)) importExistingLocalMemoryIntoClone(ctx, workDir, nil, clone, "agt_alice", nil) require.NoFileExists(t, filepath.Join(workDir, "MEMORY.md")) } func readFileString(t *testing.T, path string) string { t.Helper() data, err := os.ReadFile(path) require.NoError(t, err) return string(data) } // remoteOriginURLForClone reads the clone's origin URL. func remoteOriginURLForClone(t *testing.T, dir string) string { t.Helper() repo, err := git.PlainOpen(dir) require.NoError(t, err) remote, err := repo.Remote("origin") require.NoError(t, err) urls := remote.Config().URLs require.NotEmpty(t, urls) return urls[0] } // commitFileInWorkTree writes a new commit on the current branch. func commitFileInWorkTree(t *testing.T, dir, path, content string) { t.Helper() repo, err := git.PlainOpen(dir) require.NoError(t, err) wt, err := repo.Worktree() require.NoError(t, err) require.NoError(t, os.MkdirAll(filepath.Dir(filepath.Join(dir, path)), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(dir, path), []byte(content), 0o644)) _, err = wt.Add(path) require.NoError(t, err) _, err = wt.Commit("test commit: "+path, &git.CommitOptions{}) require.NoError(t, err) } // TestRefreshMemoryAccessRebindsWhenMemoryIDChangesReplaceLocal verifies the // dispatcher rebind path: when the server reports a new MemoryID for an agent // whose session already has a watcher, RefreshMemoryAccess stops the watcher, // backs up the prior clone, and clones fresh from the new binding. func TestRefreshMemoryAccessRebindsWhenMemoryIDChangesReplaceLocal(t *testing.T) { t.Parallel() originA := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin-a")) originB := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin-b")) workDir := t.TempDir() agentID := "agt_rebind_replace" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_alpha", OriginURL: originA, Branch: "main", }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, agent: agent.Config{Name: "Rebind Replace"}, binding: conversationBinding{ channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam", }, workDir: workDir, } d.sessionManager.sessions[agentID] = sess d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.watcher, "precondition: watcher attached") memoryDir := sess.memoryClone.Path() commitFileInWorkTree(t, memoryDir, "alpha-note.md", "alpha only\n") require.Equal(t, originA, remoteOriginURLForClone(t, memoryDir)) // Server switches the binding to a different memory with ReplaceLocal=true. provider.set(agentID, MemoryAccess{ MemoryID: "mem_bravo", OriginURL: originB, Branch: "main", ReplaceLocalContentOnSwitch: true, }) d.RefreshMemoryAccess(context.Background(), []string{agentID}) require.NotNil(t, sess.watcher, "watcher should reattach after rebind") require.NotNil(t, sess.memoryClone, "clone handle should be populated after rebind") require.Equal(t, "mem_bravo", sess.memoryClone.MemoryID(), "clone should reflect new memory id") require.Equal(t, originB, remoteOriginURLForClone(t, memoryDir), "origin should repoint to B") require.FileExists(t, filepath.Join(memoryDir, "MEMORY.md"), "fresh clone from B should contain MEMORY.md") require.Equal(t, "# Origin memory\n\nInitial content.\n", readFileString(t, filepath.Join(memoryDir, "MEMORY.md"))) _, err := os.Stat(filepath.Join(memoryDir, "alpha-note.md")) require.Error(t, err) require.True(t, os.IsNotExist(err), "alpha-note.md from prior binding should not survive fresh clone") entries, err := os.ReadDir(filepath.Dir(memoryDir)) require.NoError(t, err) foundBackup := false for _, entry := range entries { if strings.HasPrefix(entry.Name(), "memory.backup-") { foundBackup = true break } } require.True(t, foundBackup, "prior clone should have been backed up") } // TestRefreshMemoryAccessRebindsWhenMemoryIDChangesRepoint verifies the // repoint path: when ReplaceLocalContentOnSwitch=false, the dispatcher keeps // the local clone, repoints its origin to the new binding, and force-pushes // the local history to the new origin. func TestRefreshMemoryAccessRebindsWhenMemoryIDChangesRepoint(t *testing.T) { t.Parallel() originA := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin-a")) originB := makeBareOriginRepoForMemory(t, filepath.Join(t.TempDir(), "origin-b")) workDir := t.TempDir() agentID := "agt_rebind_repoint" provider := &fakeMemoryAccessProvider{} provider.set(agentID, MemoryAccess{ MemoryID: "mem_alpha", OriginURL: originA, Branch: "main", }) d := newDispatcherWithMemoryProvider(t, provider) sess := &interactiveSession{ agentID: agentID, agent: agent.Config{Name: "Rebind Repoint"}, binding: conversationBinding{ channelID: "#agentteam", targetScopeType: "channel", conversationKey: "#agentteam", }, workDir: workDir, } d.sessionManager.sessions[agentID] = sess d.attachMemoryWatcher(context.Background(), sess, agentID, workDir) t.Cleanup(func() { stopMemoryWatcher(sess) }) require.NotNil(t, sess.watcher, "precondition: watcher attached") memoryDir := sess.memoryClone.Path() commitFileInWorkTree(t, memoryDir, "retained.md", "local bravo note\n") require.Equal(t, originA, remoteOriginURLForClone(t, memoryDir)) // Server switches the binding to a different memory, this time keeping the // local history (ReplaceLocal=false). provider.set(agentID, MemoryAccess{ MemoryID: "mem_bravo", OriginURL: originB, Branch: "main", }) d.RefreshMemoryAccess(context.Background(), []string{agentID}) require.NotNil(t, sess.watcher, "watcher should reattach after rebind") require.NotNil(t, sess.memoryClone, "clone handle should be populated after rebind") require.Equal(t, "mem_bravo", sess.memoryClone.MemoryID(), "clone should reflect new memory id") require.Equal(t, originB, remoteOriginURLForClone(t, memoryDir), "origin should repoint to B") require.Equal(t, "local bravo note\n", readFileString(t, filepath.Join(memoryDir, "retained.md")), "local working tree content must survive repoint") // Origin B should now carry the local commit (force-pushed). bRepo, err := git.PlainOpen(strings.TrimPrefix(originB, "file://")) require.NoError(t, err) bHead, err := bRepo.Head() require.NoError(t, err) bHeadCommit, err := bRepo.CommitObject(bHead.Hash()) require.NoError(t, err) require.Contains(t, bHeadCommit.Message, "test commit: retained.md") // Retention state must reflect the new binding. state, err := vault.ReadRetentionState(memoryDir) require.NoError(t, err) require.Equal(t, "mem_bravo", state.MemoryID) require.Equal(t, agentID, state.AgentID) require.Empty(t, state.RevokedAt) }