feat(loops): Run a skill instead of instructions

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",

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,

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),
+        ),
+    ]

products/tasks/backend/migrations/max_migration.txt

@@ -1 +1 @@
-0071_alter_origin_product_mcp_analytics
+0072_loop_skill_bundles

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

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."""
 

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.",