Push editor project files into ALL mapped sandboxes for this user (or a specific name if provided). This enables a project-level "Sync sandbox" action to refresh multiple live sandboxes at once.
(request: SyncRequest)
| 54 | |
| 55 | @router.post("/sync") |
| 56 | async def sync_existing_sandbox(request: SyncRequest) -> dict[str, Any]: |
| 57 | """Push editor project files into ALL mapped sandboxes for this user (or a specific name if provided). |
| 58 | |
| 59 | This enables a project-level "Sync sandbox" action to refresh multiple live sandboxes at once. |
| 60 | """ |
| 61 | oidc_token = await get_vercel_oidc_token() |
| 62 | os.environ["VERCEL_OIDC_TOKEN"] = oidc_token |
| 63 | |
| 64 | mappings = {} |
| 65 | mappings = await get_user_project_sandboxes(request.user_id, request.project_id) |
| 66 | if not mappings: |
| 67 | raise HTTPException(status_code=404, detail="no sandboxes mapped for user") |
| 68 | |
| 69 | # Filter project once (respect ignore rules server-side) |
| 70 | is_ignored = make_ignore_predicate(request.project or {}) |
| 71 | filtered: dict[str, str] = { |
| 72 | p: c for p, c in (request.project or {}).items() if (not is_ignored(p)) or (p in {".gitignore", ".agentignore"}) |
| 73 | } |
| 74 | |
| 75 | targets: dict[str, str] = mappings |
| 76 | if request.name: |
| 77 | sid = mappings.get(request.name) |
| 78 | if not sid: |
| 79 | return {"ok": False, "error": f"sandbox not found for name '{request.name}'"} |
| 80 | targets = {request.name: sid} |
| 81 | |
| 82 | results: dict[str, Any] = {} |
| 83 | total_writes = 0 |
| 84 | for name, sid in targets.items(): |
| 85 | try: |
| 86 | sandbox = await Sandbox.get(sandbox_id=sid) |
| 87 | files_payload = [] |
| 88 | for path, content in filtered.items(): |
| 89 | try: |
| 90 | files_payload.append({"path": path, "content": content.encode("utf-8")}) |
| 91 | except Exception: |
| 92 | files_payload.append({"path": path, "content": bytes(str(content), "utf-8")}) |
| 93 | wrote = 0 |
| 94 | touched_paths: list[str] = [] |
| 95 | if files_payload: |
| 96 | for i in range(0, len(files_payload), 64): |
| 97 | chunk = files_payload[i : i + 64] |
| 98 | await sandbox.write_files(chunk) |
| 99 | wrote += len(chunk) |
| 100 | try: |
| 101 | # accumulate paths for touch to bump mtimes and trigger watchers |
| 102 | for e in chunk: |
| 103 | p = e.get("path") |
| 104 | if isinstance(p, str): |
| 105 | touched_paths.append(p) |
| 106 | except Exception: |
| 107 | pass |
| 108 | total_writes += wrote |
| 109 | # Best-effort: update mtimes for written files to trigger file watchers |
| 110 | try: |
| 111 | if touched_paths: |
| 112 | # quote paths safely and touch them |
| 113 | def _sh_quote(p: str) -> str: |
nothing calls this directly
no test coverage detected