merge: feat/phase1 into task/9

docs/ARCHITECTURE.md

@@ -16,7 +16,7 @@ user layer or files a Changes-inbox event.
 |---|---|---|
 | `apps/server` | `Bun.serve`: `/api/v1`, `/ingest/<plugin>`, job worker, static web | planned |
 | `apps/web` | React app (HTML imports) | planned |
-| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`), and `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`), `jobs` (queue, worker, scheduler), `registry` (`loadPlugins`), `connections` (CRUD, plugin context, `syncHandler`) | exists (partial) |
+| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`), `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`), `jobs` (queue, worker, scheduler), `registry` (`loadPlugins`), `connections` (CRUD, plugin context, `syncHandler`) | exists (partial) |
 | `packages/db` | Kysely over one Bun.SQL dialect (SQLite + Postgres), ids, JSON helpers, migrations, `forEachDialect` test harness | exists (schema pending) |
 | `packages/plugin-sdk` | plugin contract, `SourceRecord`, http/rate-limit helpers, fixture test harness | planned |
 | `plugins/<id>` | one source integration each (hardcover, trakt, mangabaka, listenbrainz, lastfm, csv) | planned |
@@ -32,8 +32,9 @@ Sync lifecycle (`packages/core/src/sync/`; steps 1–2 are `jobs.ts` and `connec
    - Every other record goes through `applyRecord`: hash its normalised form; same hash as
      `source_records.current_hash` → only `last_seen` moves. Otherwise a new `source_record_versions` row
      (gzipped `{normalized, raw}`) is written, and:
-     - **new record**: `matchWork` (own transaction), then in one apply transaction create/patch the entry or
+     - **new record**: `matchWork` (own transaction, result committed on the record row so retries reuse it),
-       engagement (de-duplicating plays across sources) and write an `added` `auto_applied` event;
+       then in one apply transaction create/patch the entry, engagement (de-duplicating plays across sources)
+       or list, and write an `added` `auto_applied` event;
      - **changed record**: fields that differ from the previous version are decided against `field_sources`
        → applied (`changed` `auto_applied`) or filed as `pending` `changed`/`conflict` events. Mapped records
        are updated through `source_records.target_*`, never re-matched;

docs/DECISIONS.md

@@ -41,6 +41,76 @@ No index: `grep -n '^## ' docs/DECISIONS.md` lists every entry (date, title, lin
   within 7 days; else `changes`. A failed run is rethrown so the worker retries it.
 **Consequences:** Reload re-imports plugin modules with `?v=<Date.now()>`; old module instances are not freed by Bun.
 
+
+## 2026-10-30: Sync engine: hash composition, anti-flip-flop, no revive, root ancestor, dedupe
+**Status:** accepted
+**Context:** Task 8 built the change engine (`packages/core/src/sync/`). Spec §8.3 fixes the outcomes; these
+are the mechanics a maintainer would otherwise reverse-engineer.
+**Decision:**
+- **Hash composition.** sha256 of the stable-key-ordered JSON of `{kind, work, entry|engagement|listen|list,
+  deleted, mergedIntoKey}`. `key` and `raw` are excluded; `deleted` is always a boolean and `mergedIntoKey`
+  always present (null), so an omitted flag and an explicit `false` hash alike. A version row stores
+  base64(`Bun.gzipSync`) of `{normalized, raw}`: the normalised record is what the next change is diffed
+  against (field-level diff = previous version vs. incoming), the raw payload is kept for the record.
+- **Per-field decision.** Only fields that differ from the previous version are considered, and of those
+  only values that differ from the current STTracker value. `field_sources[f]` unset or this connection's
+  actor → applied; `"user"` → `pending changed {field, from: current, to: incoming}`; another source →
+  `pending conflict {field, current, currentSource, incoming, incomingSource}`. A new record landing on an
+  entry that already exists goes through the same per-field decision, so a first sync never overwrites
+  what the user or another source wrote. Unchanged values are never sent, so a plugin never takes
+  ownership of a field it did not change.
+- **Anti-flip-flop and superseding.** A pending `changed`/`conflict` for the same (record, field, incoming
+  value) is never filed twice. A different newer value for the same (record, field) resolves the older
+  pending event as `ignored` with `payload.supersededBy = <new event id>`, so at most one is pending per
+  field. If the source comes round to the value STTracker already holds, its pending events for that field
+  are resolved `ignored` with `payload.reason = "source_agrees"`. `keep_mine` re-stamps
+  `field_sources[f] = "user"`, so later values from any source become a user-decided `changed`, never
+  another source-vs-source conflict.
+- **No revive.** If the mapped entry/engagement is soft-deleted, the record is still snapshotted but nothing
+  is applied and no event is filed; a new record on a work whose entry the user deleted adds nothing.
+  Only the user revives.
+- **Root ancestor.** Creating an entry on a child work also creates a `status "none"` entry (plugin actor,
+  empty `field_sources`) on the root ancestor work, unless the user has one, live or deleted.
+- **Mapped records are never re-matched.** Only new records go through `matchWork`. Its result is committed
+  at once as the record's row (`current_hash ""`, `target_entity "work"`), so a failed apply is retried
+  against the same work and an ID-less hint never leaks a second draft. A changed work hint refreshes the
+  mapped work via `updateWorkFromHint` (shared works only). `moveEntryToWork` re-points `source_records`
+  from a merged-away entry to the survivor in the same transaction, so a user's merge never orphans a source.
+- **Engagement dedupe.** Among live engagements on the entry not already sourced by this connection:
+  finish dates match when both are datetimes ≤ 24 h apart as instants, or when both are day precision and
+  their dates (first 10 chars, computed with `Date.UTC`, no timezone shift) are ≤ 1 day apart; or start
+  datetimes are the same instant. A match attaches an `engagement_sources` row and fills fields that are
+  empty and that no source has set. Every other differing field goes through the per-field decision
+  (so a disagreeing rating is a `conflict`), except the date that identified the play, whose difference is
+  within the tolerance by definition.
+- **Removal.** Absence counts only after a `full` pull whose iteration finished and whose records all
+  applied without error; any record error fails the run and skips removal. A removed record that comes
+  back clears `deleted_at` and retires its still-pending `removed` event as `ignored`. A merged-away record
+  (`merged_into_key` set, or a pending `merged` event) is expected to vanish: no removal, ever. Accepting
+  `removed` on an engagement other sources still report only deletes this connection's
+  `engagement_sources` row; the engagement is soft-deleted when no source remains. Run counts include only
+  records whose transaction committed.
+- **Listens skip snapshots.** Listen records are not stored as `source_records` (they are high-volume and
+  append-only). `listens` insert-or-ignore is the idempotence; they are never removed. Per batch of 500:
+  one lookup of existing keys, each distinct recording resolved once, one multi-row insert.
+- **Lists.** A new `list` record creates a user list (kind `list`, description `Imported from <plugin id>`),
+  items matched outside the transaction (an ID-less item first reuses the user's own private work with the
+  same media type and title). Lists have no `field_sources`; ownership is derived from `user_history`:
+  `name`/`description` belong to whoever last changed them, `items` to the user once any user edit exists.
+  A user-owned field becomes a pending `changed` (for items, the whole incoming item set); otherwise the
+  fields are applied and items replaced (`replaceListItems`, one history row).
+- **Merge accept.** The target work is the merge target record's mapped entry's work, else `matchWork` on
+  its latest hint. Admin-merge events (`toWorkId`, no connection) resolve through the same `resolveChange`.
+- **Gone targets.** Accept/keep_mine on an event whose target is deleted (or merged away with no live
+  survivor, found through the user's work alias) resolves it `ignored` with `payload.reason = "target_gone"`.
+
+**Alternatives:**
+- Diffing against the live entry instead of the previous version would re-file decided conflicts each run.
+- Hashing `raw` would turn every cosmetic API change into a new version.
+- Keeping one run "ok" despite failing records risks marking those records removed.
+
+**Consequences:** Concurrent runs of one connection are not locked here (Task 9's worker; `docs/TODO.md`).
+
 ## 2026-10-30: Export format `sttracker-export/v1`: per-user zip, camelCase JSONL, deleted rows included, no secrets
 **Status:** accepted
 **Context:** Task 15 (`packages/core/src/{export,backup}.ts`) fixed the on-disk bundle format that backups and restore share.
@@ -67,54 +137,6 @@ No index: `grep -n '^## ' docs/DECISIONS.md` lists every entry (date, title, lin
   are built in `<stamp>.tmp` then renamed (modes 0700/0600), and prune to the newest `keep` (default 14).
 **Consequences:** Any change to these files after release needs `v2` and a migration path for old bundles.
 
-## 2026-10-30: Sync engine: hash composition, anti-flip-flop, no revive, root ancestor, dedupe
-**Status:** accepted
-**Context:** Task 8 built the change engine (`packages/core/src/sync/`). Spec §8.3 fixes the outcomes; these
-are the mechanics a maintainer would otherwise reverse-engineer.
-**Decision:**
-- **Hash composition.** sha256 of the stable-key-ordered JSON of `{kind, work, entry|engagement|listen|list,
-  deleted, mergedIntoKey}`. `key` and `raw` are excluded; `deleted` is always a boolean and `mergedIntoKey`
-  always present (null), so an omitted flag and an explicit `false` hash alike. A version row stores
-  base64(`Bun.gzipSync`) of `{normalized, raw}`: the normalised record is what the next change is diffed
-  against (field-level diff = previous version vs. incoming), the raw payload is kept for the record.
-- **Per-field decision.** Only fields that differ from the previous version are considered, and of those
-  only values that differ from the current STTracker value. `field_sources[f]` unset or this connection's
-  actor → applied; `"user"` → `pending changed {field, from: current, to: incoming}`; another source →
-  `pending conflict {field, current, currentSource, incoming, incomingSource}`. A new record landing on an
-  entry that already exists goes through the same per-field decision, so a first sync never overwrites
-  what the user or another source wrote. Unchanged values are never sent, so a plugin never takes
-  ownership of a field it did not change.
-- **Anti-flip-flop.** A pending `changed`/`conflict` for the same (record, field, incoming value) is never
-  filed twice. `keep_mine` re-stamps `field_sources[f] = "user"`, so later values from any source become a
-  user-decided `changed`, never another source-vs-source conflict.
-- **No revive.** If the mapped entry/engagement is soft-deleted, the record is still snapshotted but nothing
-  is applied and no event is filed; a new record on a work whose entry the user deleted adds nothing.
-  Only the user revives.
-- **Root ancestor.** Creating an entry on a child work also creates a `status "none"` entry (plugin actor,
-  empty `field_sources`) on the root ancestor work, unless the user has one, live or deleted.
-- **Mapped records are never re-matched.** Only new records go through `matchWork`. A changed work hint
-  refreshes the mapped work via `updateWorkFromHint` (shared works only).
-- **Engagement dedupe.** Among live engagements on the entry not already sourced by this connection:
-  finish dates match when both are datetimes ≤ 24 h apart as instants, or when both are day precision and
-  their dates (first 10 chars, computed with `Date.UTC`, no timezone shift) are ≤ 1 day apart; or start
-  datetimes are the same instant. A match attaches an `engagement_sources` row and fills only fields that
-  are empty and that no source has set; the first source's values otherwise stand.
-- **Removal.** Absence counts only after a `full` pull whose iteration finished and whose records all
-  applied without error; any record error fails the run and skips removal. A removed record that comes
-  back clears `deleted_at` and retires its still-pending `removed` event as `ignored`.
-- **Listens skip snapshots.** Listen records are not stored as `source_records` (they are high-volume and
-  append-only). `listens` insert-or-ignore is the idempotence; they are never removed.
-- **Merge accept.** The target work is the merge target record's mapped entry's work, else `matchWork` on
-  its latest hint. Admin-merge events (`toWorkId`, no connection) resolve through the same `resolveChange`.
-
-**Alternatives:**
-- Diffing against the live entry instead of the previous version would re-file decided conflicts each run.
-- Hashing `raw` would turn every cosmetic API change into a new version.
-- Keeping one run "ok" despite failing records risks marking those records removed.
-
-**Consequences:** `list` records are snapshotted but not applied, and removal of a de-duplicated engagement
-deletes it even if another source still reports it (both in `docs/TODO.md`).
-
 ## 2026-10-30: User layer: per-user moves merge entries, keep aliases, and roll the library up by hierarchy
 **Status:** accepted
 **Context:** Task 6 built the user layer (`packages/core/src/{entries,engagements,history,lists,copies}.ts`).

docs/TODO.md

@@ -27,20 +27,13 @@ Small, specific, independently actionable. Remove only when actually done.
 - [ ] Plugins: reload re-imports entry files only; edits to a plugin's other modules need a server restart.
 - [ ] Plugins: rate limiting is per run (each run builds its own client), not per plugin (spec §11); share a limiter per plugin in Phase 2.
 - [ ] Plugins: `ctx.log` reaches `source_runs.log` only after the run ends, so a crash mid-run loses it.
+- [ ] Sync: two concurrent runs of one connection outside the job queue (e.g. a manual run) can still race on `source_records`.
+  Queued syncs are serialised by `lockKey`.
 - [ ] Restore: a new child work whose parent+position collides with an existing child of a re-matched parent fails
   with `ImportError("conflict")` instead of being merged or repositioned.
 - [ ] Restore: an unlink of identical attachment bytes between the pre-write and the commit could remove the file;
   mitigated by re-writing after commit, not locked.
 - [ ] Export bundles hold attachment bytes in memory; very large libraries need a streaming zip.
-- [ ] Sync: `list` records are snapshotted but not applied to `lists` (Trakt/Hardcover emit them). Decide
-  the mapping (one list per record, items by matched work) and apply it through the change engine.
-- [ ] Sync: accepting a `removed` event on an engagement de-duplicated across sources soft-deletes it even if
-  another connection still reports it; consider detaching that connection's `engagement_sources` row instead.
-- [ ] Sync: pending `changed`/`conflict` events for a value the source has since moved past stay pending;
-  consider superseding them when a newer value for the same field arrives.
-- [ ] Sync: two concurrent runs of the same connection (outside the job queue, e.g. a manual run) can both insert a new
-  `source_records` row for one key. Queued syncs are serialised by `lockKey`.
-- [ ] Sync: `ingestListens` does one lookup and one insert per listen; batch it for large backfills.
 - [ ] Admin merge does not merge colliding children recursively; colliding children stay under the tombstoned parent.
 - [ ] Matching: a shared parent resolved without external IDs or position (title/year reuse in
   `packages/core/src/matching.ts`) has no unique key, so concurrent matches can create it twice.

packages/core/src/entries.ts

@@ -579,6 +579,16 @@ async function moveEntry(
     await moveEngagements(k, userId, src.id, dst.id, actor);
     await writeEntry(k, src, {}, actor, { extra: { deleted_at: nowIso() } });
   }
+  if (result.id !== src.id) {
+    // The moved entry merged into another one: source records that mapped to it keep syncing into the
+    // survivor. Engagements keep their ids when re-parented, so their mappings stay valid.
+    await k
+      .updateTable("source_records")
+      .set({ target_id: result.id })
+      .where("target_entity", "=", "entry")
+      .where("target_id", "=", src.id)
+      .execute();
+  }
   if (opts.undo) return result;
 
   await repointCopiesAndListItems(k, userId, src.workId, target.id, actor);

packages/core/src/lists.ts

@@ -265,6 +265,28 @@ export async function addListItem(
   );
 }
 
+/** Replace a list's items with `next`, in order (one history row). Used by the sync engine. */
+export async function replaceListItems(
+  db: Db,
+  userId: string,
+  listId: string,
+  next: { workId: string; note?: string | null }[],
+  actor: Actor = "user",
+  tx?: Executor,
+): Promise<void> {
+  await inTx(db, tx, (k) =>
+    mutate(k, userId, listId, actor, async () => {
+      const rows: ItemRow[] = [];
+      for (const [i, it] of next.entries()) {
+        const work = await resolveVisibleWork(k, userId, it.workId);
+        rows.push({ id: newId(), list_id: listId, work_id: work.id, position: i, note: it.note == null ? null : checkText("note", it.note) });
+      }
+      await k.deleteFrom("list_items").where("list_id", "=", listId).execute();
+      if (rows.length > 0) await k.insertInto("list_items").values(rows).execute();
+    }),
+  );
+}
+
 function clampIndex(i: number, max: number): number {
   if (typeof i !== "number" || !Number.isInteger(i)) throw new ValidationError("position", "not_integer");
   return Math.max(0, Math.min(i, max));

packages/core/src/sync/apply.ts

@@ -1,11 +1,13 @@
+import { sql } from "kysely";
 import type { Db } from "@sttracker/db/db";
 import { newId, nowIso } from "@sttracker/db/ids";
 import { fromJson, toJson } from "@sttracker/db/json";
 import type { ChangeStatus, ChangeType, Database } from "@sttracker/db/schema";
-import type { EngagementFields, EntryFields, PartialDate, SourceRecord } from "@sttracker/plugin-sdk/types";
+import type { EngagementFields, EntryFields, PartialDate, SourceRecord, WorkHint } from "@sttracker/plugin-sdk/types";
 import { addEngagement, type EngagementPatch, updateEngagement } from "../engagements.ts";
 import { type EntryPatch, fromStoredDate, resolveVisibleWork, upsertEntry } from "../entries.ts";
 import type { Actor, Executor } from "../history.ts";
+import { createList, replaceListItems, updateList } from "../lists.ts";
 import { matchWork } from "../matching.ts";
 import { resolveAliasIn, updateWorkFromHintIn } from "../works.ts";
 import { decodeVersion, encodeVersion, hashRecord, type NormalizedRecord, normalizeRecord, sameValue } from "./snapshot.ts";
@@ -31,14 +33,33 @@ export interface SyncState {
   db: Db;
   userId: string;
   connectionId: string;
+  pluginId: string;
   runId: string;
   actor: Actor;
   counts: RunCounts;
   log: string[];
 }
 
+/**
+ * Run `fn` against scratch counts and log, and fold them into the run's only when `fn` resolves, so
+ * the run summary reflects committed work only.
+ */
+export async function committed(st: SyncState, fn: (local: SyncState) => Promise<void>): Promise<void> {
+  const local: SyncState = { ...st, counts: emptyCounts(), log: [] };
+  await fn(local);
+  for (const key of Object.keys(local.counts) as (keyof RunCounts)[]) st.counts[key] += local.counts[key];
+  st.log.push(...local.log);
+}
+
 type SourceRecordRow = Database["source_records"];
 type Fields = Record<string, unknown>;
+type ListItemValue = { workId: string; note: string | null };
+
+/**
+ * `source_records.current_hash` of a record that has been matched (its work id stored as
+ * `target_entity "work"`) but whose apply has not committed yet. A retry reuses the stored work.
+ */
+const PENDING_HASH = "";
 
 // ---------------------------------------------------------------------------------------------
 // Field mapping between SourceRecord fields and the user layer's patch keys
@@ -71,6 +92,14 @@ export function engagementPatchFrom(f: EngagementFields | Record<string, unknown
   return out;
 }
 
+/** A list record's own fields (items are matched separately). */
+function listPatchFrom(l: Record<string, unknown> | undefined): Fields {
+  const out: Fields = {};
+  if (typeof l?.name === "string") out.name = l.name;
+  if (typeof l?.description === "string") out.description = l.description;
+  return out;
+}
+
 /** An entry row's current values under the sync-managed patch keys. */
 export function entryValues(r: Database["entries"]): Fields {
   return {
@@ -95,10 +124,39 @@ export function engagementValues(r: Database["engagements"]): Fields {
   };
 }
 
+/** A list's current values: name, description and ordered items. */
+export async function listValues(k: Executor, r: Database["lists"]): Promise<Fields> {
+  const items = await k.selectFrom("list_items").select(["work_id", "note"]).where("list_id", "=", r.id).orderBy("position").orderBy("id").execute();
+  return { name: r.name, description: r.description, items: items.map((i): ListItemValue => ({ workId: i.work_id, note: i.note })) };
+}
+
+/**
+ * Lists carry no `field_sources`, so ownership is read from their history: `name`/`description` belong
+ * to whoever last changed them, and `items` to the user once any user edit exists, else to the source.
+ */
+export async function listFieldSources(k: Executor, listId: string, sourceActor: Actor): Promise<Record<string, string>> {
+  const rows = await k
+    .selectFrom("user_history")
+    .select(["after", "actor"])
+    .where("entity", "=", "list")
+    .where("entity_id", "=", listId)
+    .orderBy("created_at")
+    .orderBy("id")
+    .execute();
+  const fs: Record<string, string> = { items: sourceActor };
+  for (const r of rows) {
+    const after = fromJson<Fields>(r.after, {});
+    for (const f of ["name", "description"]) if (f in after) fs[f] = r.actor;
+    if (r.actor === "user") fs.items = "user";
+  }
+  return fs;
+}
+
 function patchOf(n: NormalizedRecord | undefined): Fields {
   if (!n) return {};
   if (n.kind === "entry") return entryPatchFrom(n.entry);
   if (n.kind === "engagement") return engagementPatchFrom(n.engagement);
+  if (n.kind === "list") return listPatchFrom(n.list);
   return {};
 }
 
@@ -109,12 +167,12 @@ function patchOf(n: NormalizedRecord | undefined): Fields {
 async function writeEvent(
   k: Executor,
   st: SyncState,
-  e: { recordId: string; type: ChangeType; status: ChangeStatus; targetEntity: string; targetId: string | null; payload: Fields },
+  e: { id?: string; recordId: string; type: ChangeType; status: ChangeStatus; targetEntity: string; targetId: string | null; payload: Fields },
 ): Promise<void> {
   await k
     .insertInto("change_events")
     .values({
-      id: newId(),
+      id: e.id ?? newId(),
       user_id: st.userId,
       connection_id: st.connectionId,
       run_id: st.runId,
@@ -133,35 +191,44 @@ async function writeEvent(
   if (e.status === "pending" && e.type === "conflict") st.counts.conflicts++;
 }
 
-/**
+const incomingOf = (p: Fields) => ("incoming" in p ? p.incoming : p.to);
- * Anti-flip-flop: a pending changed/conflict event for the same (record, field, incoming value) is
+
- * never filed twice.
+/** Pending changed/conflict events of a record for one field. */
- */
+async function pendingFieldEvents(k: Executor, recordId: string, field: string) {
-async function hasPendingFieldEvent(k: Executor, recordId: string, field: string, incoming: unknown): Promise<boolean> {
   const rows = await k
     .selectFrom("change_events")
-    .select("payload")
+    .select(["id", "payload"])
     .where("record_id", "=", recordId)
     .where("status", "=", "pending")
     .where("type", "in", ["changed", "conflict"])
     .execute();
-  return rows.some((r) => {
+  return rows.map((r) => ({ id: r.id, payload: fromJson<Fields>(r.payload, {}) })).filter((r) => r.payload.field === field);
-    const p = fromJson<Fields>(r.payload, {});
+}
-    return p.field === field && sameValue("incoming" in p ? p.incoming : p.to, incoming);
+
-  });
+/** Resolve stale pending events as `ignored`, recording why in their payload. */
+async function retire(k: Executor, stale: { id: string; payload: Fields }[], extra: Fields): Promise<void> {
+  for (const s of stale) {
+    await k
+      .updateTable("change_events")
+      .set({ status: "ignored", resolved_at: nowIso(), payload: toJson({ ...s.payload, ...extra }) })
+      .where("id", "=", s.id)
+      .execute();
+  }
 }
 
 interface Target {
-  entity: "entry" | "engagement";
+  entity: "entry" | "engagement" | "list";
   id: string;
   current: Fields;
   fieldSources: Record<string, string>;
 }
 
 /**
- * Decide each incoming field against the target's `field_sources` (spec §8.3):
+ * Decide each incoming field against the target's field ownership (spec §8.3):
  * unset or this connection → applied; `"user"` → pending `changed`; another source → pending `conflict`.
  * Values equal to the current one are skipped, so the plugin never takes ownership of a field it did not change.
+ * Anti-flip-flop: a pending event for the same (record, field, incoming value) is never filed twice, and a
+ * newer incoming value supersedes an older pending one (`ignored`, `payload.supersededBy`).
  * Returns the patch that may be applied, with each applied field's previous value.
  */
 async function decideFields(
@@ -175,16 +242,24 @@ async function decideFields(
   const from: Fields = {};
   for (const [field, value] of Object.entries(incoming)) {
     const current = t.current[field];
-    if (sameValue(current, value)) continue;
+    const pending = await pendingFieldEvents(k, recordId, field);
+    if (sameValue(current, value)) {
+      // The source now agrees with what STTracker holds: its older proposals are moot.
+      await retire(k, pending, { reason: "source_agrees" });
+      continue;
+    }
     const source = t.fieldSources[field];
     if (source === undefined || source === st.actor) {
       patch[field] = value;
       from[field] = current ?? null;
       continue;
     }
-    if (await hasPendingFieldEvent(k, recordId, field, value)) continue;
+    if (pending.some((p) => sameValue(incomingOf(p.payload), value))) continue;
+    const id = newId();
+    await retire(k, pending, { supersededBy: id });
     if (source === "user") {
       await writeEvent(k, st, {
+        id,
         recordId,
         type: "changed",
         status: "pending",
@@ -194,6 +269,7 @@ async function decideFields(
       });
     } else {
       await writeEvent(k, st, {
+        id,
         recordId,
         type: "conflict",
         status: "pending",
@@ -206,25 +282,30 @@ async function decideFields(
   return { patch, from };
 }
 
-async function applyPatch(k: Executor, st: SyncState, t: Target, patch: Fields): Promise<void> {
+/** Write an allowed patch to the target as this connection's actor. */
+export async function applyPatch(k: Executor, db: Db, userId: string, t: Pick<Target, "entity" | "id">, patch: Fields, actor: Actor): Promise<void> {
   if (Object.keys(patch).length === 0) return;
   if (t.entity === "entry") {
     const row = await k.selectFrom("entries").select("work_id").where("id", "=", t.id).executeTakeFirstOrThrow();
-    await upsertEntry(st.db, st.userId, row.work_id, patch as Partial<EntryPatch>, st.actor, k);
+    await upsertEntry(db, userId, row.work_id, patch as Partial<EntryPatch>, actor, k);
+  } else if (t.entity === "engagement") {
+    await updateEngagement(db, userId, t.id, patch as Partial<EngagementPatch>, actor, k);
   } else {
-    await updateEngagement(st.db, st.userId, t.id, patch as Partial<EngagementPatch>, st.actor, k);
+    const { items, ...own } = patch;
+    if (Object.keys(own).length > 0) await updateList(db, userId, t.id, own, actor, k);
+    if (items !== undefined) await replaceListItems(db, userId, t.id, items as ListItemValue[], actor, k);
   }
 }
 
 /** The live target of a mapped record, or null (unmapped, gone, or soft-deleted: never revived). */
-async function loadTarget(k: Executor, userId: string, entity: string | null, id: string | null): Promise<Target | null> {
+async function loadTarget(k: Executor, st: SyncState, entity: string | null, id: string | null): Promise<Target | null> {
   if (!id) return null;
   if (entity === "entry") {
     const r = await k
       .selectFrom("entries")
       .selectAll()
       .where("id", "=", id)
-      .where("user_id", "=", userId)
+      .where("user_id", "=", st.userId)
       .where("deleted_at", "is", null)
       .executeTakeFirst();
     return r ? { entity, id, current: entryValues(r), fieldSources: fromJson(r.field_sources, {}) } : null;
@@ -235,12 +316,22 @@ async function loadTarget(k: Executor, userId: string, entity: string | null, id
       .innerJoin("entries as e", "e.id", "g.entry_id")
       .selectAll("g")
       .where("g.id", "=", id)
-      .where("e.user_id", "=", userId)
+      .where("e.user_id", "=", st.userId)
       .where("g.deleted_at", "is", null)
       .where("e.deleted_at", "is", null)
       .executeTakeFirst();
     return r ? { entity, id, current: engagementValues(r), fieldSources: fromJson(r.field_sources, {}) } : null;
   }
+  if (entity === "list") {
+    const r = await k
+      .selectFrom("lists")
+      .selectAll()
+      .where("id", "=", id)
+      .where("user_id", "=", st.userId)
+      .where("deleted_at", "is", null)
+      .executeTakeFirst();
+    return r ? { entity, id, current: await listValues(k, r), fieldSources: await listFieldSources(k, id, st.actor) } : null;
+  }
   return null;
 }
 
@@ -320,7 +411,10 @@ export function sameStartInstant(a: PartialDate, b: PartialDate): boolean {
   return Number.isNaN(x) || Number.isNaN(y) ? a.value === b.value : x === y;
 }
 
-/** A live engagement on the entry that is the same play, not already reported by this connection. */
+/**
+ * A live engagement on the entry that is the same play, not already reported by this connection.
+ * Also returns the date field that matched (its values differ only within the dedupe tolerance).
+ */
 async function findDuplicateEngagement(k: Executor, st: SyncState, entryId: string, incoming: Fields) {
   const fin = incoming.finishedAt as PartialDate | null | undefined;
   const start = incoming.startedAt as PartialDate | null | undefined;
@@ -347,8 +441,8 @@ async function findDuplicateEngagement(k: Executor, st: SyncState, entryId: stri
     const v = engagementValues(r);
     const f = v.finishedAt as PartialDate | null;
     const s = v.startedAt as PartialDate | null;
-    if (fin && f && finishedWithin24h(fin, f)) return r;
+    if (fin && f && finishedWithin24h(fin, f)) return { row: r, matchedOn: "finishedAt" };
-    if (start && s && sameStartInstant(start, s)) return r;
+    if (start && s && sameStartInstant(start, s)) return { row: r, matchedOn: "startedAt" };
   }
   return null;
 }
@@ -356,72 +450,147 @@ async function findDuplicateEngagement(k: Executor, st: SyncState, entryId: stri
 const isEmptyValue = (v: unknown) =>
   v === null || v === undefined || (typeof v === "object" && !Array.isArray(v) && Object.keys(v as object).length === 0);
 
+// ---------------------------------------------------------------------------------------------
+// List items
+// ---------------------------------------------------------------------------------------------
+
+/**
+ * Match list item hints to works, outside any transaction. An ID-less hint first reuses the user's own
+ * live private work with the same media type and title, so re-matching never piles up drafts.
+ */
+async function matchListItems(st: SyncState, items: { work: WorkHint; note?: string }[]): Promise<ListItemValue[]> {
+  const out: ListItemValue[] = [];
+  const seen = new Set<string>();
+  for (const it of items) {
+    let workId: string | undefined;
+    if (Object.keys(it.work.externalIds).length === 0) {
+      workId = (
+        await st.db.k
+          .selectFrom("works")
+          .select("id")
+          .where("owner_user_id", "=", st.userId)
+          .where("merged_into_id", "is", null)
+          .where("media_type", "=", it.work.mediaType)
+          .where(sql<boolean>`lower(title) = ${it.work.title.toLowerCase()}`)
+          .orderBy("id")
+          .executeTakeFirst()
+      )?.id;
+    }
+    workId ??= (await matchWork(st.db, st.userId, it.work, st.actor)).workId;
+    if (seen.has(workId)) continue;
+    seen.add(workId);
+    out.push({ workId, note: it.note ?? null });
+  }
+  return out;
+}
+
 // ---------------------------------------------------------------------------------------------
 // Records
 // ---------------------------------------------------------------------------------------------
 
-/** Snapshot one non-listen record and apply it. Call outside any transaction (matchWork has its own). */
+/**
-export async function applyRecord(st: SyncState, rec: SourceRecord): Promise<void> {
+ * Snapshot one non-listen record and apply it. Call outside any transaction (matchWork has its own).
-  const hash = hashRecord(rec);
+ * Counts land in `st` only once the record's work has committed.
-  const existing = await st.db.k
+ */
-    .selectFrom("source_records")
+export function applyRecord(st: SyncState, rec: SourceRecord): Promise<void> {
-    .selectAll()
+  return committed(st, async (local) => {
-    .where("connection_id", "=", st.connectionId)
+    const hash = hashRecord(rec);
-    .where("key", "=", rec.key)
+    const existing = await local.db.k
-    .executeTakeFirst();
+      .selectFrom("source_records")
-  if (!existing) return addRecord(st, rec, hash);
+      .selectAll()
-  if (existing.current_hash === hash) {
+      .where("connection_id", "=", local.connectionId)
-    await st.db.k.transaction().execute(async (tx) => {
+      .where("key", "=", rec.key)
-      await tx.updateTable("source_records").set({ last_seen: nowIso() }).where("id", "=", existing.id).execute();
+      .executeTakeFirst();
-      if (existing.deleted_at !== null && rec.deleted !== true) await reappeared(tx, existing);
+    if (!existing || existing.current_hash === PENDING_HASH) return addRecord(local, rec, hash, existing ?? null);
-    });
+    if (existing.current_hash === hash) {
-    return;
+      await local.db.k.transaction().execute(async (tx) => {
-  }
+        await tx.updateTable("source_records").set({ last_seen: nowIso() }).where("id", "=", existing.id).execute();
-  return changeRecord(st, existing, rec, hash);
+        if (existing.deleted_at !== null && rec.deleted !== true) await reappeared(tx, existing);
-}
+      });
-
-async function addRecord(st: SyncState, rec: SourceRecord, hash: string): Promise<void> {
-  const applicable = (rec.kind === "entry" || rec.kind === "engagement") && rec.deleted !== true && !rec.mergedIntoKey;
-  // Matching runs in its own transaction (it retries on unique violations), before the apply transaction.
-  const workId =
-    applicable && (rec.kind === "entry" || rec.kind === "engagement") ? (await matchWork(st.db, st.userId, rec.work, st.actor)).workId : null;
-  await st.db.k.transaction().execute(async (tx) => {
-    const now = nowIso();
-    const recordId = newId();
-    await tx
-      .insertInto("source_records")
-      .values({
-        id: recordId,
-        connection_id: st.connectionId,
-        key: rec.key,
-        kind: rec.kind,
-        current_hash: hash,
-        first_seen: now,
-        last_seen: now,
-        deleted_at: rec.deleted === true ? now : null,
-        merged_into_key: rec.mergedIntoKey ?? null,
-        target_entity: rec.kind === "entry" || rec.kind === "engagement" ? rec.kind : null,
-        target_id: null,
-      })
-      .execute();
-    await tx.insertInto("source_record_versions").values({ id: newId(), record_id: recordId, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
-    if (!applicable || workId === null) {
-      if (rec.kind === "list") st.log.push(`record ${rec.key}: list records are stored but not applied`);
       return;
     }
-    const targetId =
+    return changeRecord(local, existing, rec, hash);
-      rec.kind === "entry"
+  });
-        ? await addEntryRecord(tx, st, recordId, rec.key, workId, entryPatchFrom(rec.entry))
+}
-        : rec.kind === "engagement"
+
-          ? await addEngagementRecord(tx, st, recordId, rec.key, workId, engagementPatchFrom(rec.engagement))
+async function addRecord(st: SyncState, rec: SourceRecord, hash: string, pendingRow: SourceRecordRow | null): Promise<void> {
-          : null;
+  const applicable = rec.deleted !== true && !rec.mergedIntoKey && rec.kind !== "listen";
+  let recordId = pendingRow?.id ?? null;
+  let workId: string | null = null;
+  let items: ListItemValue[] = [];
+  if (applicable && (rec.kind === "entry" || rec.kind === "engagement")) {
+    if (pendingRow?.target_entity === "work" && pendingRow.target_id) {
+      workId = pendingRow.target_id;
+    } else {
+      // Matching runs in its own transaction (it retries on unique violations). Its result is committed
+      // right away, so a failed apply is retried against the same work instead of a fresh draft.
+      workId = (await matchWork(st.db, st.userId, rec.work, st.actor)).workId;
+      recordId = await persistMatched(st, rec, workId, recordId);
+    }
+  } else if (applicable && rec.kind === "list") {
+    items = await matchListItems(st, rec.list.items);
+  }
+
+  await st.db.k.transaction().execute(async (tx) => {
+    const now = nowIso();
+    const row = {
+      current_hash: hash,
+      last_seen: now,
+      deleted_at: rec.deleted === true ? now : null,
+      merged_into_key: rec.mergedIntoKey ?? null,
+      target_entity: rec.kind === "listen" ? null : rec.kind,
+      target_id: null,
+    };
+    if (recordId) {
+      await tx.updateTable("source_records").set(row).where("id", "=", recordId).execute();
+    } else {
+      recordId = newId();
+      await tx
+        .insertInto("source_records")
+        .values({ id: recordId, connection_id: st.connectionId, key: rec.key, kind: rec.kind, first_seen: now, ...row })
+        .execute();
+    }
+    const id = recordId;
+    await tx.insertInto("source_record_versions").values({ id: newId(), record_id: id, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
+    if (!applicable) return;
+    let targetId: string | null = null;
+    if (rec.kind === "entry" && workId) targetId = await addEntryRecord(tx, st, id, rec.key, workId, entryPatchFrom(rec.entry));
+    else if (rec.kind === "engagement" && workId) targetId = await addEngagementRecord(tx, st, id, rec.key, workId, engagementPatchFrom(rec.engagement));
+    else if (rec.kind === "list") targetId = await addListRecord(tx, st, id, rec.key, rec.list.name, items);
     if (targetId) {
-      await tx.updateTable("source_records").set({ target_id: targetId }).where("id", "=", recordId).execute();
+      await tx.updateTable("source_records").set({ target_id: targetId }).where("id", "=", id).execute();
       st.counts.added++;
     }
   });
 }
 
+/** Commit the record row with its matched work (`target_entity "work"`, pending hash) before applying. */
+async function persistMatched(st: SyncState, rec: SourceRecord, workId: string, recordId: string | null): Promise<string> {
+  const now = nowIso();
+  if (recordId) {
+    await st.db.k.updateTable("source_records").set({ target_entity: "work", target_id: workId, last_seen: now }).where("id", "=", recordId).execute();
+    return recordId;
+  }
+  const id = newId();
+  await st.db.k
+    .insertInto("source_records")
+    .values({
+      id,
+      connection_id: st.connectionId,
+      key: rec.key,
+      kind: rec.kind,
+      current_hash: PENDING_HASH,
+      first_seen: now,
+      last_seen: now,
+      deleted_at: null,
+      merged_into_key: null,
+      target_entity: "work",
+      target_id: workId,
+    })
+    .execute();
+  return id;
+}
+
 async function addEntryRecord(k: Executor, st: SyncState, recordId: string, key: string, workId: string, incoming: Fields): Promise<string | null> {
   const live = await liveEntryFor(k, st, workId, incoming);
   if (!live) {
@@ -433,7 +602,7 @@ async function addEntryRecord(k: Executor, st: SyncState, recordId: string, key:
     // The user (or another source) already has this entry: decide field by field.
     const t: Target = { entity: "entry", id: live.row.id, current: entryValues(live.row), fieldSources: fromJson(live.row.field_sources, {}) };
     const { patch } = await decideFields(k, st, recordId, t, incoming);
-    await applyPatch(k, st, t, patch);
+    await applyPatch(k, st.db, st.userId, t, patch, st.actor);
     applied = Object.keys(patch);
   }
   await writeEvent(k, st, {
@@ -455,25 +624,32 @@ async function addEngagementRecord(k: Executor, st: SyncState, recordId: string,
   }
   const dup = await findDuplicateEngagement(k, st, live.row.id, incoming);
   let engagementId: string;
-  let applied: string[];
   let payload: Fields;
   if (dup) {
-    // Same play from another source: attach, and only fill fields nobody has set yet.
+    // Same play from another source: attach. Fields nobody has set yet are filled; any other differing
-    const current = engagementValues(dup);
+    // field goes through the per-field rules (conflict / pending changed). The date that identified the
-    const sources = fromJson<Record<string, string>>(dup.field_sources, {});
+    // play differs only within the dedupe tolerance and is left alone.
+    const t: Target = {
+      entity: "engagement",
+      id: dup.row.id,
+      current: engagementValues(dup.row),
+      fieldSources: fromJson(dup.row.field_sources, {}),
+    };
     const fill: Fields = {};
+    const rest: Fields = {};
     for (const [f, v] of Object.entries(incoming)) {
-      if (sources[f] === undefined && isEmptyValue(current[f]) && !isEmptyValue(v)) fill[f] = v;
+      if (f === dup.matchedOn) continue;
+      if (t.fieldSources[f] === undefined && isEmptyValue(t.current[f]) && !isEmptyValue(v)) fill[f] = v;
+      else rest[f] = v;
     }
-    if (Object.keys(fill).length > 0) await updateEngagement(st.db, st.userId, dup.id, fill as Partial<EngagementPatch>, st.actor, k);
+    const { patch } = await decideFields(k, st, recordId, t, rest);
-    engagementId = dup.id;
+    await applyPatch(k, st.db, st.userId, t, { ...fill, ...patch }, st.actor);
-    applied = Object.keys(fill);
+    engagementId = dup.row.id;
-    payload = { key, kind: "engagement", fields: applied, dedupedInto: dup.id };
+    payload = { key, kind: "engagement", fields: Object.keys({ ...fill, ...patch }), dedupedInto: dup.row.id };
   } else {
     const eg = await addEngagement(st.db, st.userId, live.row.id, incoming as Partial<EngagementPatch>, st.actor, k);
     engagementId = eg.id;
-    applied = Object.keys(incoming);
+    payload = { key, kind: "engagement", fields: Object.keys(incoming) };
-    payload = { key, kind: "engagement", fields: applied };
   }
   await k
     .insertInto("engagement_sources")
@@ -484,6 +660,20 @@ async function addEngagementRecord(k: Executor, st: SyncState, recordId: string,
   return engagementId;
 }
 
+async function addListRecord(k: Executor, st: SyncState, recordId: string, key: string, name: string, items: ListItemValue[]): Promise<string> {
+  const list = await createList(st.db, st.userId, { name, description: `Imported from ${st.pluginId}`, kind: "list" }, st.actor, k);
+  if (items.length > 0) await replaceListItems(st.db, st.userId, list.id, items, st.actor, k);
+  await writeEvent(k, st, {
+    recordId,
+    type: "added",
+    status: "auto_applied",
+    targetEntity: "list",
+    targetId: list.id,
+    payload: { key, kind: "list", fields: ["name", "description", "items"] },
+  });
+  return list.id;
+}
+
 async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: SourceRecord, hash: string): Promise<void> {
   const prevRow = await st.db.k
     .selectFrom("source_record_versions")
@@ -495,18 +685,23 @@ async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: Sourc
     .executeTakeFirst();
   const prev = prevRow ? decodeVersion(prevRow.raw_gz).normalized : undefined;
   const next = normalizeRecord(rec);
+  // List items are matched before the transaction (matchWork has its own), and only when they changed.
+  const itemsChanged = rec.kind === "list" && !rec.deleted && !sameValue(prev?.list?.items, next.list?.items);
+  const items = itemsChanged && rec.kind === "list" ? await matchListItems(st, rec.list.items) : null;
 
   await st.db.k.transaction().execute(async (tx) => {
     const now = nowIso();
+    // A merged record that is later reported deleted stays "merged": its disappearance is expected.
+    const mergedInto = rec.mergedIntoKey ?? (rec.deleted === true ? existing.merged_into_key : null);
     await tx.insertInto("source_record_versions").values({ id: newId(), record_id: existing.id, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
     await tx
       .updateTable("source_records")
-      .set({ current_hash: hash, last_seen: now, merged_into_key: rec.mergedIntoKey ?? null })
+      .set({ current_hash: hash, last_seen: now, merged_into_key: mergedInto })
       .where("id", "=", existing.id)
       .execute();
 
     if (rec.deleted === true) {
-      if (existing.deleted_at === null) await markRemoved(tx, st, existing);
+      if (existing.deleted_at === null) await markRemoved(tx, st, { ...existing, merged_into_key: mergedInto });
       return;
     }
     st.counts.changed++;
@@ -514,14 +709,15 @@ async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: Sourc
 
     if (rec.mergedIntoKey && rec.mergedIntoKey !== (prev?.mergedIntoKey ?? null)) await fileMerged(tx, st, existing, rec.mergedIntoKey);
 
-    const target = await loadTarget(tx, st.userId, existing.target_entity, existing.target_id);
+    const target = await loadTarget(tx, st, existing.target_entity, existing.target_id);
     if (!target) return; // unmapped, or the user deleted it: never revived
     const before = patchOf(prev);
     const after = patchOf(next);
     const incoming: Fields = {};
     for (const [f, v] of Object.entries(after)) if (!sameValue(before[f], v)) incoming[f] = v;
+    if (items) incoming.items = items;
     const { patch, from } = await decideFields(tx, st, existing.id, target, incoming);
-    await applyPatch(tx, st, target, patch);
+    await applyPatch(tx, st.db, st.userId, target, patch, st.actor);
     for (const [field, to] of Object.entries(patch)) {
       await writeEvent(tx, st, {
         recordId: existing.id,
@@ -541,6 +737,7 @@ async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: Sourc
 }
 
 async function targetWorkId(k: Executor, t: Target): Promise<string | null> {
+  if (t.entity === "list") return null;
   const entryId =
     t.entity === "entry"
       ? t.id
@@ -561,11 +758,24 @@ async function reappeared(k: Executor, rec: SourceRecordRow): Promise<void> {
     .execute();
 }
 
-/** Set `source_records.deleted_at` and file a pending `removed` event when there is a live target. */
+/**
+ * Set `source_records.deleted_at` and file a pending `removed` event when there is a live target.
+ * A record merged into another (its `merged_into_key` set, or a pending `merged` event) is expected
+ * to vanish: no removal is counted or filed for it, ever.
+ */
 export async function markRemoved(k: Executor, st: SyncState, rec: SourceRecordRow): Promise<void> {
   await k.updateTable("source_records").set({ deleted_at: nowIso() }).where("id", "=", rec.id).execute();
+  if (rec.merged_into_key) return;
+  const pendingMerge = await k
+    .selectFrom("change_events")
+    .select("id")
+    .where("record_id", "=", rec.id)
+    .where("type", "=", "merged")
+    .where("status", "=", "pending")
+    .executeTakeFirst();
+  if (pendingMerge) return;
   st.counts.removed++;
-  const target = await loadTarget(k, st.userId, rec.target_entity, rec.target_id);
+  const target = await loadTarget(k, st, rec.target_entity, rec.target_id);
   if (!target) return;
   const already = await k
     .selectFrom("change_events")
@@ -588,8 +798,8 @@ export async function markRemoved(k: Executor, st: SyncState, rec: SourceRecordR
 /** A pending `merged` event on the entry this record maps to (an engagement record's entry). */
 async function fileMerged(k: Executor, st: SyncState, rec: SourceRecordRow, intoKey: string): Promise<void> {
   st.counts.merged++;
-  const target = await loadTarget(k, st.userId, rec.target_entity, rec.target_id);
+  const target = await loadTarget(k, st, rec.target_entity, rec.target_id);
-  if (!target) return;
+  if (!target || target.entity === "list") return;
   const entryId =
     target.entity === "entry"
       ? target.id

packages/core/src/sync/changes.ts

@@ -1,14 +1,16 @@
 import type { Db } from "@sttracker/db/db";
 import { nowIso } from "@sttracker/db/ids";
-import { fromJson } from "@sttracker/db/json";
+import { fromJson, toJson } from "@sttracker/db/json";
 import type { ChangeStatus, ChangeType, Database } from "@sttracker/db/schema";
 import type { WorkHint } from "@sttracker/plugin-sdk/types";
 import { type EngagementPatch, softDeleteEngagement, updateEngagement } from "../engagements.ts";
 import { type EntryPatch, moveEntryToWork, softDeleteEntry, upsertEntry } from "../entries.ts";
 import { NotFoundError, ValidationError } from "../errors.ts";
 import type { Actor, Executor } from "../history.ts";
+import { softDeleteList } from "../lists.ts";
 import { matchWork } from "../matching.ts";
-import { engagementValues, entryValues } from "./apply.ts";
+import { resolveAliasIn } from "../works.ts";
+import { applyPatch, engagementValues, entryValues } from "./apply.ts";
 import { decodeVersion } from "./snapshot.ts";
 
 export interface ChangeEvent {
@@ -65,10 +67,12 @@ const STATUS_FOR = { accept: "accepted", ignore: "ignored", keep_mine: "kept_min
 /**
  * Resolve a pending event.
  * - `accept`: changed/conflict applies the incoming value as the event's plugin actor (which then owns
- *   the field); removed soft-deletes the target; merged moves the entry to the merge target's work.
+ *   the field); removed soft-deletes the target (an engagement other sources still report only loses
+ *   this connection's `engagement_sources` row); merged moves the entry to the merge target's work.
  * - `keep_mine`: changed/conflict set `field_sources[field] = "user"`, so that source stops overriding it.
  * - `ignore`: data untouched.
- * All three set `status` and `resolved_at`.
+ * All three set `status` and `resolved_at`. Accept/keep_mine on a target that is gone (deleted, or merged
+ * away with no live survivor) resolves the event as `ignored` with `payload.reason = "target_gone"`.
  */
 export async function resolveChange(
   db: Db,
@@ -76,7 +80,7 @@ export async function resolveChange(
   changeId: string,
   action: "accept" | "ignore" | "keep_mine",
 ): Promise<void> {
-  if (!(action in STATUS_FOR)) throw new ValidationError("action", "invalid");
+  if (!Object.hasOwn(STATUS_FOR, action)) throw new ValidationError("action", "invalid");
   const row = await db.k.selectFrom("change_events").selectAll().where("id", "=", changeId).where("user_id", "=", userId).executeTakeFirst();
   if (!row) throw new NotFoundError("change");
   if (row.status !== "pending") throw new ValidationError("status", "not_pending");
@@ -96,19 +100,26 @@ export async function resolveChange(
       .executeTakeFirst();
     if (Number(claimed.numUpdatedRows) === 0) throw new ValidationError("status", "not_pending");
     if (action === "ignore") return;
-    if (ev.type === "changed" || ev.type === "conflict") {
+    const isField = ev.type === "changed" || ev.type === "conflict";
-      const field = String(ev.payload.field);
+    if (action === "keep_mine" && !isField) return;
-      if (action === "accept") {
+
-        const value = ev.type === "conflict" ? ev.payload.incoming : ev.payload.to;
+    const targetId = await liveTargetId(tx, userId, ev);
-        await setField(db, tx, userId, ev, field, value, actor);
+    if (!targetId) {
-      } else {
+      await tx
-        await setField(db, tx, userId, ev, field, undefined, "user");
+        .updateTable("change_events")
-      }
+        .set({ status: "ignored", payload: toJson({ ...ev.payload, reason: "target_gone" }) })
+        .where("id", "=", ev.id)
+        .execute();
       return;
     }
-    if (action !== "accept") return;
+    if (isField) {
-    if (ev.type === "removed") return removeTarget(db, tx, userId, ev, actor);
+      const field = String(ev.payload.field);
-    if (ev.type === "merged" && mergeTarget) return moveTo(db, tx, userId, ev, mergeTarget, actor);
+      const value = action === "keep_mine" ? undefined : ev.type === "conflict" ? ev.payload.incoming : ev.payload.to;
+      await setField(db, tx, userId, ev.targetEntity, targetId, field, value, action === "keep_mine" ? "user" : actor);
+      return;
+    }
+    if (ev.type === "removed") return removeTarget(db, tx, userId, ev, targetId, actor);
+    if (ev.type === "merged" && mergeTarget) return moveTo(db, tx, userId, targetId, mergeTarget, actor);
   });
 }
 
@@ -118,51 +129,29 @@ async function actorFor(db: Db, connectionId: string | null): Promise<Actor> {
   return c ? `plugin:${c.plugin_id}:${c.id}` : "user";
 }
 
-/** Write `value` to the target's field as `actor`; with `value` undefined, re-stamp the current value. */
+/**
-async function setField(db: Db, k: Executor, userId: string, ev: ChangeEvent, field: string, value: unknown, actor: Actor): Promise<void> {
+ * The event's target if it is still live. A soft-deleted entry that was merged into another (the user's
-  if (!ev.targetId) throw new NotFoundError(ev.targetEntity);
+ * work alias leads to a work with a live entry) resolves to that survivor. Otherwise null.
+ */
+async function liveTargetId(k: Executor, userId: string, ev: ChangeEvent): Promise<string | null> {
+  if (!ev.targetId) return null;
   if (ev.targetEntity === "entry") {
-    const r = await k
+    const r = await k.selectFrom("entries").select(["id", "work_id", "deleted_at"]).where("id", "=", ev.targetId).where("user_id", "=", userId).executeTakeFirst();
+    if (!r) return null;
+    if (r.deleted_at === null) return r.id;
+    const to = await resolveAliasIn(k, userId, r.work_id);
+    if (to === r.work_id) return null;
+    const survivor = await k
       .selectFrom("entries")
-      .selectAll()
+      .select("id")
-      .where("id", "=", ev.targetId)
       .where("user_id", "=", userId)
+      .where("work_id", "=", to)
       .where("deleted_at", "is", null)
       .executeTakeFirst();
-    if (!r) throw new NotFoundError("entry");
+    return survivor?.id ?? null;
-    const v = value === undefined ? entryValues(r)[field] : value;
-    await upsertEntry(db, userId, r.work_id, { [field]: v } as Partial<EntryPatch>, actor, k);
-    return;
   }
   if (ev.targetEntity === "engagement") {
     const r = await k
-      .selectFrom("engagements as g")
-      .innerJoin("entries as e", "e.id", "g.entry_id")
-      .selectAll("g")
-      .where("g.id", "=", ev.targetId)
-      .where("e.user_id", "=", userId)
-      .executeTakeFirst();
-    if (!r) throw new NotFoundError("engagement");
-    const v = value === undefined ? engagementValues(r)[field] : value;
-    await updateEngagement(db, userId, r.id, { [field]: v } as Partial<EngagementPatch>, actor, k);
-    return;
-  }
-  throw new ValidationError("targetEntity", "unsupported");
-}
-
-async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent, actor: Actor): Promise<void> {
-  if (!ev.targetId) return;
-  if (ev.targetEntity === "entry") {
-    const live = await k
-      .selectFrom("entries")
-      .select("id")
-      .where("id", "=", ev.targetId)
-      .where("user_id", "=", userId)
-      .where("deleted_at", "is", null)
-      .executeTakeFirst();
-    if (live) await softDeleteEntry(db, userId, live.id, actor, k);
-  } else if (ev.targetEntity === "engagement") {
-    const live = await k
       .selectFrom("engagements as g")
       .innerJoin("entries as e", "e.id", "g.entry_id")
       .select("g.id")
@@ -171,7 +160,51 @@ async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent
       .where("g.deleted_at", "is", null)
       .where("e.deleted_at", "is", null)
       .executeTakeFirst();
-    if (live) await softDeleteEngagement(db, userId, live.id, actor, k);
+    return r?.id ?? null;
+  }
+  if (ev.targetEntity === "list") {
+    const r = await k
+      .selectFrom("lists")
+      .select("id")
+      .where("id", "=", ev.targetId)
+      .where("user_id", "=", userId)
+      .where("deleted_at", "is", null)
+      .executeTakeFirst();
+    return r?.id ?? null;
+  }
+  return null;
+}
+
+/** Write `value` to a live target's field as `actor`; with `value` undefined, re-stamp the current value. */
+async function setField(db: Db, k: Executor, userId: string, entity: string, id: string, field: string, value: unknown, actor: Actor): Promise<void> {
+  if (entity === "entry") {
+    const r = await k.selectFrom("entries").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+    const v = value === undefined ? entryValues(r)[field] : value;
+    await upsertEntry(db, userId, r.work_id, { [field]: v } as Partial<EntryPatch>, actor, k);
+  } else if (entity === "engagement") {
+    const r = await k.selectFrom("engagements").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+    const v = value === undefined ? engagementValues(r)[field] : value;
+    await updateEngagement(db, userId, id, { [field]: v } as Partial<EngagementPatch>, actor, k);
+  } else if (entity === "list") {
+    // Lists have no field_sources: ownership comes from history, so keep_mine needs no write.
+    if (value === undefined) return;
+    await applyPatch(k, db, userId, { entity: "list", id }, { [field]: value }, actor);
+  } else {
+    throw new ValidationError("targetEntity", "unsupported");
+  }
+}
+
+async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent, targetId: string, actor: Actor): Promise<void> {
+  if (ev.targetEntity === "entry") return softDeleteEntry(db, userId, targetId, actor, k);
+  if (ev.targetEntity === "list") return softDeleteList(db, userId, targetId, actor, k);
+  if (ev.targetEntity === "engagement") {
+    // The same play may still be reported by other sources: detach this one, delete only when none remain.
+    if (ev.connectionId) {
+      await k.deleteFrom("engagement_sources").where("engagement_id", "=", targetId).where("connection_id", "=", ev.connectionId).execute();
+    }
+    const left = await k.selectFrom("engagement_sources").select("id").where("engagement_id", "=", targetId).executeTakeFirst();
+    if (left) return;
+    await softDeleteEngagement(db, userId, targetId, actor, k);
   }
 }
 
@@ -191,7 +224,9 @@ async function mergeTargetWork(db: Db, userId: string, ev: ChangeEvent, actor: A
     .where("key", "=", intoKey)
     .executeTakeFirst();
   if (!target) throw new ValidationError("intoKey", "unknown_record");
-  if (target.target_id) {
+  // A record matched but not yet applied stores its work directly.
+  if (target.target_entity === "work" && target.target_id) return target.target_id;
+  if (target.target_id && (target.target_entity === "entry" || target.target_entity === "engagement")) {
     const entryId =
       target.target_entity === "entry"
         ? target.target_id
@@ -214,25 +249,9 @@ async function mergeTargetWork(db: Db, userId: string, ev: ChangeEvent, actor: A
   return (await matchWork(db, userId, hint, actor)).workId;
 }
 
-async function moveTo(db: Db, k: Executor, userId: string, ev: ChangeEvent, workId: string, actor: Actor): Promise<void> {
+/** Move the live entry to the work; `moveEntryToWork` re-points source records if it merges. */
-  if (!ev.targetId) return;
+async function moveTo(db: Db, k: Executor, userId: string, entryId: string, workId: string, actor: Actor): Promise<void> {
-  const entry = await k
+  const entry = await k.selectFrom("entries").select("work_id").where("id", "=", entryId).executeTakeFirstOrThrow();
-    .selectFrom("entries")
-    .select(["id", "work_id"])
-    .where("id", "=", ev.targetId)
-    .where("user_id", "=", userId)
-    .where("deleted_at", "is", null)
-    .executeTakeFirst();
-  if (!entry) throw new NotFoundError("entry");
   if (entry.work_id === workId) return;
-  const result = await moveEntryToWork(db, userId, entry.id, workId, actor, k);
+  await moveEntryToWork(db, userId, entryId, workId, actor, k);
-  if (result.id !== entry.id) {
-    // The entry merged into an existing one: records that mapped to it now map to the survivor.
-    await k
-      .updateTable("source_records")
-      .set({ target_id: result.id })
-      .where("target_entity", "=", "entry")
-      .where("target_id", "=", entry.id)
-      .execute();
-  }
 }

packages/core/src/sync/listens.ts

@@ -31,10 +31,14 @@ async function trackWork(db: Db, recording: string, l: ListenFields, actor: Acto
   }
 }
 
+/** Listens per query/transaction. 500 rows × 12 columns stays well under both dialects' parameter limits. */
+const BATCH = 500;
+
 /**
  * Append listens (spec §6.4): insert-or-ignore on (user, source, source_key), where `source_key` is the
  * record key. A work is linked only through `mbids.recording`. Non-listen records are ignored.
- * No change events are written.
+ * No change events are written. Per batch: one lookup of existing keys, each distinct recording resolved
+ * once (outside the transaction; `createWork` has its own), then one multi-row insert.
  */
 export async function ingestListens(
   db: Db,
@@ -43,38 +47,48 @@ export async function ingestListens(
   records: SourceRecord[],
   actor: Actor = `plugin:${source}:ingest`,
 ): Promise<{ inserted: number }> {
+  const listens = records.filter((r): r is Extract<SourceRecord, { kind: "listen" }> => r.kind === "listen");
   let inserted = 0;
-  for (const r of records) {
+  for (let i = 0; i < listens.length; i += BATCH) {
-    if (r.kind !== "listen") continue;
+    const batch = new Map<string, ListenFields>();
-    const exists = await db.k
+    for (const r of listens.slice(i, i + BATCH)) if (!batch.has(r.key)) batch.set(r.key, r.listen);
+    const existing = await db.k
       .selectFrom("listens")
-      .select("id")
+      .select("source_key")
       .where("user_id", "=", userId)
       .where("source", "=", source)
-      .where("source_key", "=", r.key)
+      .where("source_key", "in", [...batch.keys()])
-      .executeTakeFirst();
+      .execute();
-    if (exists) continue;
+    for (const e of existing) batch.delete(e.source_key);
-    const l = r.listen;
+    if (batch.size === 0) continue;
-    const recording = l.mbids?.recording;
+
-    const workId = recording ? await trackWork(db, recording, l, actor) : null;
+    const works = new Map<string, string>();
-    const res = await db.k
+    for (const l of batch.values()) {
-      .insertInto("listens")
+      const rec = l.mbids?.recording;
-      .values({
+      if (rec && !works.has(rec)) works.set(rec, await trackWork(db, rec, l, actor));
-        id: newId(),
+    }
-        user_id: userId,
+    const now = nowIso();
-        listened_at: l.listenedAt,
+    const rows = [...batch].map(([key, l]) => ({
-        artist: l.artist,
+      id: newId(),
-        track: l.track,
+      user_id: userId,
-        album: l.album ?? null,
+      listened_at: l.listenedAt,
-        duration_ms: l.durationMs ?? null,
+      artist: l.artist,
-        mbids: toJson(l.mbids ?? {}),
+      track: l.track,
-        work_id: workId,
+      album: l.album ?? null,
-        source,
+      duration_ms: l.durationMs ?? null,
-        source_key: r.key,
+      mbids: toJson(l.mbids ?? {}),
-        created_at: nowIso(),
+      work_id: l.mbids?.recording ? works.get(l.mbids.recording)! : null,
-      })
+      source,
-      .onConflict((oc) => oc.columns(["user_id", "source", "source_key"]).doNothing())
+      source_key: key,
-      .executeTakeFirst();
+      created_at: now,
+    }));
+    const res = await db.k.transaction().execute((tx) =>
+      tx
+        .insertInto("listens")
+        .values(rows)
+        .onConflict((oc) => oc.columns(["user_id", "source", "source_key"]).doNothing())
+        .executeTakeFirst(),
+    );
     inserted += Number(res.numInsertedOrUpdatedRows ?? 0n);
   }
   return { inserted };

packages/core/src/sync/run.ts

@@ -5,7 +5,7 @@ import type { ConnectionStatus, Database, RunMode } from "@sttracker/db/schema";
 import type { PluginContext, SourcePlugin, SourceRecord } from "@sttracker/plugin-sdk/types";
 import { NotFoundError, ValidationError } from "../errors.ts";
 import type { Actor } from "../history.ts";
-import { applyRecord, emptyCounts, markRemoved, type RunCounts, type SyncState } from "./apply.ts";
+import { applyRecord, committed, emptyCounts, markRemoved, type RunCounts, type SyncState } from "./apply.ts";
 import { ingestListens } from "./listens.ts";
 
 /** A `source_connections` row with JSON parsed. Credentials stay encrypted; the caller's `makeContext` decrypts. */
@@ -92,7 +92,7 @@ export async function runSync(deps: SyncDeps, plugin: SourcePlugin, connectionId
     .execute();
 
   const actor: Actor = `plugin:${conn.pluginId}:${conn.id}`;
-  const st: SyncState = { db, userId: conn.userId, connectionId: conn.id, runId, actor, counts: emptyCounts(), log: [] };
+  const st: SyncState = { db, userId: conn.userId, connectionId: conn.id, pluginId: conn.pluginId, runId, actor, counts: emptyCounts(), log: [] };
   const seen = new Set<string>();
   const errors: string[] = [];
   let pullError: string | null = null;
@@ -140,7 +140,7 @@ export async function runSync(deps: SyncDeps, plugin: SourcePlugin, connectionId
         .execute();
       for (const r of live) {
         if (seen.has(r.key)) continue;
-        await db.k.transaction().execute((tx) => markRemoved(tx, st, r));
+        await committed(st, (local) => db.k.transaction().execute((tx) => markRemoved(tx, local, r)));
       }
     } catch (e) {
       errors.push(`removal: ${message(e)}`);

packages/core/test/sync/helpers.ts

@@ -108,3 +108,7 @@ export async function entryFor(db: Db, userId: string, isbn: string) {
     .executeTakeFirst();
   return row ? { ...row, fieldSources: JSON.parse(row.field_sources) as Record<string, string> } : undefined;
 }
+
+export function listRec(key: string, name: string, items: { work: WorkHint; note?: string }[], description?: string): SourceRecord {
+  return { key, kind: "list", list: { name, items, ...(description !== undefined ? { description } : {}) }, raw: { key } };
+}

packages/core/test/sync/listens.test.ts

@@ -45,6 +45,29 @@ forEachDialect("listens", (getDb) => {
     expect(await ingestListens(db, bob, "listenbrainz", records.slice(0, 1))).toEqual({ inserted: 1 });
   });
 
+  test("1200 listens ingest correctly and idempotently", async () => {
+    const db = getDb();
+    const userId = await makeUser(db);
+    const records = Array.from({ length: 1200 }, (_, i) =>
+      listenRec(`k${i}`, {
+        listenedAt: `2024-01-01T00:00:${String(i % 60).padStart(2, "0")}Z`,
+        artist: "A",
+        track: `T${i % 7}`,
+        ...(i % 3 === 0 ? { mbids: { recording: `rec-${i % 7}` } } : {}),
+      }),
+    );
+    // A duplicate key inside the same call is inserted once.
+    records.push(records[5]!);
+    expect(await ingestListens(db, userId, "lb", records)).toEqual({ inserted: 1200 });
+    expect(await count(db, "listens")).toBe(1200);
+    expect(await count(db, "works")).toBe(7);
+    const linked = await db.k.selectFrom("listens").select("work_id").where("work_id", "is not", null).execute();
+    expect(linked).toHaveLength(400);
+    expect(await ingestListens(db, userId, "lb", records)).toEqual({ inserted: 0 });
+    expect(await count(db, "listens")).toBe(1200);
+    expect(await count(db, "works")).toBe(7);
+  });
+
   test("runSync routes listens to ingestListens and counts them", async () => {
     const db = getDb();
     const userId = await makeUser(db);

packages/core/test/sync/run.test.ts

@@ -1,6 +1,7 @@
 import { expect, test } from "bun:test";
 import { forEachDialect } from "@sttracker/db/testing";
-import { softDeleteEntry, upsertEntry } from "../../src/entries.ts";
+import { moveEntryToWork, softDeleteEntry, upsertEntry } from "../../src/entries.ts";
+import { updateList } from "../../src/lists.ts";
 import { listChanges, resolveChange } from "../../src/sync/changes.ts";
 import { runSync } from "../../src/sync/run.ts";
 import { NotFoundError, ValidationError } from "../../src/errors.ts";
@@ -15,8 +16,10 @@ import {
   fakePlugin,
   type FakeSource,
   listenRec,
+  listRec,
   makeConnection,
   makeUser,
+  makeWork,
 } from "./helpers.ts";
 
 forEachDialect("sync engine", (getDb) => {
@@ -295,26 +298,42 @@ forEachDialect("sync engine", (getDb) => {
     expect(conflicts).toHaveLength(1);
     expect(conflicts[0]!.payload).toMatchObject({ field: "rating", current: 80, currentSource: actorA, incoming: 60, incomingSource: actorB });
 
-    // Rerun both: nothing new. Then B flips 60 → 70 → 60: the conflict for 60 is not filed twice.
+    // Rerun both: nothing new.
     await runSync(deps(db), pluginA, connA, "full");
     await runSync(deps(db), pluginB, connB, "full");
     expect(await events(db, { type: "conflict" })).toHaveLength(1);
+
+    // B moves 60 → 70: the stale pending conflict is superseded, not left beside the new one.
     srcB.records = [entryRec("b", book("1"), { rating: 70, status: "completed" })];
     await runSync(deps(db), pluginB, connB, "full");
+    let all = await events(db, { type: "conflict" });
+    expect(all.map((c) => [c.payload.incoming, c.status])).toEqual([[60, "ignored"], [70, "pending"]]);
+    expect(all[0]!.payload.supersededBy).toBe(all[1]!.id);
+    expect(all[0]!.resolved_at).not.toBeNull();
+
+    // And back 70 → 60: still exactly one pending conflict for the field.
     srcB.records = [entryRec("b", book("1"), { rating: 60, status: "completed" })];
     await runSync(deps(db), pluginB, connB, "full");
     conflicts = await events(db, { type: "conflict", status: "pending" });
-    expect(conflicts.map((c) => c.payload.incoming)).toEqual([60, 70]);
+    expect(conflicts.map((c) => c.payload.incoming)).toEqual([60]);
 
     await resolveChange(db, userId, conflicts[0]!.id, "keep_mine");
     const kept = (await entryFor(db, userId, "1"))!;
     expect(kept.rating).toBe(80);
     expect(kept.fieldSources.rating).toBe("user");
-    expect((await events(db, { type: "conflict" }))[0]!.status).toBe("kept_mine");
+    all = await events(db, { type: "conflict" });
+    expect(all.find((c) => c.id === conflicts[0]!.id)!.status).toBe("kept_mine");
 
+    // Reruns file nothing; a NEW value from B is now a user-decided pending change, not a conflict.
     await runSync(deps(db), pluginA, connA, "full");
     await runSync(deps(db), pluginB, connB, "full");
-    expect(await events(db, { type: "conflict" })).toHaveLength(2);
+    expect(await events(db, { type: "conflict" })).toHaveLength(3);
+    srcB.records = [entryRec("b", book("1"), { rating: 65, status: "completed" })];
+    await runSync(deps(db), pluginB, connB, "full");
+    expect(await events(db, { type: "conflict" })).toHaveLength(3);
+    const changed = await events(db, { type: "changed", status: "pending" });
+    expect(changed).toHaveLength(1);
+    expect(changed[0]!.payload).toMatchObject({ field: "rating", from: 80, to: 65 });
     expect((await entryFor(db, userId, "1"))!.rating).toBe(80);
   });
 
@@ -373,6 +392,189 @@ forEachDialect("sync engine", (getDb) => {
     expect((await events(db, { type: "merged" }))[0]!.status).toBe("accepted");
   });
 
+  test("a deduped play with a disagreeing value files a conflict instead of dropping it", async () => {
+    const { db, userId, connectionId: connA, src: srcA, plugin: pluginA, actor: actorA } = await setup("alpha");
+    const connB = await makeConnection(db, userId, "beta");
+    const pluginB = fakePlugin("beta", { records: [engagementRec("r", book("1"), { finishedAt: { value: "2024-03-01", precision: "day" }, rating: 90 })] });
+    srcA.records = [engagementRec("p", book("1"), { finishedAt: { value: "2024-03-01T20:00:00Z", precision: "day" }, rating: 80 })];
+    await runSync(deps(db), pluginA, connA, "full");
+    const s = await runSync(deps(db), pluginB, connB, "full");
+    expect(s.counts.conflicts).toBe(1);
+    const egs = await db.k.selectFrom("engagements").selectAll().execute();
+    expect(egs).toHaveLength(1);
+    expect(egs[0]!.rating).toBe(80);
+    expect(egs[0]!.finished_at).toBe("2024-03-01T20:00:00Z"); // the matching date is not a conflict
+    const c = await events(db, { type: "conflict", status: "pending" });
+    expect(c).toHaveLength(1);
+    expect(c[0]!.target_entity).toBe("engagement");
+    expect(c[0]!.payload).toMatchObject({ field: "rating", current: 80, currentSource: actorA, incoming: 90 });
+  });
+
+  test("a merged-away record that vanishes files no removal; the survivor is untouched", async () => {
+    const { db, userId, connectionId, src, plugin } = await setup();
+    src.records = [entryRec("x", book("1"), { status: "completed" })];
+    await runSync(deps(db), plugin, connectionId, "full");
+    src.records = [entryRec("x", book("1"), { status: "completed" }, { mergedIntoKey: "y" }), entryRec("y", book("2"), {})];
+    await runSync(deps(db), plugin, connectionId, "full");
+    // Before the user decides, x vanishing files nothing either.
+    src.records = [entryRec("y", book("2"), {})];
+    expect((await runSync(deps(db), plugin, connectionId, "full")).counts.removed).toBe(0);
+    expect(await events(db, { type: "removed" })).toEqual([]);
+
+    const [merged] = await events(db, { type: "merged", status: "pending" });
+    await resolveChange(db, userId, merged!.id, "accept");
+    const survivor = (await entryFor(db, userId, "2"))!;
+    const s = await runSync(deps(db), plugin, connectionId, "full");
+    expect(s.counts.removed).toBe(0);
+    expect(await events(db, { type: "removed" })).toEqual([]);
+    const after = (await entryFor(db, userId, "2"))!;
+    expect(after.deleted_at).toBeNull();
+    expect(after.status).toBe(survivor.status);
+  });
+
+  test("an id-less record whose apply fails is retried against the same draft", async () => {
+    const { db, userId, connectionId, src, plugin } = await setup();
+    const hint = { title: "Obscure Zine", mediaType: "book" as const, externalIds: {} };
+    src.records = [entryRec("csv-1", hint, { rating: 150 })];
+    const failed = await runSync(deps(db), plugin, connectionId, "full");
+    expect(failed.status).toBe("failed");
+    expect(failed.counts.added).toBe(0);
+    src.records = [entryRec("csv-1", hint, { rating: 60 })];
+    const ok = await runSync(deps(db), plugin, connectionId, "full");
+    expect(ok.status).toBe("ok");
+    expect(ok.counts.added).toBe(1);
+    const drafts = await db.k.selectFrom("works").select("id").where("owner_user_id", "=", userId).execute();
+    expect(drafts).toHaveLength(1);
+    const e = await db.k.selectFrom("entries").selectAll().where("user_id", "=", userId).execute();
+    expect(e.map((x) => [x.work_id, x.rating])).toEqual([[drafts[0]!.id, 60]]);
+  });
+
+  test("a record-level error fails the run and skips removal", async () => {
+    const { db, connectionId, src, plugin } = await setup();
+    src.records = [entryRec("a", book("1"), { status: "completed" }), entryRec("b", book("2"), { status: "planned" })];
+    await runSync(deps(db), plugin, connectionId, "full");
+    src.records = [entryRec("a", book("1"), { status: "completed", rating: 999 })];
+    const s = await runSync(deps(db), plugin, connectionId, "full");
+    expect(s.status).toBe("failed");
+    expect(s.error).toContain("record a");
+    expect(s.counts.removed).toBe(0);
+    expect(s.counts.changed).toBe(0); // the failed record's work rolled back, and is not counted
+    expect(await events(db, { type: "removed" })).toEqual([]);
+  });
+
+  test("after the user move-merges an entry, the source keeps syncing into the survivor", async () => {
+    const { db, userId, connectionId, src, plugin, actor } = await setup();
+    src.records = [entryRec("a", book("1"), { rating: 50 })];
+    await runSync(deps(db), plugin, connectionId, "full");
+    const e1 = (await entryFor(db, userId, "1"))!;
+    const w2 = await makeWork(db, { title: "Other" });
+    const e2 = await upsertEntry(db, userId, w2, { status: "planned" }, "user");
+    await moveEntryToWork(db, userId, e1.id, w2, "user");
+    const rec = await db.k.selectFrom("source_records").selectAll().where("key", "=", "a").executeTakeFirstOrThrow();
+    expect(rec.target_id).toBe(e2.id);
+
+    src.records = [entryRec("a", book("1"), { rating: 60 })];
+    const s = await runSync(deps(db), plugin, connectionId, "full");
+    expect(s.counts.autoApplied).toBe(1);
+    const survivor = await db.k.selectFrom("entries").selectAll().where("id", "=", e2.id).executeTakeFirstOrThrow();
+    expect(survivor.rating).toBe(60);
+    expect(JSON.parse(survivor.field_sources).rating).toBe(actor);
+  });
+
+  test("accepting removal of a play another source still reports only detaches this source", async () => {
+    const { db, userId, connectionId: connA, src: srcA, plugin: pluginA } = await setup("alpha");
+    const connB = await makeConnection(db, userId, "beta");
+    const srcB: FakeSource = { records: [engagementRec("r", book("1"), { finishedAt: { value: "2024-03-01", precision: "day" } })] };
+    const pluginB = fakePlugin("beta", srcB);
+    srcA.records = [engagementRec("p", book("1"), { finishedAt: { value: "2024-03-01", precision: "day" } })];
+    await runSync(deps(db), pluginA, connA, "full");
+    await runSync(deps(db), pluginB, connB, "full");
+    const [eg] = await db.k.selectFrom("engagements").selectAll().execute();
+
+    srcA.records = [];
+    await runSync(deps(db), pluginA, connA, "full");
+    const [removedA] = await events(db, { type: "removed", status: "pending" });
+    expect(removedA!.target_id).toBe(eg!.id);
+    await resolveChange(db, userId, removedA!.id, "accept");
+    let row = await db.k.selectFrom("engagements").selectAll().where("id", "=", eg!.id).executeTakeFirstOrThrow();
+    expect(row.deleted_at).toBeNull();
+    const left = await db.k.selectFrom("engagement_sources").select("connection_id").where("engagement_id", "=", eg!.id).execute();
+    expect(left.map((x) => x.connection_id)).toEqual([connB]);
+
+    srcB.records = [];
+    await runSync(deps(db), pluginB, connB, "full");
+    const [removedB] = await events(db, { type: "removed", status: "pending" });
+    await resolveChange(db, userId, removedB!.id, "accept");
+    row = await db.k.selectFrom("engagements").selectAll().where("id", "=", eg!.id).executeTakeFirstOrThrow();
+    expect(row.deleted_at).not.toBeNull();
+  });
+
+  test("accepting a change whose target is gone resolves it as ignored (target_gone)", async () => {
+    const { db, userId, connectionId, src, plugin } = await setup();
+    src.records = [entryRec("a", book("1"), { rating: 80 })];
+    await runSync(deps(db), plugin, connectionId, "full");
+    const e = (await entryFor(db, userId, "1"))!;
+    await upsertEntry(db, userId, e.work_id, { rating: 90 }, "user");
+    src.records = [entryRec("a", book("1"), { rating: 70 })];
+    await runSync(deps(db), plugin, connectionId, "full");
+    const [pending] = await events(db, { type: "changed", status: "pending" });
+    await softDeleteEntry(db, userId, e.id, "user");
+    await resolveChange(db, userId, pending!.id, "accept");
+    const [ev] = await events(db, { type: "changed" });
+    expect(ev!.status).toBe("ignored");
+    expect(ev!.payload.reason).toBe("target_gone");
+    expect(ev!.resolved_at).not.toBeNull();
+  });
+
+  test("list records: create, update, user-edited → pending, removed", async () => {
+    const { db, userId, connectionId, src, plugin, actor } = await setup();
+    const draftHint = { title: "Zine", mediaType: "book" as const, externalIds: {} };
+    src.records = [listRec("l1", "Favourites", [{ work: book("1") }, { work: draftHint, note: "rare" }])];
+    const s1 = await runSync(deps(db), plugin, connectionId, "full");
+    expect(s1.counts.added).toBe(1);
+    const list = await db.k.selectFrom("lists").selectAll().where("user_id", "=", userId).executeTakeFirstOrThrow();
+    expect([list.name, list.description, list.kind]).toEqual(["Favourites", "Imported from fake", "list"]);
+    const itemsOf = async () =>
+      (await db.k.selectFrom("list_items as i").innerJoin("works as w", "w.id", "i.work_id").select(["w.title", "i.note"]).where("i.list_id", "=", list.id).orderBy("i.position").execute()).map((r) => [r.title, r.note]);
+    expect(await itemsOf()).toEqual([["Book 1", null], ["Zine", "rare"]]);
+    const rec = await db.k.selectFrom("source_records").selectAll().where("key", "=", "l1").executeTakeFirstOrThrow();
+    expect([rec.target_entity, rec.target_id]).toEqual(["list", list.id]);
+
+    // Source update, list never touched by the user: name and items applied. The ID-less item reuses its draft.
+    src.records = [listRec("l1", "Top", [{ work: draftHint, note: "rare" }, { work: book("2") }])];
+    await runSync(deps(db), plugin, connectionId, "full");
+    expect((await db.k.selectFrom("lists").select("name").where("id", "=", list.id).executeTakeFirstOrThrow()).name).toBe("Top");
+    expect(await itemsOf()).toEqual([["Zine", "rare"], ["Book 2", null]]);
+    expect(await db.k.selectFrom("works").select("id").where("owner_user_id", "=", userId).execute()).toHaveLength(1);
+    const applied = await events(db, { type: "changed", status: "auto_applied" });
+    expect(applied.map((e) => e.payload.field).sort()).toEqual(["items", "name"]);
+
+    // The user edits the list: further source changes become pending.
+    await updateList(db, userId, list.id, { name: "Mine" }, "user");
+    src.records = [listRec("l1", "Top 2", [{ work: book("3") }])];
+    const s3 = await runSync(deps(db), plugin, connectionId, "full");
+    expect(s3.counts.pending).toBe(2);
+    expect((await db.k.selectFrom("lists").select("name").where("id", "=", list.id).executeTakeFirstOrThrow()).name).toBe("Mine");
+    expect(await itemsOf()).toEqual([["Zine", "rare"], ["Book 2", null]]);
+    const pending = await events(db, { type: "changed", status: "pending" });
+    expect(pending.map((e) => e.payload.field).sort()).toEqual(["items", "name"]);
+    const itemsEv = pending.find((e) => e.payload.field === "items")!;
+    expect((itemsEv.payload.to as unknown[]).length).toBe(1);
+    await resolveChange(db, userId, itemsEv.id, "accept");
+    expect(await itemsOf()).toEqual([["Book 3", null]]);
+    const hist = await db.k.selectFrom("user_history").select("actor").where("entity_id", "=", list.id).orderBy("created_at", "desc").orderBy("id", "desc").executeTakeFirstOrThrow();
+    expect(hist.actor).toBe(actor);
+
+    // Removed → pending; accept soft-deletes the list.
+    src.records = [];
+    await runSync(deps(db), plugin, connectionId, "full");
+    const [removed] = await events(db, { type: "removed", status: "pending" });
+    expect(removed!.target_entity).toBe("list");
+    expect((await db.k.selectFrom("lists").select("deleted_at").where("id", "=", list.id).executeTakeFirstOrThrow()).deleted_at).toBeNull();
+    await resolveChange(db, userId, removed!.id, "accept");
+    expect((await db.k.selectFrom("lists").select("deleted_at").where("id", "=", list.id).executeTakeFirstOrThrow()).deleted_at).not.toBeNull();
+  });
+
   test("resolveChange is scoped to the user", async () => {
     const { db, userId, connectionId, src, plugin } = await setup();
     src.records = [entryRec("a", book("1"), { status: "completed" })];