diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py
index 807288bbc911..0b192e17fda6 100644
--- a/products/tasks/backend/facade/api.py
+++ b/products/tasks/backend/facade/api.py
@@ -1728,6 +1728,10 @@ def _sync_automation_schedule(automation: TaskAutomation) -> None:
         "snapshot_mount_path",
         "workflow_id",
         "pending_dispatch",
+        # Written once at loop fire time; seeding copies these storage paths into the
+        # run's artifact prefix, so a PATCHable value would be an arbitrary
+        # object-storage read (and write-location) primitive.
+        "skill_bundle_seeds",
         "cancel_requested_at",
         "cancel_requested_by_user_id",
         "cancel_source",
diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py
index 29381f53e607..6952d4ed05db 100644
--- a/products/tasks/backend/facade/loops.py
+++ b/products/tasks/backend/facade/loops.py
@@ -99,6 +99,7 @@
         "connectors",
         "context_target",
         "triggers",
+        "skill_bundles",
     }
 )
 _PAUSE_FIELD = "enabled"
@@ -212,6 +213,19 @@ class LoopContextTargetDTO:
     outputs: LoopContextOutputsDTO = Field(default_factory=LoopContextOutputsDTO)
 
 
+@dataclass(frozen=True)
+class LoopSkillBundleDTO:
+    """A skill bundle attached to a loop, sans storage internals. `content_sha256` lets the
+    client detect when its local copy of the skill has drifted from the stored snapshot."""
+
+    id: str
+    skill_name: str
+    skill_source: str
+    size: int
+    content_sha256: str
+    uploaded_at: str
+
+
 @dataclass(frozen=True)
 class LoopTriggerDTO:
     """A single loop trigger. `config` shape depends on `type` — see LOOPS.md `LoopTrigger`."""
@@ -257,6 +271,7 @@ class LoopDTO:
     updated_at: datetime
     context_target: LoopContextTargetDTO | None = None
     triggers: list[LoopTriggerDTO] = Field(default_factory=list)
+    skill_bundles: list[LoopSkillBundleDTO] = Field(default_factory=list)
 
 
 @dataclass(frozen=True)
@@ -425,9 +440,29 @@ def _loop_to_dto(loop: Loop) -> LoopDTO:
         updated_at=loop.updated_at,
         context_target=_context_target_dto(loop.context_target),
         triggers=[_trigger_to_dto(trigger) for trigger in triggers],
+        skill_bundles=_skill_bundle_dtos(loop.skill_bundles),
     )
 
 
+def _skill_bundle_dtos(entries: list | None) -> list[LoopSkillBundleDTO]:
+    dtos: list[LoopSkillBundleDTO] = []
+    for entry in entries or []:
+        if not isinstance(entry, dict):
+            continue
+        metadata = entry.get("metadata") or {}
+        dtos.append(
+            LoopSkillBundleDTO(
+                id=str(entry.get("id", "")),
+                skill_name=str(metadata.get("skill_name", "")),
+                skill_source=str(metadata.get("skill_source", "")),
+                size=int(entry.get("size", 0)),
+                content_sha256=str(metadata.get("content_sha256", "")),
+                uploaded_at=str(entry.get("uploaded_at", "")),
+            )
+        )
+    return dtos
+
+
 def _parse_uuid(value: Any) -> UUID | None:
     if not value:
         return None
@@ -560,7 +595,7 @@ def _authorize_update(loop: Loop, user: User | None, validated_data: dict) -> No
             return
         raise LoopPermissionError(
             "Only the loop owner may change identity-bearing configuration (instructions, "
-            "repositories, connectors, behaviors, model config, or triggers)."
+            "repositories, connectors, behaviors, model config, triggers, or skill bundles)."
         )
 
 
@@ -955,6 +990,274 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_
     return _loop_to_dto(loop)
 
 
