sync

Incremental sync: the cursor/watermark contract that makes a source updated in place instead of polled whole, and the three ways a source is captured — polling-diff (the portable baseline every connector gets), provider-delta (a vendor token carried, compared and resumed, never interpreted) and log (a backend change-log position). The PURE half is thirteen Ring 1 verbs: the declaration checker, the effective [capture] of a verb, the baseline watermark, the next poll's intent, the dedup key (an identity plus the store's own content address), the page diff, the advance — the one function every refusal reaches (a regression, an epoch change, a stall, a run bound, an overlap, and the kit's host-escape, header-injection and cursor-length refusals) — the epoch read, the reseed plan, the horizon prune, the journal event of a delta, the audit record of a run and the operator's status. No store, no journal, no clock and no capability: $now, the watermark, the index and the page are always arguments, so every decision grades offline. The advance is max(version) minus overlap, never a clock; a delta names a state, never a patch; dedup is by content address over a declared identity and cannot be turned off. The EXECUTING half is seven platform-group verbs — open (resolve a source in the host, open its connector through the kit, bind the journal and the store; every boot refusal), close, run (one capture: the lease and the load, the plan, the read through the kit's walk, the diff, the append of every delta to the adapter stream, the advance, the ONE compare-and-swap append of the watermark on the source's own journal stream — a second writer refused CXER1114 — and one audit record), status, pause, resume and reseed (a reason, the deployment's grant and a person). mode=log executes over a sqlite source's change table — installed from the declaration at open by the kit, read in ordinal order, every insert, update and delete in the order the source committed it, a lost or reset position refused CXER6413 — over a postgres source's WAL through logical decoding and a mysql source's binary log through the replication protocol (cx-platform-connector#27), and is refused CXER6401 at open over any backend with no log.

sync:validate

[$sync:validate] ($source::element $opts::map {}) -> element — Check a [capture], a [gateway] carrying one, or a whole [feature] against sync.md §2.1: [findings [finding code= where= message=]…] by default, the first finding raised as its own err under {strict: true}. An opts map that would raise a §8 bound refuses CXER6400 whatever strict says.

      [?lib 'cx-platform/sync']
[?let [= $f [$sync:validate [gateway name=pg kind=sql rung=complete-ordered
              [capture mode=log kind=ordinal key='/id' log=logical-slot]]]]
  [probe [findings [$count [?for [in $it $f/finding] [yield $it]]]]]]

      [probe [findings 0]]

sync:capture-of

[$sync:capture-of] ($gateway::element $verb::element) -> any — The EFFECTIVE [capture] for one verb: the verb's own where it carries one, otherwise the gateway's, otherwise the absence channel — an ordinary non-incremental verb, not an error.

      [?lib 'cx-platform/sync']
[?let [= $g [gateway name=acme-api kind=http rung=snapshot-diff
              [capture mode=polling-diff kind=instant key='/id' version='/updated_at' overlap=5m]]]
      [= $v [verb name=list-events effect=observe via=acme-api
              [capture mode=provider-delta kind=opaque key='/id' cursor-path='/next']]]
      [= $c [$sync:capture-of $g $v]]
  [probe [mode [$string $c@mode]] [kind [$string $c@kind]]]]

      [probe [mode provider-delta] [kind opaque]]

sync:watermark-init

[$sync:watermark-init] ($capture::element $source::string $tenant::string) -> element — The baseline [watermark] for a first run: the declaration's mode and kind, seq=0, status=:complete, no cursor, no epoch, an empty [horizon].

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=provider-delta kind=ordinal key='/uid' cursor-path='/highest-modseq']]
      [= $w [$sync:watermark-init $c 'mail/imap' 't1']]
  [probe [mode $w@mode] [kind $w@kind] [tenant [$string $w@tenant]]]]

      [probe [mode :provider-delta] [kind :ordinal] [tenant t1]]

sync:plan-poll

