Upload 12 files

src/agentcore/orchestrator.py

@@ -184,6 +184,22 @@ def analytics_log_status() -> dict:
     }
 
 
+async def _send_with_retry(worker_url: str, token: str, payload: dict,
+                           attempts: int = 2, backoff: float = 2.0):
+    """_send_log_payload with one retry after a short backoff. Connectivity to
+    workers.dev from this host drops for short windows; a single immediate
+    retry recovers a meaningful fraction of those without involving the buffer."""
+    last_exc = None
+    for i in range(attempts):
+        try:
+            return await _send_log_payload(worker_url, token, payload)
+        except Exception as exc:
+            last_exc = exc
+            if i + 1 < attempts:
+                await asyncio.sleep(backoff)
+    raise last_exc
+
+
 async def _send_log_payload(worker_url: str, token: str, payload: dict):
     """POST one analytics payload.
 
@@ -208,6 +224,34 @@ async def _send_log_payload(worker_url: str, token: str, payload: dict):
     return resp.status_code, report
 
 
+_FLUSHER_STARTED = {"on": False}
+
+
+def _ensure_periodic_flusher(worker_url: str, token: str) -> None:
+    """Start (once) a background loop that retries buffered rows every 90s, so
+    parked writes land as soon as connectivity returns instead of waiting for
+    the next user query. The task holds a strong reference via _spawn_background.
+    Lost on container restart - like the buffer itself - which the keep-alive
+    pings are there to make rare."""
+    if _FLUSHER_STARTED["on"]:
+        return
+
+    async def _loop():
+        while True:
+            await asyncio.sleep(90)
+            if _PENDING_LOGS:
+                try:
+                    n = await _flush_pending(worker_url, token, max_items=10)
+                    if n:
+                        _record_attempt("flushed", f"{n} buffered row(s) delivered")
+                except Exception:
+                    pass
+
+    if _spawn_background(_loop()):
+        _FLUSHER_STARTED["on"] = True
+        logger.info("Analytics flusher started (90s interval)")
+
+
 async def _flush_pending(worker_url: str, token: str, max_items: int = 5):
     """Re-attempt parked payloads oldest-first; stop at the first failure so a
     still-blocked Worker is not hammered."""
@@ -272,7 +316,7 @@ def _log_agent_query(question: str, response: AgentResponse, start_time: float)
                 except Exception:
                     pass
                 try:
-                    status, report = await _send_log_payload(worker_url, _token, _payload)
+                    status, report = await _send_with_retry(worker_url, _token, _payload)
                 except Exception as _exc:
                     _PENDING_LOGS.append(_payload)
                     _record_attempt("unreachable", f"{type(_exc).__name__}: {_exc}")
@@ -320,6 +364,8 @@ def _log_agent_query(question: str, response: AgentResponse, start_time: float)
 
             if not _spawn_background(_post_to_worker()):
                 logger.warning("D1 logging skipped: no running event loop")
+            else:
+                _ensure_periodic_flusher(worker_url, _token)
     except Exception:
         pass