+MAX_LOOP_SKILL_BUNDLES = 10
+# Decoded per-bundle ceiling. The request is JSON with base64 content and the endpoint
+# caps the declared body at 20MB before parsing, leaving roughly 15MB of decoded budget
+# per request — 10MB keeps this advertised limit honestly reachable within it instead of
+# promising the run-artifact 30MB that a JSON upload can never hit.
+MAX_LOOP_SKILL_BUNDLE_SIZE_BYTES = 10 * 1024 * 1024
+# Expansion ceilings, checked against the zip's central directory before anything is
+# stored. The sandbox extracts every stored bundle on every fire, so a zip bomb would
+# wedge each scheduled run; these mirror the client bundler's own build-time caps.
+MAX_LOOP_SKILL_BUNDLE_FILES = 1000
+MAX_LOOP_SKILL_BUNDLE_UNCOMPRESSED_BYTES = 30 * 1024 * 1024
+# Aggregate expansion ceiling across every bundle in one replace. The sandbox extracts
+# the whole set on every fire, so without this a loop could legitimately carry
+# MAX_LOOP_SKILL_BUNDLES * 30MB of highly-compressible content and inflate ~300MB per run.
+MAX_LOOP_SKILL_BUNDLES_TOTAL_UNCOMPRESSED_BYTES = 60 * 1024 * 1024
+# Ceiling on the zip central directory itself, checked from the trailer before ZipFile
+# parses it. This is what actually bounds parse-time ZipInfo allocation (zipfile walks
+# the directory by size, not by the forgeable entry-count field). 300KB comfortably fits
+# MAX_LOOP_SKILL_BUNDLE_FILES entries with long names while capping a hostile directory
+# at a few thousand records instead of hundreds of thousands.
+MAX_LOOP_SKILL_BUNDLE_CENTRAL_DIR_BYTES = 300 * 1024
+
+
+def _zip_trailer(content_bytes: bytes) -> tuple[int, int] | None:
+    """(entry count, central directory size in bytes) from the zip's
+    end-of-central-directory trailer, without materializing per-entry metadata. Mirrors
+    how `zipfile` locates the trailer (last well-formed record within the max 64KB
+    comment window), so the values validated here are the ones `ZipFile` will use.
+    Returns None when no trailer is found. Like `zipfile._EndRecData`, only the LAST
+    signature occurrence is considered — retrying earlier ones would both diverge from
+    what ZipFile will parse and let a tail stuffed with fake signatures force repeated
+    64KB scans. A last occurrence that fails validation falls through to ZipFile, which
+    rejects the archive as bad. The directory SIZE is the load-bearing number:
+    `ZipFile.infolist()` walks exactly that many directory bytes (one ZipInfo per
+    record, each at least 46 bytes) and ignores the count field, so bounding the size
+    bounds parse-time allocation even when the count field lies. Zip64 archives report
+    sentinel maxima in both fields, which exceed the caps and are rejected."""
+    eocd_size = 22
+    tail = content_bytes[-(eocd_size + 65535) :]
+    offset = tail.rfind(b"PK\x05\x06")
+    if offset == -1:
+        return None
+    record = tail[offset:]
+    if len(record) < eocd_size:
+        return None
+    comment_length = int.from_bytes(record[20:22], "little")
+    if len(record) != eocd_size + comment_length:
+        return None
+    entry_count = int.from_bytes(record[10:12], "little")
+    central_dir_size = int.from_bytes(record[12:16], "little")
+    return entry_count, central_dir_size
+
+
+def _delete_skill_bundle_objects(loop_id: UUID, paths: list[str]) -> None:
+    """Best-effort removal of loop skill bundle objects. Failures are logged, not raised:
+    a leaked object is recoverable garbage, while failing the caller's operation over
+    cleanup is not. Only for objects nothing references (this request's own discarded
+    uploads); superseded objects go through `_expire_skill_bundle_objects` instead."""
+    if not paths:
+        return
+    from posthog.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path
+
+    try:
+        object_storage.delete_objects(paths)
+    except Exception as exc:
+        logger.warning(
+            "loop.skill_bundle_cleanup_failed",
+            extra={"loop_id": str(loop_id), "paths": paths, "error": str(exc)},
+        )
+
+
+def _expire_skill_bundle_objects(loop_id: UUID, team_id: int, paths: list[str]) -> None:
+    """Retire superseded bundle objects via a short retention tag rather than deleting
+    them outright: a fire that committed just before this write may still be running its
+    post-commit S3 copy from these paths, and an immediate delete would fail that run
+    for no reason. The lifecycle rule reaps them after the grace day."""
+    if not paths:
+        return
+    from posthog.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path
+
+    for path in paths:
+        try:
+            object_storage.tag(path, {"ttl_days": "1", "team_id": str(team_id)})
+        except Exception as exc:
+            logger.warning(
+                "loop.skill_bundle_expire_failed",
+                extra={"loop_id": str(loop_id), "path": path, "error": str(exc)},
+            )
+
+
+def replace_loop_skill_bundles(
+    loop_id: str | UUID, team_id: int, user: User | None, *, bundles: list[dict]
+) -> LoopDTO | None:
+    """Replace the loop's attached skill bundles wholesale — the client sends the full
+    declarative set on every save, so there is no per-bundle add/remove surface. Bundle
+    bytes land under the loop's own S3 prefix (not any run's, so run retention never
+    reaps them). Every bundle is validated before the first byte is written, a failed
+    write deletes what this request already wrote, and the manifest swap happens under
+    a row lock with superseded paths read from the locked row — so neither a partial
+    failure nor a concurrent replace strands unreferenced objects. Owner-gated like
+    every other identity-bearing field. Returns `None` if the loop is not
+    found/reachable."""
+    import io  # noqa: PLC0415
+    import hashlib  # noqa: PLC0415
+    import zipfile  # noqa: PLC0415
+
+    from django.utils import timezone as django_timezone  # noqa: PLC0415
+
+    from posthog.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path
+
+    from products.tasks.backend.logic.services.staged_artifacts import get_safe_artifact_name  # noqa: PLC0415
+
+    if len(bundles) > MAX_LOOP_SKILL_BUNDLES:
+        raise LoopValidationError(f"A loop can carry at most {MAX_LOOP_SKILL_BUNDLES} skill bundles.")
+
+    loop = _fetch_loop_for_write(loop_id, team_id, user)
+    if loop is None:
+        return None
+    _authorize_update(loop, user, {"skill_bundles": bundles})
+
+    decoded: list[tuple[dict, bytes]] = []
+    total_uncompressed = 0
+    for bundle in bundles:
+        try:
+            content_bytes = base64.b64decode(bundle["content_base64"], validate=True)
+        except (ValueError, TypeError):
+            raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' is not valid base64.")
+        if len(content_bytes) > MAX_LOOP_SKILL_BUNDLE_SIZE_BYTES:
+            raise LoopValidationError(
+                f"Skill bundle for '{bundle['skill_name']}' exceeds the "
+                f"{MAX_LOOP_SKILL_BUNDLE_SIZE_BYTES // (1024 * 1024)}MB limit."
+            )
+        if hashlib.sha256(content_bytes).hexdigest() != bundle["content_sha256"]:
+            raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' does not match its declared sha256.")
+        # Trailer prechecks run BEFORE ZipFile parses the archive: infolist()
+        # materializes a ZipInfo per directory record, so a small zip carrying a huge
+        # central directory would otherwise allocate tens of MB of metadata just to be
+        # rejected. The directory-size cap is the real allocation bound; the count is a
+        # cheap fast-fail for honest oversized archives.
+        trailer = _zip_trailer(content_bytes)
+        if trailer is None:
+            raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' is not a valid zip archive.")
+        entry_count, central_dir_size = trailer
+        if entry_count > MAX_LOOP_SKILL_BUNDLE_FILES or central_dir_size > MAX_LOOP_SKILL_BUNDLE_CENTRAL_DIR_BYTES:
+            raise LoopValidationError(
+                f"Skill bundle for '{bundle['skill_name']}' contains more than {MAX_LOOP_SKILL_BUNDLE_FILES} files."
+            )
+        # Central-directory metadata only — nothing is decompressed here. The parsed
+        # entry list is re-checked against the cap in case the trailer undercounted:
+        # a lying count can shave the fast-fail, but the size cap above already bounds
+        # how much ZipFile can allocate getting here.
+        try:
+            with zipfile.ZipFile(io.BytesIO(content_bytes)) as archive:
+                archive_entries = archive.infolist()
+        except zipfile.BadZipFile:
+            raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' is not a valid zip archive.")
+        if len(archive_entries) > MAX_LOOP_SKILL_BUNDLE_FILES:
+            raise LoopValidationError(
+                f"Skill bundle for '{bundle['skill_name']}' contains more than {MAX_LOOP_SKILL_BUNDLE_FILES} files."
+            )
+        uncompressed_total = sum(entry.file_size for entry in archive_entries)
+        if uncompressed_total > MAX_LOOP_SKILL_BUNDLE_UNCOMPRESSED_BYTES:
+            raise LoopValidationError(
+                f"Skill bundle for '{bundle['skill_name']}' expands to more than "
+                f"{MAX_LOOP_SKILL_BUNDLE_UNCOMPRESSED_BYTES // (1024 * 1024)}MB."
+            )
+        total_uncompressed += uncompressed_total
+        if total_uncompressed > MAX_LOOP_SKILL_BUNDLES_TOTAL_UNCOMPRESSED_BYTES:
+            raise LoopValidationError(
+                "The loop's skill bundles together expand to more than "
+                f"{MAX_LOOP_SKILL_BUNDLES_TOTAL_UNCOMPRESSED_BYTES // (1024 * 1024)}MB."
+            )
+        decoded.append((bundle, content_bytes))
+
+    prefix = loop.get_skill_bundle_s3_prefix()
+    entries: list[dict] = []
+    written_paths: list[str] = []
+    try:
+        for bundle, content_bytes in decoded:
+            bundle_id = uuid4().hex
+            safe_name = get_safe_artifact_name(bundle["file_name"])
+            storage_path = f"{prefix}/{bundle_id[:8]}_{safe_name}"
+            object_storage.write(storage_path, content_bytes, {"ContentType": "application/zip"})
+            written_paths.append(storage_path)
+            try:
+                object_storage.tag(storage_path, {"team_id": str(loop.team_id)})
+            except Exception as exc:
+                logger.warning(
+                    "loop.skill_bundle_tag_failed",
+                    extra={"loop_id": str(loop.id), "storage_path": storage_path, "error": str(exc)},
+                )
+            entries.append(
+                {
+                    "id": bundle_id,
+                    "name": safe_name,
+                    "type": "skill_bundle",
+                    "source": "posthog_code_skill",
+                    "size": len(content_bytes),
+                    "content_type": "application/zip",
+                    "storage_path": storage_path,
+                    "uploaded_at": django_timezone.now().isoformat(),
+                    "metadata": {
+                        "skill_name": bundle["skill_name"],
+                        "skill_source": bundle["skill_source"],
+                        "content_sha256": bundle["content_sha256"],
+                        "bundle_format": bundle["bundle_format"],
+                        "schema_version": 1,
+                    },
+                }
+            )
+    except Exception:
+        _delete_skill_bundle_objects(loop.id, written_paths)
+        raise
+
+    # Previous paths come from the row read under the lock, not the earlier fetch: if a
+    # concurrent replace committed in between, its objects are what this swap supersedes
+    # and must be the ones deleted, or they'd be stranded with no manifest referencing them.
+    # A delete that committed in between is re-checked the same way — its lock-holder
+    # already cleared the manifest, so this request must discard its own uploads instead
+    # of resurrecting bundles on a deleted loop.
+    previous_paths: list[str] = []
+    lost_delete_race = False
+    denied: LoopPermissionError | None = None
+    try:
+        with transaction.atomic():
+            locked = Loop.objects.unscoped().select_for_update().get(pk=loop.pk)
+            if locked.deleted:
+                lost_delete_race = True
+            else:
+                try:
+                    # Ownership can change (takeover) while the bundle bytes were uploading,
+                    # so the swap is re-authorized against the row as locked — otherwise a
+                    # former owner's in-flight replace lands a skill that every future fire
+                    # executes under the new owner's credentials.
+                    _authorize_update(locked, user, {"skill_bundles": bundles})
+                except LoopPermissionError as exc:
+                    denied = exc
+                else:
+                    previous_paths = [
+                        entry["storage_path"]
+                        for entry in (locked.skill_bundles or [])
+                        if isinstance(entry, dict) and entry.get("storage_path")
+                    ]
+                    locked.skill_bundles = entries
+                    locked.save(update_fields=["skill_bundles", "updated_at"])
+    except Exception:
+        # A failed swap (lock timeout, database error) rolls back the manifest but not
+        # the objects this request already wrote — release them before surfacing.
+        _delete_skill_bundle_objects(loop.id, written_paths)
+        raise
+
+    if lost_delete_race:
+        _delete_skill_bundle_objects(loop.id, written_paths)
+        return None
+
+    if denied is not None:
+        _delete_skill_bundle_objects(loop.id, written_paths)
+        raise denied
+
+    new_paths = {entry["storage_path"] for entry in entries}
+    _expire_skill_bundle_objects(
+        loop.id, loop.team_id, [path for path in previous_paths if path.startswith(prefix) and path not in new_paths]
+    )
+
+    loop.refresh_from_db()
+    return _loop_to_dto(loop)
+
+
 def soft_delete_loop(loop_id: str | UUID, team_id: int, user: User | None) -> bool:
     loop = _fetch_loop_for_write(loop_id, team_id, user)
     if loop is None:
@@ -969,8 +1272,22 @@ def soft_delete_loop(loop_id: str | UUID, team_id: int, user: User | None) -> bo
     elif not (_is_owner(loop, user) or _is_team_admin(loop, user)):
         raise LoopPermissionError("Only the owner or a project admin may delete a loop.")
 
-    loop.deleted = True
-    loop.save(update_fields=["deleted", "updated_at"])
+    # A deleted loop never fires again, so its skill bundle objects are dead weight with
+    # no retention TTL — release them now and clear the manifest so the row stays honest.
+    # The paths are read under the same row lock the skill-bundle replace takes, so a
+    # replace racing this delete either commits first (its objects are what we read and
+    # remove here) or locks after us, sees `deleted`, and cleans up its own uploads.
+    with transaction.atomic():
+        locked = Loop.objects.unscoped().select_for_update().get(pk=loop.pk)
+        bundle_paths = [
+            entry["storage_path"]
+            for entry in (locked.skill_bundles or [])
+            if isinstance(entry, dict) and entry.get("storage_path")
+        ]
+        locked.deleted = True
+        locked.skill_bundles = []
+        locked.save(update_fields=["deleted", "skill_bundles", "updated_at"])
+    _expire_skill_bundle_objects(loop.id, loop.team_id, bundle_paths)
     loop_service.delete_loop_schedules(loop)
     return True
 
@@ -1242,6 +1559,9 @@ def _loop_runs_page(loop: Loop, team_id: int, *, cursor: str | None, limit: int)
     "LoopRunDTO",
     "LoopRunPageDTO",
     "LoopScheduleSyncStatus",
+    "LoopSkillBundleDTO",
+    "MAX_LOOP_SKILL_BUNDLES",
+    "MAX_LOOP_SKILL_BUNDLE_SIZE_BYTES",
     "LoopTriggerDTO",
     "LoopTriggerType",
     "LoopVisibility",
@@ -1266,6 +1586,7 @@ def _loop_runs_page(loop: Loop, team_id: int, *, cursor: str | None, limit: int)
     "pause_loops_for_removed_member",
     "pause_loops_referencing_integrations",
     "preview_loop",
