conflicts: converge divergent edits via contested-freeze, notify both edges (#158)

agent/agent.go

@@ -47,11 +47,18 @@ type SyncRunner func(ctx context.Context, vol *config.Volume, destName string) S
 
 // SyncRunReport is the SyncRunner result surfaced to the scheduler.
 // Kept minimal — anything richer belongs in the runs table the CLI
-// inspects via `squirrel runs`, not on the agent's hot path.
+// inspects via `squirrel runs`, not on the agent's hot path. Conflicts
+// and Contested carry the peer-sync divergence counts so the scheduler
+// logs a conflict signal on the initiating machine, not only the hub
+// (#158, F27): Conflicts is how many paths this run resolved by
+// preserving the loser and landing this node's bytes live, Contested how
+// many the receiver refused because a prior freeze still stands.
 type SyncRunReport struct {
-	RunID  int64
-	Status string
-	Err    error
+	RunID     int64
+	Status    string
+	Err       error
+	Conflicts int
+	Contested int
 }
 
 // Config configures one agent listener. Fields are validated by New; all

agent/contested_test.go

@@ -0,0 +1,190 @@
+package agent
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/mbertschler/squirrel/store"
+	"github.com/mbertschler/squirrel/syncproto"
+	"github.com/zeebo/blake3"
+)
+
+// seedLocalWrite writes content to path and records a present index row
+// attributed to a *local* index run (no peer linkage), so a divergent
+// initiator hash classifies as a conflict — the precondition for a freeze.
+func (f *preStageFixture) seedLocalWrite(t *testing.T, ctx context.Context, path string, content []byte) []byte {
+	t.Helper()
+	abs := filepath.Join(f.vol.Path, path)
+	if err := os.MkdirAll(filepath.Dir(abs), 0o755); err != nil {
+		t.Fatalf("mkdir for %s: %v", path, err)
+	}
+	if err := os.WriteFile(abs, content, 0o644); err != nil {
+		t.Fatalf("write %s: %v", path, err)
+	}
+	sum := blake3.Sum256(content)
+	if err := f.store.Upsert(ctx, store.FileRow{
+		VolumeID: f.volID, Path: path, Blake3: sum[:], SizeBytes: int64(len(content)),
+		MtimeNs: store.NowNs(), Status: store.StatusPresent,
+		FirstSeenRunID: f.localRun, LastSeenRunID: f.localRun, IndexedAtNs: store.NowNs(),
+	}, nil); err != nil {
+		t.Fatalf("Upsert %s: %v", path, err)
+	}
+	return sum[:]
+}
+
+// freshSession opens another receiver run and returns a session bound to
+// it at the given protocol version, so a test can drive successive /plan
+// exchanges the way successive cadence ticks would.
+func (f *preStageFixture) freshSession(t *testing.T, protocol int) *peerSession {
+	t.Helper()
+	run, err := f.store.BeginPeerSyncRun(context.Background(), f.volID, f.peerID, 7, "peerA")
+	if err != nil {
+		t.Fatalf("BeginPeerSyncRun: %v", err)
+	}
+	sess := f.newSession()
+	sess.receiverRunID = run
+	sess.protocolVersion = protocol
+	return sess
+}
+
+// countConflictRunDirs returns how many run-<id>/ subdirectories exist
+// under .squirrel-conflicts — one per distinct conflict-preserving run, so
+// the F27 unbounded-growth regression shows up as this number climbing.
+func countConflictRunDirs(t *testing.T, volPath string) int {
+	t.Helper()
+	entries, err := os.ReadDir(filepath.Join(volPath, ConflictsDirName))
+	if err != nil {
+		if os.IsNotExist(err) {
+			return 0
+		}
+		t.Fatalf("read conflicts dir: %v", err)
+	}
+	return len(entries)
+}
+
+// TestContestedFreezeStopsPingPong is the #158 acceptance at the /plan
+// level: the first conflict preserves the loser and freezes the path; a
+// later divergent re-assertion is refused with `contested` (no second
+// copy minted); and the frozen winner re-asserting is still allowed
+// through so a dropped /close can re-land it.
+func TestContestedFreezeStopsPingPong(t *testing.T) {
+	ctx := context.Background()
+	f := newPreStageFixture(t)
+
+	contentX := []byte("homepc's version, indexed locally on the hub")
+	contentZ := []byte("laptop's divergent version — wins the first conflict")
+	contentW := []byte("homepc re-asserts yet another edit")
+	f.seedLocalWrite(t, ctx, "doc.md", contentX)
+
+	// Tick 1: laptop's divergent bytes conflict with the local write.
+	sess1 := f.freshSession(t, syncproto.ProtocolVersionContested)
+	plan1, err := f.router.planSession(ctx, sess1, []syncproto.IndexEntry{
+		{Path: "doc.md", Blake3Hex: blakeHex(contentZ), SizeBytes: int64(len(contentZ))},
+	})
+	if err != nil {
+		t.Fatalf("planSession tick 1: %v", err)
+	}
+	if d := sess1.dispositions["doc.md"].disposition; d != syncproto.DispositionConflict {
+		t.Fatalf("tick 1 disposition = %q, want conflict", d)
+	}
+	if len(plan1.Conflicts) != 1 {
+		t.Fatalf("tick 1 conflicts = %d, want 1", len(plan1.Conflicts))
+	}
+	if got := countConflictRunDirs(t, f.vol.Path); got != 1 {
+		t.Fatalf("conflict run dirs after tick 1 = %d, want 1", got)
+	}
+	// The path is now frozen with laptop's bytes as the winner.
+	latch, err := f.store.GetContestedPath(ctx, f.volID, "doc.md")
+	if err != nil {
+		t.Fatalf("GetContestedPath: %v", err)
+	}
+	if blakeHex(contentZ) != hexOrEmpty(latch.LiveBlake3) {
+		t.Fatalf("latch winner = %s, want hash(Z) %s", hexOrEmpty(latch.LiveBlake3), blakeHex(contentZ))
+	}
+
+	// Tick 2: homepc re-asserts a third version. The freeze refuses it —
+	// no bytes move, no second copy is minted (the F27 fix).
+	sess2 := f.freshSession(t, syncproto.ProtocolVersionContested)
+	plan2, err := f.router.planSession(ctx, sess2, []syncproto.IndexEntry{
+		{Path: "doc.md", Blake3Hex: blakeHex(contentW), SizeBytes: int64(len(contentW))},
+	})
+	if err != nil {
+		t.Fatalf("planSession tick 2: %v", err)
+	}
+	if d := sess2.dispositions["doc.md"].disposition; d != syncproto.DispositionContested {
+		t.Fatalf("tick 2 disposition = %q, want contested", d)
+	}
+	if len(plan2.Conflicts) != 0 {
+		t.Fatalf("tick 2 conflicts = %d, want 0 (frozen, not re-conflicted)", len(plan2.Conflicts))
+	}
+	if len(plan2.Contested) != 1 || plan2.Contested[0].Path != "doc.md" {
+		t.Fatalf("tick 2 contested = %+v, want one doc.md entry", plan2.Contested)
+	}
+	if plan2.Contested[0].PreservedAtPath == "" || plan2.Contested[0].LiveBlake3Hex != blakeHex(contentZ) {
+		t.Fatalf("tick 2 contested detail = %+v, want winner Z + preserved path", plan2.Contested[0])
+	}
+	if got := countConflictRunDirs(t, f.vol.Path); got != 1 {
+		t.Fatalf("conflict run dirs after tick 2 = %d, want 1 (no new copy minted)", got)
+	}
+
+	// Tick 3: the frozen winner re-asserts its own bytes — allowed through
+	// (Transfer), so a /close that never committed can re-land the winner.
+	sess3 := f.freshSession(t, syncproto.ProtocolVersionContested)
+	if _, err := f.router.planSession(ctx, sess3, []syncproto.IndexEntry{
+		{Path: "doc.md", Blake3Hex: blakeHex(contentZ), SizeBytes: int64(len(contentZ))},
+	}); err != nil {
+		t.Fatalf("planSession tick 3: %v", err)
+	}
+	if d := sess3.dispositions["doc.md"].disposition; d != syncproto.DispositionTransfer {
+		t.Fatalf("tick 3 disposition = %q, want transfer (winner re-asserting)", d)
+	}
+}
+
+// TestContestedFreezeVersionGated confirms the disposition is gated: on
+// one and the same frozen state, a session negotiated below
+// ProtocolVersionContested falls back to the legacy `conflict` it
+// understands, while a contested-capable session gets the freeze. Older
+// peers stay safe.
+func TestContestedFreezeVersionGated(t *testing.T) {
+	ctx := context.Background()
+	f := newPreStageFixture(t)
+	contentX := f.seedLocalWrite(t, ctx, "doc.md", []byte("hub-local content"))
+
+	// Freeze doc.md while its local-write bytes stay live (the winner), so
+	// both sessions below classify against an identical present row.
+	if err := f.store.RaiseContested(ctx, store.ContestedPath{
+		VolumeID: f.volID, Path: "doc.md", LiveBlake3: contentX, RaisedRunID: f.localRun,
+	}); err != nil {
+		t.Fatalf("RaiseContested: %v", err)
+	}
+	divergent := blakeHex([]byte("a divergent edit"))
+
+	// An older initiator (Merkle-walk protocol) must get the legacy
+	// conflict, not the contested disposition it can't interpret.
+	legacy := f.freshSession(t, syncproto.ProtocolVersionMerkleWalk)
+	plan, err := f.router.planSession(ctx, legacy, []syncproto.IndexEntry{
+		{Path: "doc.md", Blake3Hex: divergent, SizeBytes: 16},
+	})
+	if err != nil {
+		t.Fatalf("planSession legacy: %v", err)
+	}
+	if d := legacy.dispositions["doc.md"].disposition; d != syncproto.DispositionConflict {
+		t.Fatalf("legacy disposition = %q, want conflict (version gate)", d)
+	}
+	if len(plan.Contested) != 0 {
+		t.Fatalf("legacy plan.Contested = %+v, want empty for an older peer", plan.Contested)
+	}
+
+	// A contested-capable initiator, same frozen state, gets refused.
+	modern := f.freshSession(t, syncproto.ProtocolVersionContested)
+	if _, err := f.router.planSession(ctx, modern, []syncproto.IndexEntry{
+		{Path: "doc.md", Blake3Hex: divergent, SizeBytes: 16},
+	}); err != nil {
+		t.Fatalf("planSession modern: %v", err)
+	}
+	if d := modern.dispositions["doc.md"].disposition; d != syncproto.DispositionContested {
+		t.Fatalf("modern disposition = %q, want contested (version gate)", d)
+	}
+}

