fix: fence uncertain approved write replays

CHANGELOG.md

@@ -2,6 +2,16 @@
 
 All notable changes to this project will be documented in this file.
 
+## Unreleased
+
+### Fixed
+
+- Atomically fence an approved write from replay once its outbound Odoo mutation
+  starts, so simultaneous callers produce at most one mutation. If the response
+  is lost, `execute_approved_write` now reports
+  `external_result_uncertain` with `retry_safe: false` and requires destination
+  verification instead of allowing the same approval token to execute again.
+
 ## [1.3.0] - 2026-07-29
 
 ### Changed

README.md

@@ -490,6 +490,13 @@ Writes are intentionally boring.
 
 Odoo access rules, record rules, and server-side constraints still decide the final result.
 
+Once execution starts, an approval token is fenced from replay. If Odoo may
+have committed a write but its response is lost, the tool returns
+`code: "external_result_uncertain"` with `retry_safe: false` and refuses a
+second execution with that approval. Verify the destination state in Odoo
+before deliberately creating a new approval. This is duplicate-resistant
+session-local behavior, not an exactly-once guarantee.
+
 Batch creates go through the same gates: pass `values_list` (one dict per
 record, max 100) to `preview_write`/`validate_write` — execution maps to a
 single atomic Odoo `create(vals_list)` call. Per-record differing `write`

docs/architecture.md

@@ -83,6 +83,15 @@ flowchart LR
 
 `validate_write` stores an executable approval only when validation used trusted, non-empty live Odoo `fields_get` metadata. Client-provided or shape-only metadata can explain issues, but it does not authorize execution.
 
+Immediately before the outbound mutation, `execute_approved_write` atomically
+claims the approval as `external_attempt_started`. Concurrent callers receive
+`write_execution_in_progress`. A successful response consumes the claim. If the
+call raises after it may have reached Odoo, the approval moves to
+`external_result_uncertain` and cannot be replayed in that server session.
+Callers must read back destination state or request human review before
+creating a new approval. This fence prevents a blind duplicate; it cannot
+prove exactly-once execution or reconcile destination state automatically.
+
 ## Local addon scanning
 
 `scan_addons_source` scans files from configured `ODOO_ADDONS_PATHS` roots without importing addon code. Explicit paths must live inside those configured roots. This keeps source inspection deterministic and avoids executing arbitrary addon imports.

src/odoo_mcp/server_core.py

@@ -66,6 +66,10 @@ class AppContext:
     _clients_lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
     schema_cache: Any = field(default_factory=_build_schema_cache)
     write_approvals: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+    write_approvals_lock: threading.Lock = field(
+        default_factory=threading.Lock,
+        repr=False,
+    )
 
     @property
     def odoo(self) -> OdooClient:
@@ -283,7 +287,10 @@ def _sweep_expired_write_approvals(app_context: AppContext, now: float) -> None:
     expired_tokens = [
         token
         for token, record in app_context.write_approvals.items()
-        if now > float(record.get("expires_at", 0))
+        if (
+            str(record.get("execution_status") or "validated") == "validated"
+            and now > float(record.get("expires_at", 0))
+        )
     ]
     for token in expired_tokens:
         app_context.write_approvals.pop(token, None)
