ack BYOC leases only after callback

Author: abrichr

docs/BYOC_CONNECTOR.md

@@ -28,7 +28,7 @@ poll     -> POST /api/connector/poll        long-poll; lease the next queued job
 execute  -> openadapt-flow run ...          the governed admission gate + Replayer,
                                              against the CUSTOMER'S own storage
 callback -> POST /api/internal/run-callback  PHI-free status/metrics
-ack      -> POST /api/connector/ack          release the lease (done|failed)
+ack      -> POST /api/connector/ack          release only after callback acceptance
 ```
 
 ## Install and run
@@ -86,6 +86,12 @@ refuses any bundle that is not certified, identity-armed, effect-verified, and
 encrypted — those engine gates are unchanged, so identity checks, effect
 verification, and halt-don't-guess all remain intact.
 
+The callback is retried a bounded three times. The Connector ACKs a job only
+after the control plane accepts its outcome. If delivery still fails, the lease
+is left unacknowledged; Cloud marks expiry as `lease_expired_uncertain` and
+never blindly re-offers it, so an operator checks the real effect before
+authorizing a fresh run.
+
 ## Enabling the lane (control plane)
 
 The lane is off by default. An operator enables it with

openadapt_flow/connector/client.py

@@ -118,17 +118,17 @@ def run_callback(self, body: dict[str, Any], *, run_token: Optional[str]) -> Non
         """POST PHI-free run status/metrics via the existing callback boundary.
 
         Authenticated by the run-scoped ``x-run-token`` delivered in the job
-        (proves this run; forbids forging another's status). Best-effort:
-        observability/status must not crash the loop, but a hard transport error
-        propagates so the caller records it.
+        (proves this run; forbids forging another's status). Every non-2xx
+        response propagates: the caller must not ACK a lease whose authoritative
+        run outcome the control plane rejected or did not receive.
         """
         headers = {"content-type": "application/json"}
         if run_token:
             headers["x-run-token"] = run_token
         resp = self._client.post(
             "/api/internal/run-callback", json=body, headers=headers
         )
-        if resp.status_code >= 500:
+        if not 200 <= resp.status_code < 300:
             raise ConnectorClientError(
                 f"run-callback failed: {resp.status_code} {resp.text[:300]}"
             )

openadapt_flow/connector/daemon.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import time
 from typing import Any, Callable, Optional
 
 from openadapt_flow.connector.client import ConnectorClient
@@ -25,6 +26,7 @@
 #: storage_factory(job) -> CustomerStorage. Injected in tests; the default builds
 #: from settings + the job's backend hint.
 StorageFactory = Callable[[ByocJob], CustomerStorage]
+CALLBACK_ATTEMPTS = 3
 
 
 def phi_free_callback_body(job: ByocJob, result: ExecutionResult) -> dict[str, Any]:
@@ -81,16 +83,23 @@ def handle_job(
             "failed", {}, None, job.report_ref(), f"{type(exc).__name__}"
         )
 
-    # PHI-free callback (best effort; a transport error is caught so we still ack
-    # and release the lease rather than stranding the job).
-    try:
-        client.run_callback(
-            phi_free_callback_body(job, result), run_token=job.run_token
-        )
-    except Exception:  # noqa: BLE001 - callback failure must not strand the lease
-        pass
-
-    if job.lease_job_id:
+    # The callback is the authoritative run outcome. Retry briefly for a
+    # transient transport failure, but NEVER ACK the lease unless Cloud accepted
+    # it. An unacked lease expires to `lease_expired_uncertain` and is not
+    # automatically re-offered, preventing blind duplicate actuation after a
+    # crash or partition.
+    callback_accepted = False
+    callback_body = phi_free_callback_body(job, result)
+    for attempt in range(CALLBACK_ATTEMPTS):
+        try:
+            client.run_callback(callback_body, run_token=job.run_token)
+            callback_accepted = True
+            break
+        except Exception:  # noqa: BLE001 - bounded retry, then leave lease unacked
+            if attempt + 1 < CALLBACK_ATTEMPTS:
+                time.sleep(0.25 * (2**attempt))
+
+    if job.lease_job_id and callback_accepted:
         ack_status = "failed" if result.status == "failed" else "done"
         try:
             client.ack(job.lease_job_id, ack_status, result.error)
@@ -101,6 +110,7 @@ def handle_job(
         "job_id": job.lease_job_id,
         "run_id": job.run_id,
         "status": result.status,
+        "callback_accepted": callback_accepted,
     }
 
 

tests/test_connector.py

@@ -408,6 +408,8 @@ def __init__(self):
         self.tokens = {}  # token -> org_id
         self.jobs = []  # queued jobs (each {id, org_id, payload, status, leased_by})
         self.callbacks = []
+        self.callback_attempts = 0
+        self.callback_status = 200
         self.acks = []
         self._n = 0
 
@@ -458,6 +460,9 @@ def handler(self, request: httpx.Request) -> httpx.Response:
             return httpx.Response(200, json={"ok": True})
 
         if path == "/api/internal/run-callback":
+            self.callback_attempts += 1
+            if self.callback_status != 200:
+                return httpx.Response(self.callback_status, json={"ok": False})
             self.callbacks.append({"headers": dict(request.headers), "body": body})
             return httpx.Response(200, json={"ok": True})
 
@@ -520,6 +525,41 @@ def test_full_loop_dispatch_execute_callback_ack():
     assert cp.acks[0]["status"] == "done"
 
 
+@pytest.mark.parametrize("callback_status", [400, 503])
+def test_callback_rejection_is_retried_but_never_acked(callback_status, monkeypatch):
+    monkeypatch.setattr("openadapt_flow.connector.daemon.time.sleep", lambda _: None)
+    cp = StubControlPlane()
+    cp.callback_status = callback_status
+    cp.enqueue("org_demo", _payload())
+    transport = httpx.MockTransport(cp.handler)
+    client = ConnectorClient("https://app.test", transport=transport)
+    client.enroll(enrollment_secret="s", org_id="org_demo", name="n")
+    settings = ConnectorSettings(
+        control_plane_url="https://app.test",
+        org_id="org_demo",
+        token=client.token,
+        poll_wait_s=0,
+    )
+
+    result = run_once(
+        client,
+        settings,
+        runner=_fake_success_runner(SUCCESS_REPORT),
+        storage_factory=lambda job: InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES),
+    )
+
+    assert result == {
+        "job_id": "bjob_1",
+        "run_id": _payload()["run_id"],
+        "status": "success",
+        "callback_accepted": False,
+    }
+    assert cp.callback_attempts == 3
+    assert cp.callbacks == []
+    assert cp.acks == []
+    assert cp.jobs[0]["status"] == "leased"
+
+
 def test_cross_org_isolation_connector_never_sees_another_orgs_job():
     cp = StubControlPlane()
     cp.enqueue("org_B", _payload(org_id="org_B"))  # a DIFFERENT org's job