feat(core): sync engine and changes inbox

runSync snapshots and diffs source records, applies new records and
never-edited fields, files pending changed/conflict/removed/merged
events, de-duplicates plays across sources and ingests listens.
listChanges/resolveChange drive the Changes inbox.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/core/src/sync/run.ts

@@ -0,0 +1,164 @@
+import type { Db } from "@sttracker/db/db";
+import { newId, nowIso } from "@sttracker/db/ids";
+import { fromJson, toJson } from "@sttracker/db/json";
+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 { ingestListens } from "./listens.ts";
+
+/** A `source_connections` row with JSON parsed. Credentials stay encrypted; the caller's `makeContext` decrypts. */
+export interface Connection {
+  id: string;
+  userId: string;
+  pluginId: string;
+  name: string;
+  config: Record<string, unknown>;
+  credentialsEnc: string | null;
+  scheduleMinutes: number | null;
+  cursor: string | null;
+  status: ConnectionStatus;
+  lastRunAt: string | null;
+  createdAt: string;
+}
+
+export interface SyncDeps {
+  db: Db;
+  /** Builds the plugin context (http client, credentials, cursor, logging). Injected by the caller. */
+  makeContext: (conn: Connection, runId: string) => PluginContext;
+}
+
+export type SyncMode = "full" | "changes" | { file: { name: string; bytes: Uint8Array } };
+
+export interface RunSummary {
+  runId: string;
+  status: "ok" | "failed";
+  counts: RunCounts;
+  error?: string;
+}
+
+const LISTEN_BATCH = 500;
+
+function toConnection(r: Database["source_connections"]): Connection {
+  return {
+    id: r.id,
+    userId: r.user_id,
+    pluginId: r.plugin_id,
+    name: r.name,
+    config: fromJson<Record<string, unknown>>(r.config, {}),
+    credentialsEnc: r.credentials_enc,
+    scheduleMinutes: r.schedule_minutes,
+    cursor: r.cursor,
+    status: r.status,
+    lastRunAt: r.last_run_at,
+    createdAt: r.created_at,
+  };
+}
+
+function iterate(plugin: SourcePlugin, ctx: PluginContext, mode: SyncMode): AsyncIterable<SourceRecord> {
+  const id = plugin.manifest.id;
+  if (mode === "full") {
+    if (!plugin.pull) throw new Error(`plugin "${id}" cannot pull`);
+    return plugin.pull(ctx);
+  }
+  if (mode === "changes") {
+    if (!plugin.pullChanges) throw new Error(`plugin "${id}" cannot pull changes`);
+    return plugin.pullChanges(ctx);
+  }
+  if (!plugin.importFile) throw new Error(`plugin "${id}" cannot import files`);
+  return plugin.importFile(ctx, mode.file);
+}
+
+const message = (e: unknown) => (e instanceof Error ? e.message : String(e));
+
+/**
+ * One sync run (spec §8.3): pull records, snapshot and diff each against its previous version, apply
+ * or file change events, and append listens. Absent records are marked removed only after a `full`
+ * pull whose iteration finished and whose records all applied. Any failure marks the run `failed` and
+ * the connection `error`; the summary is returned, not thrown.
+ */
+export async function runSync(deps: SyncDeps, plugin: SourcePlugin, connectionId: string, mode: SyncMode): Promise<RunSummary> {
+  const { db } = deps;
+  const row = await db.k.selectFrom("source_connections").selectAll().where("id", "=", connectionId).executeTakeFirst();
+  if (!row) throw new NotFoundError("connection");
+  if (row.plugin_id !== plugin.manifest.id) throw new ValidationError("plugin", "connection_plugin_mismatch");
+  const conn = toConnection(row);
+  const runId = newId();
+  const runMode: RunMode = typeof mode === "string" ? mode : "file";
+  await db.k
+    .insertInto("source_runs")
+    .values({ id: runId, connection_id: conn.id, mode: runMode, started_at: nowIso(), finished_at: null, status: "running", counts: toJson(emptyCounts()), log: "", error: null })
+    .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 seen = new Set<string>();
+  const errors: string[] = [];
+  let pullError: string | null = null;
+  let listens: SourceRecord[] = [];
+  const flushListens = async () => {
+    if (listens.length === 0) return;
+    const batch = listens;
+    listens = [];
+    try {
+      st.counts.listens += (await ingestListens(db, conn.userId, conn.pluginId, batch, actor)).inserted;
+    } catch (e) {
+      errors.push(`listens: ${message(e)}`);
+    }
+  };
+
+  try {
+    const ctx = deps.makeContext(conn, runId);
+    for await (const rec of iterate(plugin, ctx, mode)) {
+      st.counts.seen++;
+      if (rec.kind === "listen") {
+        listens.push(rec);
+        if (listens.length >= LISTEN_BATCH) await flushListens();
+        continue;
+      }
+      seen.add(rec.key);
+      try {
+        await applyRecord(st, rec);
+      } catch (e) {
+        errors.push(`record ${rec.key}: ${message(e)}`);
+      }
+    }
+  } catch (e) {
+    pullError = message(e);
+  }
+  await flushListens();
+
+  // Removal by absence: only a completed full pull with every record applied can prove a record is gone.
+  if (mode === "full" && pullError === null && errors.length === 0) {
+    try {
+      const live = await db.k
+        .selectFrom("source_records")
+        .selectAll()
+        .where("connection_id", "=", conn.id)
+        .where("deleted_at", "is", null)
+        .execute();
+      for (const r of live) {
+        if (seen.has(r.key)) continue;
+        await db.k.transaction().execute((tx) => markRemoved(tx, st, r));
+      }
+    } catch (e) {
+      errors.push(`removal: ${message(e)}`);
+    }
+  }
+
+  const failed = pullError !== null || errors.length > 0;
+  const error = pullError ?? (errors.length > 0 ? `${errors.length} error(s); first: ${errors[0]}` : undefined);
+  const log = [...st.log, ...errors];
+  if (pullError !== null) log.push(`pull failed: ${pullError}`);
+  const now = nowIso();
+  await db.k
+    .updateTable("source_runs")
+    .set({ finished_at: now, status: failed ? "failed" : "ok", counts: toJson(st.counts), log: log.join("\n"), error: error ?? null })
+    .where("id", "=", runId)
+    .execute();
+  const nextStatus: ConnectionStatus = failed ? "error" : conn.status === "error" ? "active" : conn.status;
+  await db.k.updateTable("source_connections").set({ last_run_at: now, status: nextStatus }).where("id", "=", conn.id).execute();
+
+  return { runId, status: failed ? "failed" : "ok", counts: st.counts, ...(error !== undefined ? { error } : {}) };
+}