@@ -308,18 +315,33 @@ def register_write_approval(
     if not token:
         return False
     now = time.time()
-    _sweep_expired_write_approvals(app_context, now)
     record: Dict[str, Any] = {
         "approval": dict(approval),
         "payload": write_approval_payload(approval),
         "validated_at": now,
         "expires_at": now + WRITE_APPROVAL_TTL_SECONDS,
+        "execution_status": "validated",
     }
     if resolved_binary_values:
         record["resolved_binary_values"] = dict(resolved_binary_values)
-    app_context.write_approvals[token] = record
-    approval["validated_at"] = now
-    approval["expires_at"] = now + WRITE_APPROVAL_TTL_SECONDS
+    with app_context.write_approvals_lock:
+        _sweep_expired_write_approvals(app_context, now)
+        existing = app_context.write_approvals.get(token)
+        existing_status = (
+            str(existing.get("execution_status") or "validated")
+            if existing is not None
+            else "validated"
+        )
+        if existing is not None and existing_status in {
+            "external_attempt_started",
+            "external_result_uncertain",
+        }:
+            approval["validated_at"] = existing.get("validated_at")
+            approval["expires_at"] = existing.get("expires_at")
+            return True
+        app_context.write_approvals[token] = record
+        approval["validated_at"] = now
+        approval["expires_at"] = now + WRITE_APPROVAL_TTL_SECONDS
     return True
 
 
@@ -328,13 +350,71 @@ def require_validated_write_approval(
 ) -> Dict[str, Any] | None:
     """Return a server-side validation record or None when it is missing/expired."""
     token = str(approval.get("token", ""))
-    record = app_context.write_approvals.get(token)
-    if record is None:
-        return None
-    if time.time() > float(record.get("expires_at", 0)):
-        app_context.write_approvals.pop(token, None)
-        return None
-    return record
+    with app_context.write_approvals_lock:
+        record = app_context.write_approvals.get(token)
+        if record is None:
+            return None
+        execution_status = str(record.get("execution_status") or "validated")
+        if (
+            execution_status == "validated"
+            and time.time() > float(record.get("expires_at", 0))
+        ):
+            app_context.write_approvals.pop(token, None)
+            return None
+        return record
+
+
+def claim_validated_write_approval(
+    app_context: AppContext,
+    approval: Dict[str, Any],
+) -> tuple[Dict[str, Any] | None, str]:
+    """Atomically claim one validated approval for its outbound attempt."""
+    token = str(approval.get("token", ""))
+    with app_context.write_approvals_lock:
+        record = app_context.write_approvals.get(token)
+        if record is None:
+            return None, "missing"
+        execution_status = str(record.get("execution_status") or "validated")
+        if (
+            execution_status == "validated"
+            and time.time() > float(record.get("expires_at", 0))
+        ):
+            app_context.write_approvals.pop(token, None)
+            return None, "missing"
+        if write_approval_payload(approval) != record.get("payload"):
+            return record, "payload_mismatch"
+        if execution_status != "validated":
+            return record, execution_status
+        record["execution_status"] = "external_attempt_started"
+        record["execution_started_at"] = time.time()
+        return record, "claimed"
+
+
+def mark_write_approval_uncertain(
+    app_context: AppContext,
+    token: str,
+    claimed_record: Dict[str, Any],
+    error: Exception,
+) -> None:
+    """Persist an uncertain result without allowing the approval to be reset."""
+    with app_context.write_approvals_lock:
+        current = app_context.write_approvals.get(token)
+        if current is not claimed_record:
+            return
+        current["execution_status"] = "external_result_uncertain"
+        current["execution_error"] = str(error)
+        current.pop("resolved_binary_values", None)
+
+
+def complete_write_approval(
+    app_context: AppContext,
+    token: str,
+    claimed_record: Dict[str, Any],
+) -> None:
+    """Consume the exact claimed record after a confirmed external response."""
+    with app_context.write_approvals_lock:
+        if app_context.write_approvals.get(token) is claimed_record:
+            app_context.write_approvals.pop(token, None)
 
 
 # ---------------------------------------------------------------------------

src/odoo_mcp/tools_write.py

@@ -45,6 +45,9 @@
     mcp,
     _app_context,
     _resolve_odoo,
+    claim_validated_write_approval,
+    complete_write_approval,
+    mark_write_approval_uncertain,
     register_write_approval,
     require_validated_write_approval,
     restrict_attachment_upload_path,
@@ -403,14 +406,20 @@ def execute_approved_write(
 ) -> Dict[str, Any]:
     """Execute create/write/unlink only after token, confirm, and env gates pass."""
     report = _execute_approved_write_gated(ctx, approval, confirm)
+    if report.get("success"):
+        outcome = "success"
+    elif report.get("code") == "external_result_uncertain":
+        outcome = "uncertain"
+    else:
+        outcome = "denied"
     safe_record_ids = [
         int(rid)
         for rid in approval.get("record_ids") or []
         if isinstance(rid, (int, str)) and str(rid).isdigit()
     ]
     record_write_event(
         "execute",
-        outcome="success" if report.get("success") else "denied",
+        outcome=outcome,
         model=str(approval.get("model") or "") or None,
         operation=str(approval.get("operation") or "") or None,
         record_ids=safe_record_ids,
@@ -421,6 +430,53 @@ def execute_approved_write(
     return report
 
 
+def _write_execution_state_error(
+    execution_status: str,
+    validation_record: Dict[str, Any] | None,
+) -> Dict[str, Any]:
+    if execution_status == "external_attempt_started":
+        return {
+            "success": False,
+            "tool": "execute_approved_write",
+            "code": "write_execution_in_progress",
+            "retry_safe": False,
+            "error": (
+                "write execution is already in progress for this approval; "
+                "do not replay it"
+            ),
+        }
+    if execution_status == "external_result_uncertain":
+        original_error = str(
+            (validation_record or {}).get("execution_error") or ""
+        )
+        detail = f" Original error: {original_error}" if original_error else ""
+        return {
+            "success": False,
+            "tool": "execute_approved_write",
+            "code": "external_result_uncertain",
+            "retry_safe": False,
+            "error": (
+                "the write may already have completed in Odoo; this approval "
+                "cannot be replayed. Verify destination state before creating "
+                f"a new approval.{detail}"
+            ),
+        }
+    if execution_status == "payload_mismatch":
+        return {
+            "success": False,
+            "tool": "execute_approved_write",
+            "error": "approval payload does not match the stored validation record",
+        }
+    return {
+        "success": False,
+        "tool": "execute_approved_write",
+        "error": (
+            "approval token has not been validated in this server session "
+            "or has expired; call validate_write first"
+        ),
+    }
+
+
 def _execute_approved_write_gated(
     ctx: Context,
     approval: Dict[str, Any],
@@ -449,6 +505,14 @@ def _execute_approved_write_gated(
                     "or has expired; call validate_write first"
                 ),
             }
+        execution_status = str(
+            validation_record.get("execution_status") or "validated"
+        )
+        if execution_status != "validated":
+            return _write_execution_state_error(
+                execution_status,
+                validation_record,
+            )
         if write_approval_payload(approval) != validation_record.get("payload"):
             return {
                 "success": False,
@@ -504,8 +568,41 @@ def _execute_approved_write_gated(
         else:
             _, odoo = app_context.get_client(approval_instance)
 
-        result = odoo.execute_method(model, operation, *args, **kwargs)
-        app_context.write_approvals.pop(str(approval.get("token", "")), None)
+        token = str(approval.get("token", ""))
+        claimed_record, claim_status = claim_validated_write_approval(
+            app_context,
+            approval,
+        )
+        if claim_status != "claimed" or claimed_record is None:
+            return _write_execution_state_error(
+                claim_status,
+                claimed_record,
+            )
+        try:
+            result = odoo.execute_method(model, operation, *args, **kwargs)
+        except Exception as execution_error:
+            mark_write_approval_uncertain(
+                app_context,
+                token,
+                claimed_record,
+                execution_error,
+            )
+            return {
+                "success": False,
+                "tool": "execute_approved_write",
+                "code": "external_result_uncertain",
+                "retry_safe": False,
+                "error": (
+                    "the write may already have completed in Odoo; this approval "
+                    "cannot be replayed. Verify destination state before creating "
+                    f"a new approval. Original error: {execution_error}"
+                ),
+            }
+        complete_write_approval(
+            app_context,
+            token,
+            claimed_record,
+        )
         return {
             "success": True,
             "tool": "execute_approved_write",

tests/test_batch_write.py

@@ -1,4 +1,5 @@
 import importlib
+import threading
 
 from odoo_mcp import agent_tools
 
@@ -13,6 +14,7 @@ def __init__(self, odoo):
         self.odoo = odoo
         self.schema_cache = {}
         self.write_approvals = {}
+        self.write_approvals_lock = threading.Lock()
 
     def get_client(self, instance=None):
         return "default", self.odoo

tests/test_field_acl_enforcement.py

@@ -2,6 +2,7 @@
 
 import importlib
 import json
+import threading
 
 import pytest
 
@@ -122,6 +123,7 @@ def __init__(self, odoo):
         self.odoo = odoo
         self.schema_cache = {}
         self.write_approvals = {}
+        self.write_approvals_lock = threading.Lock()
 
     def get_client(self, instance=None):
         return (instance or "default"), self.odoo