agent/dispatcher.go

@@ -184,7 +184,19 @@ func (d *syncDispatcher) runOne(ctx context.Context, vol *config.Volume, destNam
 		"kind", "sync", "volume", vol.Name, "destination", destName,
 		"run_id", rep.RunID, "status", status,
 		"duration_ms", duration.Milliseconds(),
+		"conflicts", rep.Conflicts, "contested", rep.Contested,
 	)
+	// A conflict signal on this (initiating) machine, not just the hub: a
+	// divergent edit was preserved or a frozen path refused, and a human
+	// must eventually resolve it (#158, F27). Logged at warn so it stands
+	// out from the routine finished line in the scheduler's own output.
+	if rep.Conflicts > 0 || rep.Contested > 0 {
+		d.logger.Warn("scheduler.conflict",
+			"kind", "sync", "volume", vol.Name, "destination", destName,
+			"run_id", rep.RunID,
+			"conflicts", rep.Conflicts, "contested", rep.Contested,
+		)
+	}
 	if rep.Err != nil {
 		d.logger.Error("scheduler.error",
 			"kind", "sync", "volume", vol.Name, "destination", destName,

agent/sync.go

@@ -401,7 +401,7 @@ func (r *peerSyncRouter) finishBegin(ctx context.Context, body syncproto.BeginRe
 // know is clamped down to ProtocolVersionMerkleWalk rather than
 // rejected, so a partial rollout doesn't break syncs.
 func negotiateProtocol(requested int) int {
-	const receiverMax = syncproto.ProtocolVersionMerkleWalk
+	const receiverMax = syncproto.ProtocolVersionContested
 	if requested <= 0 {
 		return syncproto.ProtocolVersionFlat
 	}
@@ -574,9 +574,53 @@ func (r *peerSyncRouter) planSession(ctx context.Context, sess *peerSession, ent
 		})
 	}
 	resp.Conflicts = collectConflicts(sess)
+	contested, err := r.collectContested(ctx, sess, entries)
+	if err != nil {
+		return syncproto.PlanResponse{}, err
+	}
+	resp.Contested = contested
 	return resp, nil
 }
 
+// collectContested builds the wire-format contested list from the entries
+// classified as `contested`, in the order the initiator sent them. Each
+// path is frozen, so its contested_paths latch supplies the two versions
+// and the preserved-loser location the initiator mirrors into its own
+// marker and renders. A latch that vanished between classify and here
+// (an operator resolved it mid-plan) is skipped rather than erroring —
+// the next sync re-plans cleanly.
+func (r *peerSyncRouter) collectContested(ctx context.Context, sess *peerSession, entries []syncproto.IndexEntry) ([]syncproto.ContestedDetail, error) {
+	var out []syncproto.ContestedDetail
+	for _, e := range entries {
+		if sess.dispositions[e.Path].disposition != syncproto.DispositionContested {
+			continue
+		}
+		latch, err := r.srv.store.GetContestedPath(ctx, sess.volumeID, e.Path)
+		if err != nil {
+			if store.IsNotFound(err) {
+				continue
+			}
+			return nil, fmt.Errorf("look up contested %s: %w", e.Path, err)
+		}
+		out = append(out, syncproto.ContestedDetail{
+			Path:               e.Path,
+			LiveBlake3Hex:      hexOrEmpty(latch.LiveBlake3),
+			PreservedBlake3Hex: hexOrEmpty(latch.PreservedBlake3),
+			PreservedAtPath:    latch.PreservedAtPath,
+		})
+	}
+	return out, nil
+}
+
+// hexOrEmpty renders a raw digest as lowercase hex, or "" for a nil digest
+// (the "unknown winner" latch case), keeping the wire field omittable.
+func hexOrEmpty(b []byte) string {
+	if len(b) == 0 {
+		return ""
+	}
+	return hex.EncodeToString(b)
+}
+
 // handlePlanFolders implements POST /v1/sync/plan-folders: look up the
 // receiver's folder rows for each requested path and return the
 // (shallow, deep, direct children) tuple. The endpoint is the
@@ -804,21 +848,31 @@ func collectConflicts(sess *peerSession) []syncproto.ConflictDetail {
 // strategy is "off" (initiator opted out).
 func (r *peerSyncRouter) classify(ctx context.Context, sess *peerSession, relPath string, entry *sessionEntry) (string, error) {
 	existing, err := r.srv.store.GetByPath(ctx, sess.volumeID, relPath)
-	if err != nil {
-		if store.IsNotFound(err) {
-			return r.classifyMissingPath(ctx, sess, entry)
-		}
+	liveFound := err == nil && existing.Status == store.StatusPresent
+	if err != nil && !store.IsNotFound(err) {
 		return "", err
 	}
-	if existing.Status != store.StatusPresent {
-		// A live row in 'missing' status is treated as "no file here";
-		// the initiator's bytes are new content. Dedup still applies —
-		// the bytes may exist at another path.
-		return r.classifyMissingPath(ctx, sess, entry)
-	}
-	if bytesEqual(existing.Blake3, entry.blake3) {
+	if liveFound && bytesEqual(existing.Blake3, entry.blake3) {
 		return syncproto.DispositionAlreadyCorrect, nil
 	}
+	// The initiator's bytes differ from any live row here (or there is no
+	// live row). Before deciding Supersede/Conflict/Transfer, check the
+	// contested freeze: a path frozen by a prior conflict refuses a
+	// divergent re-assertion from any peer, so the flip-flop stops at the
+	// first conflict instead of minting a fresh copy every cadence tick.
+	// The frozen winner re-asserting its own digest is *not* refused — it
+	// falls through so a /close that never committed can re-establish it.
+	if contested, err := r.contestedRefusal(ctx, sess, relPath, entry); err != nil {
+		return "", err
+	} else if contested {
+		return syncproto.DispositionContested, nil
+	}
+	if !liveFound {
+		// No live row (or a 'missing'/'offloaded' one): the initiator's
+		// bytes are new content here. Dedup still applies — the bytes may
+		// exist at another path.
+		return r.classifyMissingPath(ctx, sess, entry)
+	}
 	// Different blake3 at this path. Provenance decides the verdict
 	// (Supersede / Conflict). Content at a path takes precedence over
 	// a cross-path dedup — we don't let dedup paper over a divergence
@@ -831,6 +885,28 @@ func (r *peerSyncRouter) classify(ctx context.Context, sess *peerSession, relPat
 	return disp, nil
 }
 
+// contestedRefusal reports whether relPath is frozen and this initiator's
+// bytes should be refused with the `contested` disposition. It returns
+// false — letting normal classification proceed — when the path isn't
+// frozen, when the initiator's digest matches the frozen winner (the
+// winner re-asserting, which must be allowed to re-land), or when the
+// session negotiated a protocol older than ProtocolVersionContested (an
+// initiator that can't interpret the disposition falls back to the legacy
+// `conflict`, keeping older peers safe per the version gate).
+func (r *peerSyncRouter) contestedRefusal(ctx context.Context, sess *peerSession, relPath string, entry *sessionEntry) (bool, error) {
+	if sess.protocolVersion < syncproto.ProtocolVersionContested {
+		return false, nil
+	}
+	liveBlake3, contested, err := r.srv.store.IsPathContested(ctx, sess.volumeID, relPath)
+	if err != nil {
+		return false, err
+	}
+	if !contested {
+		return false, nil
+	}
+	return !bytesEqual(entry.blake3, liveBlake3), nil
+}
+
 // classifyMissingPath is the no-live-row branch of classify: try to
 // satisfy the path from existing content elsewhere in the volume
 // (CopyFromExisting), or fall back to Transfer. Strategy "off" skips
@@ -1188,6 +1264,22 @@ func (r *peerSyncRouter) preStageConflicts(ctx context.Context, sess *peerSessio
 			return fmt.Errorf("record conflict pre-stage for %s: %w", path, err)
 		}
 		entry.preservedAtPath = preservedRel
+		// Freeze the path so the next cadence tick refuses a divergent
+		// re-assertion instead of minting another `.squirrel-conflicts/`
+		// copy (#158, F27). The winner is the initiator's bytes (landing
+		// live on /close); the loser is the just-preserved prior row.
+		// Idempotent: a path already frozen keeps its original latch.
+		if err := r.srv.store.RaiseContested(ctx, store.ContestedPath{
+			VolumeID:        sess.volumeID,
+			Path:            path,
+			LiveBlake3:      entry.blake3,
+			PreservedBlake3: entry.priorRow.Blake3,
+			PreservedAtPath: preservedRel,
+			PeerNodeID:      sess.peerNodeID,
+			RaisedRunID:     sess.receiverRunID,
+		}); err != nil {
+			return fmt.Errorf("raise contested for %s: %w", path, err)
+		}
 	}
 	return nil
 }