Give Drukbox a valid Idempotency-Key for every run id

backend/druks/sandbox/client.py

@@ -1,5 +1,7 @@
 import asyncio
+import hashlib
 import logging
+import re
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime, timedelta
@@ -336,8 +338,21 @@ def _api(self) -> SandboxAPI:
         )
 
 
+# What Drukbox accepts as an Idempotency-Key.
+_IDEMPOTENCY_KEY = re.compile(r"[A-Za-z0-9_\-:.]{1,255}")
+
+
 def provisioning_key(*parts: str) -> str:
-    return ":".join(part for part in parts if part)
+    """The Idempotency-Key of a box: the parts joined with `:`. A key that Drukbox
+    refuses (a DBOS schedule run id carries `+00:00`, and an id can be long) becomes
+    its accepted characters plus a hash of the whole key, so two keys never collide
+    and a retry of the same run still finds its box. An accepted key stays as is."""
+    key = ":".join(part for part in parts if part)
+    if _IDEMPOTENCY_KEY.fullmatch(key):
+        return key
+    digest = hashlib.sha256(key.encode()).hexdigest()[:16]
+    readable = re.sub(r"[^A-Za-z0-9_\-:.]", "-", key)[: 255 - len(digest) - 1]
+    return f"{readable}.{digest}"
 
 
 async def _upload_helper_script(host: Host) -> None:

backend/tests/test_provisioning_key.py

@@ -0,0 +1,30 @@
+import re
+
+from druks.sandbox.client import provisioning_key
+
+# What Drukbox's Idempotency-Key header accepts.
+ACCEPTED = re.compile(r"[A-Za-z0-9_\-:.]{1,255}")
+
+
+def test_a_schedule_run_id_gives_a_key_that_drukbox_accepts():
+    run_id = "sched-dependency_updates.find_tickets-trigger-2026-10-05T19:26:34.645715+00:00"
+    key = provisioning_key(run_id, "step-1")
+    assert ACCEPTED.fullmatch(key)
+    assert key == provisioning_key(run_id, "step-1")
+
+
+def test_keys_that_differ_only_in_a_refused_character_stay_different():
+    plus = provisioning_key("sched-x-2026-10-05T19:26:34+00:00", "step")
+    minus = provisioning_key("sched-x-2026-10-05T19:26:34-00:00", "step")
+    assert plus != minus
+    assert ACCEPTED.fullmatch(plus)
+
+
+def test_an_accepted_key_stays_as_it_is():
+    assert provisioning_key("wf-1", "workflow", "", "anthropic.1") == "wf-1:workflow:anthropic.1"
+
+
+def test_a_long_key_is_cut_to_drukbox_s_limit():
+    key = provisioning_key("w" * 300, "step")
+    assert ACCEPTED.fullmatch(key)
+    assert key != provisioning_key("w" * 301, "step")