From 9c01c8e4e0e603b637dfe72c905e20709e2d3a7e Mon Sep 17 00:00:00 2001 From: Dinesh Pandiyan Date: Sun, 11 Oct 2026 00:16:25 +1100 Subject: [PATCH] fix: collect selected checkout for sync and hooks --- README.md | 9 +- cmd/diffx/hook.go | 2 +- cmd/diffx/producer.go | 48 +--- cmd/diffx/producer_test.go | 25 ++- cmd/diffx/sync.go | 98 ++++++++- cmd/diffx/sync_test.go | 303 +++++++++++++++++++++++++- docs/architecture.md | 1 + docs/development.md | 5 +- docs/plugins.md | 14 +- docs/system.md | 17 +- internal/collector/boundaries_test.go | 62 ++---- internal/collector/discovery.go | 43 +--- internal/submission/outbox.go | 58 ++++- internal/submission/outbox_test.go | 54 +++++ 14 files changed, 563 insertions(+), 176 deletions(-) create mode 100644 internal/submission/outbox_test.go diff --git a/README.md b/README.md index d7c91875..af85ec63 100644 --- a/README.md +++ b/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. diff --git a/cmd/diffx/hook.go b/cmd/diffx/hook.go index f583fa10..1bd046dc 100644 --- a/cmd/diffx/hook.go +++ b/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 } diff --git a/cmd/diffx/producer.go b/cmd/diffx/producer.go index 7559fba2..0f4ea5cf 100644 --- a/cmd/diffx/producer.go +++ b/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...) -} diff --git a/cmd/diffx/producer_test.go b/cmd/diffx/producer_test.go index e0835e6b..c59a5310 100644 --- a/cmd/diffx/producer_test.go +++ b/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") } } diff --git a/cmd/diffx/sync.go b/cmd/diffx/sync.go index e39dc118..5d877481 100644 --- a/cmd/diffx/sync.go +++ b/cmd/diffx/sync.go @@ -6,6 +6,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "encoding/json" "errors" "fmt" "io" @@ -13,6 +14,7 @@ import ( "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[:]) +} diff --git a/cmd/diffx/sync_test.go b/cmd/diffx/sync_test.go index ef7ddc59..bc1f3731 100644 --- a/cmd/diffx/sync_test.go +++ b/cmd/diffx/sync_test.go @@ -2,22 +2,28 @@ package main import ( "bytes" + "crypto/sha256" + "encoding/hex" "io" "net/http" "net/http/httptest" "os" "path/filepath" + "reflect" "strings" "sync/atomic" "testing" + "github.com/flexdinesh/diffx/internal/hooks" "github.com/flexdinesh/diffx/internal/ingestion" "github.com/flexdinesh/diffx/internal/remoteserver" + "github.com/flexdinesh/diffx/internal/review" "github.com/flexdinesh/diffx/internal/reviewstore" + "github.com/flexdinesh/diffx/internal/submission" "github.com/flexdinesh/diffx/internal/webui" ) -func TestSyncPrintDebugCommitsAllWorktreesAndRecoversRemovedCheckout(t *testing.T) { +func TestSyncPrintDebugCommitsSelectedCheckoutAndRecoversRemovedCheckout(t *testing.T) { root := cliRepository(t) linked := filepath.Join(t.TempDir(), "linked") producerGit(t, root, "worktree", "add", "-b", "linked", linked) @@ -77,8 +83,8 @@ func TestSyncPrintDebugCommitsAllWorktreesAndRecoversRemovedCheckout(t *testing. t.Fatal(err) } contexts, err := store.ObservationContexts(owner.ID, 10, 0, "", ingestion.Filter{}) - if err != nil || len(contexts) != 2 { - t.Fatalf("worktree publication: %d %v", len(contexts), err) + if err != nil || len(contexts) != 1 { + t.Fatalf("selected checkout publication: %d %v", len(contexts), err) } if err := run(t.Context(), []string{"sync", "--retry"}, nil, io.Discard, io.Discard); err == nil || !strings.Contains(err.Error(), "no pending") { t.Fatalf("acknowledged uploads remained pending: %v", err) @@ -104,3 +110,294 @@ func TestSyncRequiresRemoteAndConfigMasksToken(t *testing.T) { t.Fatalf("token output: %s", output.String()) } } + +// Exercise the real queue, worker and stored catalog while counting uploads. +type syncFixture struct { + store *reviewstore.Store + ownerID string + server *httptest.Server + uploads atomic.Int64 + online atomic.Bool +} + +func newSyncFixture(t *testing.T) *syncFixture { + t.Helper() + t.Setenv("XDG_STATE_HOME", t.TempDir()) + t.Setenv("DIFFX_CONFIG_PATH", filepath.Join(t.TempDir(), "config.json")) + t.Setenv("DIFFX_RUNTIME_DIR", t.TempDir()) + t.Setenv("DIFFX_SOURCE_ID", "machine") + token := strings.Repeat("b", 40) + t.Setenv("DIFFX_TOKEN", token) + store, err := reviewstore.Open("") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + handler, closeServices, err := remoteserver.Handler(t.Context(), store, "admin", token, webui.Assets()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = closeServices() }) + owner, err := store.AuthenticateToken(token) + if err != nil { + t.Fatal(err) + } + fixture := &syncFixture{store: store, ownerID: owner.ID} + fixture.online.Store(true) + fixture.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !fixture.online.Load() { + http.Error(w, "offline", http.StatusServiceUnavailable) + return + } + if r.Method == http.MethodPost && r.URL.Path == "/api/v2/ingestion-jobs" { + fixture.uploads.Add(1) + } + handler.ServeHTTP(w, r) + })) + t.Cleanup(fixture.server.Close) + t.Setenv("DIFFX_SERVER_URL", fixture.server.URL) + return fixture +} + +func TestSyncAndHooksCollectOneCheckoutAndDeduplicate(t *testing.T) { + root := cliRepository(t) + producerGit(t, root, "branch", "main") + linked := filepath.Join(t.TempDir(), "linked") + producerGit(t, root, "worktree", "add", "-b", "linked", linked) + filename := filepath.Join(linked, "file.txt") + if err := os.WriteFile(filename, []byte("committed\n"), 0600); err != nil { + t.Fatal(err) + } + producerGit(t, linked, "commit", "-am", "feature change") + subdir := filepath.Join(linked, "subdir") + if err := os.Mkdir(subdir, 0700); err != nil { + t.Fatal(err) + } + t.Chdir(subdir) + fixture := newSyncFixture(t) + sync := func(wantUploads int64) string { + t.Helper() + var output bytes.Buffer + if err := run(t.Context(), []string{"sync"}, nil, &output, io.Discard); err != nil { + t.Fatal(err) + } + if got := fixture.uploads.Load(); got != wantUploads { + t.Fatalf("uploads = %d, want %d", got, wantUploads) + } + return output.String() + } + sync(1) + contexts, err := fixture.store.ObservationContexts(fixture.ownerID, 10, 0, "", ingestion.Filter{}) + if err != nil || len(contexts) != 1 { + t.Fatalf("selected checkout contexts: %#v %v", contexts, err) + } + originalID := contexts[0].ID + metadata := contexts[0].Metadata + if metadata == nil || metadata.Root != linked || metadata.Comparison == nil || metadata.Comparison.Kind != "branch" || metadata.Comparison.BaseRef != "refs/heads/main" { + t.Fatalf("selected checkout lost merge-base comparison: %+v", metadata) + } + local, err := collectInitialInput(t.Context(), options{directory: subdir, sourceID: "machine"}, initialInput{Kind: "worktree", Path: subdir}) + if err != nil { + t.Fatal(err) + } + stored, err := fixture.store.ObservationSnapshot(fixture.ownerID, originalID, review.DiffAll) + if err != nil || len(stored.Files) != len(local.Scopes[0].Snapshot.Files) || len(stored.Files) != 1 { + t.Fatalf("local/remote selected checkout diff: %+v %v", stored, err) + } + if output := sync(1); output != "Synced 0 observations (1 unchanged)\n" { + t.Fatalf("unchanged output: %q", output) + } + + // Hook sessions must associate with the shared review, then skip repeats. + hookState := t.TempDir() + hook := func(session string, wantUploads int64) { + t.Helper() + engine := newHookEngine(hookState) + events, err := engine.Expand(t.Context(), hooks.Event{Path: subdir, InputPath: subdir, Agent: "codex", RunID: session, Base: "auto"}) + if err != nil || len(events) != 1 || events[0].Path != linked { + t.Fatalf("hook checkout selection: %+v %v", events, err) + } + job := "" + engine.Launch = func(value string) error { job = value; return nil } + if err := engine.Schedule(events[0]); err != nil { + t.Fatal(err) + } + if err := engine.Run(t.Context(), job); err != nil { + t.Fatal(err) + } + if got := fixture.uploads.Load(); got != wantUploads { + t.Fatalf("hook uploads = %d, want %d", got, wantUploads) + } + } + hook("A", 2) + hook("A", 2) + hook("B", 3) + sync(3) + for _, session := range []string{"A", "B"} { + contexts, err := fixture.store.ObservationContexts(fixture.ownerID, 10, 0, "", ingestion.Filter{Harness: "codex", SessionID: session}) + if err != nil || len(contexts) != 1 || contexts[0].ID != originalID { + t.Fatalf("session %s duplicated or lost review: %+v %v", session, contexts, err) + } + } + + // Changed contents upload; returning to old contents restores its stream head. + if err := os.WriteFile(filename, []byte("working\n"), 0600); err != nil { + t.Fatal(err) + } + sync(4) + if err := os.WriteFile(filename, []byte("committed\n"), 0600); err != nil { + t.Fatal(err) + } + sync(5) + contexts, err = fixture.store.ObservationContexts(fixture.ownerID, 10, 0, "", ingestion.Filter{}) + if err != nil || len(contexts) != 2 { + t.Fatalf("changed content history: %+v %v", contexts, err) + } + for _, context := range contexts { + if context.ID == originalID && context.Stale { + t.Fatal("recollected content stayed stale") + } + } + if err := fixture.store.DeleteContext(fixture.ownerID, originalID); err != nil { + t.Fatal(err) + } + sync(6) + contexts, err = fixture.store.ObservationContexts(fixture.ownerID, 10, 0, "", ingestion.Filter{}) + if err != nil || len(contexts) != 2 { + t.Fatalf("deleted context suppressed fresh collection: %+v %v", contexts, err) + } +} + +func TestSyncReusesPendingPayloadAndRefreshesStaleReplay(t *testing.T) { + root := cliRepository(t) + fixture := newSyncFixture(t) + fixture.online.Store(false) + sync := func() error { + return run(t.Context(), []string{"sync", root}, nil, io.Discard, io.Discard) + } + if err := sync(); err == nil { + t.Fatal("offline sync succeeded") + } + path, err := reviewstore.DefaultPath() + if err != nil { + t.Fatal(err) + } + digest := sha256.Sum256([]byte(os.Getenv("DIFFX_TOKEN"))) + route := fixture.server.URL + ":" + hex.EncodeToString(digest[:]) + pending := func() []submission.Pending { + t.Helper() + box, err := submission.Open(t.Context(), filepath.Dir(path), route) + if err != nil { + t.Fatal(err) + } + defer box.Close() + items, err := box.Pending() + if err != nil || len(items) != 1 { + t.Fatalf("pending duplicate: %+v %v", items, err) + } + return items + } + first := pending()[0] + if err := sync(); err == nil { + t.Fatal("offline repeat succeeded") + } + if second := pending()[0]; !reflect.DeepEqual(first, second) { + t.Fatal("unchanged retry replaced immutable payload") + } + // A newer observation from another producer must not be superseded by an old retry. + if err := os.WriteFile(filepath.Join(root, "file.txt"), []byte("other producer\n"), 0600); err != nil { + t.Fatal(err) + } + other, err := collectSubmission(t.Context(), options{directory: root, sourceID: "machine"}) + if err != nil { + t.Fatal(err) + } + if _, err := fixture.store.Ingest(fixture.ownerID, other); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(root, "file.txt"), []byte("after\n"), 0600); err != nil { + t.Fatal(err) + } + fixture.online.Store(true) + if err := sync(); err != nil { + t.Fatal(err) + } + if fixture.uploads.Load() != 2 { + t.Fatalf("stale retry needed fresh submission: %d uploads", fixture.uploads.Load()) + } + contexts, err := fixture.store.ObservationContexts(fixture.ownerID, 10, 0, "", ingestion.Filter{}) + if err != nil || len(contexts) != 2 { + t.Fatalf("stale retry duplicated history: %+v %v", contexts, err) + } + for _, context := range contexts { + if context.Metadata != nil && context.Metadata.CollectedAt == first.Request.Metadata.CollectedAt && context.Stale { + t.Fatal("matching fresh collection did not advance old replay") + } + } + if err := sync(); err != nil || fixture.uploads.Load() != 2 { + t.Fatalf("acknowledged replay uploaded again: %v, %d uploads", err, fixture.uploads.Load()) + } +} + +func TestSyncRecordsNewSessionMetadataWithoutDuplicatingReviews(t *testing.T) { + root := cliRepository(t) + fixture := newSyncFixture(t) + for _, step := range []struct { + session, name string + uploads int64 + }{ + {"A", "First", 1}, {"A", "First", 1}, + {"B", "Second", 2}, {"B", "Renamed", 3}, {"B", "Renamed", 3}, + } { + if err := run(t.Context(), []string{"sync", root, "--harness", "codex", "--run-id", step.session, "--session-name", step.name}, nil, io.Discard, io.Discard); err != nil { + t.Fatal(err) + } + if got := fixture.uploads.Load(); got != step.uploads { + t.Fatalf("session %s/%s: %d uploads, want %d", step.session, step.name, got, step.uploads) + } + contexts, err := fixture.store.ObservationContexts(fixture.ownerID, 10, 0, "", ingestion.Filter{Harness: "codex", SessionID: step.session, SessionName: step.name}) + if err != nil || len(contexts) != 1 { + t.Fatalf("session association: %+v %v", contexts, err) + } + } + contexts, err := fixture.store.ObservationContexts(fixture.ownerID, 10, 0, "", ingestion.Filter{}) + if err != nil || len(contexts) != 1 { + t.Fatalf("session uploads duplicated snapshot: %+v %v", contexts, err) + } +} + +func TestSyncKeepsForeignPendingPayloadAndPublishesFreshCollection(t *testing.T) { + root := cliRepository(t) + fixture := newSyncFixture(t) + request, err := collectSubmission(t.Context(), options{directory: root, sourceID: "machine", trigger: "manual"}) + if err != nil { + t.Fatal(err) + } + path, err := reviewstore.DefaultPath() + if err != nil { + t.Fatal(err) + } + digest := sha256.Sum256([]byte(os.Getenv("DIFFX_TOKEN"))) + route := fixture.server.URL + ":" + hex.EncodeToString(digest[:]) + box, err := submission.Open(t.Context(), filepath.Dir(path), route) + if err != nil { + t.Fatal(err) + } + if err := box.Save(submission.Pending{Request: request, StateID: "old-database"}); err != nil { + t.Fatal(err) + } + box.Close() + err = run(t.Context(), []string{"sync", root}, nil, io.Discard, io.Discard) + if err == nil || !strings.Contains(err.Error(), "another server database") || fixture.uploads.Load() != 1 { + t.Fatalf("fresh collection blocked or old payload rerouted: %v, %d uploads", err, fixture.uploads.Load()) + } + box, err = submission.Open(t.Context(), filepath.Dir(path), route) + if err != nil { + t.Fatal(err) + } + defer box.Close() + pending, err := box.Pending() + if err != nil || len(pending) != 1 || !reflect.DeepEqual(pending[0].Request, request) || pending[0].StateID != "old-database" { + t.Fatalf("foreign retry mutated or lost: %+v %v", pending, err) + } +} diff --git a/docs/architecture.md b/docs/architecture.md index 4fb7e14d..f50b017c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -19,6 +19,7 @@ canonical record of package boundaries and architectural decisions. Source support and review policy resolve snapshot capabilities. A mode never invents contents, scopes or refresh support. Queue presence at the composition root enables remote admission; no queue means no admission routes. +Local, manual sync and hook producers collect only the selected checkout. Stored observations cannot refresh from Git. Private local control supports authenticated status/shutdown only. Replacement diff --git a/docs/development.md b/docs/development.md index fe911ec4..6c6402fd 100644 --- a/docs/development.md +++ b/docs/development.md @@ -100,8 +100,9 @@ mise run dev:server -- --no-browser ``` `dev --fixture` collects its fixture before starting in the foreground and does -not use personal lifecycle discovery. `sync` collects the originating checkout and all registered -worktrees against the default branch merge base, plus working changes. Use +not use personal lifecycle discovery. `sync` collects only the selected checkout +against the default branch merge base, plus working changes. Hooks collect only +the triggering checkout. Use `--base HEAD` for working changes only, and `--branch` for explicit object-only recovery. `sync` submits to one configured destination and exits. Redirected stdin (`git diff | diffx`) starts a foreground local process with diff --git a/docs/plugins.md b/docs/plugins.md index e93c9ab5..caa7bb20 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -163,8 +163,8 @@ It retains patches and bounded full before/after contents for future full-file viewing. It does not attribute edits to a turn or agent; completion is a trigger, and the worker may capture edits made after that trigger. -Nested hook directories resolve to the checkout root. Discovery includes that -checkout and all registered live worktrees. +Nested hook directories resolve to the checkout root. Collection includes only +that checkout; other registered worktrees are not inspected. Non-Git input is ignored; child repositories and branches without checkouts are excluded. Explicit `sync --branch` can recover committed changes after worktree removal. Discovery never fetches, switches branches or creates worktrees. @@ -237,8 +237,8 @@ diffx sync --path /path/to/repository --branch feature --base main --no-browser diffx sync --path /path/to/checkout --base HEAD --no-browser ``` -Manual sync defaults to auto and collects all registered worktrees, returning -the originating checkout's URL. `--base HEAD` selects working changes only. +Manual sync defaults to auto and collects only the selected checkout. `--print` +returns the configured server URL. `--base HEAD` selects working changes only. `--branch` reads only the named branch's Git objects, defaults its base to auto, and ignores the current checkout's dirty files. Deleted uncommitted files cannot be recovered unless a payload was captured first. Renamed/deleted @@ -272,5 +272,7 @@ printf '%s\n' '{"cwd":"/path/to/checkout","session_id":"session-id","session_nam Hooks default to `--base auto`, never open a browser, and return after scheduling. Scheduling success does not confirm ingestion; hooks log failures and exit successfully. Inspect the dashboard or hook log for delivery state. Manual -`diffx sync` always submits, waits for receipts and reports failures -directly. Manual and hook collection use the same auto baseline. +`diffx sync` waits for receipts and reports failures directly. Both manual sync +and hooks skip unchanged uploads only after acknowledgement and server freshness +checks; new session metadata is submitted. Manual and hook collection use the +same auto baseline. diff --git a/docs/system.md b/docs/system.md index c8522810..8b991dee 100644 --- a/docs/system.md +++ b/docs/system.md @@ -42,11 +42,12 @@ The home-directory database survives process termination. Only one local process owns it; another invocation prompts before authenticated graceful replacement. Noninteractive callers use `--replace`. There is no collector daemon. -`diffx sync` collects the originating checkout and registered worktrees once, +`diffx sync` collects the selected checkout once, saves immutable pending payloads, and submits them over authenticated HTTP. The remote server binary owns durable queue admission, the ingestion worker, shared application services and persistent storage. Sync waits for committed results. -Plugins schedule finite remote-only collector invocations with harness metadata. +Plugins schedule finite remote-only collection of the triggering checkout with +harness metadata. `diffx-server` runs directly or as the same binary inside a container. It resolves deployment config without reading or writing personal CLI settings. Container @@ -63,8 +64,9 @@ and review operations. Their lifecycle and submission delivery differ. Collection resolves the baseline and all/staged/unstaged scopes together with full-content identity and bounded immutable previews. It never fetches refs. -Local collection selects one checkout; remote sync discovers all registered -worktrees. Explicit `--branch` recovery remains available for remote collection. +Local, sync and hook collection select one checkout, resolving nested paths to +its root. Other registered worktrees are not collected. Explicit `--branch` +recovery remains available for remote collection. Local composition calls `contextservice.Ingest` directly. Remote collectors POST to `/api/v2/ingestion-jobs`, receiving 202 with a durable job/status URL. @@ -189,6 +191,13 @@ retry. A lost response can mean the transaction committed; replay resolves that uncertainty without creating a second observation. Validation and conflicting requests fail explicitly rather than being retried as transport errors. +Manual sync suppresses unchanged uploads only with an acknowledgement less than +24 hours old, a matching complete fingerprint and session metadata, the same +server database identity, and a still-current server context. Matching pending +payloads reuse their original submission ID; stale saved collections require a +fresh submission to advance the stream head. Expired or deleted contexts are +recollected rather than suppressed. + Hook workers persist bounded immutable pending payloads and acknowledgements. They retry after subsequent triggers or `collector retry`; no perpetual upload process or guarantee that an agent hook will run exists. Local acknowledgements diff --git a/internal/collector/boundaries_test.go b/internal/collector/boundaries_test.go index 0009f1a7..0ed393cd 100644 --- a/internal/collector/boundaries_test.go +++ b/internal/collector/boundaries_test.go @@ -2,7 +2,6 @@ package collector import ( "errors" - "fmt" "os" "path/filepath" "testing" @@ -124,7 +123,7 @@ func TestComparisonCanonicalRefsAndBaselineChanges(t *testing.T) { } } -func TestWorktreeDiscoveryOriginFirstWithoutWorkspaceOrBranchRecovery(t *testing.T) { +func TestCheckoutDiscoverySelectsOnlyInputWithoutWorkspaceOrBranchRecovery(t *testing.T) { root := branchRepository(t) git(t, root, "branch", "recoverable") linked := filepath.Join(t.TempDir(), "linked") @@ -133,60 +132,25 @@ func TestWorktreeDiscoveryOriginFirstWithoutWorkspaceOrBranchRecovery(t *testing if err := os.Mkdir(subdir, 0o700); err != nil { t.Fatal(err) } - result, err := DiscoverWorktrees(t.Context(), subdir, "machine", "") - if err != nil || len(result.Sources) != 2 || len(result.Diagnostics) != 0 { - t.Fatalf("registered worktrees: %+v %v", result, err) + result, err := DiscoverCheckout(t.Context(), subdir, "machine", "") + if err != nil || len(result.Sources) != 1 || len(result.Diagnostics) != 0 { + t.Fatalf("selected checkout: %+v %v", result, err) } - if result.Sources[0].Path != linked || result.Sources[1].Path != root || result.Sources[0].Identity == result.Sources[1].Identity { - t.Fatalf("origin/worktree identities lost: %+v", result.Sources) + source := result.Sources[0] + if source.Path != linked || source.InputPath != subdir || source.Base != "auto" || source.Branch != "" || source.Identity == "" { + t.Fatalf("selected checkout lost identity or trigger: %+v", source) } - for _, source := range result.Sources { - if source.InputPath != subdir || source.Base != "auto" || source.Branch != "" { - t.Fatalf("implicit branch recovery or lost trigger directory: %+v", source) - } - } - if result, err := DiscoverWorktrees(t.Context(), filepath.Dir(linked), "machine", "auto"); !errors.Is(err, diffsource.ErrNotRepository) || len(result.Sources) != 0 { + if result, err := DiscoverCheckout(t.Context(), filepath.Dir(linked), "machine", "auto"); !errors.Is(err, diffsource.ErrNotRepository) || len(result.Sources) != 0 { t.Fatalf("non-Git workspace silently scanned: %+v %v", result, err) } - git(t, root, "worktree", "remove", "--force", linked) - result, err = DiscoverWorktrees(t.Context(), root, "machine", "HEAD") - if err != nil || len(result.Sources) != 1 || result.Sources[0].Base != "HEAD" { - t.Fatalf("removed worktree recovered by default: %+v %v", result, err) - } -} - -func TestWorktreeDiscoveryReportsUnavailableCheckout(t *testing.T) { - root := branchRepository(t) - linked := filepath.Join(t.TempDir(), "linked") - git(t, root, "worktree", "add", "-b", "other", linked) if err := os.RemoveAll(linked); err != nil { t.Fatal(err) } - result, err := DiscoverWorktrees(t.Context(), root, "machine", "auto") - if err != nil || len(result.Sources) != 1 || len(result.Diagnostics) != 1 { - t.Fatalf("missing checkout needs diagnostic without false source: %+v %v", result, err) + result, err = DiscoverCheckout(t.Context(), root, "machine", "HEAD") + if err != nil || len(result.Sources) != 1 || len(result.Diagnostics) != 0 || result.Sources[0].Path != root || result.Sources[0].Base != "HEAD" { + t.Fatalf("unrelated missing checkout affected selection: %+v %v", result, err) } -} - -func TestWorktreeDiscoveryIncludesEveryRegisteredCheckout(t *testing.T) { - root := branchRepository(t) - worktrees := t.TempDir() - // The registered Git list is finite and must not inherit workspace scanning's - // unrelated safety bound. No checkout files are needed for discovery. - for index := 0; index < discoveryLimit; index++ { - linked := filepath.Join(worktrees, fmt.Sprintf("linked-%03d", index)) - git(t, root, "worktree", "add", "--detach", "--no-checkout", linked) - } - result, err := DiscoverWorktrees(t.Context(), root, "machine", "auto") - if err != nil || len(result.Diagnostics) != 0 || len(result.Sources) != discoveryLimit+1 { - t.Fatalf("registered checkout list truncated: %d sources, diagnostics %v, err %v", len(result.Sources), result.Diagnostics, err) - } - if result.Sources[0].Path != root { - t.Fatalf("origin not first: %+v", result.Sources[0]) - } - for _, source := range result.Sources { - if source.InputPath != root || source.Branch != "" { - t.Fatalf("registered checkout lost trigger directory: %+v", source) - } + if result, err := DiscoverCheckout(t.Context(), linked, "machine", "auto"); err == nil || len(result.Sources) != 0 { + t.Fatalf("unavailable selected checkout accepted: %+v %v", result, err) } } diff --git a/internal/collector/discovery.go b/internal/collector/discovery.go index 3891559f..82b9f4c9 100644 --- a/internal/collector/discovery.go +++ b/internal/collector/discovery.go @@ -53,11 +53,9 @@ func SourceIdentity(ctx context.Context, root, sourceID, branch, base string) (s return hash("collector-source-v1", sourceID, identity.RepositoryKey, checkout, identity.BranchID, base), nil } -// DiscoverWorktrees collects only the input repository's current checkout and -// registered worktrees. Non-Git input returns ErrNotRepository; callers decide -// whether that is an error or an ignored hook event. Recovery and workspace -// scanning belong to the explicit legacy Discover policy below. -func DiscoverWorktrees(ctx context.Context, input, sourceID, base string) (Discovery, error) { +// DiscoverCheckout resolves only the selected live checkout. Non-Git input +// returns ErrNotRepository; callers decide whether to ignore a hook event. +func DiscoverCheckout(ctx context.Context, input, sourceID, base string) (Discovery, error) { var result Discovery input, err := filepath.Abs(input) if err != nil { @@ -70,42 +68,11 @@ func DiscoverWorktrees(ctx context.Context, input, sourceID, base string) (Disco if err != nil { return result, err } - _, worktrees, err := diffsource.RepositoryWorktrees(ctx, source.Root()) + identity, err := SourceIdentity(ctx, source.Root(), sourceID, "", base) if err != nil { return result, err } - // The originating checkout comes first, even when invoked from a subdirectory - // of a linked worktree. It determines the URL returned by manual review. - roots := []string{source.Root()} - for _, worktree := range worktrees { - if worktree.Root != source.Root() { - roots = append(roots, worktree.Root) - } - } - seen := make(map[string]string) - for _, root := range roots { - if err := ctx.Err(); err != nil { - return result, err - } - if _, err := os.Stat(root); err != nil { - result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("worktree unavailable %s: %v", root, err)) - continue - } - identity, err := SourceIdentity(ctx, root, sourceID, "", base) - if err != nil { - if ctx.Err() != nil { - return result, ctx.Err() - } - result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("identify worktree %s: %v", root, err)) - continue - } - if previous, exists := seen[identity]; exists { - result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("collector identity copied across %s and %s; duplicate source skipped, use separate source IDs", previous, root)) - continue - } - seen[identity] = root - result.Sources = append(result.Sources, DiscoverySource{InputPath: input, Path: root, Base: base, Identity: identity}) - } + result.Sources = []DiscoverySource{{InputPath: input, Path: source.Root(), Base: base, Identity: identity}} return result, nil } diff --git a/internal/submission/outbox.go b/internal/submission/outbox.go index f41b7d88..9c57e9e4 100644 --- a/internal/submission/outbox.go +++ b/internal/submission/outbox.go @@ -20,6 +20,15 @@ type Pending struct { StateID string `json:"stateId,omitempty"` } +// Acknowledgement permits upload suppression only while its server context +// remains current. Fingerprints are supplied by the collector composition. +type Acknowledgement struct { + Fingerprint string `json:"fingerprint"` + StateID string `json:"stateId"` + ContextID string `json:"contextId"` + At time.Time `json:"at"` +} + type Outbox struct { directory string lock *processlock.Lock @@ -52,11 +61,54 @@ func (box *Outbox) path(id string) string { return filepath.Join(box.directory, hex.EncodeToString(sum[:])+".json") } func (box *Outbox) Save(pending Pending) error { - raw, err := json.Marshal(pending) + return box.write(box.path(pending.Request.SubmissionID), pending) +} + +func (box *Outbox) acknowledgementPath(request ingestion.Request) string { + metadata := request.Metadata + branch := metadata.BranchID + if branch == "" { + branch = metadata.Branch + } + data, _ := json.Marshal([]string{metadata.SourceID, metadata.RepositoryKey, metadata.CheckoutKey, branch, ingestion.ComparisonPolicy(metadata), metadata.Agent, metadata.RunID}) + sum := sha256.Sum256(data) + return filepath.Join(box.directory, "acknowledgements", hex.EncodeToString(sum[:])+".json") +} + +func (box *Outbox) Acknowledgement(request ingestion.Request) (Acknowledgement, error) { + raw, err := os.ReadFile(box.acknowledgementPath(request)) + if errors.Is(err, os.ErrNotExist) { + return Acknowledgement{}, nil + } if err != nil { + return Acknowledgement{}, err + } + var ack Acknowledgement + if err := json.Unmarshal(raw, &ack); err != nil { + return Acknowledgement{}, err + } + if time.Since(ack.At) >= 24*time.Hour { + return Acknowledgement{}, nil + } + return ack, nil +} + +func (box *Outbox) Acknowledge(request ingestion.Request, fingerprint, stateID, contextID string) error { + if fingerprint == "" { + return nil + } + return box.write(box.acknowledgementPath(request), Acknowledgement{Fingerprint: fingerprint, StateID: stateID, ContextID: contextID, At: time.Now()}) +} + +func (box *Outbox) write(path string, value interface{}) error { + raw, err := json.Marshal(value) + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { return err } - file, err := os.CreateTemp(box.directory, ".pending-*") + file, err := os.CreateTemp(filepath.Dir(path), ".pending-*") if err != nil { return err } @@ -72,7 +124,7 @@ func (box *Outbox) Save(pending Pending) error { if err := file.Close(); err != nil { return err } - return os.Rename(file.Name(), box.path(pending.Request.SubmissionID)) + return os.Rename(file.Name(), path) } func (box *Outbox) Remove(id string) error { return os.Remove(box.path(id)) } func (box *Outbox) Pending() ([]Pending, error) { diff --git a/internal/submission/outbox_test.go b/internal/submission/outbox_test.go new file mode 100644 index 00000000..9271f794 --- /dev/null +++ b/internal/submission/outbox_test.go @@ -0,0 +1,54 @@ +package submission + +import ( + "testing" + "time" + + "github.com/flexdinesh/diffx/internal/ingestion" +) + +func TestAcknowledgementExpiryAndStreamIsolation(t *testing.T) { + root := t.TempDir() + box, err := Open(t.Context(), root, "route") + if err != nil { + t.Fatal(err) + } + request := ingestion.Request{Metadata: ingestion.Metadata{SourceID: "source", RepositoryKey: "repo", CheckoutKey: "checkout", BranchID: "branch", Agent: "codex", RunID: "A"}} + if err := box.Acknowledge(request, "fingerprint", "database", "context"); err != nil { + t.Fatal(err) + } + box.Close() + box, err = Open(t.Context(), root, "route") + if err != nil { + t.Fatal(err) + } + defer box.Close() + ack, err := box.Acknowledgement(request) + if err != nil || ack.Fingerprint != "fingerprint" || ack.StateID != "database" || ack.ContextID != "context" { + t.Fatalf("durable acknowledgement: %+v %v", ack, err) + } + for _, mutate := range []func(*ingestion.Metadata){ + func(m *ingestion.Metadata) { m.SourceID = "other" }, + func(m *ingestion.Metadata) { m.CheckoutKey = "other" }, + func(m *ingestion.Metadata) { m.BranchID = "other" }, + func(m *ingestion.Metadata) { m.RunID = "B" }, + func(m *ingestion.Metadata) { m.Agent = "claude" }, + func(m *ingestion.Metadata) { m.Comparison = &ingestion.Comparison{Kind: "branch", BaseRef: "main"} }, + } { + other := request + mutate(&other.Metadata) + if ack, err := box.Acknowledgement(other); err != nil || ack != (Acknowledgement{}) { + t.Fatalf("acknowledgement leaked across stream/session: %+v %v", ack, err) + } + } + ack.At = time.Now().Add(-24*time.Hour - time.Minute) + if err := box.write(box.acknowledgementPath(request), ack); err != nil { + t.Fatal(err) + } + if ack, err := box.Acknowledgement(request); err != nil || ack != (Acknowledgement{}) { + t.Fatalf("expired acknowledgement suppressed retention refresh: %+v %v", ack, err) + } + if items, err := box.Pending(); err != nil || len(items) != 0 { + t.Fatalf("acknowledgements parsed as pending uploads: %+v %v", items, err) + } +}