[$sync:plan-poll] ($capture::element $watermark::element $intent::element) -> element — The [intent] for the next read: the watermark's cursor placed in the declared since-param= slot and nothing else changed; no since-param= (an unfiltered poll) or no cursor yet (a baseline) answers the intent unchanged.

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=polling-diff kind=instant key='/id' version='/updated_at' since-param=updated-since overlap=5m]]
      [= $w [watermark source='acme/acme-api' tenant='t1' mode=:polling-diff kind=:instant cursor='2026-09-13T09:54:11Z' seq=40 status=:complete]]
      [= $i [$sync:plan-poll $c $w [do 'acme/list-contacts']]]
  [probe [since [$string [$first $i/updated-since]]]]]

      [probe [since '2026-09-13T09:54:11Z']]

sync:dedup-key

[$sync:dedup-key] ($capture::element $record::any) -> element — One record's key: the identity the key= path yields over the record, and the store's content address of its canonical bytes. A path yielding no node, a non-scalar, an empty string or over 1 KiB refuses CXER6408.

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=polling-diff kind=instant key='/id' version='/updated_at' overlap=5m]]
      [= $k [$sync:dedup-key $c [row [id 'c-17'] [email 'ada@example.test']]]]
  [probe [identity [$string $k@identity]] [tagged [$starts-with [$string $k@address] 'sha2-256:']]]]

      [probe [identity c-17] [tagged true]]

sync:diff-page

[$sync:diff-page] ($capture::element $index::element $page::element $now::datetime $opts::map {}) -> element — Diff one page against the dedup index — each element child of the page is one record: [diff suppressed=N [delta …]… [index …]], creates, changes carrying prior=, suppressions counted and not emitted, deletes only on an unfiltered read that names its known identities, a record over 16 MiB dead-lettered as op=:refused. Two records with one identity refuse the page CXER6408.

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=polling-diff kind=instant key='/id' version='/updated_at' overlap=5m]]
      [= $ix [index source='acme/acme-api' count=0 from='2026-09-01T00:00:00Z']]
      [= $pg [rows [row [id 'c-17']]]]
      [= $d [$sync:diff-page $c $ix $pg 2026-09-13T10:05:00Z]]
  [probe [op [$first [?for [in $it $d/delta] [yield $it]]]@op] [suppressed $d@suppressed]]]

      [probe [op :create] [suppressed 0]]

sync:advance

[$sync:advance] ($capture::element $watermark::element $observed::element $stop::element) -> element — The next [watermark], or the refusal: the cursor advanced under kind='s ordering (an instant to max(version) minus overlap, never $now), seq= counted, a stall counted and refused past §8's bound, a non-completing stop left :in-flight. Every refusal the module owns is reachable here, with no socket.

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=polling-diff kind=instant key='/id' version='/updated_at' since-param=updated-since overlap=5m]]
      [= $w [watermark source='acme/acme-api' tenant='t1' mode=:polling-diff kind=:instant cursor='2026-09-13T09:00:00Z' seq=4 status=:complete]]
      [= $o [observed max-version='2026-09-13T10:04:11Z' records=18]]
      [= $n [$sync:advance $c $w $o [stop reason=:absent]]]
  [probe [cursor [$string $n@cursor]]]]

      [probe [cursor '2026-09-13T09:59:11Z']]

sync:epoch-of

[$sync:epoch-of] ($capture::element $response::element) -> any — The source's epoch from epoch-path= over one decoded [response], or the absence channel where the capture declares none.

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=provider-delta kind=ordinal key='/uid' cursor-path='/highest-modseq' epoch-path='/uidvalidity']]
  [probe [epoch [$string [$sync:epoch-of $c [response status=200 [body '{"uidvalidity":8,"highest-modseq":980}']]]]]]]

      [probe [epoch '8']]

sync:reseed-plan

