fix: collect selected checkout for sync and hooks

README.md

@@ -65,8 +65,9 @@ diffx sync --debug
 diffx sync --retry
 ```
 
-Sync waits for all observations to commit. `--print` prints only the configured
-server URL on success; progress/errors use stderr. `--debug` uses one updating
+Sync collects only the selected checkout and waits for pending observations to
+commit. Unchanged uploads are skipped after server freshness checks. `--print`
+prints only the configured server URL on success; progress/errors use stderr. `--debug` uses one updating
 terminal line, or plain lines when redirected. Failed uploads remain private,
 immutable pending submissions; `sync --retry` recovers them without the checkout.
 Cancelling a wait does not cancel an accepted server job.
@@ -104,8 +105,8 @@ codex plugin add diffx@diffx
 ```
 
 Configure a remote server and token first. Plugins are remote-only.
-The plugin returns after scheduling; a detached Go worker collects the checkout
-and all registered worktrees using the same branch comparison as manual review.
+The plugin returns after scheduling; a detached Go worker collects only the
+triggering checkout using the same branch comparison as manual review.
 Non-Git directories are ignored. Child repositories and branches without live
 checkouts are excluded; use `sync --branch` for explicit object-only recovery.
 Persistent Git identities survive checkout moves and branch renames.

cmd/diffx/hook.go

@@ -256,7 +256,7 @@ func newHookEngine(directory string) hooks.Engine {
 		if err != nil {
 			return nil, err
 		}
-		discovery, err := collector.DiscoverWorktrees(ctx, event.Path, sourceID, event.Base)
+		discovery, err := collector.DiscoverCheckout(ctx, event.Path, sourceID, event.Base)
 		if errors.Is(err, diffsource.ErrNotRepository) {
 			return nil, nil
 		}

cmd/diffx/producer.go

@@ -1,14 +1,6 @@
 package main
 
-import (
-	"context"
-	"errors"
-	"fmt"
-
-	"github.com/flexdinesh/diffx/internal/collector"
-	"github.com/flexdinesh/diffx/internal/config"
-	"github.com/flexdinesh/diffx/internal/ingestion"
-)
+import "github.com/flexdinesh/diffx/internal/config"
 
 func resolvedCollectorSettings(values options) (options, error) {
 	saved, err := config.LoadFile(values.configFile)
@@ -23,41 +15,3 @@ func resolvedCollectorSettings(values options) (options, error) {
 	}
 	return values, nil
 }
-
-func collectSubmissions(ctx context.Context, values options) ([]ingestion.Request, error) {
-	if values.branch != "" {
-		request, err := collectSubmission(ctx, values)
-		if err != nil {
-			return nil, err
-		}
-		return []ingestion.Request{request}, nil
-	}
-	settings, err := collectionOptions(values)
-	if err != nil {
-		return nil, err
-	}
-	discovery, err := collector.DiscoverWorktrees(ctx, values.directory, settings.SourceID, values.base)
-	if err != nil {
-		return nil, err
-	}
-	requests := make([]ingestion.Request, 0, len(discovery.Sources))
-	failures := make([]error, 0)
-	for _, diagnostic := range discovery.Diagnostics {
-		failures = append(failures, errors.New(diagnostic))
-	}
-	for _, source := range discovery.Sources {
-		child := values
-		child.directory, child.base = source.Path, source.Base
-		// Keep the triggering checkout distinct from each discovered root.
-		settings.SubmissionID = newSubmissionID()
-		settings.Base = source.Base
-		settings.TriggerRoot = source.InputPath
-		request, err := collector.Collect(ctx, child.directory, settings)
-		if err != nil {
-			failures = append(failures, fmt.Errorf("collect %s: %w", source.Path, err))
-			continue
-		}
-		requests = append(requests, request)
-	}
-	return requests, errors.Join(failures...)
-}

cmd/diffx/producer_test.go

@@ -7,7 +7,6 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
-	"strings"
 	"testing"
 
 	"github.com/flexdinesh/diffx/internal/contextservice"
@@ -53,10 +52,16 @@ func TestCollectorDestinationConfigAndExplicitLocalOverride(t *testing.T) {
 	}
 }
 
