diff --git a/internal/models/item.go b/internal/models/item.go index 0ce879147..95f33c65f 100644 --- a/internal/models/item.go +++ b/internal/models/item.go @@ -1334,6 +1334,16 @@ type ItemUpdate struct { // bracketing version. Internal-only (`json:"-"`): the store honours it, but // no HTTP client can set it. Only consulted when content actually changes. ForceVersion bool `json:"-"` + // ExternalContent is the body an update sends to the item's LIVE + // collaborative document instead of to items.content: the designated- + // applier path, taken while a tab has the item open (BUG-3327). The row + // write runs with Content nil, so without this the store never reached + // its version write and the edit was in no version. The store versions + // ExternalContent exactly as it would Content (same throttle, attribution + // and reverse patch of the body before the edit) and never writes it to + // the row; the tab's next flush does. Internal-only, and ignored when + // Content is set. + ExternalContent *string `json:"-"` // ActorUserID is the request's user, stamped on a version row this update // writes (PLAN-2348 U2). Internal-only; empty for system writers. ActorUserID string `json:"-"` diff --git a/internal/server/bug3327_applier_version_test.go b/internal/server/bug3327_applier_version_test.go new file mode 100644 index 000000000..e3a3ce2a0 --- /dev/null +++ b/internal/server/bug3327_applier_version_test.go @@ -0,0 +1,204 @@ +package server + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// BUG-3327: a body edit that takes the designated-applier path (a tab has the +// item open) wrote NO version row: the row write runs with Content nil, and +// the store's version write lives only on the content path. An agent's edit +// made while someone had the item open was therefore missing from History. +// It must leave a version of the body it replaced, attributed to the writer, +// resolvable BEFORE the tab's flush lands in items.content. +func TestBUG3327_ApplierPathBodyEditLeavesAVersion(t *testing.T) { + srv := testServerWithCollab(t) + ts := httptest.NewServer(srv) + t.Cleanup(ts.Close) + slug := createWSWithCollections(t, srv) + item := createTaskWithFields(t, srv, slug, "Item", `{"status":"open"}`) + + // Long and nearly identical bodies, so a reverse patch is smaller than + // the full body and the store would choose it: the mutant this test must + // catch is the applier path writing a patch against a stale row. + shared := strings.Repeat("A paragraph that both bodies share, line after line.\n", 40) + previous := shared + "PREVIOUS ending, written before any tab connected" + next := shared + "NEW ending, sent by an agent through the applier" + if rr := doRequest(srv, "PATCH", "/api/v1/workspaces/"+slug+"/items/"+item.Slug, + map[string]interface{}{"content": previous}); rr.Code != http.StatusOK { + t.Fatalf("seed PATCH: %d %s", rr.Code, rr.Body.String()) + } + + conn, resp, err := dialCollab(t, ts.URL, item.ID, nil, "") + if err != nil { + status := "" + if resp != nil { + status = resp.Status + } + t.Fatalf("dialCollab: %v (%s)", err, status) + } + stop := applierEcho(t, conn) + t.Cleanup(stop) + waitForApplierPath(t, srv, slug, item.Slug, item.ID) + + before, err := srv.store.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + + // An AGENT's edit while the user's tab is open: the case History lost. + rr := doRequestWithHeaders(srv, "PATCH", "/api/v1/workspaces/"+slug+"/items/"+item.Slug, + map[string]interface{}{"content": next}, + map[string]string{"X-Pad-Agent": "test-agent"}) + if rr.Code != http.StatusOK { + t.Fatalf("applier-path PATCH: %d %s", rr.Code, rr.Body.String()) + } + if !strings.Contains(rr.Body.String(), "applied_pending_flush") { + t.Fatalf("the PATCH did not take the applier path (no content_outcome): %s", rr.Body.String()) + } + + after, err := srv.store.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + if len(after) != len(before)+1 { + t.Fatalf("versions %d -> %d: the applier-path edit left no version row", len(before), len(after)) + } + + // Resolve against the row as it stands NOW, before any flush: the row + // still holds `previous`, and the new version must still read as it. + row, err := srv.store.GetItem(item.ID) + if err != nil { + t.Fatal(err) + } + resolved, err := srv.store.ListItemVersionsResolved(item.ID, row.Content) + if err != nil { + t.Fatal(err) + } + newest := resolved[0] + if newest.Content != previous { + t.Errorf("newest version resolves to %q, want the replaced body %q", newest.Content, previous) + } + if newest.CreatedBy != "agent" { + t.Errorf("newest version created_by = %q, want agent", newest.CreatedBy) + } + + // The tab's flush lands the new body in the row (codex review). It must + // not store the replaced body a second time, which would put the agent's + // change in History under the tab user's row; and the chain must still + // resolve, with the agent's row's change reading previous -> next. + if rr := doRequest(srv, "PATCH", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"?source=collab-snapshot", + map[string]interface{}{"content": next}); rr.Code != http.StatusOK { + t.Fatalf("flush PATCH: %d %s", rr.Code, rr.Body.String()) + } + flushed, err := srv.store.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + if len(flushed) != len(after) { + t.Errorf("versions %d -> %d across the flush: the replaced body was stored twice", len(after), len(flushed)) + } + row, err = srv.store.GetItem(item.ID) + if err != nil { + t.Fatal(err) + } + if row.Content != next { + t.Fatalf("flush did not land: row = %q", row.Content) + } + resolved, err = srv.store.ListItemVersionsResolved(item.ID, row.Content) + if err != nil { + t.Fatal(err) + } + if resolved[0].Content != previous || resolved[0].CreatedBy != "agent" { + t.Errorf("after the flush, newest version = %q by %q, want the replaced body by agent", resolved[0].Content, resolved[0].CreatedBy) + } + d, err := srv.store.GetItemVersionDiff(item.ID, resolved[0].ID, row.Content) + if err != nil { + t.Fatal(err) + } + if d.Before != previous || d.After != next { + t.Errorf("the agent's version records %q -> %q, want previous -> next", d.Before, d.After) + } + // Older versions still resolve: the seed write's version is the body + // before `previous`, which was empty. + if last := resolved[len(resolved)-1]; last.Content != "" { + t.Errorf("oldest version resolves to %q, want the empty initial body", last.Content) + } +} + +// The flush dedupe is limited to flushes (codex review, round 2): a user's +// A->B (versioned) then B->A (throttled, same writer) leaves the newest row +// holding A, and a DIFFERENT writer's A->C must still get its own row rather +// than have its change read as the first writer's. +func TestBUG3327_DedupeSkipsOnlyFlushes(t *testing.T) { + srv := testServer(t) + slug := createWSWithCollections(t, srv) + item := createTaskWithFields(t, srv, slug, "Item", `{"status":"open"}`) + patch := func(content string, headers map[string]string) { + t.Helper() + rr := doRequestWithHeaders(srv, "PATCH", "/api/v1/workspaces/"+slug+"/items/"+item.Slug, + map[string]interface{}{"content": content}, headers) + if rr.Code != http.StatusOK { + t.Fatalf("PATCH %q: %d %s", content, rr.Code, rr.Body.String()) + } + } + agent := map[string]string{"X-Pad-Agent": "test-agent"} + patch("body A", nil) // the user writes A + patch("body B", agent) // a different writer: versions A as a full body + patch("body A", agent) // same writer within the hour: throttled, no row + before, err := srv.store.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + if len(before) == 0 || before[0].IsDiff || before[0].Content != "body A" { + t.Fatalf("fixture: the newest version must be a full-body \"body A\", got %+v", before) + } + patch("body C", nil) // the user again: a different writer from the newest row + after, err := srv.store.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + if len(after) != len(before)+1 { + t.Fatalf("versions %d -> %d: the user's A->C was deduped against the agent's row", len(before), len(after)) + } +} + +// Codex review, round 3: an ORDINARY flush (no applier edit before it) after a +// throttled revert to a versioned body must still version. Only an +// applier-path row, which postdates the item's last direct content write, may +// be treated as the flush's duplicate. +func TestBUG3327_FlushAfterThrottledRevertStillVersions(t *testing.T) { + srv := testServer(t) + slug := createWSWithCollections(t, srv) + item := createTaskWithFields(t, srv, slug, "Item", `{"status":"open"}`) + patch := func(path, content string, headers map[string]string) { + t.Helper() + rr := doRequestWithHeaders(srv, "PATCH", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+path, + map[string]interface{}{"content": content}, headers) + if rr.Code != http.StatusOK { + t.Fatalf("PATCH %q: %d %s", content, rr.Code, rr.Body.String()) + } + } + agent := map[string]string{"X-Pad-Agent": "test-agent"} + patch("", "body A", nil) + patch("", "body B", agent) // versions A as a full body + // The revert must land in a LATER second than A's version, or this is the + // documented same-second residual rather than the case under test. + time.Sleep(1100 * time.Millisecond) + patch("", "body A", agent) // throttled: no row, but content_flushed_at moves + before, err := srv.store.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + patch("?source=collab-snapshot", "body C", nil) // an ordinary tab flush + after, err := srv.store.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + if len(after) != len(before)+1 { + t.Fatalf("versions %d -> %d: an ordinary flush was treated as an applier duplicate", len(before), len(after)) + } +} diff --git a/internal/server/handlers_items_content_route.go b/internal/server/handlers_items_content_route.go index 90a843e81..38c2bfc04 100644 --- a/internal/server/handlers_items_content_route.go +++ b/internal/server/handlers_items_content_route.go @@ -282,6 +282,12 @@ func (s *Server) applierFirstWrite( // write is gated on Content != nil, so a nil here means the row write touches // neither items.content nor the version chain's content bracket. rowInput.Content = nil + // ...but the edit still owes a version of the body it replaces (BUG-3327): + // it used to write none, so an agent's edit made while a tab had the item + // open was missing from History. ExternalContent versions it in this same + // row write's transaction, before the live document changes, and is never + // written to the row. + rowInput.ExternalContent = &content updated, uerr := s.store.UpdateItemWithParentLink(item.ID, rowInput, openChildrenPrecheck, parentLink) if uerr != nil { diff --git a/internal/store/bug3327_flush_dedupe_test.go b/internal/store/bug3327_flush_dedupe_test.go new file mode 100644 index 000000000..ec5fe37a4 --- /dev/null +++ b/internal/store/bug3327_flush_dedupe_test.go @@ -0,0 +1,60 @@ +package store + +import ( + "testing" + + "github.com/PerpetualSoftware/pad/internal/models" +) + +// BUG-3327: a collab-snapshot flush skips the version of a body the newest +// version already holds ONLY when that newest row is recognised as the +// applier edit's: content_flushed_at present and not newer than it. A NULL +// content_flushed_at (migration 052 leaves it so on items with op-log rows) +// is no evidence, so the flush versions (codex review, round 4). +func TestBUG3327_FlushDedupeNeedsTheFlushTimestamp(t *testing.T) { + for _, tc := range []struct { + name string + nullFlushed bool + wantDelta int + }{ + {"applier row recognised: flush skips the duplicate", false, 0}, + {"no content_flushed_at: no evidence, flush versions", true, 1}, + } { + t.Run(tc.name, func(t *testing.T) { + s := testStore(t) + ws := createTestWorkspace(t, s, "Dedupe") + coll := createTestCollection(t, s, ws.ID, "Tasks") + item := createTestItem(t, s, ws.ID, coll.ID, "Subject", "body A") + + // The applier-path edit: versions body A as a full body, the row + // keeps body A. + next := "body B" + if _, err := s.UpdateItemWithParentLink(item.ID, models.ItemUpdate{ + ExternalContent: &next, LastModifiedBy: "agent", VersionSource: "cli", + }, nil, nil); err != nil { + t.Fatal(err) + } + if tc.nullFlushed { + if _, err := s.db.Exec(s.q(`UPDATE items SET content_flushed_at = NULL WHERE id = ?`), item.ID); err != nil { + t.Fatal(err) + } + } + before, err := s.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + if _, err := s.UpdateItemWithParentLink(item.ID, models.ItemUpdate{ + Content: &next, LastModifiedBy: "user", VersionSource: "collab-snapshot", + }, nil, nil); err != nil { + t.Fatal(err) + } + after, err := s.ListItemVersions(item.ID) + if err != nil { + t.Fatal(err) + } + if got := len(after) - len(before); got != tc.wantDelta { + t.Fatalf("flush added %d version rows, want %d", got, tc.wantDelta) + } + }) + } +} diff --git a/internal/store/items.go b/internal/store/items.go index 08d008c92..52d11bbbb 100644 --- a/internal/store/items.go +++ b/internal/store/items.go @@ -2789,9 +2789,16 @@ func (s *Store) updateItemWithParentLinkOnce( ts := now() - // Create version if content is changing + // Create version if content is changing. The new body is Content, or on + // the applier path ExternalContent, which goes to the live document + // rather than the row (BUG-3327): either way the body before the edit + // is what a version must keep. bodyEditedWithoutVersion := false - if input.Content != nil && *input.Content != existing.Content { + newBody := input.Content + if newBody == nil { + newBody = input.ExternalContent + } + if newBody != nil && *newBody != existing.Content { createdBy := input.LastModifiedBy if createdBy == "" { createdBy = "user" @@ -2828,14 +2835,36 @@ func (s *Store) updateItemWithParentLinkOnce( if err != nil { return nil, fmt.Errorf("check version throttle: %w", err) } + // BUG-3327: an applier-path edit already stored this exact body as + // a full-body version, and the tab's flush that follows would store + // it a second time. That second row would then carry the agent's + // change in History under the tab user's name, leaving the agent's + // own row as an empty change. So a FLUSH (collab-snapshot) skips a + // version whose body the newest version already holds verbatim. + // Only a flush: any other writer's edit owes its own row even when + // an earlier throttled edit brought the body back to a versioned + // one (codex review), or that writer's change would read as the + // earlier writer's. + if shouldVersion && source == "collab-snapshot" { + dup, derr := s.newestVersionIsUnflushedApplierRow(tx, id, existing.Content) + if derr != nil { + return nil, fmt.Errorf("check duplicate version: %w", derr) + } + shouldVersion = !dup + } } if shouldVersion { vid := newID() versionContent := existing.Content isDiff := false - patch := diff.CreateReversePatch(existing.Content, *input.Content) - if diff.IsDiffSmaller(patch, existing.Content) { + patch := diff.CreateReversePatch(existing.Content, *newBody) + // A reverse patch is resolved backwards from the ROW's body, which + // on the applier path (ExternalContent, BUG-3327) is still the old + // body until the tab's flush lands: a patch would resolve against + // the wrong base in that window. The full body resolves the same + // whatever the row holds. + if input.Content != nil && diff.IsDiffSmaller(patch, existing.Content) { versionContent = patch isDiff = true } @@ -2847,7 +2876,7 @@ func (s *Store) updateItemWithParentLinkOnce( // under the item lock (this runs in the update's own tx). // PLAN-2348 U2: an update row holds the body BEFORE its edit, so // its recorded change is that body → the one replacing it. - added, removed := diff.LineCounts(existing.Content, *input.Content) + added, removed := diff.LineCounts(existing.Content, *newBody) _, err = tx.Exec(s.q(` INSERT INTO item_versions (id, item_id, content, change_summary, created_by, source, is_diff, created_at, version_seq, user_id, lines_added, lines_removed, is_create) @@ -5597,6 +5626,53 @@ func buildItemSort(sort string, dialect Dialect) string { // and the pool test's item leg was a TITLE-only update, which never reaches // this branch. A test that exercises one arm of an optional path is not // evidence about the other arm. +// newestVersionIsUnflushedApplierRow reports whether the item's newest +// version is the row an applier-path edit wrote (BUG-3327) for exactly +// body, still waiting for the tab's flush to land the new body in the row. +// Such a row is recognised positively, not by its content alone (codex +// review): it is a full-body, non-create row, and it is NEWER than the +// item's content_flushed_at. Every direct content write bumps +// content_flushed_at in the same write as its version, so only an +// applier-path version (whose row write leaves content alone) can postdate +// it. A throttled revert to an earlier versioned body moves +// content_flushed_at past that version, so it does not match (but see the +// same-second residual below). +func (s *Store) newestVersionIsUnflushedApplierRow(q rowQueryer, itemID, body string) (bool, error) { + var content, createdAt string + var flushedAt sql.NullString + var isDiff, isCreate bool + err := q.QueryRow(s.q(` + SELECT v.content, v.is_diff, v.is_create, v.created_at, i.content_flushed_at + FROM item_versions v + JOIN items i ON i.id = v.item_id + WHERE v.item_id = ? + ORDER BY v.created_at DESC, v.version_seq DESC + LIMIT 1 + `), itemID).Scan(&content, &isDiff, &isCreate, &createdAt, &flushedAt) + if err == sql.ErrNoRows { + return false, nil + } + if err != nil { + return false, err + } + if isDiff || isCreate || content != body { + return false, nil + } + // No timestamp is no evidence (codex review): migration 052 leaves + // content_flushed_at NULL on items with op-log rows, and treating that as + // "unflushed" would let an ordinary flush lose its version. Without it the + // skip does not apply, and the cost is a duplicate row, never a lost one. + if !flushedAt.Valid || flushedAt.String == "" { + return false, nil + } + // Not-before rather than after: both are stored at one-second + // resolution, and an applier edit routinely lands in the same second as + // the last direct write. The residual is a versioned A->B and a throttled + // revert B->A inside ONE second, then a flush: that flush's version is + // skipped and its change reads under A->B's row. + return !parseTime(createdAt).Before(parseTime(flushedAt.String)), nil +} + func (s *Store) shouldCreateItemVersion(q rowQueryer, itemID, actor, source string) (bool, error) { var createdBy, src, createdAt string err := q.QueryRow(s.q(`