Upload 12 files

src/agentcore/orchestrator.py

@@ -185,13 +185,27 @@ def analytics_log_status() -> dict:
 
 
 async def _send_log_payload(worker_url: str, token: str, payload: dict):
-    """POST one analytics payload. Raises on transport failure; returns status."""
+    """POST one analytics payload.
+
+    Returns (status_code, report) where report is the Worker's own JSON verdict.
+    This matters because the Worker answers HTTP 200 even when the D1 INSERT
+    itself failed, reporting the problem in a 'warning' field - so a status code
+    alone cannot distinguish a stored row from a dropped one.
+    """
     from src.services.http_client import make_client as _mk
     headers = {"Authorization": f"Bearer {token}"} if token else {}
     async with _mk(timeout=10) as client:
         resp = await client.post(worker_url.rstrip("/") + "/log",
                                  json=payload, headers=headers)
-    return resp.status_code
+    report = {}
+    try:
+        body = resp.json()
+        if isinstance(body, dict):
+            report = {k: body.get(k) for k in ("status", "logged", "schema", "warning")
+                      if body.get(k) is not None}
+    except Exception:
+        report = {"raw": resp.text[:120]}
+    return resp.status_code, report
 
 
 async def _flush_pending(worker_url: str, token: str, max_items: int = 5):
@@ -201,10 +215,10 @@ async def _flush_pending(worker_url: str, token: str, max_items: int = 5):
     while _PENDING_LOGS and sent < max_items:
         payload = _PENDING_LOGS[0]
         try:
-            status = await _send_log_payload(worker_url, token, payload)
+            status, report = await _send_log_payload(worker_url, token, payload)
         except Exception:
             return sent
-        if status >= 400:
+        if status >= 400 or report.get("logged") is False:
             return sent
         _PENDING_LOGS.popleft()
         sent += 1
@@ -258,7 +272,7 @@ def _log_agent_query(question: str, response: AgentResponse, start_time: float)
                 except Exception:
                     pass
                 try:
-                    status = await _send_log_payload(worker_url, _token, _payload)
+                    status, report = await _send_log_payload(worker_url, _token, _payload)
                 except Exception as _exc:
                     _PENDING_LOGS.append(_payload)
                     _record_attempt("unreachable", f"{type(_exc).__name__}: {_exc}")
@@ -286,8 +300,22 @@ def _log_agent_query(question: str, response: AgentResponse, start_time: float)
                     _PENDING_LOGS.append(_payload)
                     _record_attempt("http_error", f"HTTP {status}")
                     logger.warning(f"D1 logging failed: HTTP {status}; buffered")
+                elif report.get("logged") is False:
+                    # Worker answered 200 but told us the row was not stored.
+                    _PENDING_LOGS.append(_payload)
+                    _record_attempt("worker_rejected",
+                                    f"HTTP {status} {report.get('warning') or report}")
+                    logger.warning(
+                        f"D1 insert failed Worker-side: {report.get('warning') or report}"
+                    )
+                elif report.get("warning"):
+                    # Stored, but via the legacy fallback schema - surface it.
+                    _record_attempt("ok_with_warning",
+                                    f"HTTP {status} schema={report.get('schema')} "
+                                    f"{report.get('warning')}")
+                    logger.warning(f"D1 logged with warning: {report.get('warning')}")
                 else:
-                    _record_attempt("ok", f"HTTP {status}")
+                    _record_attempt("ok", f"HTTP {status} schema={report.get('schema')}")
                     logger.debug("D1 logged OK")
 
             if not _spawn_background(_post_to_worker()):