+    "replace_loop_skill_bundles",
     "repository_accessible_via_integration",
     "sandbox_environment_queryset",
     "soft_delete_loop",
diff --git a/products/tasks/backend/logic/services/loop_runs.py b/products/tasks/backend/logic/services/loop_runs.py
index 8161ad6ca6b8..c0ed0dcdda11 100644
--- a/products/tasks/backend/logic/services/loop_runs.py
+++ b/products/tasks/backend/logic/services/loop_runs.py
@@ -490,6 +490,150 @@ def _team_rate_capped(loop: Loop) -> bool:
     return fire_count >= LOOP_TEAM_RATE_CAP_PER_DAY
 
 
+def _seed_skill_bundles_and_dispatch(
+    *,
+    task_run: TaskRun,
+    team_id: int,
+    user_id: int | None,
+    task_id: str,
+    run_id: str,
+    create_pr: bool,
+    posthog_mcp_scopes: PosthogMcpScopes,
+) -> None:
+    """Post-commit tail of a fire: seed the run's snapshotted skill bundles, then
+    dispatch the Temporal workflow. Seeding copies S3 objects, and
+    ``_fire_loop_committed`` holds the team-wide advisory lock for its whole transaction
+    — an external call under that lock would serialize every fire for the team behind S3
+    latency (the same reason the usage gate runs before the lock and dispatch runs after
+    commit). Post-commit there is no rollback to lean on, so a failed seed compensates
+    instead: the run is terminalized as failed (feeding the loop's failure bookkeeping)
+    and the workflow is never dispatched, so the agent can't run with silently missing
+    skills."""
+    try:
+        _seed_skill_bundle_artifacts(task_run)
+    except Exception as exc:
+        logger.exception(
+            "loop_run.skill_bundle_seed_failed",
+            extra={"task_run_id": run_id, "error": str(exc)},
+        )
+        from products.tasks.backend.temporal.client import (  # noqa: PLC0415 — breaks the temporal.client -> loop_runs import cycle
+            _terminalize_unstarted_task_run,
+        )
+
+        _terminalize_unstarted_task_run(run_id, "Failed to stage the loop's skill bundles for this run")
+        return
+
+    _execute_task_processing_workflow_for_loop(
+        team_id=team_id,
+        user_id=user_id,
+        task_id=task_id,
+        run_id=run_id,
+        create_pr=create_pr,
+        posthog_mcp_scopes=posthog_mcp_scopes,
+    )
+
+
+def ensure_loop_skill_bundles_seeded(task_run: TaskRun) -> bool:
+    """Idempotently seed a loop run's skill bundles ahead of an out-of-band dispatch.
+
+    The orphaned-QUEUED reconciler recovers runs whose create-time ``on_commit`` callback
+    was lost — the same callback that seeds skill bundles — so re-dispatching without this
+    check would silently start the run with its skills missing. Seeds strictly from the
+    snapshot captured on the run at fire time, never from the live loop: the loop's
+    bundles (and its owner) may have changed since, and the run executes with the
+    credentials captured at fire time. Returns False when seeding was needed but failed;
+    the caller must not dispatch (the next sweep retries)."""
+    task = task_run.task
+    if task.origin_product != Task.OriginProduct.LOOP or task.loop_id is None:
+        return True
+    already_seeded = any(
+        isinstance(entry, dict) and entry.get("type") == "skill_bundle" for entry in (task_run.artifacts or [])
+    )
+    if already_seeded:
+        return True
+    try:
+        _seed_skill_bundle_artifacts(task_run)
+    except Exception:
+        logger.exception(
+            "loop_run.skill_bundle_seed_failed",
+            extra={"task_run_id": str(task_run.id)},
+        )
+        return False
+    return True
+
+
+def _snapshot_skill_bundle_seeds(loop: Loop) -> list[dict]:
+    """The bundle entries a fire freezes onto its run's state. Seeding always reads this
+    snapshot rather than the loop row, so a later replace or ownership takeover can't
+    swap which skills an already-created run installs."""
+    return [entry for entry in (loop.skill_bundles or []) if isinstance(entry, dict) and entry.get("storage_path")]
+
+
+def _seed_skill_bundle_artifacts(task_run: TaskRun) -> None:
+    """Copy the run's snapshotted skill bundles into place: S3 objects under the run's
+    artifact prefix plus matching ``skill_bundle`` manifest entries, so the sandbox
+    agent-server installs them exactly like bundles a client uploaded at task creation.
+    Raises on a failed copy; the caller compensates. Already-copied objects are deleted
+    best-effort before re-raising, since nothing would ever reference them."""
+    state = task_run.state if isinstance(task_run.state, dict) else {}
+    seeds = state.get("skill_bundle_seeds")
+    bundles = [entry for entry in (seeds or []) if isinstance(entry, dict) and entry.get("storage_path")]
+    if not bundles:
+        return
+
+    from posthog.storage import object_storage  # noqa: PLC0415 — keep storage deps off the fire path import
+
+    from products.tasks.backend.logic.services.staged_artifacts import get_safe_artifact_name  # noqa: PLC0415
+
+    # `state` is a client-PATCHable surface (the seeds key is server-protected, but this
+    # copy primitive must not trust it anyway): every source path has to sit under the
+    # owning loop's own bundle prefix, and the target segments are re-sanitized, or a
+    # forged entry would read or write arbitrary bucket keys via the recovery path.
+    loop_id = task_run.task.loop_id
+    if loop_id is None:
+        raise ValueError("skill bundle seeds on a run without a loop")
+    loop_prefix = f"{Loop.skill_bundle_s3_prefix_for(task_run.team_id, loop_id)}/"
+
+    run_prefix = task_run.get_artifact_s3_prefix()
+    manifest = list(task_run.artifacts or [])
+    copied_paths: list[str] = []
+    try:
+        for bundle in bundles:
+            entry = dict(bundle)
+            source_path = str(entry["storage_path"])
+            if not source_path.startswith(loop_prefix):
+                raise ValueError("skill bundle seed escapes the loop's storage prefix")
+            entry_id = str(entry["id"])
+            if not re.fullmatch(r"[0-9a-f]{8,64}", entry_id):
+                raise ValueError("skill bundle seed has a malformed id")
+            entry["name"] = get_safe_artifact_name(str(entry["name"]))
+            target_path = f"{run_prefix}/{entry_id[:8]}_{entry['name']}"
+            object_storage.copy(source_path, target_path)
+            copied_paths.append(target_path)
+            try:
+                object_storage.tag(target_path, {"ttl_days": "30", "team_id": str(task_run.team_id)})
+            except Exception as exc:
+                logger.warning(
+                    "loop_run.skill_bundle_tag_failed",
+                    extra={"task_run_id": str(task_run.id), "storage_path": target_path, "error": str(exc)},
+                )
+            entry["storage_path"] = target_path
+            manifest.append(entry)
+
+        task_run.artifacts = manifest
+        task_run.save(update_fields=["artifacts", "updated_at"])
+    except Exception:
+        if copied_paths:
+            try:
+                object_storage.delete_objects(copied_paths)
+            except Exception as cleanup_exc:
+                logger.warning(
+                    "loop_run.skill_bundle_seed_cleanup_failed",
+                    extra={"task_run_id": str(task_run.id), "paths": copied_paths, "error": str(cleanup_exc)},
+                )
+        raise
+
+
 def _create_loop_task_and_run(loop: Loop, trigger: LoopTrigger | None, trigger_context: str) -> tuple[Task, TaskRun]:
     repository: str | None = None
     github_integration_id: int | None = None
@@ -571,6 +715,13 @@ def _create_loop_task_and_run(loop: Loop, trigger: LoopTrigger | None, trigger_c
         "slack_thread_context": None,
         "workflow_id_prefix": None,
     }
+    # Freeze the bundle set on the run itself: seeding (post-commit and reconciler
+    # recovery alike) reads only this snapshot, so a later replace or ownership takeover
+    # can't change which skills this run installs under its captured credentials.
+    bundle_seeds = _snapshot_skill_bundle_seeds(loop)
+    if bundle_seeds:
+        extra_state["skill_bundle_seeds"] = bundle_seeds
+
     task_run = task.create_run(mode="background", extra_state=extra_state)
 
     team_id = loop.team_id
