feat(witan-core): send and check the Omnigraph-Http-Api contract header

mcp/servers/witan-code/tests/test_store.py

@@ -636,6 +636,8 @@ def getresponse(self):
         class _Response:
             status = 404
             will_close = False
+            # Unstamped, as a 0.11.0 server's responses are.
+            headers = http_module.http.client.HTTPMessage()
 
             def read(self):
                 return json.dumps(

packages/witan-core/changelog.d/20261008_133544_blarghmatey_omnigraph_013_http_api_header.md

@@ -0,0 +1,11 @@
+### Added
+
+- The pooled HTTP transport sends `Omnigraph-Http-Api: 0.13` on every request.
+  omnigraph-server 0.13 answers `400 api_contract_mismatch` without it; 0.11
+  ignores it.
+- The transport checks the same header on responses before decoding the body.
+  A response stamped with another contract is terminal. Once the pinned
+  storage format reaches 14, a response with no stamp is too (a successful
+  one, or any refusal of a write), because without the server's stamp a
+  429/503/409 no longer proves nothing was applied. A read that failed
+  unstamped keeps its ordinary classification, since repeating a read is safe.

packages/witan-core/tests/test_omnigraph_http.py

@@ -58,17 +58,28 @@ def __init__(
         body: str = "",
         will_close: bool = False,
         headers: dict | None = None,
+        stamped: bool = True,
     ):
+        """``stamped`` adds the contract header an omnigraph-server 0.13
+        puts on every response. Pass ``False`` for a 0.11 server or a proxy
+        answering in the server's place."""
         self.status = status
         self.will_close = will_close
         self._body = body.encode()
-        self._headers = headers or {}
+        # A real HTTPMessage, so repeated headers and case-insensitive lookup
+        # behave as they do on a live response. A list value repeats the header.
+        self.headers = http.client.HTTPMessage()
+        if stamped:
+            self.headers.add_header(ogh.HTTP_API_CONTRACT_HEADER, ogh.HTTP_API_CONTRACT)
+        for name, value in (headers or {}).items():
+            for item in value if isinstance(value, list) else [value]:
+                self.headers.add_header(name, item)
 
     def read(self) -> bytes:
         return self._body
 
     def getheader(self, name: str, default=None):
-        return self._headers.get(name, default)
+        return self.headers.get(name, default)
 
 
 class FakeConnection:
@@ -634,6 +645,119 @@ def test_no_token_sends_no_authorization_header(_fake_http):
     assert "Authorization" not in _fake_http.created[0].requests[0]["headers"]
 
 
+# ── the HTTP API contract header (omnigraph 0.12+) ───────────────────
+
+
+@pytest.fixture
+def contract_required(monkeypatch):
+    """The state after the pin reaches format 14: an unstamped response is a
+    violation. Until then a 0.11 server legitimately sends none."""
+    monkeypatch.setattr(ogh, "RESPONSE_CONTRACT_REQUIRED", True)
+
+
+def test_every_request_carries_the_contract_header(_fake_http):
+    _fake_http.script = [ok({"rows": []}), ok({}), ok({"graphs": []})]
+    transport = ogh.PooledTransport("http://host:8080")
+    transport.query("council", "q", {}, "t")
+    transport.mutate("council", "q", {}, "t")
+    transport.graphs("t")
+
+    sent = [r["headers"] for r in _fake_http.created[0].requests]
+    assert [h[ogh.HTTP_API_CONTRACT_HEADER] for h in sent] == ["0.13"] * 3
+
+
+def test_an_unstamped_response_is_accepted_from_a_0_11_server(_fake_http, monkeypatch):
+    monkeypatch.setattr(ogh, "RESPONSE_CONTRACT_REQUIRED", False)
+    _fake_http.script = [ok({"rows": []}, stamped=False)]
+
+    outcome = ogh.PooledTransport("http://host:8080").query("council", "q", {}, None)
+
+    assert outcome.kind == ogh.OK
+
+
+@pytest.mark.parametrize("values", ["0.12", ["0.13", "0.13"], "0.13, 0.13"])
+def test_a_response_on_another_contract_is_terminal_even_for_a_read(_fake_http, values):
+    """Present but wrong, which no pin gates: the body is not decoded."""
+    _fake_http.script = [
+        ok(
+            {"rows": [{"m.slug": "x"}]},
+            stamped=False,
+            headers={ogh.HTTP_API_CONTRACT_HEADER: values},
+        )
+    ]
+
+    outcome = ogh.PooledTransport("http://host:8080").query("council", "q", {}, None)
+
+    assert outcome.kind == ogh.CONTRACT_VIOLATION
+    assert outcome.body == ""
+    assert "effects are unknown" in outcome.error
+
+
+@pytest.mark.parametrize("idempotent", [True, False])
+def test_a_refusal_on_another_contract_is_not_classified_by_its_status(
+    _fake_http, idempotent
+):
+    """A 503 stamped 0.14 is not our server asking us to wait."""
+    _fake_http.script = [
+        FakeResponse(
+            503,
+            "loading",
+            stamped=False,
+            headers={ogh.HTTP_API_CONTRACT_HEADER: "0.14"},
+        )
+    ]
+    transport = ogh.PooledTransport("http://host:8080")
+
+    outcome = transport.post("/graphs/council/query", {}, None, idempotent=idempotent)
+
+    assert outcome.kind == ogh.CONTRACT_VIOLATION
+
+
+def test_an_unstamped_success_is_not_a_result(_fake_http, contract_required):
+    _fake_http.script = [ok({"rows": []}, stamped=False)]
+
+    outcome = ogh.PooledTransport("http://host:8080").query("council", "q", {}, None)
+
+    assert outcome.kind == ogh.CONTRACT_VIOLATION
+    assert "found none" in outcome.error
+
+
+@pytest.mark.parametrize("status", [409, 429, 502, 503])
+def test_an_unstamped_refusal_of_a_write_is_never_retryable(
+    _fake_http, contract_required, status
+):
+    """Each of these is retried when the server says it, because the server
+    saying it proves nothing was applied. A proxy saying it proves nothing."""
+    _fake_http.script = [FakeResponse(status, "upstream reset", stamped=False)]
+
+    outcome = ogh.PooledTransport("http://host:8080").mutate("council", "q", {}, None)
+
+    assert outcome.kind == ogh.CONTRACT_VIOLATION
+    assert outcome.status == status
+    assert "effects are unknown" in outcome.error
+
+
+def test_a_stamped_refusal_of_a_write_keeps_its_classification(
+    _fake_http, contract_required
+):
+    _fake_http.script = [err(503, "graph is loading", code="graph_unavailable")]
+
+    outcome = ogh.PooledTransport("http://host:8080").mutate("council", "q", {}, None)
+
+    assert outcome.kind == ogh.UNAVAILABLE
+
+
+def test_an_unstamped_failure_of_a_read_keeps_its_classification(
+    _fake_http, contract_required
+):
+    """A proxy's 503 while the server restarts. Repeating a read is safe."""
+    _fake_http.script = [FakeResponse(503, "no healthy upstream", stamped=False)]
+
+    outcome = ogh.PooledTransport("http://host:8080").query("council", "q", {}, None)
+
+    assert outcome.kind == ogh.UNAVAILABLE
+
+
 # ── routing: what must NOT take the HTTP path ────────────────────────
 
 
@@ -1058,6 +1182,35 @@ def test_recovery_required_on_a_write_is_indeterminate_and_terminal(
     assert len(_fake_http.created[0].requests) == 1
 
 
+def test_an_unstamped_refusal_of_a_write_is_indeterminate_and_terminal(
+    monkeypatch, tmp_path, _fake_http, contract_required
+):
+    """One response scripted: a retry would exhaust the script. Raised as
+    WriteIndeterminate so a caller that re-reads on that does so here too."""
+    monkeypatch.setattr(og.time, "sleep", lambda *_: None)
+    _fake_http.script = [FakeResponse(503, "no healthy upstream", stamped=False)]
+    client = _client(
+        monkeypatch, "http://host:8080", _mutation_dir(tmp_path), graph_id="council"
+    )
+
+    with pytest.raises(og.WriteIndeterminate, match="contract stamp"):
+        client.change("mutations.gq", "claim", {"slug": "t-1"})
+
+    assert len(_fake_http.created[0].requests) == 1
+
+
+def test_an_unstamped_success_of_a_read_is_an_error_not_rows(
+    monkeypatch, queries_dir, _fake_http, contract_required
+):
+    _fake_http.script = [ok({"rows": [{"m.slug": "x"}]}, stamped=False)]
+    client = _client(monkeypatch, "http://host:8080", queries_dir, graph_id="council")
+
+    with pytest.raises(RuntimeError, match="effects are unknown"):
+        client.read("read.gq", "find_memory", {"slug": "x"})
+
+    assert len(_fake_http.created[0].requests) == 1
+
+
 def test_recovery_required_on_a_write_is_never_a_lost_race(
     monkeypatch, tmp_path, _fake_http
 ):

packages/witan-core/witan_core/omnigraph.py

@@ -2581,6 +2581,18 @@ def _retry_loop(
                     f"will not clear it:\n{err.strip()}",
                     operation_id,
                 ) from None
+            if kind == _http.CONTRACT_VIOLATION and is_write:
+                # The response did not carry exactly the stamp of the
+                # omnigraph-server we speak to (none, another contract's, or
+                # more than one), so it says nothing about whether the write
+                # was applied. A read falls through to the RuntimeError below.
+                raise WriteIndeterminate(
+                    f"omnigraph {label} got a response that did not carry "
+                    f"exactly the expected HTTP contract stamp. ITS OUTCOME IS "
+                    f"INDETERMINATE: the "
+                    f"write may or may not have been applied. Re-read before "
+                    f"retrying:\n{err.strip()}"
+                ) from None
             if kind == _http.RECOVERY_REQUIRED:
                 # ★ A WRITE STOPS HERE. See `WriteIndeterminate`: this one
                 # response covers both an effect-free bystander barrier and

packages/witan-core/witan_core/omnigraph_http.py

@@ -50,6 +50,8 @@
 import urllib.parse
 from typing import NamedTuple
 
+from .omnigraph_install import _OMNIGRAPH_INTERNAL_SCHEMA
+
 # Classification vocabulary shared with the subprocess path. `_execute` maps a
 # subprocess's stderr onto these same names, so a condition is handled
 # identically no matter which transport reported it.
@@ -152,6 +154,12 @@
 #: one legible failure into a repair that fails the same way.
 STORE_QUARANTINED = "store_quarantined"
 
+#: The response did not carry the server's HTTP contract stamp, or carried
+#: another contract's (see :func:`contract_violation`). Never retried. Its own
+#: kind, not FATAL, so the policy loop can raise ``WriteIndeterminate`` for a
+#: write: nothing in such a response says whether the write was applied.
+CONTRACT_VIOLATION = "contract_violation"
+
 #: The conditional-write precondition header (upstream #470). Sent raw and
 #: exactly once, and ONLY to the dedicated routes below — the ordinary
 #: ``/mutate`` rejects it outright rather than ignoring an unknown header,
@@ -161,6 +169,28 @@
 #: Path suffix selecting the conditional variant of a write route.
 IF_GRAPH_COMMIT_SUFFIX = "/if-graph-commit"
 
+#: The HTTP contract discriminator omnigraph-server 0.12 introduced. Every
+#: protected route (everything this module calls; ``/healthz`` and ``/readyz``
+#: are public) answers ``400 api_contract_mismatch`` unless the request
+#: carries exactly one of these with exactly the server's value, and the
+#: server stamps the same header on every response, errors included.
+HTTP_API_CONTRACT_HEADER = "Omnigraph-Http-Api"
+
+#: The contract this client speaks. Upstream keeps it separate from the
+#: package version (``omnigraph-api-types``: "independent of the package
+#: version and graph-storage stamp"), so it is its own constant and not
+#: derived from ``_OMNIGRAPH_VERSION``. 0.12.0 required ``0.12`` and 0.13.0
+#: requires ``0.13``. A 0.11 server ignores the header.
+HTTP_API_CONTRACT = "0.13"
+
+#: The storage format of the servers that stamp :data:`HTTP_API_CONTRACT_HEADER`
+#: on responses (observed on 0.13.0, format 14). A 0.11 server (format 9)
+#: sends none, so an absent header is tolerated until the pinned format
+#: reaches this. Moving ``_OMNIGRAPH_INTERNAL_SCHEMA`` is what turns the
+#: requirement on.
+_CONTRACT_STAMPED_FROM_SCHEMA = 14
+RESPONSE_CONTRACT_REQUIRED = _OMNIGRAPH_INTERNAL_SCHEMA >= _CONTRACT_STAMPED_FROM_SCHEMA
+
 # A pooled connection idle longer than this is closed and reopened rather than
 # reused. It is not a performance knob — it is what keeps the connect-failure
 # classification honest.
@@ -372,6 +402,33 @@ def error_message(status: int, body: str) -> str:
     return f"HTTP {status}{f': {stripped}' if stripped else ''}"
 
 
+def contract_violation(values: list[str]) -> str | None:
+    """Why a response's contract header is unacceptable, or None if it is fine.
+
+    Upstream's rule (``docs/user/operations/server.md`` § HTTP contract): check
+    the header before decoding the body, and a missing or incompatible one
+    "means unknown effects, never permission to replay". A response without it
+    came from something other than the omnigraph-server we speak to: a proxy
+    answering in its place, or a server on another contract.
+
+    :param values: Every :data:`HTTP_API_CONTRACT_HEADER` value on the response.
+    :returns: A message naming the violation, or ``None``.
+    :rtype: str | None
+    """
+    if values == [HTTP_API_CONTRACT]:
+        return None
+    if not values:
+        if not RESPONSE_CONTRACT_REQUIRED:
+            return None
+        found = "none"
+    else:
+        found = ", ".join(repr(value) for value in values)
+    return (
+        f"response did not carry exactly one {HTTP_API_CONTRACT_HEADER}: "
+        f"{HTTP_API_CONTRACT} header (found {found})"
+    )
+
+
 class PooledTransport:
     """Keep-alive connections to one omnigraph-server, one per calling thread.
 
@@ -524,7 +581,10 @@ def _send(
           is the stale-keep-alive case, and it is a fast path, not a different
           safety rule.
         """
-        headers = {"Accept": "application/json"}
+        headers = {
+            "Accept": "application/json",
+            HTTP_API_CONTRACT_HEADER: HTTP_API_CONTRACT,
+        }
         if body is not None:
             headers["Content-Type"] = "application/json"
             headers["Content-Length"] = str(len(body.encode()))
@@ -596,8 +656,32 @@ def _send(
             if response.will_close:
                 self._discard()
 
+            succeeded = 200 <= response.status < 300
+            contract_values = response.headers.get_all(HTTP_API_CONTRACT_HEADER) or []
+            violation = contract_violation(contract_values)
+            if violation is not None and (
+                succeeded or not idempotent or contract_values
+            ):
+                # Checked before the body is decoded or the status classified.
+                # Without the server's stamp, a 2xx body is not a result and a
+                # 429/503/409 is not the server saying "nothing was applied",
+                # which is the premise every retryable classification rests on.
+                # So a write is terminal here whatever the status. A read that
+                # failed with no stamp at all falls through to the ordinary
+                # classification below, since repeating a read is safe (a
+                # proxy's 503 while the server restarts). A stamp naming
+                # another contract is terminal for a read too: waiting does not
+                # change which server answers.
+                return Outcome(
+                    kind=CONTRACT_VIOLATION,
+                    error=(
+                        f"{violation}; this request's effects are unknown and "
+                        f"it was not retried (HTTP {response.status})"
+                    ),
+                    status=response.status,
+                )
             text = raw.decode("utf-8", errors="replace")
-            if 200 <= response.status < 300:
+            if succeeded:
                 return Outcome(kind=OK, body=text, status=response.status)
             message = error_message(response.status, text)
             return Outcome(