[$sync:reseed-plan] ($capture::element $watermark::element $reason::string $opts::map {}) -> element — What a reseed invalidates and keeps: [reseed cursor=:reset epoch=:reset index=:keep|:clear seq=:preserve reason=…], the index arm from identities=. An empty reason, or an agent: actor, refuses CXER6410 — a reseed rests on a person.

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=provider-delta kind=ordinal key='/uid' cursor-path='/highest-modseq' epoch-path='/uidvalidity' identities=epoch-scoped]]
      [= $w [watermark source='mail/imap' tenant='t1' mode=:provider-delta kind=:ordinal cursor='918' epoch='7' seq=9 status=:complete]]
      [= $p [$sync:reseed-plan $c $w 'uidvalidity changed']]
  [probe [cursor $p@cursor] [epoch $p@epoch] [index $p@index] [seq $p@seq]]]

      [probe [cursor :reset] [epoch :reset] [index :clear] [seq :preserve]]

sync:horizon-prune

[$sync:horizon-prune] ($index::element $now::datetime $bounds::map {}) -> element — Prune the dedup index by count AND by age, oldest first, whichever binds first; a bound above its §8 ceiling refuses CXER6400.

      [?lib 'cx-platform/sync']
[?let [= $ix [index source='acme/acme-api' count=3 from='2026-09-01T00:00:00Z'
               [at identity='c-1' address='sha2-256:9b8fb6a4563c44a1c8dc955577c9233a7c7234e64e7a9e8f37cf54e9d9fc4105' at='2026-09-01T00:00:00Z']
               [at identity='c-2' address='sha2-256:5e37ac2dd1db3e512e726d440d4a8952bb1122b8b18e71a7467b87cc849bf354' at='2026-09-02T00:00:00Z']
               [at identity='c-3' address='sha2-256:526e18f6700e4caeaf9c8ac84b97ed4d877b9aa9a6c79b714641f2edbc5f5236' at='2026-09-03T00:00:00Z']]]
      [= $p [$sync:horizon-prune $ix 2026-09-04T00:00:00Z {max-entries: 2}]]
  [probe [count $p@count] [first [$string [$first [?for [in $it $p/at] [yield $it]]]@identity]]]]

      [probe [count 2] [first c-2]]

sync:delta-record

[$sync:delta-record] ($capture::element $delta::element $facts::map {}) -> element — The journal event for one delta: a STATE, never a patch — the record carried through unchanged and its content address computed.

      [?lib 'cx-platform/sync']