@@ -579,7 +730,8 @@ def _create_loop_task_and_run(loop: Loop, trigger: LoopTrigger | None, trigger_c
     run_id = str(task_run.id)
 
     transaction.on_commit(
-        lambda: _execute_task_processing_workflow_for_loop(
+        lambda: _seed_skill_bundles_and_dispatch(
+            task_run=task_run,
             team_id=team_id,
             user_id=user_id,
             task_id=task_id,
diff --git a/products/tasks/backend/migrations/0072_loop_skill_bundles.py b/products/tasks/backend/migrations/0072_loop_skill_bundles.py
new file mode 100644
index 000000000000..113a384deaee
--- /dev/null
+++ b/products/tasks/backend/migrations/0072_loop_skill_bundles.py
@@ -0,0 +1,15 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("tasks", "0071_alter_origin_product_mcp_analytics"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="loop",
+            name="skill_bundles",
+            field=models.JSONField(blank=True, default=list),
+        ),
+    ]
diff --git a/products/tasks/backend/migrations/max_migration.txt b/products/tasks/backend/migrations/max_migration.txt
index ba43db339840..e12dffe2bd4e 100644
--- a/products/tasks/backend/migrations/max_migration.txt
+++ b/products/tasks/backend/migrations/max_migration.txt
@@ -1 +1 @@
-0071_alter_origin_product_mcp_analytics
+0072_loop_skill_bundles
diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py
index 1016d4f0966e..8c1859bf6184 100644
--- a/products/tasks/backend/models.py
+++ b/products/tasks/backend/models.py
@@ -1145,6 +1145,10 @@ class OverlapPolicy(models.TextChoices):
     # Drives feed placement (each run's Task.channel) and the context.md / canvas publish contract
     # injected into every run's prompt. See products/tasks/docs/LOOPS.md.
     context_target = models.JSONField(default=dict, blank=True)
+    # Skill bundles attached at save time: zipped local skills whose manifest entries (same shape
+    # as TaskRun.artifacts entries, type "skill_bundle", bytes in object storage under
+    # get_skill_bundle_s3_prefix()) are copied into every fired run so the sandbox installs them.
+    skill_bundles = models.JSONField(default=list, blank=True)
     internal = models.BooleanField(
         default=False,
         help_text="If true, this loop is for internal use and should not be exposed to end users.",
@@ -1175,6 +1179,16 @@ class Meta:
     def __str__(self):
         return self.name
 
+    @staticmethod
+    def skill_bundle_s3_prefix_for(team_id: int, loop_id: "uuid.UUID | str") -> str:
+        """Base prefix for a loop's skill bundle objects in S3, computable from ids so
+        seeding can validate snapshot paths without loading the row."""
+        tasks_folder = settings.OBJECT_STORAGE_TASKS_FOLDER
+        return f"{tasks_folder}/artifacts/team_{team_id}/loop_{loop_id}"
+
+    def get_skill_bundle_s3_prefix(self) -> str:
+        return Loop.skill_bundle_s3_prefix_for(self.team_id, self.id)
+
     def _get_before_update(self, **kwargs: Any) -> "Loop | None":
         # ModelActivityMixin's prior-state lookup goes through `objects` (the fail-closed
         # TeamScopedManager). Loop saves happen from webhook handlers and Temporal activities
diff --git a/products/tasks/backend/presentation/serializers_loops.py b/products/tasks/backend/presentation/serializers_loops.py
index 89608e00e054..c30af7ce33c0 100644
--- a/products/tasks/backend/presentation/serializers_loops.py
+++ b/products/tasks/backend/presentation/serializers_loops.py
@@ -29,6 +29,10 @@
     get_models_for_runtime_adapter,
     get_reasoning_effort_error,
 )
+from products.tasks.backend.presentation.serializers import (
+    TASK_RUN_SKILL_BUNDLE_FORMAT_CHOICES,
+    TASK_RUN_SKILL_SOURCE_CHOICES,
+)
 
 
 class LoopRepositoryEntrySerializer(serializers.Serializer):
@@ -549,6 +553,11 @@ class Meta:
         dataclass = loops_facade.LoopRepositoryEntryDTO
 
 
+class LoopSkillBundleResponseSerializer(DataclassSerializer):
+    class Meta:
+        dataclass = loops_facade.LoopSkillBundleDTO
+
+
 class LoopSerializer(DataclassSerializer):
     """Detail/create/update response for a loop, including its triggers."""
 
@@ -560,6 +569,9 @@ class LoopSerializer(DataclassSerializer):
         allow_null=True, required=False, help_text="Context this loop is attached to, or null when unattached."
     )
     triggers = LoopTriggerSerializer(many=True, help_text="Triggers attached to this loop.")
+    skill_bundles = LoopSkillBundleResponseSerializer(
+        many=True, help_text="Skill bundles attached to this loop, seeded into every fired run."
+    )
 
     class Meta:
         dataclass = loops_facade.LoopDTO
@@ -592,6 +604,7 @@ class Meta:
             "created_at",
             "updated_at",
             "triggers",
+            "skill_bundles",
         ]
 
 
@@ -654,6 +667,34 @@ class Meta:
         dataclass = loops_facade.LoopPreviewDTO
 
 
+class LoopSkillBundleUploadSerializer(serializers.Serializer):
+    """One zipped local skill in a skill-bundle replace request."""
+
+    file_name = serializers.CharField(
+        allow_blank=False, max_length=255, help_text="File name for the stored bundle, e.g. `my-skill.zip`."
+    )
+    skill_name = serializers.CharField(
+        allow_blank=False, max_length=255, help_text="Name of the skill inside the bundle."
+    )
+    skill_source = serializers.ChoiceField(
+        choices=TASK_RUN_SKILL_SOURCE_CHOICES, help_text="Local source the bundle was built from, such as user or repo."
+    )
+    content_sha256 = serializers.RegexField(
+        regex=r"^[a-f0-9]{64}$", help_text="SHA-256 hex digest of the bundle bytes."
+    )
+    bundle_format = serializers.ChoiceField(
+        choices=TASK_RUN_SKILL_BUNDLE_FORMAT_CHOICES, help_text="Archive format used for the bundle."
+    )
+    content_base64 = serializers.CharField(allow_blank=False, help_text="Base64-encoded bundle bytes.")
+
+
+class LoopSkillBundlesWriteSerializer(serializers.Serializer):
+    """Request body for replacing a loop's attached skill bundles wholesale. Send an empty
+    list to detach every skill."""
+
+    bundles = LoopSkillBundleUploadSerializer(many=True, allow_empty=True)
+
+
 class LoopFireRunSerializer(DataclassSerializer):
     """Response for a manual (`run/`) or external (`trigger/`) fire."""
 
diff --git a/products/tasks/backend/presentation/views/loops.py b/products/tasks/backend/presentation/views/loops.py
index bad8b355164e..9d230a320d67 100644
--- a/products/tasks/backend/presentation/views/loops.py
+++ b/products/tasks/backend/presentation/views/loops.py
@@ -29,10 +29,17 @@
     LoopRunPageSerializer,
     LoopRunsQuerySerializer,
     LoopSerializer,
+    LoopSkillBundlesWriteSerializer,
     LoopWriteSerializer,
 )
 
 MAX_LOOP_TRIGGER_PAYLOAD_BYTES = 64 * 1024
+# Whole-request ceiling for the skill_bundles replace, enforced from Content-Length
+# before request.data parses. This is the authoritative parse bound for the endpoint:
+# DRF parses JSON from the request stream, where Django's DATA_UPLOAD_MAX_MEMORY_SIZE
+# is not guaranteed to apply, so the gate must not assume a larger backstop. 20MB fits
+# one 10MB decoded bundle plus base64 overhead and dependencies.
+MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES = 20 * 1024 * 1024
 
 
 def _loop_limit_response(exc: "loops_facade.LoopLimitError") -> Response:
@@ -142,7 +149,9 @@ class LoopViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet):
     # (`runs`), so a service that triggers can also poll the outcome. Everything else (CRUD,
     # manual run, preview) stays session/PAT/OAuth-only.
     psak_allowed_actions = ["trigger", "runs"]
-    http_method_names = ["get", "post", "patch", "delete", "head", "options"]
+    # "put" is routed only by the skill_bundles action (wholesale replace); the loop
+    # resource itself stays PATCH-only.
+    http_method_names = ["get", "post", "patch", "put", "delete", "head", "options"]
     pagination_class = LoopsPagination
     # Fallback for drf-spectacular introspection only; every action declares its own
     # request/response schema via @validated_request / @extend_schema.
@@ -276,6 +285,53 @@ def destroy(self, request, pk=None, **kwargs):
             raise NotFound()
         return Response(status=status.HTTP_204_NO_CONTENT)
 
+    @extend_schema(
+        summary="Replace a loop's skill bundles",
+        description=(
+            "Replaces the loop's attached skill bundles wholesale: zipped local skills whose "
+            "contents are seeded into every fired run's sandbox. Send an empty list to detach "
+            "every skill. Owner-only on team loops, like other identity-bearing configuration."
+        ),
+        request=LoopSkillBundlesWriteSerializer,
+        responses={
+            200: LoopSerializer,
+            403: OpenApiResponse(description="Not permitted to change this loop's skills"),
+            404: OpenApiResponse(description="Loop not found"),
+            411: OpenApiResponse(description="Content-Length header missing"),
+            413: OpenApiResponse(description="Request body exceeds the skill bundle size ceiling"),
+        },
+    )
+    @action(detail=True, methods=["put"], url_path="skill_bundles", required_scopes=["loop:write"])
+    def skill_bundles(self, request, pk=None, **kwargs):
+        # Same shape as `trigger`'s payload gate: require a declared length, then reject
+        # oversized requests from it, all before request.data parses (and retains) the
+        # body. This gate is the endpoint's authoritative parse bound — see the note on
+        # MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES.
+        content_length = _content_length(request)
+        if content_length <= 0:
+            return Response(
+                {"detail": "A Content-Length header is required."},
+                status=status.HTTP_411_LENGTH_REQUIRED,
+            )
+        if content_length > MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES:
+            return Response(
+                {"detail": (f"Skill bundle request exceeds {MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES // (1024 * 1024)}MB.")},
+                status=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            )
+        serializer = LoopSkillBundlesWriteSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        try:
+            loop = loops_facade.replace_loop_skill_bundles(
+                pk, self.team_id, request.user, bundles=list(serializer.validated_data["bundles"])
+            )
+        except loops_facade.LoopPermissionError as exc:
+            raise PermissionDenied(str(exc))
+        except loops_facade.LoopValidationError as exc:
+            raise ValidationError(str(exc))
+        if loop is None:
+            raise NotFound()
+        return Response(LoopSerializer(loop).data)
+
     @extend_schema(
         summary="Run a loop manually",
         description="Manual fire from the UI. Owner-only for personal loops; any team member for team loops.",
diff --git a/products/tasks/backend/temporal/client.py b/products/tasks/backend/temporal/client.py
index 4f5a80c7971d..624201901db5 100644
--- a/products/tasks/backend/temporal/client.py
+++ b/products/tasks/backend/temporal/client.py
@@ -450,6 +450,18 @@ def redispatch_orphaned_task_run(run_id: str) -> str:
         posthog_mcp_scopes=dispatch_params.get("posthog_mcp_scopes") or _resolve_mcp_scopes(task_run),
     )
 
