diff --git a/docker/omnigraph-server.Dockerfile b/docker/omnigraph-server.Dockerfile index ed3834ad..88b434ae 100644 --- a/docker/omnigraph-server.Dockerfile +++ b/docker/omnigraph-server.Dockerfile @@ -36,8 +36,8 @@ ARG OMNIGRAPH_VERSION=0.10.0 # Kept separate because on a moving tag the two differ — see # witan_core/omnigraph_install.py :: _OMNIGRAPH_RELEASE_TAG. ARG OMNIGRAPH_RELEASE_TAG=edge -ARG OMNIGRAPH_SHA256_X86_64=68099e33941cc5c252f36d4c2a26f1dfff6b28e4eb627ba1f7bf098856d34349 -ARG OMNIGRAPH_SHA256_ARM64=d0e42176625584370a26c8a6ca43bc24447ea822483c46d971807475a5a5782a +ARG OMNIGRAPH_SHA256_X86_64=6b53f3eff3793d4012fd6b5c151c5589390699ce42661e3ce2ff1ed18b995f4c +ARG OMNIGRAPH_SHA256_ARM64=11c06689fcffa6e5fc9fbe9f632ba8e493f1f42997bbd10f09440f5e1d582aab # ── Fetch + checksum-verify the release, extract both binaries ──────────────── FROM debian:trixie-slim@sha256:3a39a0592364683e6bab97937b72cad5a8fa6dcbbee90edb3bb48c7f8e94f258 AS fetch diff --git a/docker/witan.Dockerfile b/docker/witan.Dockerfile index c53358d3..971ace01 100644 --- a/docker/witan.Dockerfile +++ b/docker/witan.Dockerfile @@ -28,8 +28,8 @@ ARG OMNIGRAPH_VERSION=0.10.0 # Kept separate because on a moving tag the two differ — see # witan_core/omnigraph_install.py :: _OMNIGRAPH_RELEASE_TAG. ARG OMNIGRAPH_RELEASE_TAG=edge -ARG OMNIGRAPH_SHA256_X86_64=68099e33941cc5c252f36d4c2a26f1dfff6b28e4eb627ba1f7bf098856d34349 -ARG OMNIGRAPH_SHA256_ARM64=d0e42176625584370a26c8a6ca43bc24447ea822483c46d971807475a5a5782a +ARG OMNIGRAPH_SHA256_X86_64=6b53f3eff3793d4012fd6b5c151c5589390699ce42661e3ce2ff1ed18b995f4c +ARG OMNIGRAPH_SHA256_ARM64=11c06689fcffa6e5fc9fbe9f632ba8e493f1f42997bbd10f09440f5e1d582aab # Keep in lockstep with witan-council's version (mcp/servers/witan/pyproject.toml # [project].version / [tool.bumpversion]); it labels the built image. ARG WITAN_VERSION=0.8.0 diff --git a/docs/_data/environment.toml b/docs/_data/environment.toml index 5ee83f49..a1d8b2eb 100644 --- a/docs/_data/environment.toml +++ b/docs/_data/environment.toml @@ -105,6 +105,10 @@ desc = "How long before nominal expiry a cached token is treated as already expi default = "~/.config/witan/tokens.json" desc = "Where both CLIs cache OIDC tokens. Shared on purpose, next to the shared config file." +[section.vars.WITAN_MERGE_WATERMARKS] +default = "~/.config/witan/merge-watermarks.json" +desc = "Where `witan migrate merge` records what each pair of stores looked like when they last agreed, so the next merge can name the nodes both sides have written since. Beside the token cache, and per-machine — losing it costs one merge's divergence report." + [section.vars.WITAN_REMOTE_CALL_BUDGET_SECONDS] default = "0" desc = "Deadline for a single remote graph call, used to decide whether to honour a server's retry hint or give up. `0` means no client-side deadline — obey the server's hints." diff --git a/docs/guides/migration-runbook.md b/docs/guides/migration-runbook.md index 2b8e5d86..2df59078 100644 --- a/docs/guides/migration-runbook.md +++ b/docs/guides/migration-runbook.md @@ -81,6 +81,11 @@ rather than in your environment. Read the decisions: `added` should be roughly the row count of your store, and `updated` should be small. A large `updated` on a first migration means slugs are colliding that shouldn't — stop there. +This run has no watermark to compare against, so it cannot tell you whether any +collision is a divergence — and being a dry run it records none either, so +step 4 is equally blind. The first run that can report divergence is a merge +*after* step 4 has succeeded. See [Divergence](#divergence). + **4. Run it:** ```bash @@ -224,7 +229,7 @@ witan migrate merge [SOURCE] [--from ] [--to ] [--target ] [--d | `--from ` | A `[targets.]` block's `server`, in place of `SOURCE`. A target with only a `remote_url` is refused — nothing local to export. | | `--to ` | A `[targets.]` block as the destination: through its deployment if it has a `remote_url`, into its `server` store if not. | | `--target ` | A destination store URI. Defaults to your configured store. Mutually exclusive with `--to`; `.jsonl` is refused (a target is a graph, not a snapshot). | -| `--dry-run` | Print the per-slug decisions, write nothing. | +| `--dry-run` | Print the per-slug decisions, write nothing. Reports divergence; records no watermark. | Notes: @@ -243,6 +248,55 @@ Notes: - Reconciliation covers nodes only. Edge rows (`Tagged`, `ParentOf`, …) have no slug and pass through unreconciled, same as raw `--mode merge`. +## Divergence + +Newest-record-wins is a whole-**record** decision, and several witan fields are +append-only logs rather than values — `WorkflowProject.description`, which +accretes status blocks, most of all. When both stores have written the same node +since they last agreed, keeping the newer record does not resolve a stale value; +it deletes the other side's text. + +Every merge therefore records a **watermark** for the pair of stores: the newest +timestamp in the source, and the newest that will be in the target once this +merge's winners land. The next merge uses it to name the nodes both sides have +written since: + +``` +2 node(s) changed on BOTH sides since the last merge (2026-08-19T19:46:00Z). +Newest-record-wins keeps one side and drops the other's edit … + WorkflowProject wp-witan-multi-user-service-deployment-dcf6ee + source 2026-08-19T19:49:00Z target 2026-08-19T22:06:00Z -> kept target +``` + +Nothing is merged for you. Reconcile the named slugs by hand — read both sides, +write the combined value to whichever store you want to win, and re-run — then +the merge resolves them on its own rule. + +- **The first merge of a pair has no watermark and says so.** That is "cannot + tell", not "nothing diverged"; until one is recorded, diff the projects you + care about yourself. +- `--dry-run` reports divergence but records no watermark: the mark describes a + target with this merge's winners in it, and a dry run wrote none of them. +- Marks live in `~/.config/witan/merge-watermarks.json` + (`$WITAN_MERGE_WATERMARKS`), beside the token cache, keyed by source store and + destination. Per-machine, and losing the file costs one merge's reporting. + Local paths are keyed by their resolved absolute path, so `graph.omni`, + `../graph.omni` and `file:///…/graph.omni` share one mark rather than three. +- **A merge that fails part-way leaves no mark.** The standing one is retired + before the first batch commits and a fresh one installed only on success, + because batches commit independently: rows from a half-finished merge are + already in the target, and a mark that predates them would read those rows as + an independent target edit. The next run says it cannot tell, which is true. + Re-run the merge to get back to a marked state. +- Each side is compared against its own mark, which keeps the source's clock + out of the target's threshold and vice versa — a laptop and a cluster do not + agree closely enough for a cross-clock comparison. One documented exception: + the rows a merge loads carry their source timestamps into the target, so the + target mark is raised to cover them (otherwise every row a merge added would + come back as a target edit). Under a source clock running ahead, that leaves + a blind window the width of the skew in which a genuine target edit is not + reported. + ## Fallback: in-cluster merge (operator) Use when the MCP tier is unavailable, or to merge on someone else's behalf. diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 4613fe5d..9a08116a 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -1056,6 +1056,12 @@ entirely. Rows only in ``source`` are always added; rows only in the target are left untouched. Repeatable — re-running against an already-merged target loads nothing new. +Each merge records a per-side watermark for the pair of stores, so the next +one can name the nodes BOTH sides have written since — the case where +newest-record-wins is not resolving a stale value but discarding somebody's +edit. Nothing is auto-merged; the divergent slugs are reported for you to +reconcile. The first merge of a pair has no watermark and says so. + **Parameters**: * `SOURCE, --source`: Store URI to merge from (local path, ``s3://``, ``file://``, or an diff --git a/docs/reference/environment.md b/docs/reference/environment.md index 65ce5e02..a6468421 100644 --- a/docs/reference/environment.md +++ b/docs/reference/environment.md @@ -54,6 +54,7 @@ token; both `witan` and `witan code` share one cache, so you log in once. | Variable | Default | Description | | --- | --- | --- | +| `WITAN_MERGE_WATERMARKS` | `~/.config/witan/merge-watermarks.json` | Where `witan migrate merge` records what each pair of stores looked like when they last agreed, so the next merge can name the nodes both sides have written since. Beside the token cache, and per-machine — losing it costs one merge's divergence report. | | `WITAN_OIDC_AUDIENCE` | — | Audience/resource to request, matching the deployment's own `WITAN_OIDC_AUDIENCE`. Sent on the device-auth and token requests so an issuer with an audience mapper stamps the right `aud` claim. | | `WITAN_OIDC_CLIENT_ID` | — | OIDC client id presented during the device grant. | | `WITAN_OIDC_EXPIRY_SKEW_SECONDS` | `90` | How long before nominal expiry a cached token is treated as already expired and refreshed. Sized so a refresh happens before a long write starts rather than partway through one. | diff --git a/docs/reference/mcp-tools/memory.md b/docs/reference/mcp-tools/memory.md index 6ad3738d..bb265547 100644 --- a/docs/reference/mcp-tools/memory.md +++ b/docs/reference/mcp-tools/memory.md @@ -301,6 +301,8 @@ carry no slug and pass through additively, exactly as they do there. | `rows` | list[object] | **required** | One batch of ``omnigraph export`` records — ``{"type": Node, "data":
{…}}`` for a node, ``{"edge": Edge, "from": …, "to": …}`` for an edge. | | `dry_run` | bool | `False` | Reconcile and return the per-row ``decisions`` **without writing
anything**. Run the whole migration this way first: it is the only way
to see which side wins each ``(type, slug)`` before the graph changes. | | `claim_from_author` | str? | `null` | The identity the *source* store wrote, when that store is your own.
Rows authored by exactly this name are restamped to the calling actor
before they are written; every other row keeps its author untouched.
Pass your local ``cfg.author`` here — the server cannot derive it,
having neither the caller's config nor their git checkout.
Without it, a migrated row keeps a name that no deployed identity can
ever match, and ``memory_delete`` refuses its own author forever
(#267). With it, the rows you migrate end up owned by the same identity
that owns everything you write afterwards. See ``_claim_authorship``
for why this matches rather than stamping unconditionally. | +| `since` | object? | `null` | The watermark recorded by the caller's *previous* merge into this graph
(``{"source_ts", "target_ts"}``). Supplied, a collision both sides have
written since then is marked ``diverged`` in the decisions — the merge
rule is unchanged, the losing edit is simply no longer discarded in
silence. Same value on every batch of one merge. | +| `watermark` | object? | `null` | The running watermark returned by the *previous batch of this same
merge*, folded into this batch's. The source is split across batches, so
no single batch sees its newest row; carrying the running value is what
lets the client end up with a mark covering the whole merge without
having to compare exported timestamps itself. ``None`` on the first
batch. | ## `claim_authorship` diff --git a/mcp/servers/witan/CHANGELOG.md b/mcp/servers/witan/CHANGELOG.md index 9c6788e7..747368b6 100644 --- a/mcp/servers/witan/CHANGELOG.md +++ b/mcp/servers/witan/CHANGELOG.md @@ -6,6 +6,47 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/) (pre-1.0: a MINOR bump may include breaking changes). +## [0.29.0] - 2026-08-24 + +### Fixed + +- **`witan migrate merge` no longer discards a divergent edit in silence.** + Reconciliation is newest-record-wins per node, which is right when one side is + simply stale. It is not the same thing when BOTH stores have written the same + node since they last agreed: there the loser's edit is deleted, and several + witan fields are append-only logs rather than values — + `WorkflowProject.description`, which accretes status blocks, above all. Hit + for real on 2026-08-19, where the local store held the fourth update to a + project description and the deployment held the fifth, and merging either way + would have dropped one permanently. + + The loss was invisible: a discarded divergence counted as `kept`, the same + bucket as the 3000-odd nodes that genuinely needed no action, and `--dry-run` + printed `kept-target` for it exactly as it did for them. + + A merge now records a watermark for the pair of stores — the newest timestamp + in the source, and the newest that will be in the target once its winners land + — in `~/.config/witan/merge-watermarks.json` (`$WITAN_MERGE_WATERMARKS`). The + next merge uses it to name every node both sides have written since, with both + timestamps and which side was kept. The merge rule is unchanged and nothing is + auto-merged; what changes is that the slugs to reconcile by hand are now + nameable. The decisions carry a `diverged` flag and the counts a `diverged` + total, which cuts across `updated`/`kept_target` rather than partitioning with + them. + + Each side is only ever compared against its own mark — a source is a laptop's + clock and a deployed target is a cluster's, and comparing across them would + read skew as divergence. The first merge of a pair has no watermark and says + so, which is "cannot tell", not "nothing diverged". `--dry-run` reports + divergence but records nothing, since it wrote none of the winners its mark + would describe. + + Over the deployment the watermark crosses as a `store_merge` parameter and is + threaded batch to batch, so the mark the client records covers the whole merge + rather than its last batch. A deployment too old to return one yields no mark + at all rather than a partial one, and the CLI says the next merge will be + blind. + ## [0.28.0] - 2026-08-21 ### Fixed diff --git a/mcp/servers/witan/docs/migration-runbook.md b/mcp/servers/witan/docs/migration-runbook.md index d943b156..63007212 100644 --- a/mcp/servers/witan/docs/migration-runbook.md +++ b/mcp/servers/witan/docs/migration-runbook.md @@ -71,6 +71,11 @@ rather than in your environment. Read the decisions: `added` should be roughly the row count of your store, and `updated` should be small. A large `updated` on a first migration means slugs are colliding that shouldn't — stop there. +This run has no watermark to compare against, so it cannot tell you whether any +collision is a divergence — and being a dry run it records none either, so +step 4 is equally blind. The first run that can report divergence is a merge +*after* step 4 has succeeded. See [Divergence](#divergence). + **4. Run it:** ```bash @@ -214,7 +219,7 @@ witan migrate merge [SOURCE] [--from ] [--to ] [--target ] [--d | `--from ` | A `[targets.]` block's `server`, in place of `SOURCE`. A target with only a `remote_url` is refused — nothing local to export. | | `--to ` | A `[targets.]` block as the destination: through its deployment if it has a `remote_url`, into its `server` store if not. | | `--target ` | A destination store URI. Defaults to your configured store. Mutually exclusive with `--to`; `.jsonl` is refused (a target is a graph, not a snapshot). | -| `--dry-run` | Print the per-slug decisions, write nothing. | +| `--dry-run` | Print the per-slug decisions, write nothing. Reports divergence; records no watermark. | Notes: @@ -233,6 +238,55 @@ Notes: - Reconciliation covers nodes only. Edge rows (`Tagged`, `ParentOf`, …) have no slug and pass through unreconciled, same as raw `--mode merge`. +## Divergence + +Newest-record-wins is a whole-**record** decision, and several witan fields are +append-only logs rather than values — `WorkflowProject.description`, which +accretes status blocks, most of all. When both stores have written the same node +since they last agreed, keeping the newer record does not resolve a stale value; +it deletes the other side's text. + +Every merge therefore records a **watermark** for the pair of stores: the newest +timestamp in the source, and the newest that will be in the target once this +merge's winners land. The next merge uses it to name the nodes both sides have +written since: + +``` +2 node(s) changed on BOTH sides since the last merge (2026-08-19T19:46:00Z). +Newest-record-wins keeps one side and drops the other's edit … + WorkflowProject wp-witan-multi-user-service-deployment-dcf6ee + source 2026-08-19T19:49:00Z target 2026-08-19T22:06:00Z -> kept target +``` + +Nothing is merged for you. Reconcile the named slugs by hand — read both sides, +write the combined value to whichever store you want to win, and re-run — then +the merge resolves them on its own rule. + +- **The first merge of a pair has no watermark and says so.** That is "cannot + tell", not "nothing diverged"; until one is recorded, diff the projects you + care about yourself. +- `--dry-run` reports divergence but records no watermark: the mark describes a + target with this merge's winners in it, and a dry run wrote none of them. +- Marks live in `~/.config/witan/merge-watermarks.json` + (`$WITAN_MERGE_WATERMARKS`), beside the token cache, keyed by source store and + destination. Per-machine, and losing the file costs one merge's reporting. + Local paths are keyed by their resolved absolute path, so `graph.omni`, + `../graph.omni` and `file:///…/graph.omni` share one mark rather than three. +- **A merge that fails part-way leaves no mark.** The standing one is retired + before the first batch commits and a fresh one installed only on success, + because batches commit independently: rows from a half-finished merge are + already in the target, and a mark that predates them would read those rows as + an independent target edit. The next run says it cannot tell, which is true. + Re-run the merge to get back to a marked state. +- Each side is compared against its own mark, which keeps the source's clock + out of the target's threshold and vice versa — a laptop and a cluster do not + agree closely enough for a cross-clock comparison. One documented exception: + the rows a merge loads carry their source timestamps into the target, so the + target mark is raised to cover them (otherwise every row a merge added would + come back as a target edit). Under a source clock running ahead, that leaves + a blind window the width of the skew in which a genuine target edit is not + reported. + ## Fallback: in-cluster merge (operator) Use when the MCP tier is unavailable, or to merge on someone else's behalf. diff --git a/mcp/servers/witan/pyproject.toml b/mcp/servers/witan/pyproject.toml index ad5b0fd2..f58ceb4d 100644 --- a/mcp/servers/witan/pyproject.toml +++ b/mcp/servers/witan/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "witan-council" -version = "0.28.0" +version = "0.29.0" description = "witan — agent memory, planning, and collaboration graph (work-coordination layer + umbrella CLI)" readme = "README.md" license = "BSD-3-Clause" @@ -213,7 +213,7 @@ packages = ["witan"] "schema" = "schema" [tool.bumpversion] -current_version = "0.28.0" +current_version = "0.29.0" allow_dirty = true [[tool.bumpversion.files]] diff --git a/mcp/servers/witan/tests/test_migrate.py b/mcp/servers/witan/tests/test_migrate.py index 6077cb17..e25976e5 100644 --- a/mcp/servers/witan/tests/test_migrate.py +++ b/mcp/servers/witan/tests/test_migrate.py @@ -2054,3 +2054,680 @@ def claim_authorship(self, *, was, apply): cli_migrate.claim_authorship(author) out = capsys.readouterr().out assert author in out, f"{author!r} did not survive rendering: {out!r}" + + +# ── Divergence reporting (tk-witan-migrate-merge-silently-drops-divergent-edi) ── +# +# Newest-record-wins is the right rule and stays the rule. What these cover is +# the case where applying it discards an edit rather than a stale value — both +# stores wrote the same node since they last agreed — which used to land in the +# `kept` bucket, indistinguishable from the thousands of nodes that genuinely +# needed no action. + + +def _node(node_type, slug, updated_at): + return {"type": node_type, "data": {"slug": slug, "updated_at": updated_at}} + + +def _nodes(*rows): + return {(r["type"], r["data"]["slug"]): r for r in rows} + + +_LAST_MERGE = {"source_ts": "2026-06-01T00:00:00Z", "target_ts": "2026-06-01T00:00:00Z"} + + +def test_no_watermark_marks_nothing_diverged(): + """Absent a watermark the answer is "cannot tell", not "nothing diverged". + + Record-level timestamps alone cannot separate a target that is merely ahead + from one that advanced alongside the source, so guessing would put every + ordinary collision in a report whose only virtue is that it is short. + """ + from witan import server as srv + + decisions, _ = srv._reconcile_nodes( + _nodes(_node("Memory", "mem-a", "2026-06-02T00:00:00Z")), + _nodes(_node("Memory", "mem-a", "2026-06-03T00:00:00Z")), + ) + + assert decisions[0]["decision"] == "kept-target" + assert "diverged" not in decisions[0] + assert srv._decision_counts(decisions)["diverged"] == 0 + + +def test_a_target_only_edit_is_not_divergence(): + """The noise case the watermark exists to exclude. + + The source has not been touched since the last merge, so keeping the + target's newer row discards nothing. On a shared graph other people write + constantly, which makes this the common shape of a `kept-target` — reported, + it would bury the real thing. + """ + from witan import server as srv + + decisions, _ = srv._reconcile_nodes( + _nodes(_node("Memory", "mem-a", "2026-05-01T00:00:00Z")), + _nodes(_node("Memory", "mem-a", "2026-06-05T00:00:00Z")), + _LAST_MERGE, + ) + + assert decisions[0]["decision"] == "kept-target" + assert "diverged" not in decisions[0] + + +def test_both_sides_written_since_the_last_merge_is_reported(): + from witan import server as srv + + decisions, winners = srv._reconcile_nodes( + _nodes(_node("WorkflowProject", "wp-x", "2026-06-02T00:00:00Z")), + _nodes(_node("WorkflowProject", "wp-x", "2026-06-03T00:00:00Z")), + _LAST_MERGE, + ) + + assert decisions[0]["diverged"] is True + # Reporting only. The rule is unchanged and so is the outcome: the target's + # row is newer, so it stays and the source row is not written. + assert decisions[0]["decision"] == "kept-target" + assert winners == [] + assert srv._decision_counts(decisions)["diverged"] == 1 + # Readable spellings ride along, because these two are the only timestamps + # a human is asked to compare and an 0.9 export gives them as epoch millis. + assert decisions[0]["source_at"] == "2026-06-02T00:00:00Z" + assert decisions[0]["target_at"] == "2026-06-03T00:00:00Z" + + +def test_epoch_millis_timestamps_are_reported_readably(): + """The representation the deployed graph actually exports. + + Both stores in a real cutover are on omnigraph >= 0.9, so without this the + divergence report hands you two 13-digit integers to compare by eye. + """ + from witan import server as srv + + decisions, _ = srv._reconcile_nodes( + _nodes(_node("WorkflowProject", "wp-x", 1780358400000)), + _nodes(_node("WorkflowProject", "wp-x", 1780444800000)), + {"source_ts": 1780272000000, "target_ts": 1780272000000}, + ) + + assert decisions[0]["diverged"] is True + assert decisions[0]["source_at"] == "2026-06-02T00:00:00Z" + assert decisions[0]["target_at"] == "2026-06-03T00:00:00Z" + # The raw values stay as exported; the readable pair is added, not swapped. + assert decisions[0]["source_ts"] == 1780358400000 + + +def test_divergence_is_reported_when_the_source_wins_too(): + """The direction that overwrites the SHARED graph, which is the worse one. + + A divergence resolved in the source's favour discards an edit somebody else + made in the deployment. It counts as `updated` — a perfectly ordinary-looking + bucket — so without the mark there is nothing at all to notice. + """ + from witan import server as srv + + decisions, winners = srv._reconcile_nodes( + _nodes(_node("WorkflowProject", "wp-x", "2026-06-05T00:00:00Z")), + _nodes(_node("WorkflowProject", "wp-x", "2026-06-02T00:00:00Z")), + _LAST_MERGE, + ) + + assert decisions[0]["decision"] == "updated" + assert decisions[0]["diverged"] is True + assert len(winners) == 1 + + +def test_diverged_count_cuts_across_the_other_buckets(): + """`diverged` overlaps `updated`/`kept_target` rather than partitioning with + them — a divergence is still resolved as one of the two.""" + from witan import server as srv + + decisions, _ = srv._reconcile_nodes( + _nodes( + _node("Memory", "mem-src-wins", "2026-06-05T00:00:00Z"), + _node("Memory", "mem-dst-wins", "2026-06-02T00:00:00Z"), + _node("Memory", "mem-new", "2026-06-05T00:00:00Z"), + ), + _nodes( + _node("Memory", "mem-src-wins", "2026-06-03T00:00:00Z"), + _node("Memory", "mem-dst-wins", "2026-06-04T00:00:00Z"), + ), + _LAST_MERGE, + ) + + counts = srv._decision_counts(decisions) + assert (counts["added"], counts["updated"], counts["kept_target"]) == (1, 1, 1) + assert counts["diverged"] == 2 + assert counts["added"] + counts["updated"] + counts["kept_target"] == len(decisions) + + +def test_the_watermark_covers_the_rows_the_merge_itself_loads(): + """A row this merge writes into the target must not read as "the target + changed" on the next run. + + The winners land carrying their own (source) timestamps, so a mark taken + from the pre-merge target alone sits below them — and every row the merge + added would come back as a divergence next time round. + """ + from witan import server as srv + + source = _nodes(_node("Memory", "mem-a", "2026-06-05T00:00:00Z")) + target = _nodes(_node("Memory", "mem-a", "2026-06-01T00:00:00Z")) + + decisions, winners = srv._reconcile_nodes(source, target) + assert decisions[0]["decision"] == "updated" + + watermark = srv._next_watermark(None, source, target, winners) + assert watermark["target_ts"] == "2026-06-05T00:00:00Z" + + # Replay: the target now holds what the merge just put there, and the + # source is untouched. Nothing diverged. + after, _ = srv._reconcile_nodes( + source, _nodes(_node("Memory", "mem-a", "2026-06-05T00:00:00Z")), watermark + ) + assert "diverged" not in after[0] + + +def test_a_source_clock_ahead_opens_a_blind_window_on_the_target(): + """The one place the per-side clock rule is bent, pinned with its cost. + + Winner timestamps are folded into the target mark so a merge's own rows do + not read as target edits next run. That mixes the source's clock into the + target's threshold, so a source running ahead pushes the mark into the + target's future and hides a genuine target edit made inside the skew. + + Asserting the CURRENT behaviour, not the desired one. If this starts + reporting True, the trade has been removed (per-record baselines) and the + docstring in `_next_watermark` should go with it. + """ + from witan import server as srv + + # Source machine is an hour ahead of the target's clock. + source = _nodes(_node("Memory", "mem-a", "2026-06-01T12:00:00Z")) + target = _nodes(_node("Memory", "mem-a", "2026-06-01T10:00:00Z")) + _, winners = srv._reconcile_nodes(source, target) + watermark = srv._next_watermark(None, source, target, winners) + assert watermark["target_ts"] == "2026-06-01T12:00:00Z" # source's clock + + # A real, independent target edit at 11:30 target-time — after the merge. + skewed, _ = srv._reconcile_nodes( + _nodes(_node("Memory", "mem-a", "2026-06-01T12:30:00Z")), + _nodes(_node("Memory", "mem-a", "2026-06-01T11:30:00Z")), + watermark, + ) + assert "diverged" not in skewed[0], "the blind window closed — update the docs" + + # Same shape with the clocks agreeing reports correctly, which is what + # makes the miss above attributable to skew rather than to the rule. + source = _nodes(_node("Memory", "mem-b", "2026-06-01T10:00:00Z")) + target = _nodes(_node("Memory", "mem-b", "2026-06-01T09:00:00Z")) + _, winners = srv._reconcile_nodes(source, target) + synced, _ = srv._reconcile_nodes( + _nodes(_node("Memory", "mem-b", "2026-06-01T12:30:00Z")), + _nodes(_node("Memory", "mem-b", "2026-06-01T11:30:00Z")), + srv._next_watermark(None, source, target, winners), + ) + assert synced[0]["diverged"] is True + + +def test_the_watermark_accumulates_across_batches(): + """The MCP path splits the source, so no single batch sees its newest row. + + Carrying the running mark forward is what lets the client end up with one + covering the whole merge without comparing exported timestamps itself — + including when the newest row is not in the last batch. + """ + from witan import server as srv + + first = _nodes(_node("Memory", "mem-a", "2026-06-09T00:00:00Z")) + second = _nodes(_node("Memory", "mem-b", "2026-06-02T00:00:00Z")) + + carried = srv._next_watermark(None, first, {}, []) + carried = srv._next_watermark(carried, second, {}, []) + + assert carried["source_ts"] == "2026-06-09T00:00:00Z" + + +def test_each_side_is_compared_against_its_own_mark(): + """Never source-vs-target. The source is a laptop's clock and the target a + cluster's; comparing across them would read skew as divergence.""" + from witan import server as srv + + # The target's whole history sits "before" the source's, as it would under a + # few hours of clock skew. Neither side has moved since its own mark. + decisions, _ = srv._reconcile_nodes( + _nodes(_node("Memory", "mem-a", "2026-06-01T12:00:00Z")), + _nodes(_node("Memory", "mem-a", "2026-06-01T02:00:00Z")), + {"source_ts": "2026-06-01T12:00:00Z", "target_ts": "2026-06-01T02:00:00Z"}, + ) + + assert "diverged" not in decisions[0] + + +def test_watermark_file_round_trips_and_replaces_by_pair(tmp_path, monkeypatch): + from witan import merge_watermark as mw + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + + assert mw.read("/store.omni", "https://witan.example") is None + + assert mw.write( + "/store.omni", "https://witan.example", {"source_ts": 1, "target_ts": 2} + ) + entry = mw.read("/store.omni", "https://witan.example") + assert (entry["source_ts"], entry["target_ts"]) == (1, 2) + + # Re-merging the same pair replaces its mark rather than accumulating one + # per run, while a different destination keeps its own. + mw.write("/store.omni", "https://witan.example", {"source_ts": 3, "target_ts": 4}) + mw.write("/store.omni", "/other.omni", {"source_ts": 9, "target_ts": 9}) + assert mw.read("/store.omni", "https://witan.example")["source_ts"] == 3 + assert mw.read("/store.omni", "/other.omni")["source_ts"] == 9 + assert len(mw._load()) == 2 + + +def test_a_corrupt_watermark_file_reads_as_absent(tmp_path, monkeypatch): + """Fails soft in the direction that loses the report, never the merge. + + A half-written or hand-mangled hint file must not take down a cutover; the + cost of ignoring it is one run without divergence reporting, and the next + successful merge rewrites it. + """ + from witan import merge_watermark as mw + + marks = tmp_path / "marks.json" + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(marks)) + + for garbage in ('{"version": 1, "pairs"', "null", "[]", '{"version": 99}'): + marks.write_text(garbage) + assert mw.read("/store.omni", "/target.omni") is None + + assert mw.write("/store.omni", "/target.omni", {"source_ts": 1, "target_ts": 1}) + assert mw.read("/store.omni", "/target.omni") is not None + + +class _FakeMergeProvider: + """Stands in for either merge provider — the server module or the proxy.""" + + remote_url = "https://witan.example/graphs/council" + + def __init__(self): + self.since_seen = [] + + def merge_store(self, source, *, target, dry_run, source_author, since): + self.since_seen.append(since) + return { + "target": self.remote_url, + # `kept_target: 1` with an empty decision list is a shape a real + # result cannot have — the counts are derived FROM the decisions — + # and faking it hid the "this merge carried nothing" early return. + "decisions": [ + { + "type": "Memory", + "slug": "mem-collided", + "decision": "kept-target", + "source_ts": "2026-05-01T00:00:00Z", + "target_ts": "2026-06-02T00:00:00Z", + } + ], + "added": 0, + "updated": 0, + "kept_target": 1, + "diverged": 0, + "rows_loaded": 0, + "watermark": { + "source_ts": "2026-06-01T00:00:00Z", + "target_ts": "2026-06-02T00:00:00Z", + }, + } + + +def test_merge_records_its_watermark_and_replays_it_next_run(tmp_path, monkeypatch): + from witan.cli import migrate as cli_migrate + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + provider = _FakeMergeProvider() + monkeypatch.setattr( + cli_migrate, "_merge_destination", lambda to, t: (provider, None) + ) + monkeypatch.setattr(cli_migrate, "_merge_source_author", lambda f: "pytest") + + cli_migrate._merge("/store.omni", None, False) + cli_migrate._merge("/store.omni", None, False) + + assert provider.since_seen[0] is None + assert provider.since_seen[1] == { + "source_ts": "2026-06-01T00:00:00Z", + "target_ts": "2026-06-02T00:00:00Z", + } + # Exactly the two marks. The stored entry also records which pair it + # describes, and a merge has no business sending a deployment the local + # path its rows came from. + assert "source" not in provider.since_seen[1] + + +def test_a_dry_run_does_not_record_a_watermark(tmp_path, monkeypatch): + """The mark describes a target with this merge's winners in it, and a dry + run wrote none of them. Recording it would tell the next run that everything + up to here had already been merged.""" + from witan import merge_watermark as mw + from witan.cli import migrate as cli_migrate + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + provider = _FakeMergeProvider() + monkeypatch.setattr( + cli_migrate, "_merge_destination", lambda to, t: (provider, None) + ) + monkeypatch.setattr(cli_migrate, "_merge_source_author", lambda f: "pytest") + + cli_migrate._merge("/store.omni", None, True) + + assert mw.read("/store.omni", provider.remote_url) is None + + +def test_the_watermark_key_matches_between_the_read_and_the_write(): + """Both ends of a merge look the destination up the same way. + + The read happens before the call and the write after it, so a key derived + from the result would silently miss whatever normalization the callee did to + the address — and every merge would look like a first merge. + """ + from witan.cli import migrate as cli_migrate + + class _Module: + client = type("C", (), {"graph_uri": "/configured/graph.omni"})() + + assert ( + cli_migrate._destination_key(_FakeMergeProvider(), None) + == "https://witan.example/graphs/council" + ) + assert cli_migrate._destination_key(_Module(), None) == "/configured/graph.omni" + assert cli_migrate._destination_key(_Module(), "/explicit.omni") == "/explicit.omni" + + +def test_one_store_spelled_several_ways_shares_one_mark(tmp_path, monkeypatch): + """A pair keyed by what the caller typed is keyed by the wrong thing. + + `/tmp/g.omni`, `../tmp/g.omni` and `file:///tmp/g.omni` are one store; a + mark that misses across them measures the next merge against a graph that is + not the one being merged. + """ + from witan import merge_watermark as mw + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + store = tmp_path / "g.omni" + store.mkdir() + + mw.write(str(store), "/target.omni", {"source_ts": 1, "target_ts": 2}) + + monkeypatch.chdir(tmp_path) + for spelling in (str(store), f"file://{store}", "g.omni", "./g.omni"): + entry = mw.read(spelling, "/target.omni") + assert entry is not None, f"{spelling!r} missed its own mark" + assert entry["source_ts"] == 1 + + +def test_a_relative_path_from_two_directories_is_two_stores(tmp_path, monkeypatch): + """The other half of the same bug: one key must not cover two stores. + + `graph.omni` run from two different working directories names two different + graphs, and sharing a mark between them reports divergence against somebody + else's history.""" + from witan import merge_watermark as mw + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + for name in ("a", "b"): + (tmp_path / name).mkdir() + (tmp_path / name / "graph.omni").mkdir() + + monkeypatch.chdir(tmp_path / "a") + mw.write("graph.omni", "/target.omni", {"source_ts": 1, "target_ts": 1}) + + monkeypatch.chdir(tmp_path / "b") + assert mw.read("graph.omni", "/target.omni") is None + + +def test_a_watermark_missing_a_side_is_refused_and_reads_as_absent( + tmp_path, monkeypatch +): + """The worst possible stored value: truthy, so it suppresses the "cannot + tell" notice, but unparseable on both sides, so it detects nothing. Silence + that reads as "nothing diverged" is the one outcome this feature exists to + prevent.""" + from witan import merge_watermark as mw + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + + assert mw.is_usable({"source_ts": 1, "target_ts": 2}) + assert not mw.is_usable({"source_ts": None, "target_ts": None}) + assert not mw.is_usable({"source_ts": 1, "target_ts": None}) + assert not mw.is_usable(None) + + assert ( + mw.write("/s.omni", "/t.omni", {"source_ts": None, "target_ts": None}) is False + ) + assert mw.read("/s.omni", "/t.omni") is None + + +def test_a_non_utf8_watermark_file_still_fails_soft(tmp_path, monkeypatch): + """`read_text` raises UnicodeDecodeError before `json.loads` is reached, and + that is a ValueError but not an OSError — so it escaped the narrower catch + and took down a merge from a module whose docstring promises to fail soft.""" + from witan import merge_watermark as mw + + marks = tmp_path / "marks.json" + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(marks)) + marks.write_bytes(b'{"version": 1, "pairs": [\xff\xfe invalid utf-8 ]}') + + assert mw.read("/s.omni", "/t.omni") is None + assert mw.write("/s.omni", "/t.omni", {"source_ts": 1, "target_ts": 1}) + assert mw.read("/s.omni", "/t.omni") is not None + + +def test_a_partial_merge_leaves_no_mark_rather_than_a_stale_one(tmp_path, monkeypatch): + """Batches commit independently, so a merge that dies part-way has already + put rows in the target. A mark predating those rows reads them as an + independent target edit and reports divergence on rows nothing but the + failed merge ever wrote.""" + from witan import merge_watermark as mw + from witan.cli import migrate as cli_migrate + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + mw.write("/s.omni", "/t.omni", {"source_ts": 1, "target_ts": 1}) + + class _Exploding: + remote_url = "/t.omni" + + def merge_store(self, source, *, target, dry_run, source_author, since): + # Stands in for a merge that commits some batches and then dies. + raise RuntimeError("data tier went away mid-merge") + + monkeypatch.setattr( + cli_migrate, "_merge_destination", lambda to, t: (_Exploding(), None) + ) + monkeypatch.setattr(cli_migrate, "_merge_source_author", lambda f: "pytest") + + with pytest.raises(SystemExit): + cli_migrate._merge("/s.omni", None, False) + + assert mw.read("/s.omni", "/t.omni") is None + + +def test_a_dry_run_keeps_the_standing_mark(tmp_path, monkeypatch): + """Retiring the mark is for runs that WRITE. A dry run commits nothing, so + the standing mark still describes the target accurately and throwing it away + would blind the next real merge for no reason.""" + from witan import merge_watermark as mw + from witan.cli import migrate as cli_migrate + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + mw.write("/s.omni", _FakeMergeProvider.remote_url, {"source_ts": 1, "target_ts": 1}) + + provider = _FakeMergeProvider() + monkeypatch.setattr( + cli_migrate, "_merge_destination", lambda to, t: (provider, None) + ) + monkeypatch.setattr(cli_migrate, "_merge_source_author", lambda f: "pytest") + + cli_migrate._merge("/s.omni", None, True) + + assert mw.read("/s.omni", _FakeMergeProvider.remote_url) is not None + + +def test_a_first_dry_run_does_not_promise_the_next_merge_can_report(capsys): + """A dry run records no mark, so the real merge after it is blind too. The + earliest run that can report is the one after that.""" + from witan.cli import migrate as cli_migrate + + result = {"decisions": [], "updated": 0, "kept_target": 3} + + cli_migrate._report_divergence(result, None, dry_run=True) + dry = capsys.readouterr().out + assert "blind" in dry + + cli_migrate._report_divergence(result, None, dry_run=False) + real = capsys.readouterr().out + assert "the next one will report" in real + + +def test_divergence_report_names_the_slugs_and_which_side_was_kept(capsys): + """Naming the slugs is the deliverable — the manual reconcile takes minutes + once you know where to look, and the summary counts never tell you.""" + from witan.cli import migrate as cli_migrate + + result = { + "decisions": [ + { + "type": "WorkflowProject", + "slug": "wp-witan-multi-user", + "decision": "kept-target", + "source_ts": "2026-08-19T19:49:00Z", + "target_ts": "2026-08-19T22:06:00Z", + "diverged": True, + }, + { + "type": "Memory", + "slug": "mem-untouched", + "decision": "kept-target", + "source_ts": "2026-01-01T00:00:00Z", + "target_ts": "2026-06-01T00:00:00Z", + }, + ] + } + + cli_migrate._report_divergence( + result, {"merged_at": "2026-08-19T19:46:00Z"}, dry_run=True + ) + out = capsys.readouterr().out + + assert "wp-witan-multi-user" in out + assert "kept target" in out + assert "mem-untouched" not in out + + +def test_a_first_merge_says_it_cannot_report_divergence(capsys): + """Silence would read as "nothing diverged", which is the exact confusion + this whole feature exists to remove.""" + from witan.cli import migrate as cli_migrate + + cli_migrate._report_divergence( + {"decisions": [], "updated": 0, "kept_target": 3}, None, dry_run=True + ) + + assert "watermark" in capsys.readouterr().out + + +def test_a_first_merge_that_only_adds_says_nothing(capsys): + """The note is about collisions it cannot judge. With none, it is noise on + the run that needs it least — a fresh cutover into an empty graph.""" + from witan.cli import migrate as cli_migrate + + cli_migrate._report_divergence( + {"decisions": [], "updated": 0, "kept_target": 0}, None, dry_run=True + ) + + assert capsys.readouterr().out == "" + + +def test_a_merge_that_carried_nothing_does_not_warn_about_the_deployment( + tmp_path, monkeypatch, capsys +): + """An empty source reports no mark because there was nothing to mark, not + because the deployment is too old. Warning would send someone hunting a + problem that isn't there.""" + from witan.cli import migrate as cli_migrate + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + cli_migrate._record_watermark( + "/store.omni", "/target.omni", {"decisions": [], "rows_loaded": 0} + ) + + assert capsys.readouterr().out == "" + + +def test_a_merge_that_moved_rows_but_got_no_mark_warns(tmp_path, monkeypatch, capsys): + """The pre-0.29.0 deployment case. A merge that plainly did something and + still returned no mark means the next one cannot report divergence, and + saying nothing would leave that to be discovered by not being told.""" + from witan.cli import migrate as cli_migrate + + monkeypatch.setenv("WITAN_MERGE_WATERMARKS", str(tmp_path / "marks.json")) + cli_migrate._record_watermark( + "/store.omni", + "/target.omni", + {"decisions": [{"slug": "mem-a"}], "rows_loaded": 1}, + ) + + # A short fragment: the console hard-wraps to the terminal width, so a + # longer phrase can arrive with a newline through the middle of it. + assert "No usable merge watermark" in capsys.readouterr().out + + +@requires_omnigraph +def test_merge_reports_divergence_against_two_real_stores(server, tmp_path): + """The 2026-08-19 incident, replayed against actual stores. + + Both sides advance after a merge; the second merge resolves the collision + newest-record-wins and, before this, reported the discarded edit as `kept` — + the same bucket as every node that needed nothing. The point of the assert + is not that a count went up but that the losing slug is now nameable. + """ + from witan import config as cfg_mod + from witan import graph as graph_mod + from witan import server as srv + + slug = "mem-diverged-d1d1d1" + source = graph_mod.OmnigraphClient( + _init_store(tmp_path / "source.omni"), cfg_mod.load().queries_dir + ) + _insert_memory( + source, slug=slug, content="agreed", updated_at="2026-06-01T00:00:00Z" + ) + + first = srv.merge_store(source.graph_uri) + assert first["diverged"] == 0 + watermark = first["watermark"] + assert watermark["source_ts"] and watermark["target_ts"] + + # Both stores are written after that merge — the shape a merge is most + # likely to be run in, since writes going somewhere they should not is + # what prompts one. + _insert_memory( + source, slug=slug, content="edited locally", updated_at="2026-06-02T00:00:00Z" + ) + _insert_memory( + srv.client, + slug=slug, + content="edited in the graph", + updated_at="2026-06-03T00:00:00Z", + ) + + blind = srv.merge_store(source.graph_uri, dry_run=True) + told = srv.merge_store(source.graph_uri, dry_run=True, since=watermark) + + # Same decision either way; the difference is entirely whether the loss is + # visible. Without the watermark it is one of the `kept_target` rows. + assert blind["kept_target"] == told["kept_target"] == 1 + assert blind["diverged"] == 0 + assert told["diverged"] == 1 + assert [d["slug"] for d in told["decisions"] if d.get("diverged")] == [slug] diff --git a/mcp/servers/witan/tests/test_remote_proxy.py b/mcp/servers/witan/tests/test_remote_proxy.py index 6c0ad334..3351aece 100644 --- a/mcp/servers/witan/tests/test_remote_proxy.py +++ b/mcp/servers/witan/tests/test_remote_proxy.py @@ -1008,3 +1008,135 @@ async def _declaring_branch() -> set[str]: f"filled in client-side) or something the caller states outright (add to " f"_BRANCH_IS_EXPLICIT). No longer registered: {classified - declared}" ) + + +@requires_omnigraph +def test_remote_merge_reports_divergence_across_the_mcp_tier(proxy, server, tmp_path): + """Divergence has to survive the transport, not just the in-process path. + + The watermark crosses as a tool parameter and the mark comes back the same + way, so this also pins that a dict parameter round-trips through the MCP + binding — a merge that silently dropped it would report nothing and look + exactly like a clean merge. + """ + from witan import config as cfg_mod + from witan import graph as graph_mod + from witan import server as srv + + from .test_migrate import _init_store, _insert_memory + + slug = "mem-remote-diverged-7c7c7c" + source = graph_mod.OmnigraphClient( + _init_store(tmp_path / "diverge.omni"), cfg_mod.load().queries_dir + ) + _insert_memory( + source, slug=slug, content="agreed", updated_at="2026-06-01T00:00:00Z" + ) + + first = proxy.merge_store(source.graph_uri) + assert first["diverged"] == 0 + watermark = first["watermark"] + assert watermark["source_ts"] and watermark["target_ts"] + + _insert_memory( + source, slug=slug, content="edited locally", updated_at="2026-06-02T00:00:00Z" + ) + _insert_memory( + srv.client, + slug=slug, + content="edited in the deployment", + updated_at="2026-06-03T00:00:00Z", + ) + + told = proxy.merge_store(source.graph_uri, dry_run=True, since=watermark) + + assert told["diverged"] == 1 + assert [d["slug"] for d in told["decisions"] if d.get("diverged")] == [slug] + + +def test_the_watermark_is_carried_between_batches_and_since_is_not( + proxy, tmp_path, monkeypatch +): + """Two marks that look alike and must not be confused. + + `since` is the LAST merge's mark and is the same on every batch of this one. + `watermark` is THIS merge's running mark and has to be threaded call to + call — the source is split across batches, so the value the caller finally + records comes from the last one and must have accumulated the rest. + """ + from witan.remote import proxy as proxy_mod + + export = tmp_path / "handover.jsonl" + export.write_text( + '{"type": "Memory", "data": {"slug": "mem-one", "updated_at": 1}}\n' + '{"type": "Memory", "data": {"slug": "mem-two", "updated_at": 2}}\n' + ) + + seen: list[tuple] = [] + + def _one_row_per_batch(records, max_bytes=None, *args, **kwargs): + return [[row] for row in records] + + def _fake_store_merge(*, rows, dry_run, claim_from_author, since, watermark): + seen.append((since, watermark)) + return { + "decisions": [], + "added": len(rows), + "updated": 0, + "kept_target": 0, + "diverged": 0, + "rows_loaded": len(rows), + "watermark": {"source_ts": len(seen), "target_ts": len(seen)}, + } + + monkeypatch.setattr(proxy_mod, "chunk_records", _one_row_per_batch) + monkeypatch.setattr(proxy, "store_merge", _fake_store_merge) + result = proxy.merge_store( + str(export), since={"source_ts": "mark", "target_ts": "mark"} + ) + + assert len(seen) == 2 + assert [s for s, _ in seen] == [{"source_ts": "mark", "target_ts": "mark"}] * 2 + assert [w for _, w in seen] == [None, {"source_ts": 1, "target_ts": 1}] + assert result["watermark"] == {"source_ts": 2, "target_ts": 2} + + +def test_a_deployment_that_reports_no_watermark_yields_none( + proxy, tmp_path, monkeypatch +): + """Keeping the last batch's mark would describe a fraction of the merge as + the whole of it, and the next run would compare against it and see nothing. + None is the honest answer — the CLI says so and asks for a hand diff.""" + from witan.remote import proxy as proxy_mod + + export = tmp_path / "handover.jsonl" + export.write_text( + '{"type": "Memory", "data": {"slug": "mem-one", "updated_at": 1}}\n' + '{"type": "Memory", "data": {"slug": "mem-two", "updated_at": 2}}\n' + ) + + calls: list[int] = [] + + def _one_row_per_batch(records, max_bytes=None, *args, **kwargs): + return [[row] for row in records] + + def _fake_store_merge(*, rows, dry_run, claim_from_author, since, watermark): + calls.append(1) + result = { + "decisions": [], + "added": 1, + "updated": 0, + "kept_target": 0, + "diverged": 0, + "rows_loaded": 1, + } + # Only the first batch answers; an older server answers on neither. + if len(calls) == 1: + result["watermark"] = {"source_ts": 1, "target_ts": 1} + return result + + monkeypatch.setattr(proxy_mod, "chunk_records", _one_row_per_batch) + monkeypatch.setattr(proxy, "store_merge", _fake_store_merge) + result = proxy.merge_store(str(export)) + + assert result["watermark"] is None diff --git a/mcp/servers/witan/witan/cli/migrate.py b/mcp/servers/witan/witan/cli/migrate.py index 4dcec35c..8bcaf8ed 100644 --- a/mcp/servers/witan/witan/cli/migrate.py +++ b/mcp/servers/witan/witan/cli/migrate.py @@ -7,6 +7,7 @@ import cyclopts +from .. import merge_watermark from ._common import _srv, console, esc, print_error, remote_proxy migrate_app = cyclopts.App( @@ -202,6 +203,93 @@ def _merge_destination(to_target: str | None, target: str | None): ) +def _destination_key(provider, target: str | None) -> str: + """How a merge destination is named in the watermark file. + + Computed from what is known BEFORE the merge — the watermark has to be read + to be passed in — and used again to write, so the two can't disagree. That + rules out ``result["target"]``, which is the same address after the callee + has normalized it. + """ + remote_url = getattr(provider, "remote_url", None) + if remote_url: + return remote_url + if target: + return target + return provider.client.graph_uri + + +def _since_marks(entry: dict | None) -> dict | None: + """Just the two timestamps out of a stored watermark entry.""" + if not entry: + return None + return {"source_ts": entry.get("source_ts"), "target_ts": entry.get("target_ts")} + + +def _report_divergence(result: dict, since: dict | None, dry_run: bool) -> None: + """Name the nodes both stores wrote, which reconciliation resolves by + discarding one side's edit. + + The report is the whole feature. A discarded divergent edit is otherwise + indistinguishable in the summary from a node that genuinely needed no + action — it lands in ``kept``, which reads as "nothing to do" — and several + witan fields (``WorkflowProject.description`` above all) are append-only + logs where losing one side loses real content. + """ + if not since: + # Only worth saying when something actually collided. A merge that only + # adds has nothing a watermark could have told us about, and the note + # would just be noise on the run that needs it least. + if result["updated"] or result["kept_target"]: + # A dry run records nothing, so the NEXT run is blind too — the + # earliest run that can report is the one after the next real + # merge. Promising otherwise here would have someone read the + # following merge's silence as "nothing diverged". + next_step = ( + "A dry run records none, so the real merge after this one is " + "blind as well; the one after that can report." + if dry_run + else "Recorded after this merge; the next one will report divergence." + ) + console.print( + "[dim]No merge watermark for this pair yet, so nothing can be " + f"said about the collisions above. {next_step}[/dim]" + ) + return + diverged = [d for d in result["decisions"] if d.get("diverged")] + if not diverged: + return + # Tense matters here. Before a dry run there is still a choice to make; + # after a real merge the losing edit is already gone from the target and + # the remedy is to put it back, not to reconsider. + consequence = ( + "Newest-record-wins will keep one side and drop the other's edit. " + "Reconcile these before you merge for real" + if dry_run + else "Newest-record-wins kept one side and DROPPED the other's edit. " + "The losing text is still in the store that lost; put the combined " + "value back by hand" + ) + console.print( + f"\n[yellow]{len(diverged)} node(s) changed on BOTH sides since the last " + f"merge[/yellow] (watermark {esc(str(since.get('merged_at')))}). " + f"{consequence} — for an append-only field (a WorkflowProject " + "description) this is lost content, not a stale value:" + ) + for d in diverged: + kept = "source" if d["decision"] == "updated" else "target" + # `source_at`/`target_at` ahead of the raw `source_ts`/`target_ts`: an + # omnigraph >= 0.9 export spells a timestamp as epoch millis, and two + # 13-digit integers are not something to eyeball against each other. + source_at = d.get("source_at") or d["source_ts"] + target_at = d.get("target_at") or d["target_ts"] + console.print( + f" {d['type']:16} {esc(d['slug'])}\n" + f" source {esc(str(source_at))} target {esc(str(target_at))}" + f" -> kept {kept}" + ) + + def _merge( source: str | None, target: str | None, @@ -211,12 +299,23 @@ def _merge( ) -> None: source = _merge_source(source, from_target) s, target = _merge_destination(to_target, target) + key = _destination_key(s, target) + since = merge_watermark.read(source, key) + # Held in memory for THIS run's report, then retired from the file before + # the first batch commits — see `_invalidate_watermark`. + if not dry_run: + _invalidate_watermark(source, key) try: result = s.merge_store( source, target=target, dry_run=dry_run, source_author=_merge_source_author(from_target), + # The two marks only. The stored entry also carries which pair it + # describes and when it was taken, which is this machine's business + # — a merge sends the deployment its rows, not the local path they + # came from. + since=_since_marks(since), ) except RuntimeError as exc: print_error(exc) @@ -230,6 +329,11 @@ def _merge( f"{result['added']} to add, {result['updated']} to update, " f"{result['kept_target']} kept (target already newer-or-equal)." ) + # Deliberately not recorded: the watermark describes a graph with this + # merge's winners in it, and a dry run wrote none of them. Storing it + # would tell the next run that everything up to here had already been + # merged, when nothing had. + _report_divergence(result, since, dry_run=True) return console.print( @@ -238,6 +342,59 @@ def _merge( f"{result['kept_target']} kept (target already newer-or-equal), " f"{result['rows_loaded']} rows loaded." ) + _report_divergence(result, since, dry_run=False) + _record_watermark(source, key, result) + + +def _invalidate_watermark(source: str, key: str) -> None: + """Drop the standing mark before a real merge writes anything. + + A merge is not atomic — its batches commit independently — so a run that + dies part-way leaves rows in the target that the standing mark predates. + Left in place, the next run reads exactly those rows as an independent + target edit and reports divergence on rows nothing but the failed merge + ever wrote (reproduced against `_reconcile_nodes` before this was added). + + So the old mark is retired first and the new one installed only on success. + A crashed merge then leaves no mark, and the next run says it cannot tell — + which is true of a graph whose last write was half a merge. + """ + merge_watermark.forget(source, key) + + +def _record_watermark(source: str, key: str, result: dict) -> None: + """Store the mark this merge just established, and report honestly when it + could not. + + The old mark is already gone by now (`_invalidate_watermark`), so every path + out of here that does not write one leaves the pair unmarked. That is the + safe direction — the next run says it cannot tell rather than measuring + against a mark that predates rows already in the target — but it is never + silent unless there was genuinely nothing to mark. + """ + watermark = result.get("watermark") + # A merge that carried no rows at all reports no mark either, and there is + # nothing wrong with that — warning would send someone looking for a + # deployment problem that is not there. + if not result["decisions"] and not result["rows_loaded"]: + return + if not merge_watermark.is_usable(watermark): + # Either an older deployment that returns no mark, or one missing a + # side. Both are unusable, and both leave the next merge blind, so they + # get one message rather than a distinction the reader cannot act on. + console.print( + "[yellow]No usable merge watermark came back, so the next merge " + "cannot report divergence. Check the deployment is on witan-council " + "0.29.0 or later; until then, diff the projects you care about by " + "hand before merging again.[/yellow]" + ) + return + if not merge_watermark.write(source, key, watermark): + console.print( + f"[yellow]Could not write {esc(str(merge_watermark.path()))} — the " + "merge itself is unaffected, but the next one will not be able to " + "report divergence.[/yellow]" + ) def _migrate_storage(old_binary: str | None, yes: bool) -> None: @@ -337,6 +494,12 @@ def merge( target are left untouched. Repeatable — re-running against an already-merged target loads nothing new. + Each merge records a per-side watermark for the pair of stores, so the next + one can name the nodes BOTH sides have written since — the case where + newest-record-wins is not resolving a stale value but discarding somebody's + edit. Nothing is auto-merged; the divergent slugs are reported for you to + reconcile. The first merge of a pair has no watermark and says so. + Parameters ---------- source: diff --git a/mcp/servers/witan/witan/merge_watermark.py b/mcp/servers/witan/witan/merge_watermark.py new file mode 100644 index 00000000..9452a90e --- /dev/null +++ b/mcp/servers/witan/witan/merge_watermark.py @@ -0,0 +1,192 @@ +"""Where a merge records what both stores looked like when they last agreed. + +``witan migrate merge`` reconciles newest-record-wins per node, and a +record-level timestamp cannot tell "the target is simply ahead" from "both sides +advanced independently since the last merge". The second is a divergence — one +side's edit is about to be discarded — and without something written down at +merge time it is indistinguishable from the first (see +tk-witan-migrate-merge-silently-drops-divergent-edi-1f1453). + +So each merge stores a **per-side high-water mark**: the newest comparison +timestamp present in the source, and the newest that will be in the target once +this merge's winners land. Next merge, a node whose source timestamp is past the +source mark *and* whose target timestamp is past the target mark was written on +both sides, and is reported instead of silently resolved. + +The two marks are compared against their own store, not against each other — the +source is a laptop's clock and the target is a cluster's, and a cross-clock +comparison would invent divergences out of skew alone. One documented exception, +with its cost stated, lives in ``witan.server._next_watermark``: the rows a merge +loads carry source timestamps into the target, so the target mark is raised to +cover them, which leaves a blind window the width of any source-ahead skew. + +Client-side state, deliberately. The pairing is "this store, that deployment", +which only the machine holding the source knows; the deployment sees a batch of +rows and cannot say which store they came from. + +Everything here fails soft. No file, an unreadable one, a truncated one: all mean +"no watermark", which costs the divergence report and nothing else — a merge must +not fail because a hint file is missing. +""" + +from __future__ import annotations + +import json +import os +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +DEFAULT_PATH = Path.home() / ".config" / "witan" / "merge-watermarks.json" +"""Where merge watermarks are kept, overridable with ``WITAN_MERGE_WATERMARKS``. + +Alongside ``tokens.json`` (``witan_core.remote.oidc.DEFAULT_CACHE_PATH``) rather +than inside the store: the store is the thing being merged, and a watermark that +travelled with it would be restored by the very export/load cycle it exists to +describe.""" + +_VERSION = 1 + + +def path() -> Path: + override = os.environ.get("WITAN_MERGE_WATERMARKS") + return Path(override).expanduser() if override else DEFAULT_PATH + + +def canonical(uri: str) -> str: + """The spelling a store is keyed under, so two names for it share a mark. + + A pair is keyed by what the caller typed, and a caller types a store many + ways: `graph.omni` from two different working directories is two stores + under one key, while `/tmp/graph.omni`, `../tmp/graph.omni` and + `file:///tmp/graph.omni` are one store under three. Either way the mark + describes a graph that is not the one being merged, which is worse than + having no mark at all — it reports divergence against someone else's + history. + + Local paths resolve to an absolute real path (symlinks included, since a + store reached through a symlink is the same store). Remote URIs are left + alone apart from a trailing slash: their authority and path are already + canonical, and rewriting them risks changing which graph they name. + """ + if uri.startswith("file://"): + uri = uri[len("file://") :] + elif uri.startswith(("http://", "https://", "s3://")): + return uri.rstrip("/") + return str(Path(uri).expanduser().resolve()) + + +def is_usable(watermark: dict | None) -> bool: + """Whether a mark can actually answer the divergence question. + + A mark missing either side is indistinguishable from no mark at all — + `_reconcile_nodes` parses the absent side to `None` and flags nothing — but + it is *truthy*, so storing one suppresses the "no watermark, cannot tell" + notice while detecting nothing. That combination is the one outcome this + feature exists to prevent: silence that reads as "nothing diverged". + """ + return bool(watermark) and all( + watermark.get(side) is not None for side in ("source_ts", "target_ts") + ) + + +def _load() -> list[dict]: + # ValueError, not just JSONDecodeError: a corrupt file need not be valid + # UTF-8, and `read_text` raises UnicodeDecodeError before `json.loads` is + # ever reached. It is a ValueError but not an OSError, so it escaped the + # narrower tuple and took the whole merge down from a module documented to + # fail soft. + try: + raw = json.loads(path().read_text(encoding="utf-8")) + except (OSError, ValueError): + return [] + if not isinstance(raw, dict) or raw.get("version") != _VERSION: + return [] + pairs = raw.get("pairs") + return [p for p in pairs if isinstance(p, dict)] if isinstance(pairs, list) else [] + + +def read(source: str, target: str) -> dict | None: + """The watermark from the last merge of ``source`` into ``target``, or None. + + None is the ordinary answer for a first merge, and a caller must read it as + "cannot tell", never as "nothing diverged". A stored mark that cannot answer + the question (see :func:`is_usable`) is reported as absent for the same + reason.""" + source, target = canonical(source), canonical(target) + for entry in _load(): + if entry.get("source") == source and entry.get("target") == target: + return entry if is_usable(entry) else None + return None + + +def forget(source: str, target: str) -> bool: + """Drop this pair's mark. True if the file is left without one. + + Called before a merge writes anything, because a merge is not atomic: its + batches commit independently, so a run that dies part-way leaves rows in the + target that the standing mark predates. The next run then reads those rows + as an independent target edit and reports divergence against a row nothing + but the failed merge ever wrote. + + Dropping the mark first turns that into "cannot tell", which is the honest + answer for a graph whose last write was a partial merge. A successful run + puts a fresh mark back immediately. + """ + source, target = canonical(source), canonical(target) + pairs = [ + p + for p in _load() + if not (p.get("source") == source and p.get("target") == target) + ] + return _store(pairs) + + +def write(source: str, target: str, watermark: dict) -> bool: + """Record ``watermark`` for this pair, replacing any earlier one. True if it + landed. + + Refuses a mark :func:`is_usable` rejects, rather than storing something that + reads as "we have a watermark" and answers nothing.""" + if not is_usable(watermark): + return False + source, target = canonical(source), canonical(target) + entry = { + "source": source, + "target": target, + "source_ts": watermark.get("source_ts"), + "target_ts": watermark.get("target_ts"), + "merged_at": datetime.now(timezone.utc).isoformat(), + } + pairs = [ + p + for p in _load() + if not (p.get("source") == source and p.get("target") == target) + ] + pairs.append(entry) + return _store(pairs) + + +def _store(pairs: list[dict]) -> bool: + """Replace the file with ``pairs``. True if it landed. + + Written through a temp file in the same directory so an interrupted write + cannot leave half a file behind — a corrupt watermark reads as absent, which + silently disables the divergence report for that pair until the next + successful merge rewrites it.""" + destination = path() + try: + destination.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + "w", + encoding="utf-8", + dir=destination.parent, + prefix=destination.name + ".", + delete=False, + ) as handle: + json.dump({"version": _VERSION, "pairs": pairs}, handle, indent=2) + temp_name = handle.name + os.replace(temp_name, destination) + except OSError: + return False + return True diff --git a/mcp/servers/witan/witan/remote/proxy.py b/mcp/servers/witan/witan/remote/proxy.py index 7763b4a3..f7c7b241 100644 --- a/mcp/servers/witan/witan/remote/proxy.py +++ b/mcp/servers/witan/witan/remote/proxy.py @@ -335,6 +335,17 @@ def _repo_means_detect(self, name: str) -> bool: def _branch_means_checkout(self, name: str) -> bool: return name in _BRANCH_IS_CHECKOUT + @property + def remote_url(self) -> str: + """The deployment this proxy addresses. + + Public because a caller sometimes has to name the destination *before* + making the call — ``witan migrate merge`` keys its merge watermark on + the pair of stores, and has to look one up before it has a result to + read the target off. Nothing else about the transport is exposed. + """ + return self._url + def _unreachable_hint(self) -> str: # Name the setting that is actually in play, read off the resolver's # own record of which source won (`url_source`) rather than inferred @@ -373,6 +384,7 @@ def merge_store( target: str | None = None, dry_run: bool = False, source_author: str | None = None, + since: dict | None = None, ) -> dict: """Merge a local store into the deployment, as the logged-in user. @@ -387,6 +399,13 @@ def merge_store( server resolves it from its own configuration — a client never names a store address, the same rule ADR-0005 (c) applies to witan-code's writes. Passing one is refused rather than ignored. + + ``since`` is the previous merge's watermark for this pair of stores; it + goes out unchanged on every batch, and the running mark the batches + return comes back as ``watermark`` for the caller to record. The client + never compares an exported timestamp itself — the server folds each + batch into the running value, which is what makes the returned mark + cover the whole merge rather than its last batch. """ if target is not None: raise RemoteToolUnavailable( @@ -397,7 +416,18 @@ def merge_store( ) decisions: list[dict] = [] - totals = {"added": 0, "updated": 0, "kept_target": 0, "rows_loaded": 0} + totals = { + "added": 0, + "updated": 0, + "kept_target": 0, + "diverged": 0, + "rows_loaded": 0, + } + # Threaded batch to batch rather than reduced at the end: the source is + # split across batches, so no single result covers it, and folding the + # running value in server-side keeps every timestamp comparison on the + # side that knows how an export spells a timestamp. + watermark: dict | None = None with _source_export(source) as export: # MCP_LOAD_MAX_BYTES, not the default: these rows ride as a JSON # tool parameter, so the binding ceiling is the MCP session's 4 MiB @@ -417,7 +447,11 @@ def merge_store( for index, batch in enumerate(batches): try: result = self.store_merge( - rows=batch, dry_run=dry_run, claim_from_author=claim_from + rows=batch, + dry_run=dry_run, + claim_from_author=claim_from, + since=since, + watermark=watermark, ) except RemotePayloadTooLarge as exc: # Only here is the caller known to be mid-batch, so only @@ -442,12 +476,21 @@ def merge_store( decisions.extend(result.get("decisions") or []) for key in totals: totals[key] += result.get(key, 0) + # An older deployment returns no watermark at all. Keeping the + # last one we did get would mark a fraction of the merge as the + # whole of it, which is worse than having none: the next run + # would compare against it and report nothing. + batch_watermark = result.get("watermark") + watermark = ( + batch_watermark if isinstance(batch_watermark, dict) else None + ) return { "dry_run": dry_run, "merged": not dry_run, "target": self._url, "decisions": decisions, + "watermark": watermark, **totals, } diff --git a/mcp/servers/witan/witan/server.py b/mcp/servers/witan/witan/server.py index 3cf92733..f115c83c 100644 --- a/mcp/servers/witan/witan/server.py +++ b/mcp/servers/witan/witan/server.py @@ -1766,6 +1766,7 @@ def _lines() -> Iterator[dict]: def _reconcile_nodes( source_nodes: dict[tuple[str, str], dict], target_nodes: dict[tuple[str, str], dict], + since: dict | None = None, ) -> tuple[list[dict], list[dict]]: """Newest-record-wins reconciliation: ``(decisions, rows to write)``. @@ -1774,7 +1775,18 @@ def _reconcile_nodes( (``store_merge``, where the source rows arrive over the wire and the target is the deployed graph). They must not drift: a row's fate should not depend on which transport carried it. - """ + + ``since`` is the previous merge's watermark for this pair of stores + (``{"source_ts", "target_ts"}``, see :mod:`witan.merge_watermark`). Given + one, a decision that resolved a node BOTH sides have written since then is + additionally marked ``diverged``. That does not change who wins — the rule is + still newest-record-wins — it changes whether the loser's edit disappears + quietly. Without a watermark nothing is marked: record-level timestamps alone + cannot tell a divergence from a target that is merely ahead, and guessing + would put every ordinary node in the report. + """ + since_src = _parse_ts((since or {}).get("source_ts")) + since_dst = _parse_ts((since or {}).get("target_ts")) decisions: list[dict] = [] winners: list[dict] = [] for (row_type, slug), row in source_nodes.items(): @@ -1788,25 +1800,116 @@ def _reconcile_nodes( src_dt = _parse_ts(src_ts) dst_dt = _parse_ts(dst_ts) won = src_dt is not None and (dst_dt is None or src_dt > dst_dt) - decisions.append( - { - "type": row_type, - "slug": slug, - "decision": "updated" if won else "kept-target", - "source_ts": src_ts, - "target_ts": dst_ts, - } - ) + decision = { + "type": row_type, + "slug": slug, + "decision": "updated" if won else "kept-target", + "source_ts": src_ts, + "target_ts": dst_ts, + } + # Each side against its OWN mark. Comparing `src_dt > since_dst` would + # be comparing the source machine's clock to the target's, and any skew + # between them would read as a divergence. + if ( + since_src is not None + and since_dst is not None + and src_dt is not None + and dst_dt is not None + and src_dt > since_src + and dst_dt > since_dst + ): + decision["diverged"] = True + # Rendered here, where the parser already lives, because these are + # the only timestamps in the merge report a human is asked to + # COMPARE — and omnigraph >= 0.9 exports them as epoch millis, so + # `source_ts`/`target_ts` reach the CLI as two 13-digit integers. + # Re-parsing them there to print would mean a second `_parse_ts`. + decision["source_at"] = src_dt.isoformat() + "Z" + decision["target_at"] = dst_dt.isoformat() + "Z" + decisions.append(decision) if won: winners.append(row) return decisions, winners +def _newest(values: Iterable[str | int | float | None]) -> str | int | float | None: + """The latest of some exported timestamps, returned as exported. + + Parsed to compare — an export may be either representation, see + :func:`_parse_ts` — but returned raw, so a watermark round-trips through JSON + as the same value the store reported it as.""" + best_raw: str | int | float | None = None + best_dt: datetime | None = None + for raw in values: + parsed = _parse_ts(raw) + if parsed is not None and (best_dt is None or parsed > best_dt): + best_dt, best_raw = parsed, raw + return best_raw + + +def _row_timestamps(rows: Iterable[dict]) -> Iterator[str | int | float | None]: + """Each export node record's comparison timestamp, the same field + reconciliation compares on.""" + for row in rows: + yield _reconcile_timestamp(row.get("type", ""), row.get("data") or {}) + + +def _next_watermark( + carry: dict | None, + source_nodes: dict[tuple[str, str], dict], + target_nodes: dict[tuple[str, str], dict], + winners: list[dict], +) -> dict: + """The watermark to record once this merge's winners have landed. + + ``source_ts`` is simply the newest row in the source. ``target_ts`` has to + account for the load as well as the export: the winners are about to be + written into the target carrying their own (source) timestamps, so a mark + taken from the pre-merge target alone would sit *below* rows this very merge + put there — and on the next run every one of them would read as "the target + changed since we last agreed". + + ★ THAT IS THE ONE PLACE THE PER-SIDE CLOCK RULE IS BENT, AND IT COSTS + SOMETHING. Folding winner timestamps in mixes the SOURCE's clock into the + TARGET's threshold. With a source clock running ahead, ``target_ts`` lands + in the target's future, and a genuine target edit made inside that window + stays below the mark — so a real divergence goes unreported (measured: a + 1h-ahead source hides a target edit made 30 minutes after the merge, where + the same case with synchronised clocks reports correctly). The blind window + is exactly the skew, and both ends are normally NTP-synced. + The alternative is worse, not better: a pure ``target_ts`` reports a + divergence on every row the merge itself loaded as soon as the source + touches it again, which is the ordinary repeat-merge path rather than a + skew corner. Removing the trade entirely needs per-record state — a + fingerprint or a target-generated revision — not a different maximum. + + ``carry`` folds in the running watermark from an earlier batch of the same + merge. That is how the batched MCP path accumulates one across calls without + the client having to compare timestamps itself. + """ + return { + "source_ts": _newest( + [*_row_timestamps(source_nodes.values()), (carry or {}).get("source_ts")] + ), + "target_ts": _newest( + [ + *_row_timestamps(target_nodes.values()), + *_row_timestamps(winners), + (carry or {}).get("target_ts"), + ] + ), + } + + def _decision_counts(decisions: list[dict]) -> dict: return { "added": sum(1 for d in decisions if d["decision"] == "added"), "updated": sum(1 for d in decisions if d["decision"] == "updated"), "kept_target": sum(1 for d in decisions if d["decision"] == "kept-target"), + # Cuts ACROSS the three above rather than partitioning with them: a + # divergence is resolved as an update or a kept-target like any other + # collision, and what the count adds is that somebody's edit lost. + "diverged": sum(1 for d in decisions if d.get("diverged")), } @@ -1900,6 +2003,7 @@ def merge_store( target: str | None = None, dry_run: bool = False, source_author: str | None = None, + since: dict | None = None, ) -> dict: """Merge another store's data into this store, newest-record-wins on slug collisions. @@ -1952,10 +2056,18 @@ def merge_store( carry and the identity the writer has come from different namespaces. Merging between two local stores has no such split: both ends write ``cfg.author``, so there is nothing to reconcile. - - Returns counts (``added``/``updated``/``kept_target``) and the full - per-``(type, slug)`` decision list, plus (when not a dry run) - ``rows_loaded`` and the raw ``load`` output. + since: + The previous merge's watermark for this pair of stores + (``{"source_ts", "target_ts"}``), from :mod:`witan.merge_watermark`. + Supplied, every collision both sides have written since then is marked + ``diverged`` — the losing edit is still discarded, but no longer + silently. Omitted, nothing is marked; see ``_reconcile_nodes``. + + Returns counts (``added``/``updated``/``kept_target``/``diverged``) and the + full per-``(type, slug)`` decision list, plus (when not a dry run) + ``rows_loaded`` and the raw ``load`` output. ``watermark`` is the mark to + record for the next merge — the caller persists it, since only the caller + knows which two stores these were. """ # `_acquire_store_lock`/`_ensure_graph` build filesystem `Path`s directly # from the URI and don't strip a URI scheme — same convention as the rest @@ -2005,14 +2117,16 @@ def merge_store( source_nodes, source_edges = _parse_export(source_file) target_nodes, _ = _parse_export(target_file) - decisions, winners = _reconcile_nodes(source_nodes, target_nodes) + decisions, winners = _reconcile_nodes(source_nodes, target_nodes, since) counts = _decision_counts(decisions) + watermark = _next_watermark(None, source_nodes, target_nodes, winners) if dry_run: return { "dry_run": True, "target": target, "decisions": decisions, + "watermark": watermark, **counts, } @@ -2023,6 +2137,7 @@ def merge_store( "target": target, "decisions": decisions, "rows_loaded": 0, + "watermark": watermark, **counts, } @@ -2057,6 +2172,7 @@ def merge_store( "decisions": decisions, "rows_loaded": len(to_load), "output": load_out.strip(), + "watermark": watermark, **counts, } @@ -2091,7 +2207,11 @@ def _data_tier_outage_reads_as_retryable(): @_tool def store_merge( - rows: list[dict], dry_run: bool = False, claim_from_author: str | None = None + rows: list[dict], + dry_run: bool = False, + claim_from_author: str | None = None, + since: dict | None = None, + watermark: dict | None = None, ) -> dict: """Merge a batch of exported rows into this deployment's graph, as you. @@ -2137,6 +2257,19 @@ def store_merge( (#267). With it, the rows you migrate end up owned by the same identity that owns everything you write afterwards. See ``_claim_authorship`` for why this matches rather than stamping unconditionally. + since: + The watermark recorded by the caller's *previous* merge into this graph + (``{"source_ts", "target_ts"}``). Supplied, a collision both sides have + written since then is marked ``diverged`` in the decisions — the merge + rule is unchanged, the losing edit is simply no longer discarded in + silence. Same value on every batch of one merge. + watermark: + The running watermark returned by the *previous batch of this same + merge*, folded into this batch's. The source is split across batches, so + no single batch sees its newest row; carrying the running value is what + lets the client end up with a mark covering the whole merge without + having to compare exported timestamps itself. ``None`` on the first + batch. **Batching is the caller's job, and the caller must send every node before any edge** (``witan_core.chunking.chunk_records`` does both). Batches commit @@ -2146,14 +2279,16 @@ def store_merge( atomic, and a caller needing all-or-nothing has to arrange that itself. Returns this batch's per-row ``decisions`` plus ``added``/``updated``/ - ``kept_target`` counts, and ``rows_loaded``. With ``dry_run`` the decisions - are computed and nothing is written. + ``kept_target``/``diverged`` counts, ``rows_loaded``, and the running + ``watermark`` to hand to the next batch. With ``dry_run`` the decisions are + computed and nothing is written. """ if not rows: return { "decisions": [], "rows_loaded": 0, "dry_run": dry_run, + "watermark": watermark or {"source_ts": None, "target_ts": None}, **_decision_counts([]), } @@ -2178,8 +2313,9 @@ def store_merge( client.export_to(target_file, label="export (deployed graph)") target_nodes, _ = _parse_export(target_file) - decisions, winners = _reconcile_nodes(source_nodes, target_nodes) + decisions, winners = _reconcile_nodes(source_nodes, target_nodes, since) counts = _decision_counts(decisions) + next_watermark = _next_watermark(watermark, source_nodes, target_nodes, winners) # Intersected with the WINNERS, not reported straight off the rewrite. A # stamped row that loses reconciliation is discarded with its new author, @@ -2197,6 +2333,7 @@ def store_merge( "decisions": decisions, "rows_loaded": 0, "authorship_claimed": claimed, + "watermark": next_watermark, **counts, } @@ -2208,6 +2345,7 @@ def store_merge( "decisions": decisions, "rows_loaded": len(to_load), "authorship_claimed": claimed, + "watermark": next_watermark, **counts, } diff --git a/packages/witan-core/CHANGELOG.md b/packages/witan-core/CHANGELOG.md index 543af6a9..f0af2258 100644 --- a/packages/witan-core/CHANGELOG.md +++ b/packages/witan-core/CHANGELOG.md @@ -6,6 +6,28 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/) (pre-1.0: a MINOR bump may include breaking changes). +## [0.32.1] - 2026-08-24 + +### Changed + +- **Refreshed the pinned omnigraph `edge` digests** to the 2026-08-24T12:50Z + build (upstream `972f1666c5`), from the 2026-08-21T00:11Z one + (`62a9c3fe6b`). `edge` is force-updated on every push to upstream main, so a + digest here goes stale on upstream's schedule rather than ours; this one had, + and the checksum check was refusing the download. That is the check working, + but it fails every fresh `witan setup` and every CI run until refreshed. All + three tiers move together (installer, `docker/omnigraph-server.Dockerfile`, + `docker/witan.Dockerfile`). + + Still 0.10.0, still internal-schema 6 — read off the new binary, so this is + not a rebuild-every-graph event. Three upstream commits are in scope and only + one carries Rust (#522, change-feed candidate scans); the change feed is not + a surface witan uses. The one part of it that does touch witan's write path + is an extra `omnigraph.no_by_source_delete=v1` transaction property stamped + on every keyed write, which upstream documents as read-advisory and only its + own pruning path reads. No error-prose or vocabulary rename, checked against + the artifact as well as against the diff. + ## [0.32.0] - 2026-08-21 ### Fixed diff --git a/packages/witan-core/pyproject.toml b/packages/witan-core/pyproject.toml index 80b67feb..946b3a9b 100644 --- a/packages/witan-core/pyproject.toml +++ b/packages/witan-core/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "witan-core" -version = "0.32.0" +version = "0.32.1" description = "Shared core for the witan MCP servers (witan-council + witan-code)" readme = "README.md" license = "BSD-3-Clause" @@ -66,7 +66,7 @@ build-backend = "hatchling.build" packages = ["witan_core"] [tool.bumpversion] -current_version = "0.32.0" +current_version = "0.32.1" allow_dirty = true [[tool.bumpversion.files]] diff --git a/packages/witan-core/witan_core/omnigraph_install.py b/packages/witan-core/witan_core/omnigraph_install.py index 78de8c6f..b72d03d8 100644 --- a/packages/witan-core/witan_core/omnigraph_install.py +++ b/packages/witan-core/witan_core/omnigraph_install.py @@ -122,20 +122,41 @@ #: and confirm it against the tarball you actually downloaded (`sha256sum`) in #: the same sitting — on a moving tag the two assets can be republished a #: minute apart, and a digest read across that gap describes neither build. -#: ★ THESE ARE THE `edge` BUILD OF 2026-08-21T00:11Z (through 62a9c3fe6b), -#: NOT v0.9.0's. Refreshed from the 2026-08-20T17:18Z triple (bee47cd465) -#: after CI failed the checksum check on 2026-08-21 (agent-kit#272). Six -#: commits landed in between, all storage-layer typed-failure work -#: (docs/rfcs/0038-typed-storage-failures.md): `OmniError::Lance(String)` -#: became `OmniError::Storage(StorageFailure)`, closed over a shared -#: `omnigraph-storage` crate. Checked the diff for the two things that would -#: matter to witan — the `_RETRYABLE`/`_NEEDS_REPAIR`/`_PRECONDITION_FAILED` -#: substrings in omnigraph.py, and the `"storage: "` prose prefix witan's -#: classifier keys on — and found neither renamed: the refactor keeps -#: `STORAGE_MESSAGE_PREFIX = "storage: "` and its own new tests assert the -#: same `storage: ` rendering the old `Lance` variant produced. No -#: vocabulary or JSON-output change here, unlike the 69d292ce80/ecf1d6aedd -#: rename two refreshes back. +#: ★ THESE ARE THE `edge` BUILD OF 2026-08-24T12:50Z (through 972f1666c5), +#: NOT v0.9.0's. Refreshed from the 2026-08-21T00:11Z triple (62a9c3fe6b) +#: after CI failed the checksum check on 2026-08-24 (agent-kit#281). Three +#: commits landed in between, and only one carries any Rust: #522 +#: (`perf(changes): bound candidate scans to transaction footprints`); the +#: other two are docs and CI hygiene (#543, #549). That source diff is +#: confined to `crates/omnigraph/src/changes/*` and its instrumentation and +#: failpoints — the CHANGE FEED, which witan does not use at all: the +#: subcommands it shells out to are query/mutate/load/export/schema/commit/ +#: graphs. +#: +#: ★ ONE PART OF #522 IS NOT CHANGE-FEED-INTERNAL, AND IT IS ON WITAN'S WRITE +#: PATH. `table_store.rs` now stamps `omnigraph.no_by_source_delete=v1` on +#: EVERY keyed `merge_insert` transaction, which is every write witan makes. +#: It is an extra transaction property and nothing else: upstream documents it +#: as read-advisory ("a missing marker only forces a fall-back, never a +#: correctness change") and only the new pruning path reads it. No behaviour +#: change for us — but it is why this refresh is not simply "docs and a perf +#: fix", and why the diff was read rather than the commit subjects. +#: +#: Checked the two things every refresh here checks — the `_RETRYABLE` / +#: `_NEEDS_REPAIR` / `_PRECONDITION_FAILED` substrings in omnigraph.py and the +#: `"storage: "` prose prefix witan's classifier keys on — and found no +#: rename: no error or vocabulary source changed at all, and #549's +#: "vocabulary guard" is a CI-config tweak excluding user docs, not the kind +#: of rename 69d292ce80/ecf1d6aedd were. Confirmed against the artifact with +#: `strings`, not only against the diff. +#: +#: ★ AND A TRAP IN THAT CONFIRMATION, WORTH KNOWING BEFORE RE-RUNNING IT. +#: `manifest table version` and `ahead of manifest` do NOT appear in the +#: binary — and did not in the previous build either. They are assembled at +#: runtime from fragments, so their absence from `strings` says nothing about +#: this refresh. Check a suspicious absence against the OLD binary before +#: reading it as a regression; two of the eleven substrings look alarming and +#: always have. #: #: ★ AND `edge` MOVED THREE TIMES WHILE THE PRIOR TRIPLE WAS BEING WRITTEN — #: see the git history of this comment for that episode. That is the cost of @@ -147,22 +168,28 @@ #: #: The digests below were taken by downloading all three tarballs and hashing #: them locally, then cross-checking each against the release's published -#: `.sha256` in the same sitting. Still the 0.10.0 re-test -#: (tk-omnigraph-0-10-0-edge-halved-the-write-ceiling-r-7ba7c2); version still -#: reports 0.10.0 and internal-schema still 6, both read off this binary. +#: `.sha256` in the same sitting. Upstream head and all three asset timestamps +#: were read before AND after the downloads and were identical (head +#: 972f1666c5, assets 12:50-13:01Z), so this triple describes one build rather +#: than a window. The linux/x86_64 value is also exactly the `got` digest CI +#: reported when it refused the stale pin, which corroborates it independently +#: of this machine. Still the 0.10.0 re-test +#: (tk-omnigraph-0-10-0-edge-halved-the-write-ceiling-r-7ba7c2); version +#: reports 0.10.0 and internal-schema 6, both read off THIS binary via +#: `bin/check_omnigraph_format.py`, so this is not a rebuild-every-graph event. #: Reverting the experiment means restoring the v0.9.0 triple, which was: #: linux-x86_64 507a36f385bea073e7f284fe476befbb4cd788b32bfa85d6f4cd5e943b663197 #: linux-arm64 6742a7fcf2761cb5841a38990c38383d7a884da2c65e3e7cc884afbbf2b2d881 #: macos-arm64 69f78c93e661e8ea2b92deafe6330650a0921a003c2099b75b226482a90dc03e _OMNIGRAPH_ASSET_SHA256: dict[str, str] = { "omnigraph-linux-x86_64.tar.gz": ( - "68099e33941cc5c252f36d4c2a26f1dfff6b28e4eb627ba1f7bf098856d34349" + "6b53f3eff3793d4012fd6b5c151c5589390699ce42661e3ce2ff1ed18b995f4c" ), "omnigraph-linux-arm64.tar.gz": ( - "d0e42176625584370a26c8a6ca43bc24447ea822483c46d971807475a5a5782a" + "11c06689fcffa6e5fc9fbe9f632ba8e493f1f42997bbd10f09440f5e1d582aab" ), "omnigraph-macos-arm64.tar.gz": ( - "f30f37f0ad8084ed5b26ce5163ebac04554c89bec4259aa1763826bc1efe215f" + "d5cda3d16533154cfc66975af3bb3593cde43513e9af4047c525f41e4555d5c6" ), } _VERSION_RE = re.compile(r"\d+\.\d+\.\d+") diff --git a/uv.lock b/uv.lock index 901ede62..83643502 100644 --- a/uv.lock +++ b/uv.lock @@ -2339,7 +2339,7 @@ test = [ [[package]] name = "witan-core" -version = "0.32.0" +version = "0.32.1" source = { editable = "packages/witan-core" } [package.optional-dependencies] @@ -2413,7 +2413,7 @@ test = [ [[package]] name = "witan-council" -version = "0.28.0" +version = "0.29.0" source = { editable = "mcp/servers/witan" } dependencies = [ { name = "agent-config-kit" },