search: upload the executor command once, not on every poll

ObjectStoreExecutor.submit re-PUT the identical command.json on every
0.2s result poll, so a task running minutes wrote ~1,500 versions of one
engine-queue/search-exec/<id>/command.json. On the versioned memory
bucket that was the dominant growth: 109,715 versions, 10.3 GB (91% of
the bucket). retry_mailbox already retries a lost PUT acknowledgement
internally, so one upload per task is the correct delivery guarantee.

Guard the PUT with an `uploaded` flag: upload once (still retried on
transient failure), then poll only the result. One command.json version
per task instead of ~1,500. All submit semantics preserved.

59 executor tests pass, ruff clean.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

src/openultrasast/search/executor.py

@@ -583,13 +583,16 @@ class ObjectStoreExecutor:
         retry_end = end + RETRY_SECONDS
         last_error = ""
         phase = "result upload"
+        uploaded = False
         while time.monotonic() < end:
             try:
-                # A lost PUT acknowledgement is retried with the exact same sequence.
-                phase = "command upload"
                 before = time.monotonic()
                 try:
+                    if not uploaded:
+                        # Retry lost acknowledgements internally; successful delivery needs no further PUTs.
+                        phase = "command upload"
                         retry_mailbox(lambda timeout: self.transport.put(self.command_url, payload, timeout), deadline=retry_end)
+                        uploaded = True
                     phase = "result read"
                     raw = retry_mailbox(lambda timeout: self.transport.get(self.result_url, timeout), deadline=retry_end)
                 finally:

tests/test_search_executor.py

@@ -127,6 +127,39 @@ def test_mailbox_retry_is_idempotent(tmp_path):
     assert mailbox.writes >= 2
 
 
+@pytest.mark.parametrize("ready_after", [1, 20, 50])
+def test_command_uploaded_once_while_polling_result(monkeypatch, ready_after):
+    from openultrasast.search import executor as module
+
+    now = [0.0]
+    monkeypatch.setattr(module.time, "monotonic", lambda: now[0])
+    monkeypatch.setattr(module.time, "sleep", lambda delay: now.__setitem__(0, now[0] + delay))
+    command_url, result_url = "https://store.example/command", "https://store.example/result"
+    expected = {"seq": 1, "files": ["ready.txt"]}
+
+    class DelayedResult:
+        writes = 0
+        reads = 0
+
+        def put(self, url, data, timeout):
+            assert url == command_url
+            assert json.loads(data)["seq"] == 1
+            self.writes += 1
+
+        def get(self, url, timeout):
+            assert url == result_url
+            assert self.writes > 0
+            self.reads += 1
+            return json.dumps(expected).encode() if self.reads >= ready_after else None
+
+    transport = DelayedResult()
+    client = ObjectStoreExecutor(command_url, result_url, transport=transport)
+    assert client.submit("list_files", {}) == expected
+    assert transport.reads == ready_after
+    assert transport.writes == 1
+    assert client.seq == 1 and client.pending is None and not client.uncertain
+
+
 def test_mailbox_timeout():
     class Empty:
         def put(self, *args):
@@ -254,8 +287,11 @@ def test_entrypoint_rejects_initial_credentials(tmp_path, monkeypatch, key):
 
 def test_stale_result_is_polled_until_matching_sequence(tmp_path):
     class Stale(Mailbox):
+        reads = 0
+
         def get(self, url, timeout):
-            if self.writes == 1:
+            self.reads += 1
+            if self.reads == 1:
                 return b'{"seq": 0, "files": ["stale"]}'
             return self.result
 
@@ -266,7 +302,8 @@ def test_stale_result_is_polled_until_matching_sequence(tmp_path):
     transport = Stale(InProcessExecutor(tmp_path))
     client = ObjectStoreExecutor("https://store.example/command", "https://store.example/result", transport=transport, poll_seconds=0.001)
     assert client.submit("list_files", {})["files"] == []
-    assert transport.writes == 2
+    assert transport.reads == 2
+    assert transport.writes == 1
 
 
 def test_task_boundary_run_honors_shorter_argument_timeout(tmp_path, monkeypatch):