-func TestHookDiscoveryIgnoresNonGitAndObjectOnlyBranches(t *testing.T) {
+func TestHookDiscoverySelectsCheckoutAndIgnoresNonGitAndObjectOnlyBranches(t *testing.T) {
 	workspace := t.TempDir()
 	root := cliRepository(t)
 	producerGit(t, root, "branch", "unopened")
+	linked := filepath.Join(t.TempDir(), "linked")
+	producerGit(t, root, "worktree", "add", "-b", "other", linked)
+	subdir := filepath.Join(linked, "subdir")
+	if err := os.Mkdir(subdir, 0700); err != nil {
+		t.Fatal(err)
+	}
 	t.Setenv("DIFFX_SOURCE_ID", "machine")
 	engine := newHookEngine(t.TempDir())
 	events, err := engine.Expand(t.Context(), hooks.Event{Path: workspace, InputPath: workspace, Agent: "codex", Base: "auto"})
@@ -67,6 +72,10 @@ func TestHookDiscoveryIgnoresNonGitAndObjectOnlyBranches(t *testing.T) {
 	if err != nil || len(events) != 1 || events[0].Branch != "" {
 		t.Fatalf("registered checkouts: %#v %v", events, err)
 	}
+	events, err = engine.Expand(t.Context(), hooks.Event{Path: subdir, InputPath: subdir, Agent: "codex", Base: "auto"})
+	if err != nil || len(events) != 1 || events[0].Path != linked || events[0].InputPath != subdir || events[0].Identity == "" || !events[0].Resolved {
+		t.Fatalf("selected linked checkout: %#v %v", events, err)
+	}
 }
 
 func TestHookInitialCleanSessionsAndMissingAcknowledgement(t *testing.T) {
@@ -170,18 +179,18 @@ func TestRemoteFailureNeverStartsLocalService(t *testing.T) {
 	}
 }
 
-func TestUnavailableWorktreeNeverProducesFalseEmptyCapture(t *testing.T) {
+func TestUnavailableOtherWorktreeDoesNotAffectSelectedCheckout(t *testing.T) {
 	root := cliRepository(t)
 	linked := filepath.Join(t.TempDir(), "removed")
 	producerGit(t, root, "worktree", "add", "-b", "removed", linked)
 	if err := os.RemoveAll(linked); err != nil {
 		t.Fatal(err)
 	}
-	requests, err := collectSubmissions(t.Context(), options{directory: root, sourceID: "machine", trigger: "manual"})
-	if err == nil || !strings.Contains(err.Error(), "unavailable") {
-		t.Fatalf("missing worktree not reported: %v", err)
+	request, err := collectSubmission(t.Context(), options{directory: root, sourceID: "machine", trigger: "manual"})
+	if err != nil || request.Metadata.Root != root || len(request.Scopes[0].Snapshot.Files) == 0 {
+		t.Fatalf("selected checkout collection: %#v %v", request, err)
 	}
-	if len(requests) != 1 || requests[0].Metadata.Root != root || len(requests[0].Scopes[0].Snapshot.Files) == 0 {
-		t.Fatalf("false empty capture: %#v", requests)
+	if _, err := collectSubmission(t.Context(), options{directory: linked, sourceID: "machine", trigger: "manual"}); err == nil {
+		t.Fatal("missing selected checkout produced a capture")
 	}
 }

cmd/diffx/sync.go

@@ -6,13 +6,15 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 	"strings"
 
+	"github.com/flexdinesh/diffx/internal/collector"
 	"github.com/flexdinesh/diffx/internal/ingestion"
 	"github.com/flexdinesh/diffx/internal/reviewstore"
 	"github.com/flexdinesh/diffx/internal/submission"
@@ -85,19 +87,30 @@ func runSync(ctx context.Context, arguments []string, stdin *os.File, stdout, st
 		return err
 	}
 	var failures []error
+	var fresh ingestion.Request
+	reusedSubmission := ""
 	if !values.retry {
-		progress.show("Collect · discovering worktrees and reading changes")
+		progress.show("Collect · reading selected checkout")
 		values.progress = func(stage, detail string) { progress.show(stage + " · " + detail) }
-		requests, collectErr := collectSubmissions(ctx, values)
+		var collectErr error
+		fresh, collectErr = collectSubmission(ctx, values)
 		if collectErr != nil {
 			failures = append(failures, collectErr)
-		}
-		for _, request := range requests {
-			item := submission.Pending{Request: request}
-			if err := box.Save(item); err != nil {
-				return err
+		} else {
+			fingerprint := syncFingerprint(fresh)
+			for _, item := range pending {
+				if fingerprint != "" && syncFingerprint(item.Request) == fingerprint {
+					reusedSubmission = item.Request.SubmissionID
+					break
+				}
+			}
+			if reusedSubmission == "" {
+				item := submission.Pending{Request: fresh}
+				if err := box.Save(item); err != nil {
+					return err
+				}
+				pending = append(pending, item)
 			}
-			pending = append(pending, item)
 		}
 	}
 	if len(pending) == 0 {
@@ -114,20 +127,67 @@ func runSync(ctx context.Context, arguments []string, stdin *os.File, stdout, st
 	target.Progress = func(job ingestion.Job) {
 		progress.show(fmt.Sprintf("Ingest · %s · attempt %d", job.State, job.Attempt))
 	}
-	for index, item := range pending {
+	uploaded, unchanged := 0, 0
+	for index := 0; index < len(pending); index++ {
+		item := pending[index]
 		if item.StateID != "" && item.StateID != target.Target.Identity {
 			failures = append(failures, fmt.Errorf("submission %s belongs to another server database; retained for recovery", item.Request.SubmissionID))
+			if item.Request.SubmissionID == reusedSubmission {
+				freshItem := submission.Pending{Request: fresh}
+				if err := box.Save(freshItem); err != nil {
+					return err
+				}
+				pending = append(pending, freshItem)
+			}
 			continue
 		}
+		fingerprint := syncFingerprint(item.Request)
+		ack, err := box.Acknowledgement(item.Request)
+		if err != nil {
+			return err
+		}
+		if fingerprint != "" && ack.Fingerprint == fingerprint && ack.StateID == target.Target.Identity && ack.ContextID != "" {
+			current, err := target.Current(ctx, ack.ContextID)
+			if err != nil {
+				return err
+			}
+			if current {
+				if err := box.Remove(item.Request.SubmissionID); err != nil {
+					return err
+				}
+				unchanged++
+				continue
+			}
+		}
 		item.StateID = target.Target.Identity
 		if err := box.Save(item); err != nil {
 			return err
 		}
 		progress.show(fmt.Sprintf("Upload · %d/%d", index+1, len(pending)))
-		if _, err := target.Deliver(ctx, item.Request); err != nil {
+		receipt, err := target.Deliver(ctx, item.Request)
+		if err != nil {
 			failures = append(failures, fmt.Errorf("submission %s retained; resume with sync --retry: %w", item.Request.SubmissionID, err))
 			continue
 		}
+		uploaded++
+		if err := box.Acknowledge(item.Request, fingerprint, target.Target.Identity, receipt.ContextID); err != nil {
+			return err
+		}
+		if item.Request.SubmissionID == reusedSubmission {
+			// A retry can commit older content without becoming the stream head.
+			// Publish the fresh collection only when that saved payload is stale.
+			current, err := target.Current(ctx, receipt.ContextID)
+			if err != nil {
+				return err
+			}
+			if !current {
+				freshItem := submission.Pending{Request: fresh}
+				if err := box.Save(freshItem); err != nil {
+					return err
+				}
+				pending = append(pending, freshItem)
+			}
+		}
 		if err := box.Remove(item.Request.SubmissionID); err != nil {
 			return err
 		}
@@ -138,8 +198,24 @@ func runSync(ctx context.Context, arguments []string, stdin *os.File, stdout, st
 	progress.show("Ingest · complete")
 	if values.print {
 		fmt.Fprintln(stdout, strings.TrimRight(values.server, "/"))
+	} else if unchanged > 0 {
+		fmt.Fprintf(stdout, "Synced %d observations (%d unchanged)\n", uploaded, unchanged)
 	} else {
-		fmt.Fprintf(stdout, "Synced %d observations\n", len(pending))
+		fmt.Fprintf(stdout, "Synced %d observations\n", uploaded)
 	}
 	return nil
 }
+
+// Session metadata must reach the server even when checkout contents match.
+func syncFingerprint(request ingestion.Request) string {
+	fingerprint := collector.Fingerprint(request)
+	if fingerprint == "" {
+		return ""
+	}
+	data, err := json.Marshal([]interface{}{fingerprint, request.Metadata.Agent, request.Metadata.RunID, request.Metadata.AgentSession})
+	if err != nil {
+		return ""
+	}
+	sum := sha256.Sum256(data)
+	return hex.EncodeToString(sum[:])
+}