feat(git-sync): the container pulls origin on its own (trinity-enterprise#703)

docker/base-image/agent_server/auto_sync.py

@@ -61,43 +61,50 @@ def should_start_loop() -> bool:
     return should_run_auto_sync() or _platform_coords() is not None
 
 
-async def resolve_auto_sync_enabled(client: httpx.AsyncClient) -> bool:
-    """This cycle's gate: the owner's `auto_sync_enabled`, asked live (#3010).
+async def _resolve_flag(client: httpx.AsyncClient, path: str, field: str,
+                        fallback: bool, label: str) -> bool:
+    """One live read of a per-agent git flag, with the agent's own key.
 
     - 200 → the flag.
-    - 404 "Git not configured" → False: nothing is bound to sync to.
+    - 404 "Git not configured" → False: nothing is bound to sync with.
     - anything else (backend down, 5xx, auth refusal, a uniform 404) → the env
       fallback, which the last recreate derived from the same flag. Failing to
       the last known value keeps a platform blip from arming a disabled loop or
       silencing an enabled one.
     """
-    global _last_resolved
-    value = should_run_auto_sync()
+    value = fallback
     coords = _platform_coords()
     if coords is not None:
         backend_url, mcp_key, agent_name = coords
         try:
             resp = await client.get(
-                f"{backend_url}/api/agents/{agent_name}/git/auto-sync",
+                f"{backend_url}/api/agents/{agent_name}/git/{path}",
                 headers={"Authorization": f"Bearer {mcp_key}"},
                 timeout=_FLAG_TIMEOUT,
             )
             if resp.status_code == 200:
-                value = bool(resp.json().get("auto_sync_enabled"))
+                value = bool(resp.json().get(field))
             elif resp.status_code == 404 and _detail(resp) == "Git not configured":
                 value = False
             else:
                 logger.warning(
-                    "auto-sync: flag read returned %s; using env fallback (%s)",
-                    resp.status_code, value,
+                    "%s: flag read returned %s; using env fallback (%s)",
+                    label, resp.status_code, value,
                 )
         except (httpx.HTTPError, ValueError) as exc:
-            logger.warning("auto-sync: flag read failed (%s); using env fallback (%s)",
-                           type(exc).__name__, value)
-    _last_resolved = value
+            logger.warning("%s: flag read failed (%s); using env fallback (%s)",
+                           label, type(exc).__name__, value)
     return value
 
 
+async def resolve_auto_sync_enabled(client: httpx.AsyncClient) -> bool:
+    """This cycle's push gate: the owner's `auto_sync_enabled`, asked live (#3010)."""
+    global _last_resolved
+    _last_resolved = await _resolve_flag(
+        client, "auto-sync", "auto_sync_enabled", should_run_auto_sync(), "auto-sync")
+    return _last_resolved
+
+
 def _detail(resp: httpx.Response) -> Optional[str]:
     try:
         body = resp.json()
@@ -173,22 +180,103 @@ async def run_one_cycle(client: httpx.AsyncClient, home: Path, run_once) -> Opti
     return result
 
 
+# ---------------------------------------------------------------------------
+# trinity-enterprise#703 — the pull cycle (invariant G3: human and fleet work
+# reaches the agent within a bound). A second loop beside the push loop, gated
+# per cycle on the owner's `pull_sync_enabled`; the cycle itself is
+# `routers/git.py::_run_pull_once` (never starts under a running execution,
+# never resets over a registered one).
+# ---------------------------------------------------------------------------
+
+_last_pull_resolved: Optional[bool] = None
+
+
+def should_run_pull() -> bool:
+    """The pull cycle's env fallback (`GIT_SYNC_PULL`), derived from the DB flag
+    at every recreate."""
+    return os.getenv("GIT_SYNC_PULL", "").lower() == "true"
+
+
+async def resolve_pull_sync_enabled(client: httpx.AsyncClient) -> bool:
+    global _last_pull_resolved
+    _last_pull_resolved = await _resolve_flag(
+        client, "pull-sync", "pull_sync_enabled", should_run_pull(), "pull")
+    return _last_pull_resolved
+
+
+def current_pull_sync_enabled() -> bool:
+    return _last_pull_resolved if _last_pull_resolved is not None else should_run_pull()
+
+
+def get_pull_interval_seconds() -> int:
+    """`GIT_SYNC_PULL_INTERVAL_SECONDS`, defaulting to the push interval."""
+    raw = os.getenv("GIT_SYNC_PULL_INTERVAL_SECONDS")
+    if not raw:
+        return get_interval_seconds()
+    try:
+        value = int(raw)
+        return value if value > 0 else get_interval_seconds()
+    except ValueError:
+        logger.warning("GIT_SYNC_PULL_INTERVAL_SECONDS=%r is not an int; using the push interval", raw)
+        return get_interval_seconds()
+
+
+async def run_one_pull_cycle(client: httpx.AsyncClient, home: Path, run_once) -> Optional[dict]:
+    """One pull tick: skip without a repo or with the owner's pull flag off,
+    else run the pull in a worker thread (it runs git children)."""
+    if not (home / ".git").exists():
+        return None
+    if not await resolve_pull_sync_enabled(client):
+        logger.debug("pull: off (pull_sync_enabled=0), skipping cycle")
+        return None
+    result = await asyncio.to_thread(run_once, home)
+    logger.info("pull: %s", result.get("status"))
+    return result
+
+
+async def run_pull_loop(
+    home_dir: Optional[Path] = None, interval_seconds: Optional[int] = None
+) -> None:
+    """Background pull loop. Swallows every exception to keep running."""
+    from .routers.git import _run_pull_once  # lazy: avoids circular import
+
+    home = home_dir or _HOME_DIR
+    interval = interval_seconds if interval_seconds is not None else get_pull_interval_seconds()
+    logger.info("pull loop started (interval=%ss, home=%s)", interval, home)
+    async with httpx.AsyncClient() as client:
+        try:
+            await resolve_pull_sync_enabled(client)
+        except Exception:  # noqa: BLE001 — loop must never die
+            logger.exception("pull: initial flag read raised unexpectedly")
+        # Half an interval later than the push loop (PR #3021 review): the two
+        # start together and, on the default shared interval, would otherwise
+        # reach the repo lock on the same tick every time.
+        await asyncio.sleep(interval + interval / 2)
+        while True:
+            try:
+                await run_one_pull_cycle(client, home, _run_pull_once)
+            except Exception:  # noqa: BLE001 — loop must never die
+                logger.exception("pull cycle raised unexpectedly")
+            await asyncio.sleep(interval)
+
+
 def schedule_auto_sync_if_enabled(app) -> None:
     """Attach the loop's startup handler when auto-sync is on or can be turned
     on live (#3010); each cycle then checks the owner's flag."""
-    if not should_start_loop():
+    if not (should_start_loop() or should_run_pull()):
         logger.info(
-            "auto-sync unavailable (no GIT_SYNC_AUTO and no platform "
-            "credentials to read the per-agent flag)"
+            "auto-sync unavailable (no GIT_SYNC_AUTO / GIT_SYNC_PULL and no "
+            "platform credentials to read the per-agent flags)"
         )
         return
 
     task_ref: list[asyncio.Task] = []
 
     @app.on_event("startup")
     async def _start_auto_sync() -> None:
-        task = asyncio.create_task(run_auto_sync_loop())
-        task_ref.append(task)
+        task_ref.append(asyncio.create_task(run_auto_sync_loop()))
+        # trinity-enterprise#703: the pull loop runs beside it, gated per cycle.
+        task_ref.append(asyncio.create_task(run_pull_loop()))
 
     @app.on_event("shutdown")
     async def _stop_auto_sync() -> None: