fix: align publisher with mesa v4 native contract

migrations/0013_mesa_v4_native_publisher_contract.sql

@@ -0,0 +1,10 @@
+-- Native MESA V4 delivery contract: session routes and durable session identity.
+-- Forward-only and safe for existing data roots.
+
+ALTER TABLE mesa_target_settings ADD COLUMN session_start_path TEXT NOT NULL DEFAULT '/v4/sessions/start';
+ALTER TABLE mesa_target_settings ADD COLUMN session_end_path_template TEXT NOT NULL DEFAULT '/v4/sessions/{session_id}/end';
+ALTER TABLE mesa_deliveries ADD COLUMN remote_session_id TEXT;
+
+-- Existing stored route assumptions were deliberately marked unknown by 0008.
+-- New defaults make the authoritative V4 routes available once an operator
+-- explicitly configures/verifies the rest of the target contract.

src/mesa_legal_data/publisher/client.py

@@ -39,20 +39,18 @@ def __init__(
     def is_api_key_configured(self) -> bool:
         return bool(self._api_key and self._api_key.strip())
 
-    def _get_headers(self, idempotency_key: str | None = None) -> dict[str, str]:
+    def _get_headers(self) -> dict[str, str]:
         headers = {
             "Content-Type": "application/json",
             "Accept": "application/json",
             "User-Agent": "MESA-Legal-Data-Publisher/1.0",
         }
         if self._api_key:
-            headers["Authorization"] = f"Bearer {self._api_key}"
-        if idempotency_key:
-            headers["Idempotency-Key"] = idempotency_key
+            headers["X-API-Key"] = self._api_key
         return headers
 
     def target_safety_error(self) -> str | None:
-        """Validate the target before a request can carry Authorization."""
+        """Validate the target before a request can carry the API key."""
         try:
             parsed = urlparse(self.settings.base_url)
         except ValueError:
@@ -80,6 +78,7 @@ def is_contract_configured(self) -> bool:
             self.settings.contract_source in ("configured", "live_verified")
             and self.settings.base_url
             and self.settings.health_path.startswith("/")
+            and self.settings.session_start_path.startswith("/")
             and self.settings.publish_path.startswith("/")
             and self.settings.mutation_status_path_template.startswith("/")
             and "{mutation_id}" in self.settings.mutation_status_path_template
@@ -93,6 +92,18 @@ def _normalize_mutation_state(raw_state: Any) -> tuple[str, str | None]:
         state_upper = raw_state.strip().upper()
         if state_upper in ("ACCEPTED", "RECEIVED"):
             return MutationState.QUEUED.value, None
+        if state_upper in (
+            "EXTRACTED",
+            "VALIDATED",
+            "SQL_APPLIED",
+            "VECTOR_APPLIED",
+            "GRAPH_APPLIED",
+            "RETRY_PENDING",
+            "ROLLING_BACK",
+        ):
+            return MutationState.PROCESSING.value, None
+        if state_upper in ("DEAD_LETTER", "BLOCKED", "ROLLED_BACK"):
+            return MutationState.FAILED.value, None
         if state_upper in MutationState.__members__:
             return state_upper, None
         return MutationState.FAILED.value, f"Unknown MESA mutation state: {raw_state}"
@@ -324,33 +335,106 @@ def run_preflight_checks(
             total_canonical_bytes=total_canonical_bytes,
         )
 
-    def publish_source_chunk(
+    def start_session(self) -> dict[str, Any]:
+        """Create one MESA V4 session scoped to this delivery target."""
+        if not self.is_contract_configured:
+            return {"session_id": None, "error": "MESA HTTP contract is unknown"}
+        target_error = self.target_safety_error()
+        if target_error:
+            return {"session_id": None, "error": target_error}
+        payload = {
+            "tenant_id": self.settings.tenant_id,
+            "workspace_id": self.settings.workspace_id,
+            "dataset_ids": [self.settings.dataset_id],
+            "agent_id": self.settings.agent_id,
+        }
+        try:
+            with httpx.Client(timeout=self.timeout_seconds) as client:
+                response = client.post(
+                    f"{self.settings.base_url.rstrip('/')}{self.settings.session_start_path}",
+                    json=payload,
+                    headers=self._get_headers(),
+                )
+                if response.status_code == 201:
+                    data = response.json()
+                    session_id = data.get("session_id")
+                    if isinstance(session_id, str) and session_id:
+                        return {"session_id": session_id, "message": data.get("status", "started")}
+                    return {"session_id": None, "error": "MESA session start response omitted session_id"}
+                return {"session_id": None, "error": f"HTTP {response.status_code}: {response.text[:200]}"}
+        except Exception as exc:
+            return {"session_id": None, "error": f"Transport failure starting session: {exc}"}
+
+    def end_session(self, session_id: str) -> dict[str, Any]:
+        """End a session only once no mutation needs it for status access."""
+        if not session_id or not self.settings.session_end_path_template:
+            return {"ended": False, "error": "Missing session_id or session end route"}
+        path = self.settings.session_end_path_template.replace("{session_id}", session_id)
+        try:
+            with httpx.Client(timeout=self.timeout_seconds) as client:
+                response = client.post(f"{self.settings.base_url.rstrip('/')}{path}", headers=self._get_headers())
+                return {
+                    "ended": response.status_code in (200, 202),
+                    "error": None
+                    if response.status_code in (200, 202)
+                    else f"HTTP {response.status_code}: {response.text[:200]}",
+                }
+        except Exception as exc:
+            return {"ended": False, "error": f"Transport failure ending session: {exc}"}
+
+    def build_memory_insert_payload(
         self,
         chunk: SourceChunk,
+        *,
+        session_id: str,
         idempotency_key: str,
+        finalize_revision: bool,
     ) -> dict[str, Any]:
-        """
-        Submits a single source chunk mutation to MESA v4 HTTP API.
-        Returns dictionary with mutation_id, state (COMMITTED, QUEUED, etc.), and message.
-        """
-        payload = {
-            "tenant_id": self.settings.tenant_id,
-            "workspace_id": self.settings.workspace_id,
+        """Map a frozen source chunk to the strict V4MemoryInsertRequest shape."""
+        metadata = {
+            **chunk.metadata,
+            "mesa_data_chunk_type": chunk.chunk_type,
+            "mesa_data_char_start": chunk.char_start,
+            "mesa_data_char_end": chunk.char_end,
+            "mesa_data_content_hash": chunk.content_hash,
+        }
+        source_ref = metadata.get("authoritative_source_ref") or metadata.get("source_url")
+        if not isinstance(source_ref, str) or not source_ref.strip():
+            source_ref = f"mesa-data://releases/{metadata.get('release_id', 'current')}/documents/{chunk.document_id}/revisions/{chunk.version_id}/chunks/{chunk.chunk_id}"
+        return {
+            "session_id": session_id,
             "dataset_id": self.settings.dataset_id,
-            "agent_id": self.settings.agent_id,
             "document_id": chunk.document_id,
-            "version_id": chunk.version_id,
+            "revision_id": chunk.version_id,
             "chunk_id": chunk.chunk_id,
-            "chunk_type": chunk.chunk_type,
-            "title": chunk.title,
-            "char_start": chunk.char_start,
-            "char_end": chunk.char_end,
-            "ordinal": chunk.ordinal,
+            "title": chunk.title or f"Document {chunk.document_id} chunk {chunk.ordinal}",
+            "source_ref": source_ref,
             "content": chunk.content,
-            "content_hash": chunk.content_hash,
-            "metadata": chunk.metadata,
+            "evidence_span": "",
+            "revision_number": int(metadata.get("revision_number", 1)),
+            "chunk_ordinal": chunk.ordinal,
+            "finalize_revision": finalize_revision,
+            "supersedes_revision_id": metadata.get("supersedes_revision_id"),
+            "metadata": metadata,
+            "idempotency_key": idempotency_key,
         }
 
+    def publish_source_chunk(
+        self,
+        chunk: SourceChunk,
+        idempotency_key: str,
+        *,
+        session_id: str,
+        finalize_revision: bool,
+    ) -> dict[str, Any]:
+        """Submit one strict V4MemoryInsertRequest to MESA."""
+        payload = self.build_memory_insert_payload(
+            chunk,
+            session_id=session_id,
+            idempotency_key=idempotency_key,
+            finalize_revision=finalize_revision,
+        )
+
         if not self.is_contract_configured:
             return {
                 "mutation_id": None,
@@ -363,7 +447,7 @@ def publish_source_chunk(
             return {"mutation_id": None, "state": MutationState.FAILED.value, "message": target_error}
 
         url = f"{self.settings.base_url.rstrip('/')}{self.settings.publish_path}"
-        headers = self._get_headers(idempotency_key=idempotency_key)
+        headers = self._get_headers()
 
         try:
             with httpx.Client(timeout=self.timeout_seconds) as client:

src/mesa_legal_data/publisher/engine.py

@@ -17,6 +17,7 @@
     insert_delivery_item,
     is_chunk_already_committed,
     list_failed_delivery_items,
+    set_delivery_remote_session,
     update_delivery_item_state,
     update_delivery_progress,
 )
@@ -47,8 +48,10 @@ def target_config_sha256(settings) -> str:
         "content_limit_chars": settings.content_limit_chars,
         "contract_source": settings.contract_source,
         "health_path": settings.health_path,
+        "session_start_path": settings.session_start_path,
         "publish_path": settings.publish_path,
         "mutation_status_path_template": settings.mutation_status_path_template,
+        "session_end_path_template": settings.session_end_path_template,
     }
     canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
     return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
@@ -285,6 +288,25 @@ def build_delivery_plan(
             content_limit_chars=target_settings.content_limit_chars,
         )
 
+        provenance = conn.execute(
+            """SELECT v.revision_number, v.supersedes_version_id, a.source_url, a.source_id, a.sha256
+               FROM versions v JOIN artifacts a ON a.artifact_id = v.artifact_id
+               WHERE v.version_id = ?""",
+            (v_id,),
+        ).fetchone()
+        for chunk_index, chunk in enumerate(chunks):
+            chunk.metadata.update(
+                {
+                    "release_id": release_id,
+                    "revision_number": provenance[0] if provenance else 1,
+                    "supersedes_revision_id": provenance[1] if provenance else None,
+                    "source_url": provenance[2] if provenance else None,
+                    "source_id": provenance[3] if provenance else None,
+                    "artifact_sha256": provenance[4] if provenance else None,
+                    "is_final_chunk": chunk_index == len(chunks) - 1,
+                }
+            )
+
         for chunk in chunks:
             total_bytes += len(chunk.content.encode("utf-8"))
             is_committed = is_chunk_already_committed(
@@ -376,6 +398,26 @@ def execute_publish_delivery(
         release_manifest_sha256=release_manifest_sha256,
     )
 
+    remote_session_id: str | None = None
+    if any(not is_already_done for _, is_already_done in chunk_tuples):
+        session_res = client.start_session()
+        remote_session_id = session_res.get("session_id")
+        if not remote_session_id:
+            error = session_res.get("error") or "MESA session start failed"
+            update_delivery_progress(
+                conn,
+                delivery_id=delivery_id,
+                status=DeliveryStatus.FAILED.value,
+                committed_items=0,
+                failed_items=0,
+                skipped_items=0,
+                last_error=error,
+                finished=True,
+            )
+            conn.close()
+            raise MesaClientError(f"MESA session start failed: {error}")
+        set_delivery_remote_session(conn, delivery_id=delivery_id, remote_session_id=remote_session_id)
+
     update_delivery_progress(
         conn,
         delivery_id=delivery_id,
@@ -446,7 +488,12 @@ def execute_publish_delivery(
             )
 
             # Submit chunk to MESA v4
-            pub_res = client.publish_source_chunk(chunk, idempotency_key=idemp_key)
+            pub_res = client.publish_source_chunk(
+                chunk,
+                idempotency_key=idemp_key,
+                session_id=remote_session_id or "",
+                finalize_revision=bool(chunk.metadata.get("is_final_chunk")),
+            )
             remote_mutation_id = pub_res.get("mutation_id")
             initial_state = pub_res.get("state", MutationState.FAILED.value)
 
@@ -558,6 +605,10 @@ def execute_publish_delivery(
         last_error=last_err,
         finished=final_delivery_status not in (DeliveryStatus.AWAITING_MUTATION.value, DeliveryStatus.SENDING.value),
     )
+    if final_delivery_status == DeliveryStatus.COMMITTED.value and remote_session_id:
+        # Current MESA checks mutation status through the session, so end only
+        # after every item is terminal and no retry/poll needs it.
+        client.end_session(remote_session_id)
     conn.close()
     return {
         "delivery_id": delivery_id,
@@ -611,6 +662,15 @@ def retry_delivery_failures(
         conn.close()
         return {"delivery_id": delivery_id, "retried_count": 0, "message": "No failed items eligible for retry"}
 
+    remote_session_id = delivery.get("remote_session_id")
+    if not remote_session_id:
+        session_res = client.start_session()
+        remote_session_id = session_res.get("session_id")
+        if not remote_session_id:
+            conn.close()
+            raise MesaClientError(f"MESA session start failed: {session_res.get('error') or 'missing session_id'}")
+        set_delivery_remote_session(conn, delivery_id=delivery_id, remote_session_id=remote_session_id)
+
     retried_success = 0
     retried_failed = 0
     last_err = None
@@ -632,7 +692,12 @@ def retry_delivery_failures(
             pub_res = client.get_mutation_status(remote_mutation_id)
             final_state = pub_res.get("state", MutationState.FAILED.value)
         else:
-            pub_res = client.publish_source_chunk(chunk, idempotency_key=idemp_key)
+            pub_res = client.publish_source_chunk(
+                chunk,
+                idempotency_key=idemp_key,
+                session_id=remote_session_id,
+                finalize_revision=bool(chunk.metadata.get("is_final_chunk")),
+            )
             remote_mutation_id = pub_res.get("mutation_id")
             final_state = pub_res.get("state", MutationState.FAILED.value)
 
@@ -728,6 +793,8 @@ def retry_delivery_failures(
         last_error=last_err,
         finished=new_status != DeliveryStatus.AWAITING_MUTATION.value,
     )
+    if new_status == DeliveryStatus.COMMITTED.value and remote_session_id:
+        client.end_session(remote_session_id)
     conn.close()
 
     return {