+    # A loop run's skill bundles are seeded by the same on_commit callback whose loss
+    # this sweep recovers from, so dispatching without re-seeding would silently start
+    # the run with its skills missing. Idempotent for already-seeded runs. A failed seed
+    # is treated like any other transient recovery error: retried next sweep, with the
+    # 24h killer as the terminal backstop.
+    from products.tasks.backend.logic.services.loop_runs import (  # noqa: PLC0415 — breaks the loop_runs -> temporal.client import cycle
+        ensure_loop_skill_bundles_seeded,
+    )
+
+    if not ensure_loop_skill_bundles_seeded(task_run):
+        return "error"
+
     observe_task_run_workflow_start(task_run, outcome="attempted", reason="reconcile")
     _capture_run_feature_flags(run_id)
     try:
diff --git a/products/tasks/backend/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py
index 1af9a64ca22c..64a554103eed 100644
--- a/products/tasks/backend/tests/test_loop_runs.py
+++ b/products/tasks/backend/tests/test_loop_runs.py
@@ -537,6 +537,139 @@ def test_fire_threads_the_loops_sandbox_environment_into_run_state(self, _name,
             self.assertNotIn("sandbox_environment_id", task_run.state)
 
 
+class TestFireLoopSeedsSkillBundles(LoopRunsTestCase):
+    def _bundle_entry(self, loop: Loop, **overrides) -> dict:
+        # Seeding validates every source path against the owning loop's real prefix.
+        entry = {
+            "id": "abcdef0123456789abcdef0123456789",
+            "name": "my-skill.zip",
+            "type": "skill_bundle",
+            "source": "posthog_code_skill",
+            "size": 9,
+            "content_type": "application/zip",
+            "storage_path": f"{loop.get_skill_bundle_s3_prefix()}/abcdef01_my-skill.zip",
+            "uploaded_at": "2026-07-22T00:00:00+00:00",
+            "metadata": {
+                "skill_name": "my-skill",
+                "skill_source": "user",
+                "content_sha256": "a" * 64,
+                "bundle_format": "zip",
+                "schema_version": 1,
+            },
+        }
+        entry.update(overrides)
+        return entry
+
+    def loop_with_bundles(self, *entry_overrides: dict) -> tuple[Loop, list[dict]]:
+        loop = self.create_loop()
+        entries = [self._bundle_entry(loop, **overrides) for overrides in (entry_overrides or ({},))]
+        loop.skill_bundles = entries
+        loop.save(update_fields=["skill_bundles"])
+        return loop, entries
+
+    def fire_with_post_commit(self, loop: Loop, trigger: LoopTrigger):
+        """Fire once, executing the post-commit seed-and-dispatch tail with the workflow
+        dispatch mocked. Returns (result, mock_dispatch)."""
+        with patch(f"{LOOP_RUNS_MODULE}._execute_task_processing_workflow_for_loop") as mock_dispatch:
+            with self.captureOnCommitCallbacks(execute=True):
+                result = fire_loop(loop, trigger, "fire-1", "ctx")
+        return result, mock_dispatch
+
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.copy")
+    def test_fire_copies_loop_skill_bundles_into_the_run_manifest(self, mock_copy, mock_tag):
+        loop, (entry,) = self.loop_with_bundles()
+        trigger = self.create_trigger(loop)
+
+        result, mock_dispatch = self.fire_with_post_commit(loop, trigger)
+
+        self.assertTrue(result.created)
+        assert result.task_run_id is not None
+        task_run = TaskRun.objects.get(id=result.task_run_id)
+        expected_target = f"{task_run.get_artifact_s3_prefix()}/abcdef01_my-skill.zip"
+        mock_copy.assert_called_once_with(entry["storage_path"], expected_target)
+        seeded = [artifact for artifact in task_run.artifacts if artifact["type"] == "skill_bundle"]
+        self.assertEqual(len(seeded), 1)
+        self.assertEqual(seeded[0]["storage_path"], expected_target)
+        self.assertEqual(seeded[0]["metadata"], entry["metadata"])
+        mock_dispatch.assert_called_once()
+        # The fire freezes the bundle set on the run; recovery paths seed from this
+        # snapshot instead of the live loop.
+        self.assertEqual(task_run.state["skill_bundle_seeds"][0]["storage_path"], entry["storage_path"])
+        loop.refresh_from_db()
+        self.assertEqual(loop.skill_bundles[0]["storage_path"], entry["storage_path"])
+
+    def test_fire_without_bundles_does_not_touch_storage(self):
+        loop = self.create_loop()
+        trigger = self.create_trigger(loop)
+
+        with patch("posthog.storage.object_storage.copy") as mock_copy:
+            result, mock_dispatch = self.fire_with_post_commit(loop, trigger)
+
+        self.assertTrue(result.created)
+        mock_copy.assert_not_called()
+        mock_dispatch.assert_called_once()
+
+    def test_seeding_happens_after_commit_not_under_the_fire_lock(self):
+        # S3 copies must never run inside the fire transaction: it holds the team-wide
+        # advisory lock, so an S3 stall there would serialize every fire for the team.
+        loop, _entries = self.loop_with_bundles()
+        trigger = self.create_trigger(loop)
+
+        with patch("posthog.storage.object_storage.copy") as mock_copy:
+            with patch(f"{LOOP_RUNS_MODULE}._execute_task_processing_workflow_for_loop"):
+                with self.captureOnCommitCallbacks(execute=False) as callbacks:
+                    result = fire_loop(loop, trigger, "fire-1", "ctx")
+                mock_copy.assert_not_called()
+                for callback in callbacks:
+                    callback()
+                mock_copy.assert_called_once()
+
+        self.assertTrue(result.created)
+
+    @patch("posthog.storage.object_storage.delete_objects")
+    @patch("posthog.storage.object_storage.copy", side_effect=RuntimeError("s3 down"))
+    def test_a_failed_seed_terminalizes_the_run_and_skips_dispatch(self, mock_copy, mock_delete):
+        loop, _entries = self.loop_with_bundles()
+        trigger = self.create_trigger(loop)
+
+        result, mock_dispatch = self.fire_with_post_commit(loop, trigger)
+
+        # The fire itself committed; the compensation happens post-commit.
+        self.assertTrue(result.created)
+        assert result.task_run_id is not None
+        task_run = TaskRun.objects.get(id=result.task_run_id)
+        self.assertEqual(task_run.status, TaskRun.Status.FAILED)
+        assert task_run.error_message is not None
+        self.assertIn("skill bundles", task_run.error_message)
+        mock_dispatch.assert_not_called()
+        self.assertEqual(self.active_run_count(loop), 0)
+
+    @patch("posthog.storage.object_storage.delete_objects")
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.copy")
+    def test_a_partially_seeded_fire_deletes_the_copies_it_made(self, mock_copy, mock_tag, mock_delete):
+        # Nothing ever references a partial copy set (the run is terminalized, and any
+        # later fire mints a new run id) — without cleanup every failed attempt would
+        # strand another set of objects.
+        mock_copy.side_effect = [None, RuntimeError("s3 down")]
+        loop, _entries = self.loop_with_bundles(
+            {},
+            {"id": "fedcba9876543210fedcba9876543210", "name": "other-skill.zip"},
+        )
+        trigger = self.create_trigger(loop)
+
+        result, mock_dispatch = self.fire_with_post_commit(loop, trigger)
+
+        self.assertTrue(result.created)
+        mock_delete.assert_called_once()
+        deleted_paths = mock_delete.call_args.args[0]
+        self.assertEqual(len(deleted_paths), 1)
+        self.assertIn("abcdef01_my-skill.zip", deleted_paths[0])
+        mock_dispatch.assert_not_called()
+        self.assertEqual(self.active_run_count(loop), 0)
+
+
 class TestFireLoopContextTarget(LoopRunsTestCase):
     FOLDER_ID = "11111111-1111-1111-1111-111111111111"
     CANVAS_ID = "22222222-2222-2222-2222-222222222222"
diff --git a/products/tasks/backend/tests/test_loops_api.py b/products/tasks/backend/tests/test_loops_api.py
index 1c0ea53c5dd9..18cc02e9bce5 100644
--- a/products/tasks/backend/tests/test_loops_api.py
+++ b/products/tasks/backend/tests/test_loops_api.py
@@ -1,3 +1,7 @@
+import io
+import base64
+import hashlib
+import zipfile
 from contextlib import nullcontext
 from datetime import timedelta
 from uuid import UUID
@@ -231,6 +235,314 @@ def test_resent_trigger_without_enabled_keeps_its_current_value(self):
         self.assertFalse(response.json()["triggers"][0]["enabled"])
 
 
+class LoopSkillBundlesAPITest(LoopsAPITestCase):
+    def _skill_bundles_url(self, loop_id: str) -> str:
+        return f"{self._loop_url(loop_id)}skill_bundles/"
+
+    def _zip_bytes(self, files: dict[str, str] | None = None) -> bytes:
+        buffer = io.BytesIO()
+        with zipfile.ZipFile(buffer, "w") as archive:
+            for entry_name, entry_content in (files or {"SKILL.md": "body"}).items():
+                archive.writestr(entry_name, entry_content)
+        return buffer.getvalue()
+
+    def _bundle_payload(self, name: str = "my-skill", content: bytes | None = None) -> dict:
+        content_bytes = content if content is not None else self._zip_bytes()
+        return {
+            "file_name": f"{name}.zip",
+            "skill_name": name,
+            "skill_source": "user",
+            "content_sha256": hashlib.sha256(content_bytes).hexdigest(),
+            "bundle_format": "zip",
+            "content_base64": base64.b64encode(content_bytes).decode("ascii"),
+        }
+
+    @patch("posthog.storage.object_storage.delete_objects")
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.write")
+    def test_owner_replaces_and_clears_skill_bundles(self, mock_write, mock_tag, mock_delete):
+        loop = self._create_loop(self.owner_client)
+        content = self._zip_bytes()
+
+        replaced = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]),
+            {"bundles": [self._bundle_payload(content=content)]},
+            format="json",
+        )
+
+        self.assertEqual(replaced.status_code, status.HTTP_200_OK, replaced.content)
+        bundles = replaced.json()["skill_bundles"]
+        self.assertEqual(len(bundles), 1)
+        self.assertEqual(bundles[0]["skill_name"], "my-skill")
+        self.assertEqual(bundles[0]["skill_source"], "user")
+        self.assertEqual(bundles[0]["size"], len(content))
+        self.assertEqual(bundles[0]["content_sha256"], hashlib.sha256(content).hexdigest())
+        mock_write.assert_called_once()
+
+        row = Loop.objects.unscoped().get(id=loop["id"])
+        stored = row.skill_bundles[0]
+        self.assertEqual(stored["type"], "skill_bundle")
+        self.assertTrue(stored["storage_path"].startswith(row.get_skill_bundle_s3_prefix()))
+        self.assertEqual(stored["metadata"]["skill_name"], "my-skill")
+        first_storage_path = stored["storage_path"]
+
+        retrieved = self.owner_client.get(self._loop_url(loop["id"]))
+        self.assertEqual(retrieved.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(retrieved.json()["skill_bundles"]), 1)
+
+        cleared = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": []}, format="json")
+        self.assertEqual(cleared.status_code, status.HTTP_200_OK, cleared.content)
+        self.assertEqual(cleared.json()["skill_bundles"], [])
+        self.assertEqual(Loop.objects.unscoped().get(id=loop["id"]).skill_bundles, [])
+        # Superseded objects are expired via a grace-period tag, not deleted outright —
+        # an in-flight fire may still be copying from them.
+        mock_delete.assert_not_called()
+        mock_tag.assert_any_call(first_storage_path, {"ttl_days": "1", "team_id": str(self.team.id)})
+
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_rejects_a_sha_mismatch(self, mock_write):
+        loop = self._create_loop(self.owner_client)
+        payload = self._bundle_payload()
+        payload["content_sha256"] = "0" * 64
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": [payload]}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        self.assertIn("sha256", response.json()["detail"])
+        mock_write.assert_not_called()
+
+    def test_replace_rejects_too_many_bundles(self):
+        loop = self._create_loop(self.owner_client)
+        bundles = [self._bundle_payload(name=f"skill-{index}") for index in range(11)]
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": bundles}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+
+    @patch("posthog.storage.object_storage.write")
+    def test_a_later_invalid_bundle_prevents_any_write(self, mock_write):
+        loop = self._create_loop(self.owner_client)
+        valid = self._bundle_payload(name="first")
+        invalid = self._bundle_payload(name="second")
+        invalid["content_sha256"] = "0" * 64
+
+        response = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [valid, invalid]}, format="json"
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        mock_write.assert_not_called()
+        self.assertEqual(Loop.objects.unscoped().get(id=loop["id"]).skill_bundles, [])
+
+    @patch("posthog.storage.object_storage.delete_objects")
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.write")
+    def test_a_failed_write_deletes_the_bundles_already_written(self, mock_write, mock_tag, mock_delete):
+        loop = self._create_loop(self.owner_client)
+        mock_write.side_effect = [None, RuntimeError("s3 down")]
+
+        response = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]),
+            {"bundles": [self._bundle_payload(name="first"), self._bundle_payload(name="second")]},
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR, response.content)
+        mock_delete.assert_called_once()
+        deleted_paths = mock_delete.call_args.args[0]
+        self.assertEqual(len(deleted_paths), 1)
+        self.assertIn("first", deleted_paths[0])
+        self.assertEqual(Loop.objects.unscoped().get(id=loop["id"]).skill_bundles, [])
+
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_rejects_a_non_zip_bundle(self, mock_write):
+        loop = self._create_loop(self.owner_client)
+        payload = self._bundle_payload(content=b"not a zip archive")
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": [payload]}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        self.assertIn("not a valid zip archive", response.json()["detail"])
+        mock_write.assert_not_called()
+
+    @patch("products.tasks.backend.facade.loops.MAX_LOOP_SKILL_BUNDLE_UNCOMPRESSED_BYTES", 64)
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_rejects_a_bundle_that_expands_past_the_cap(self, mock_write):
+        loop = self._create_loop(self.owner_client)
+        payload = self._bundle_payload(content=self._zip_bytes({"SKILL.md": "x" * 1024}))
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": [payload]}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        self.assertIn("expands to more than", response.json()["detail"])
+        mock_write.assert_not_called()
+
+    @patch("products.tasks.backend.facade.loops.MAX_LOOP_SKILL_BUNDLE_FILES", 1)
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_rejects_a_bundle_with_too_many_entries(self, mock_write):
+        loop = self._create_loop(self.owner_client)
+        payload = self._bundle_payload(content=self._zip_bytes({"SKILL.md": "body", "extra.md": "more"}))
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": [payload]}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        self.assertIn("contains more than", response.json()["detail"])
+        mock_write.assert_not_called()
+
+    @patch("products.tasks.backend.facade.loops.MAX_LOOP_SKILL_BUNDLES_TOTAL_UNCOMPRESSED_BYTES", 1500)
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_rejects_bundles_that_together_expand_past_the_cap(self, mock_write):
+        loop = self._create_loop(self.owner_client)
+        bundles = [
+            self._bundle_payload(name="first", content=self._zip_bytes({"SKILL.md": "x" * 1024})),
+            self._bundle_payload(name="second", content=self._zip_bytes({"SKILL.md": "y" * 1024})),
+        ]
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": bundles}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        self.assertIn("together expand", response.json()["detail"])
+        mock_write.assert_not_called()
+
+    @patch("products.tasks.backend.facade.loops.MAX_LOOP_SKILL_BUNDLE_CENTRAL_DIR_BYTES", 10)
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_rejects_an_oversized_central_directory(self, mock_write):
+        loop = self._create_loop(self.owner_client)
+
+        response = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        mock_write.assert_not_called()
+
+    @patch("products.tasks.backend.facade.loops.MAX_LOOP_SKILL_BUNDLE_FILES", 1)
+    @patch("posthog.storage.object_storage.write")
+    def test_a_lying_trailer_count_is_caught_after_parse(self, mock_write):
+        # The trailer's entry count is attacker-controlled; forging it low shaves the
+        # fast-fail but the parsed entry list must still trip the cap.
+        content = bytearray(self._zip_bytes({"SKILL.md": "body", "extra.md": "more"}))
+        eocd = content.rfind(b"PK\x05\x06")
+        content[eocd + 8 : eocd + 12] = (1).to_bytes(2, "little") * 2
+        loop = self._create_loop(self.owner_client)
+        payload = self._bundle_payload(content=bytes(content))
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), {"bundles": [payload]}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content)
+        self.assertIn("contains more than", response.json()["detail"])
+        mock_write.assert_not_called()
+
+    def test_replace_requires_a_declared_content_length(self):
+        loop = self._create_loop(self.owner_client)
+
+        response = self.owner_client.put(self._skill_bundles_url(loop["id"]), CONTENT_LENGTH="0")
+
+        self.assertEqual(response.status_code, status.HTTP_411_LENGTH_REQUIRED, response.content)
+
+    @patch("products.tasks.backend.presentation.views.loops.MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES", 10)
+    def test_replace_rejects_an_oversized_request_up_front(self):
+        loop = self._create_loop(self.owner_client)
+
+        response = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, response.content)
+
+    @patch("posthog.storage.object_storage.delete_objects")
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_racing_an_ownership_takeover_is_denied(self, mock_write, mock_tag, mock_delete):
+        # Ownership moves to a teammate while the former owner's replace is mid-upload;
+        # the swap must re-authorize under the lock, discard its uploads and 403 rather
+        # than landing the former owner's skill on the taken-over loop.
+        loop = self._create_loop(self.owner_client, visibility="team")
+
+        def take_ownership_mid_request(*args, **kwargs):
+            Loop.objects.unscoped().filter(id=loop["id"]).update(created_by=self.peer)
+
+        mock_write.side_effect = take_ownership_mid_request
+
+        response = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN, response.content)
+        mock_delete.assert_called_once()
+        discarded_paths = mock_delete.call_args.args[0]
+        self.assertEqual(len(discarded_paths), 1)
+        self.assertEqual(Loop.objects.unscoped().get(id=loop["id"]).skill_bundles, [])
+
+    @patch("posthog.storage.object_storage.delete_objects")
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_racing_a_delete_discards_its_uploads(self, mock_write, mock_tag, mock_delete):
+        # The soft delete commits between this request's fetch and its manifest swap; the
+        # swap must notice the deleted row and discard its own fresh uploads, not
+        # resurrect bundles on a deleted loop.
+        loop = self._create_loop(self.owner_client)
+
+        def soft_delete_mid_request(*args, **kwargs):
+            Loop.objects.unscoped().filter(id=loop["id"]).update(deleted=True)
+
+        mock_write.side_effect = soft_delete_mid_request
+
+        response = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND, response.content)
+        mock_delete.assert_called_once()
+        discarded_paths = mock_delete.call_args.args[0]
+        self.assertEqual(len(discarded_paths), 1)
+        self.assertEqual(Loop.objects.unscoped().get(id=loop["id"]).skill_bundles, [])
+
+    @patch("posthog.storage.object_storage.delete_objects")
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.write")
+    def test_deleting_a_loop_releases_its_bundle_objects(self, mock_write, mock_tag, mock_delete):
+        loop = self._create_loop(self.owner_client)
+        replaced = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+        self.assertEqual(replaced.status_code, status.HTTP_200_OK, replaced.content)
+        stored_path = Loop.objects.unscoped().get(id=loop["id"]).skill_bundles[0]["storage_path"]
+
+        deleted = self.owner_client.delete(self._loop_url(loop["id"]))
+
+        self.assertEqual(deleted.status_code, status.HTTP_204_NO_CONTENT, deleted.content)
+        mock_delete.assert_not_called()
+        mock_tag.assert_any_call(stored_path, {"ttl_days": "1", "team_id": str(self.team.id)})
+        row = Loop.objects.unscoped().get(id=loop["id"])
+        self.assertTrue(row.deleted)
+        self.assertEqual(row.skill_bundles, [])
+
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.write")
+    def test_replace_is_owner_gated_on_a_team_loop(self, mock_write, mock_tag):
+        loop = self._create_loop(self.owner_client, visibility="team")
+
+        denied = self.peer_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+        self.assertEqual(denied.status_code, status.HTTP_403_FORBIDDEN, denied.content)
+
+        allowed = self.owner_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+        self.assertEqual(allowed.status_code, status.HTTP_200_OK, allowed.content)
+
+    def test_replace_on_someone_elses_personal_loop_is_a_404(self):
+        loop = self._create_loop(self.owner_client, visibility="personal")
+
+        response = self.peer_client.put(
+            self._skill_bundles_url(loop["id"]), {"bundles": [self._bundle_payload()]}, format="json"
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND, response.content)
+
+
 class LoopSafetyLimitAPITest(LoopsAPITestCase):
     def test_too_many_triggers_rejected(self):
         triggers = [
diff --git a/products/tasks/backend/tests/test_temporal_client.py b/products/tasks/backend/tests/test_temporal_client.py
index 3f3805d3afec..b840f585dbe0 100644
--- a/products/tasks/backend/tests/test_temporal_client.py
+++ b/products/tasks/backend/tests/test_temporal_client.py
@@ -10,7 +10,7 @@
 from posthog.models import Organization, Team
 from posthog.models.user import User
 
-from products.tasks.backend.models import Task, TaskRun
+from products.tasks.backend.models import Loop, Task, TaskRun
 from products.tasks.backend.temporal.client import (
     execute_task_processing_workflow,
     execute_task_processing_workflow_async,
@@ -328,6 +328,92 @@ def _run_reconcile(self, run: TaskRun, start_workflow: Mock) -> str:
         ):
             return redispatch_orphaned_task_run(str(run.id))
 
+    def _orphaned_loop_run(self, storage_path: str | None = None) -> TaskRun:
+        # A loop fire freezes its bundle set onto the run's state; seeding reads only
+        # this snapshot, never the live loop, and validates every source path against
+        # the owning loop's prefix.
+        # Direct save, like LoopRunsTestCase.create_loop: the scoped manager's create()
+        # fails closed without an ambient team scope.
+        loop = Loop(
+            team=self.team,
+            created_by=self.user,
+            name="Digest",
+            instructions="/my-skill",
+            runtime_adapter="claude",
+        )
+        loop.save()
+        self.task.origin_product = Task.OriginProduct.LOOP
+        self.task.loop = loop
+        self.task.save(update_fields=["origin_product", "loop"])
+        run = self._orphaned_run(pending_dispatch={"create_pr": False, "posthog_mcp_scopes": "read_only"})
+        run.state = {
+            **run.state,
+            "skill_bundle_seeds": [
+                {
+                    "id": "abcdef0123456789abcdef0123456789",
+                    "name": "my-skill.zip",
+                    "type": "skill_bundle",
+                    "source": "posthog_code_skill",
+                    "size": 9,
+                    "content_type": "application/zip",
+                    "storage_path": storage_path or f"{loop.get_skill_bundle_s3_prefix()}/abcdef01_my-skill.zip",
+                    "uploaded_at": "2026-07-23T00:00:00+00:00",
+                    "metadata": {
+                        "skill_name": "my-skill",
+                        "skill_source": "user",
+                        "content_sha256": "a" * 64,
+                        "bundle_format": "zip",
+                        "schema_version": 1,
+                    },
+                }
+            ],
+        }
+        run.save(update_fields=["state"])
+        return run
+
+    @patch("posthog.storage.object_storage.tag")
+    @patch("posthog.storage.object_storage.copy")
+    def test_recovery_seeds_loop_skill_bundles_before_dispatch(self, mock_copy, mock_tag) -> None:
+        # The lost on_commit callback this sweep recovers from is also what seeds skill
+        # bundles; recovery must re-seed or the run silently starts without its skills.
+        run = self._orphaned_loop_run()
+        start_workflow = AsyncMock()
+
+        outcome = self._run_reconcile(run, start_workflow)
+
+        self.assertEqual(outcome, "recovered")
+        mock_copy.assert_called_once()
+        run.refresh_from_db()
+        seeded = [artifact for artifact in run.artifacts if artifact["type"] == "skill_bundle"]
+        self.assertEqual(len(seeded), 1)
+
+    @patch("posthog.storage.object_storage.copy")
+    def test_recovery_rejects_seeds_outside_the_loops_prefix(self, mock_copy) -> None:
+        # Run state is a client-PATCHable surface; a forged seed pointing at another
+        # prefix must never be copied into the run's downloadable artifacts.
+        run = self._orphaned_loop_run(storage_path="tasks/artifacts/team_999/task_x/run_y/secret.zip")
+        start_workflow = AsyncMock()
+
+        outcome = self._run_reconcile(run, start_workflow)
+
+        self.assertEqual(outcome, "error")
+        mock_copy.assert_not_called()
+        start_workflow.assert_not_called()
+
+    @patch("posthog.storage.object_storage.copy", side_effect=RuntimeError("s3 down"))
+    def test_recovery_aborts_when_seeding_fails(self, mock_copy) -> None:
+        run = self._orphaned_loop_run()
+        start_workflow = AsyncMock()
+
+        outcome = self._run_reconcile(run, start_workflow)
+
+        # Transient like any other recovery error: left QUEUED for the next sweep, with
+        # the 24h killer as the terminal backstop — but never dispatched skill-less.
+        self.assertEqual(outcome, "error")
+        start_workflow.assert_not_called()
+        run.refresh_from_db()
+        self.assertEqual(run.status, TaskRun.Status.QUEUED)
+
     def test_recovers_orphaned_run_with_persisted_dispatch_params(self) -> None:
         run = self._orphaned_run(pending_dispatch={"create_pr": False, "posthog_mcp_scopes": "full"})
         start_workflow = AsyncMock()
diff --git a/products/tasks/frontend/generated/api.schemas.ts b/products/tasks/frontend/generated/api.schemas.ts
index 67a2494103dd..4b651c94f305 100644
--- a/products/tasks/frontend/generated/api.schemas.ts
+++ b/products/tasks/frontend/generated/api.schemas.ts
@@ -111,6 +111,15 @@ export interface LoopTriggerDTOApi {
     updated_at: string
 }
 
+export interface LoopSkillBundleDTOApi {
+    id: string
+    skill_name: string
+    skill_source: string
+    size: number
+    content_sha256: string
+    uploaded_at: string
+}
+
 /**
  * Detail/create/update response for a loop, including its triggers.
  */
@@ -156,6 +165,8 @@ export interface LoopDTOApi {
     updated_at: string
     /** Triggers attached to this loop. */
     triggers: LoopTriggerDTOApi[]
+    /** Skill bundles attached to this loop, seeded into every fired run. */
+    skill_bundles: LoopSkillBundleDTOApi[]
 }
 
 export interface PaginatedLoopDTOListApi {
@@ -598,6 +609,72 @@ export interface LoopRunPageApi {
     next_cursor: string | null
 }
 
+/**
+ * * `user` - user
+ * * `repo` - repo
+ * * `marketplace` - marketplace
+ * * `codex` - codex
+ */
+export type SkillSourceEnumApi = (typeof SkillSourceEnumApi)[keyof typeof SkillSourceEnumApi]
+
+export const SkillSourceEnumApi = {
+    User: 'user',
+    Repo: 'repo',
+    Marketplace: 'marketplace',
+    Codex: 'codex',
+} as const
+
+/**
+ * * `zip` - zip
+ */
+export type BundleFormatEnumApi = (typeof BundleFormatEnumApi)[keyof typeof BundleFormatEnumApi]
+
+export const BundleFormatEnumApi = {
+    Zip: 'zip',
+} as const
+
+/**
+ * One zipped local skill in a skill-bundle replace request.
+ */
+export interface LoopSkillBundleUploadApi {
+    /**
+     * File name for the stored bundle, e.g. `my-skill.zip`.
+     * @maxLength 255
+     */
+    file_name: string
+    /**
+     * Name of the skill inside the bundle.
+     * @maxLength 255
+     */
+    skill_name: string
+    /** Local source the bundle was built from, such as user or repo.
+     *
+     * * `user` - user
+     * * `repo` - repo
+     * * `marketplace` - marketplace
+     * * `codex` - codex */
+    skill_source: SkillSourceEnumApi
+    /**
+     * SHA-256 hex digest of the bundle bytes.
+     * @pattern ^[a-f0-9]{64}$
+     */
+    content_sha256: string
+    /** Archive format used for the bundle.
+     *
+     * * `zip` - zip */
+    bundle_format: BundleFormatEnumApi
+    /** Base64-encoded bundle bytes. */
+    content_base64: string
+}
+
+/**
+ * Request body for replacing a loop's attached skill bundles wholesale. Send an empty
+ * list to detach every skill.
+ */
+export interface LoopSkillBundlesWriteApi {
+    bundles: LoopSkillBundleUploadApi[]
+}
+
 /**
  * @nullable
  */
@@ -1090,30 +1167,6 @@ export const TaskRunDetailDTOProviderEnumApi = {
     Openai: 'openai',
 } as const
 
-/**
- * * `user` - user
- * * `repo` - repo
- * * `marketplace` - marketplace
- * * `codex` - codex
- */
-export type SkillSourceEnumApi = (typeof SkillSourceEnumApi)[keyof typeof SkillSourceEnumApi]
-
-export const SkillSourceEnumApi = {
-    User: 'user',
-    Repo: 'repo',
-    Marketplace: 'marketplace',
-    Codex: 'codex',
-} as const
-
-/**
- * * `zip` - zip
- */
-export type BundleFormatEnumApi = (typeof BundleFormatEnumApi)[keyof typeof BundleFormatEnumApi]
-
-export const BundleFormatEnumApi = {
-    Zip: 'zip',
-} as const
-
 export interface TaskRunArtifactMetadataApi {
     /**
      * Name of the local skill included in a skill_bundle artifact.
diff --git a/products/tasks/frontend/generated/api.ts b/products/tasks/frontend/generated/api.ts
index 698e0932b472..9cc59d55304a 100644
--- a/products/tasks/frontend/generated/api.ts
+++ b/products/tasks/frontend/generated/api.ts
@@ -20,6 +20,7 @@ import type {
     LoopPreviewDTOApi,
     LoopPreviewRequestApi,
     LoopRunPageApi,
+    LoopSkillBundlesWriteApi,
     LoopWriteApi,
     LoopsListParams,
     LoopsRunsRetrieveParams,
@@ -323,6 +324,28 @@ export const loopsRunsRetrieve = async (
     })
 }
 
+export const getLoopsSkillBundlesUpdateUrl = (projectId: string, id: string) => {
+    return `/api/projects/${projectId}/loops/${id}/skill_bundles/`
+}
+
+/**
+ * Replaces the loop's attached skill bundles wholesale: zipped local skills whose contents are seeded into every fired run's sandbox. Send an empty list to detach every skill. Owner-only on team loops, like other identity-bearing configuration.
+ * @summary Replace a loop's skill bundles
+ */
+export const loopsSkillBundlesUpdate = async (
+    projectId: string,
+    id: string,
+    loopSkillBundlesWriteApi: LoopSkillBundlesWriteApi,
+    options?: RequestInit
+): Promise<LoopDTOApi> => {
+    return apiMutator<LoopDTOApi>(getLoopsSkillBundlesUpdateUrl(projectId, id), {
+        ...options,
+        method: 'PUT',
+        headers: { 'Content-Type': 'application/json', ...options?.headers },
+        body: JSON.stringify(loopSkillBundlesWriteApi),
+    })
+}
+
 export const getLoopsTriggerCreateUrl = (projectId: string, id: string) => {
     return `/api/projects/${projectId}/loops/${id}/trigger/`
 }
diff --git a/products/tasks/frontend/generated/api.zod.ts b/products/tasks/frontend/generated/api.zod.ts
index 2253fea77f00..580e37f806a4 100644
--- a/products/tasks/frontend/generated/api.zod.ts
+++ b/products/tasks/frontend/generated/api.zod.ts
@@ -641,6 +641,54 @@ export const LoopsPreviewCreateBody = /* @__PURE__ */ zod.object({
         .describe('Sample trigger payload, e.g. a GitHub webhook body or an API trigger body, to render into context.'),
 })
 
+/**
+ * Replaces the loop's attached skill bundles wholesale: zipped local skills whose contents are seeded into every fired run's sandbox. Send an empty list to detach every skill. Owner-only on team loops, like other identity-bearing configuration.
+ * @summary Replace a loop's skill bundles
+ */
+export const loopsSkillBundlesUpdateBodyBundlesItemFileNameMax = 255
+
+export const loopsSkillBundlesUpdateBodyBundlesItemSkillNameMax = 255
+
+export const loopsSkillBundlesUpdateBodyBundlesItemContentSha256RegExp = new RegExp('^[a-f0-9]{64}$')
+
+export const LoopsSkillBundlesUpdateBody = /* @__PURE__ */ zod
+    .object({
+        bundles: zod.array(
+            zod
+                .object({
+                    file_name: zod
+                        .string()
+                        .max(loopsSkillBundlesUpdateBodyBundlesItemFileNameMax)
+                        .describe('File name for the stored bundle, e.g. `my-skill.zip`.'),
+                    skill_name: zod
+                        .string()
+                        .max(loopsSkillBundlesUpdateBodyBundlesItemSkillNameMax)
+                        .describe('Name of the skill inside the bundle.'),
+                    skill_source: zod
+                        .enum(['user', 'repo', 'marketplace', 'codex'])
+                        .describe(
+                            '\* `user` - user\n\* `repo` - repo\n\* `marketplace` - marketplace\n\* `codex` - codex'
+                        )
+                        .describe(
+                            'Local source the bundle was built from, such as user or repo.\n\n\* `user` - user\n\* `repo` - repo\n\* `marketplace` - marketplace\n\* `codex` - codex'
+                        ),
+                    content_sha256: zod
+                        .string()
+                        .regex(loopsSkillBundlesUpdateBodyBundlesItemContentSha256RegExp)
+                        .describe('SHA-256 hex digest of the bundle bytes.'),
+                    bundle_format: zod
+                        .enum(['zip'])
+                        .describe('\* `zip` - zip')
+                        .describe('Archive format used for the bundle.\n\n\* `zip` - zip'),
+                    content_base64: zod.string().describe('Base64-encoded bundle bytes.'),
+                })
+                .describe('One zipped local skill in a skill-bundle replace request.')
+        ),
+    })
+    .describe(
+        "Request body for replacing a loop's attached skill bundles wholesale. Send an empty\nlist to detach every skill."
+    )
+
 /**
  * Authenticated POST trigger for `type=api` triggers. Project secret API key auth (`loop:write` scope), project-wide. Request body (JSON, capped at 64 KB) becomes run context. Send an `Idempotency-Key` header to dedupe retries.
  * @summary Fire a loop externally
diff --git a/products/tasks/mcp/tools.yaml b/products/tasks/mcp/tools.yaml
index 8a6c5a85082f..31ac82fe78a8 100644
--- a/products/tasks/mcp/tools.yaml
+++ b/products/tasks/mcp/tools.yaml
@@ -148,6 +148,9 @@ tools:
             error message and output (including any PR URL), so an agent can check whether a loop's runs succeeded and
             what they produced.
         feature_flag: loops
+    loops-skill-bundles-update:
+        operation: loops_skill_bundles_update
+        enabled: false
     loops-trigger-create:
         operation: loops_trigger_create
         enabled: false
diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts
index 5f15b5dc8572..5414a333297a 100644
--- a/services/mcp/src/api/generated.ts
+++ b/services/mcp/src/api/generated.ts
@@ -37697,6 +37697,15 @@ export namespace Schemas {
       updated_at: string;
     }
 
+    export interface LoopSkillBundleDTO {
+      id: string;
+      skill_name: string;
+      skill_source: string;
+      size: number;
+      content_sha256: string;
+      uploaded_at: string;
+    }
+
     /**
      * Detail/create/update response for a loop, including its triggers.
      */
@@ -37742,6 +37751,8 @@ export namespace Schemas {
       updated_at: string;
       /** Triggers attached to this loop. */
       triggers: LoopTriggerDTO[];
+      /** Skill bundles attached to this loop, seeded into every fired run. */
+      skill_bundles: LoopSkillBundleDTO[];
     }
 
     /**
@@ -37899,6 +37910,64 @@ export namespace Schemas {
       next_cursor: string | null;
     }
 
+    /**
+     * * `user` - user
+     * * `repo` - repo
+     * * `marketplace` - marketplace
+     * * `codex` - codex
+     */
+    export type SkillSourceEnum = typeof SkillSourceEnum[keyof typeof SkillSourceEnum];
+
+
+    export const SkillSourceEnum = {
+      User: 'user',
+      Repo: 'repo',
+      Marketplace: 'marketplace',
+      Codex: 'codex',
+    } as const;
+
+    /**
+     * One zipped local skill in a skill-bundle replace request.
+     */
+    export interface LoopSkillBundleUpload {
+      /**
+         * File name for the stored bundle, e.g. `my-skill.zip`.
+         * @maxLength 255
+         */
+      file_name: string;
+      /**
+         * Name of the skill inside the bundle.
+         * @maxLength 255
+         */
+      skill_name: string;
+      /** Local source the bundle was built from, such as user or repo.
+       *
+       * * `user` - user
+       * * `repo` - repo
+       * * `marketplace` - marketplace
+       * * `codex` - codex */
+      skill_source: SkillSourceEnum;
+      /**
+         * SHA-256 hex digest of the bundle bytes.
+         * @pattern ^[a-f0-9]{64}$
+         */
+      content_sha256: string;
+      /** Archive format used for the bundle.
+       *
+       * * `zip` - zip */
+      bundle_format: BundleFormatEnum;
+      /** Base64-encoded bundle bytes. */
+      content_base64: string;
+    }
+
+    /**
+     * Request body for replacing a loop's attached skill bundles wholesale. Send an empty
+     * list to detach every skill.
+     */
+    export interface LoopSkillBundlesWrite {
+      bundles: LoopSkillBundleUpload[];
+    }
+
     export interface LoopTriggerWrite {
       /** Existing trigger id to update in place. Omit to create a new trigger. */
       id?: string;
@@ -44501,22 +44570,6 @@ export namespace Schemas {
       Openai: 'openai',
     } as const;
 
-    /**
-     * * `user` - user
-     * * `repo` - repo
-     * * `marketplace` - marketplace
-     * * `codex` - codex
-     */
-    export type SkillSourceEnum = typeof SkillSourceEnum[keyof typeof SkillSourceEnum];
-
-
-    export const SkillSourceEnum = {
-      User: 'user',
-      Repo: 'repo',
-      Marketplace: 'marketplace',
-      Codex: 'codex',
-    } as const;
-
     export interface TaskRunArtifactMetadata {
       /**
          * Name of the local skill included in a skill_bundle artifact.