[?let [= $c [capture mode=polling-diff kind=instant key='/id' version='/updated_at' overlap=5m]]
      [= $d [delta source='acme/acme-api' op=:change identity='c-17'
                   address='sha2-256:fd8d491cfe0ada074dfc8ce0253dd533f48f91ffa08725ccf5139104f19c95b9' prior='sha2-256:06dd45b664aeeb25eb319c8b6fc4d750aba76f00e018a9c73e4227dce0d1ef18'
                   observed-at='2026-09-13T10:04:11Z'
              [record [row [id 'c-17'] [email 'ada@example.test'] [name 'Ada']]]]]
      [= $e [$sync:delta-record $c $d]]
  [probe [records [$count [?for [in $it $e//record] [yield $it]]]] [patches [$count [?for [in $it $e//patch] [yield $it]]]] [op $e@op]]]

      [probe [records 1] [patches 0] [op :change]]

sync:audit-record

[$sync:audit-record] ($run::element $outcome $facts::map {}) -> element — The one [audit module=sync] record of a run: action=:capture or :reseed, decision= :announced, :refused or :denied from the outcome, and a [detail [capture-run …]] of counts and both cursors — never record content, never a URL.

      [?lib 'cx-platform/sync']
[?let [= $r [run id='r-41' source='acme/acme-api/list-contacts' tenant='t1' mode=:polling-diff kind=:instant
                 pages=3 records=412 stop=:absent status=:complete seq=41
             [counts create=11 change=7 delete=0 refused=0]]]
      [= $a [$sync:audit-record $r :complete]]
  [probe [module [$string $a@module]] [action $a@action] [decision $a@decision]]]

      [probe [module sync] [action :capture] [decision :announced]]

sync:status-of

[$sync:status-of] ($watermark::element $now::datetime) -> element — The [sync-status] an agent and an operator share: state, cursor, epoch, seq, the lag against $now, the stall count and the horizon — the watermark projected, no index and no lease.

      [?lib 'cx-platform/sync']
[?let [= $w [watermark source='acme/acme-api' tenant='t1' mode=:polling-diff kind=:instant
                       cursor='2026-09-13T09:59:11Z' seq=41 status=:paused at='2026-09-13T10:04:11Z']]
      [= $s [$sync:status-of $w 2026-09-13T10:09:11Z]]
  [probe [state $s@state]]]

      [probe [state :paused]]

sync:open

[$sync:open] ($host::element $source::string $opts::map {}) -> element — Open one source for capture: resolve <feature>/<gateway>[/<verb>] in $host, read its [capture] from the feature, open the connector handle it reads through (its credential by handle, through the kit), bind the journal and the store the watermark and its index live in and the adapter stream the deltas land on, and answer a [sync source= tenant= stream= deltas=] handle. Every boot refusal is here: CXER6400 (an invalid declaration or a raised bound), CXER6401 (mode=log over a backend with no log this engine reads — sqlite, postgres and mysql are read — or a mode or kind this engine does not page); on mode=log over sqlite, open installs the change table and its triggers through the kit (connector:log-install), and a change table the declaration does not describe is the kit's CXER6300, CXER6411 (no store and no journal), the audit sink's CXER6202; net is checked first (CXER0271).

      [?lib 'cx-platform/sync']
[$sync:open [host tenant='acme'] 'acme/acme-api']

      cx-err:CXER0271

sync:close

[$sync:close] ($sync::element) -> null — Release the handle and the connector it reads through. Idempotent: closing a closed handle is null, never an error, and close charges no capability.

      [?lib 'cx-platform/sync']
[$sync:close [sync source='acme/acme-api' tenant='acme' state=:closed]]

      null

sync:run

[$sync:run] ($sync::element $opts::map {}) -> element — One capture run, §3.1's eleven steps in order: the load at the stream's head position (the lease), the epoch, the plan, the read through connector:walk, the diff of every page, the append of every delta to the adapter stream, the advance, the index put and the watermark committed by ONE journal:append with expect-pos — a second writer is refused CXER1114 — and one audit record on every path. Answers the [run] record; {dry-run: true} decides everything and writes nothing. A paused source is refused CXER6414 before a byte is read.

      [?lib 'cx-platform/sync']
[$sync:run [sync source='acme/acme-api' tenant='acme']]

      cx-err:CXER0271

sync:status

[$sync:status] ($sync::element $opts::map {}) -> element — The persisted watermark, read fresh from its stream and projected through status-of — the one read an operator's face and an agent's share; {at: <datetime>} fixes the instant the lag is computed against.

      [?lib 'cx-platform/sync']
[$sync:status [sync source='acme/acme-api' tenant='acme']]

      cx-err:CXER0271

sync:pause

[$sync:pause] ($sync::element $opts::map {}) -> element — Stop a source without unbinding it: the watermark re-committed :paused with its cursor where it was, every later run refused CXER6414 until resume. Idempotent; answers the [sync-status].

      [?lib 'cx-platform/sync']
[$sync:pause [sync source='acme/acme-api' tenant='acme']]

      cx-err:CXER0271

sync:resume

[$sync:resume] ($sync::element $opts::map {}) -> element — Admit a paused source's runs again, the next one starting from the cursor the pause held. Idempotent; answers the [sync-status].

      [?lib 'cx-platform/sync']
[$sync:resume [sync source='acme/acme-api' tenant='acme']]

      cx-err:CXER0271

sync:reseed

[$sync:reseed] ($sync::element $reason::string $opts::map {}) -> element — A new baseline on a person's word: a reason, the deployment's grant (sync:reseed over the qualified source, decided by authz) and never an agent's face — absent any of them CXER6410. The cursor and the epoch reset, seq= preserved, the index kept or cleared as reseed-plan decides ({index: :keep | :clear} overrides it, audited), nothing rewritten; one audit record with action=:reseed. Answers the [reseed] decision.

      [?lib 'cx-platform/sync']
[$sync:reseed [sync source='acme/acme-api' tenant='acme'] 'uidvalidity changed']

      cx-err:CXER0271