From 6f4f1e26a968f7f63b24f20ef8819ae96e81d4fc Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Wed, 22 Jul 2026 23:26:23 -0700 Subject: [PATCH 01/17] add loop skill bundles seeded into fired runs --- products/tasks/backend/facade/loops.py | 138 ++++++++++++++++++ .../tasks/backend/logic/services/loop_runs.py | 33 +++++ .../migrations/0072_loop_skill_bundles.py | 15 ++ .../backend/migrations/max_migration.txt | 2 +- products/tasks/backend/models.py | 9 ++ .../backend/presentation/serializers_loops.py | 41 ++++++ .../tasks/backend/presentation/views/loops.py | 35 ++++- .../tasks/backend/tests/test_loop_runs.py | 63 ++++++++ .../tasks/backend/tests/test_loops_api.py | 100 +++++++++++++ 9 files changed, 434 insertions(+), 2 deletions(-) create mode 100644 products/tasks/backend/migrations/0072_loop_skill_bundles.py diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index 29381f53e607..f5ca05b450ac 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 @@ -955,6 +990,105 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_ return _loop_to_dto(loop) +MAX_LOOP_SKILL_BUNDLES = 10 +MAX_LOOP_SKILL_BUNDLE_SIZE_BYTES = 30 * 1024 * 1024 + + +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); superseded objects are deleted best-effort after the manifest swap. + Owner-gated like every other identity-bearing field. Returns `None` if the loop is + not found/reachable.""" + import hashlib # 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}) + + prefix = loop.get_skill_bundle_s3_prefix() + previous_paths = [ + entry["storage_path"] + for entry in (loop.skill_bundles or []) + if isinstance(entry, dict) and entry.get("storage_path") + ] + + entries: list[dict] = [] + 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.") + + 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"}) + 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": "zip", + "schema_version": 1, + }, + } + ) + + with transaction.atomic(): + loop.skill_bundles = entries + loop.save(update_fields=["skill_bundles", "updated_at"]) + + stale_paths = [path for path in previous_paths if path.startswith(prefix)] + if stale_paths: + try: + object_storage.delete_objects(stale_paths) + except Exception as exc: + logger.warning( + "loop.skill_bundle_cleanup_failed", + extra={"loop_id": str(loop.id), "paths": stale_paths, "error": str(exc)}, + ) + + 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: @@ -1242,6 +1376,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 +1403,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..93f43f6e801b 100644 --- a/products/tasks/backend/logic/services/loop_runs.py +++ b/products/tasks/backend/logic/services/loop_runs.py @@ -490,6 +490,38 @@ def _team_rate_capped(loop: Loop) -> bool: return fire_count >= LOOP_TEAM_RATE_CAP_PER_DAY +def _seed_skill_bundle_artifacts(loop: Loop, task_run: TaskRun) -> None: + """Copy the loop's stored skill bundles into the new run: 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 — a fire whose skills can't be seeded rolls back and retries + cleanly rather than running with silently missing skills.""" + bundles = [entry for entry in (loop.skill_bundles 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 + + run_prefix = task_run.get_artifact_s3_prefix() + manifest = list(task_run.artifacts or []) + for bundle in bundles: + entry = dict(bundle) + target_path = f"{run_prefix}/{str(entry['id'])[:8]}_{entry['name']}" + object_storage.copy(entry["storage_path"], 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"]) + + 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 @@ -572,6 +604,7 @@ def _create_loop_task_and_run(loop: Loop, trigger: LoopTrigger | None, trigger_c "workflow_id_prefix": None, } task_run = task.create_run(mode="background", extra_state=extra_state) + _seed_skill_bundle_artifacts(loop, task_run) team_id = loop.team_id user_id = loop.created_by_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..b00837fde535 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,11 @@ class Meta: def __str__(self): return self.name + def get_skill_bundle_s3_prefix(self) -> str: + """Base prefix for this loop's skill bundle objects in S3.""" + tasks_folder = settings.OBJECT_STORAGE_TASKS_FOLDER + return f"{tasks_folder}/artifacts/team_{self.team_id}/loop_{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..3dec2a7cc870 100644 --- a/products/tasks/backend/presentation/views/loops.py +++ b/products/tasks/backend/presentation/views/loops.py @@ -29,6 +29,7 @@ LoopRunPageSerializer, LoopRunsQuerySerializer, LoopSerializer, + LoopSkillBundlesWriteSerializer, LoopWriteSerializer, ) @@ -142,7 +143,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 +279,36 @@ 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"), + }, + ) + @action(detail=True, methods=["put"], url_path="skill_bundles", required_scopes=["loop:write"]) + def skill_bundles(self, request, pk=None, **kwargs): + 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/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py index 1af9a64ca22c..7a71a7269b1a 100644 --- a/products/tasks/backend/tests/test_loop_runs.py +++ b/products/tasks/backend/tests/test_loop_runs.py @@ -537,6 +537,69 @@ 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, **overrides) -> dict: + entry = { + "id": "abcdef0123456789abcdef0123456789", + "name": "my-skill.zip", + "type": "skill_bundle", + "source": "posthog_code_skill", + "size": 9, + "content_type": "application/zip", + "storage_path": "tasks/artifacts/team_1/loop_x/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 + + @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): + entry = self._bundle_entry() + loop = self.create_loop(skill_bundles=[entry]) + trigger = self.create_trigger(loop) + + result = fire_loop(loop, trigger, "fire-1", "ctx") + + self.assertTrue(result.created) + 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"]) + 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 = fire_loop(loop, trigger, "fire-1", "ctx") + + self.assertTrue(result.created) + mock_copy.assert_not_called() + + @patch("posthog.storage.object_storage.copy", side_effect=RuntimeError("s3 down")) + def test_fire_fails_when_a_bundle_cannot_be_seeded(self, mock_copy): + loop = self.create_loop(skill_bundles=[self._bundle_entry()]) + trigger = self.create_trigger(loop) + + with self.assertRaises(RuntimeError): + fire_loop(loop, trigger, "fire-1", "ctx") + + 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..e8823630c24f 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -1,3 +1,5 @@ +import base64 +import hashlib from contextlib import nullcontext from datetime import timedelta from uuid import UUID @@ -231,6 +233,104 @@ 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 _bundle_payload(self, name: str = "my-skill", content: bytes = b"zip-bytes") -> dict: + return { + "file_name": f"{name}.zip", + "skill_name": name, + "skill_source": "user", + "content_sha256": hashlib.sha256(content).hexdigest(), + "bundle_format": "zip", + "content_base64": base64.b64encode(content).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 = b"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, []) + mock_delete.assert_called_once_with([first_storage_path]) + + @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.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 = [ From 671a3d6016d296bd5f69b02cde6765c7b43a9cc8 Mon Sep 17 00:00:00 2001 From: "tests-posthog[bot]" <250237707+tests-posthog[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 06:31:13 +0000 Subject: [PATCH 02/17] chore: update OpenAPI generated types --- .../tasks/frontend/generated/api.schemas.ts | 101 +++++++++++++----- products/tasks/frontend/generated/api.ts | 23 ++++ products/tasks/frontend/generated/api.zod.ts | 48 +++++++++ products/tasks/mcp/tools.yaml | 3 + services/mcp/src/api/generated.ts | 85 ++++++++++++--- 5 files changed, 220 insertions(+), 40 deletions(-) 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 => { + return apiMutator(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. From 09ea81adacb60b68ef11987f454ce5857be096ce Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Wed, 22 Jul 2026 23:54:47 -0700 Subject: [PATCH 03/17] harden skill bundle storage lifecycle --- products/tasks/backend/facade/loops.py | 140 +++++++++++------- .../tasks/backend/logic/services/loop_runs.py | 50 ++++--- .../tasks/backend/presentation/views/loops.py | 3 + .../tasks/backend/tests/test_loop_runs.py | 25 ++++ .../tasks/backend/tests/test_loops_api.py | 54 +++++++ 5 files changed, 203 insertions(+), 69 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index f5ca05b450ac..da7c32313e31 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -994,15 +994,35 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_ MAX_LOOP_SKILL_BUNDLE_SIZE_BYTES = 30 * 1024 * 1024 +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.""" + 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 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); superseded objects are deleted best-effort after the manifest swap. - Owner-gated like every other identity-bearing field. Returns `None` if the loop is - not found/reachable.""" + 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 hashlib # noqa: PLC0415 from django.utils import timezone as django_timezone # noqa: PLC0415 @@ -1019,14 +1039,7 @@ def replace_loop_skill_bundles( return None _authorize_update(loop, user, {"skill_bundles": bundles}) - prefix = loop.get_skill_bundle_s3_prefix() - previous_paths = [ - entry["storage_path"] - for entry in (loop.skill_bundles or []) - if isinstance(entry, dict) and entry.get("storage_path") - ] - - entries: list[dict] = [] + decoded: list[tuple[dict, bytes]] = [] for bundle in bundles: try: content_bytes = base64.b64decode(bundle["content_base64"], validate=True) @@ -1039,51 +1052,65 @@ def replace_loop_skill_bundles( ) if hashlib.sha256(content_bytes).hexdigest() != bundle["content_sha256"]: raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' does not match its declared sha256.") + decoded.append((bundle, content_bytes)) - 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"}) - 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)}, + 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": "zip", + "schema_version": 1, + }, + } ) - 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": "zip", - "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. with transaction.atomic(): - loop.skill_bundles = entries - loop.save(update_fields=["skill_bundles", "updated_at"]) + locked = Loop.objects.unscoped().select_for_update().get(pk=loop.pk) + 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"]) - stale_paths = [path for path in previous_paths if path.startswith(prefix)] - if stale_paths: - try: - object_storage.delete_objects(stale_paths) - except Exception as exc: - logger.warning( - "loop.skill_bundle_cleanup_failed", - extra={"loop_id": str(loop.id), "paths": stale_paths, "error": str(exc)}, - ) + new_paths = {entry["storage_path"] for entry in entries} + _delete_skill_bundle_objects( + loop.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) @@ -1103,8 +1130,17 @@ 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.") + # 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. + bundle_paths = [ + entry["storage_path"] + for entry in (loop.skill_bundles or []) + if isinstance(entry, dict) and entry.get("storage_path") + ] loop.deleted = True - loop.save(update_fields=["deleted", "updated_at"]) + loop.skill_bundles = [] + loop.save(update_fields=["deleted", "skill_bundles", "updated_at"]) + _delete_skill_bundle_objects(loop.id, bundle_paths) loop_service.delete_loop_schedules(loop) return True diff --git a/products/tasks/backend/logic/services/loop_runs.py b/products/tasks/backend/logic/services/loop_runs.py index 93f43f6e801b..88e97045f1df 100644 --- a/products/tasks/backend/logic/services/loop_runs.py +++ b/products/tasks/backend/logic/services/loop_runs.py @@ -495,7 +495,10 @@ def _seed_skill_bundle_artifacts(loop: Loop, task_run: TaskRun) -> None: 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 — a fire whose skills can't be seeded rolls back and retries - cleanly rather than running with silently missing skills.""" + cleanly rather than running with silently missing skills. The database rollback + can't reclaim already-copied objects (and a retried fire mints a new run id, so + nothing would ever reference them), so those are deleted best-effort before + re-raising.""" bundles = [entry for entry in (loop.skill_bundles or []) if isinstance(entry, dict) and entry.get("storage_path")] if not bundles: return @@ -504,22 +507,35 @@ def _seed_skill_bundle_artifacts(loop: Loop, task_run: TaskRun) -> None: run_prefix = task_run.get_artifact_s3_prefix() manifest = list(task_run.artifacts or []) - for bundle in bundles: - entry = dict(bundle) - target_path = f"{run_prefix}/{str(entry['id'])[:8]}_{entry['name']}" - object_storage.copy(entry["storage_path"], 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"]) + copied_paths: list[str] = [] + try: + for bundle in bundles: + entry = dict(bundle) + target_path = f"{run_prefix}/{str(entry['id'])[:8]}_{entry['name']}" + object_storage.copy(entry["storage_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]: diff --git a/products/tasks/backend/presentation/views/loops.py b/products/tasks/backend/presentation/views/loops.py index 3dec2a7cc870..f01b21720315 100644 --- a/products/tasks/backend/presentation/views/loops.py +++ b/products/tasks/backend/presentation/views/loops.py @@ -295,6 +295,9 @@ def destroy(self, request, pk=None, **kwargs): ) @action(detail=True, methods=["put"], url_path="skill_bundles", required_scopes=["loop:write"]) def skill_bundles(self, request, pk=None, **kwargs): + # Request parsing is bounded before this runs: Django's DATA_UPLOAD_MAX_MEMORY_SIZE + # (20MB, posthog/settings/web.py) rejects larger bodies, so oversized base64 payloads + # never reach the serializer. The facade's per-bundle decoded-size cap is the backstop. serializer = LoopSkillBundlesWriteSerializer(data=request.data) serializer.is_valid(raise_exception=True) try: diff --git a/products/tasks/backend/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py index 7a71a7269b1a..1855b9a27897 100644 --- a/products/tasks/backend/tests/test_loop_runs.py +++ b/products/tasks/backend/tests/test_loop_runs.py @@ -599,6 +599,31 @@ def test_fire_fails_when_a_bundle_cannot_be_seeded(self, mock_copy): 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): + # The rollback reclaims the run row but not S3, and a retried 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")] + first = self._bundle_entry() + second = self._bundle_entry( + id="fedcba9876543210fedcba9876543210", + name="other-skill.zip", + storage_path="tasks/artifacts/team_1/loop_x/fedcba98_other-skill.zip", + ) + loop = self.create_loop(skill_bundles=[first, second]) + trigger = self.create_trigger(loop) + + with self.assertRaises(RuntimeError): + fire_loop(loop, trigger, "fire-1", "ctx") + + 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]) + self.assertEqual(self.active_run_count(loop), 0) + class TestFireLoopContextTarget(LoopRunsTestCase): FOLDER_ID = "11111111-1111-1111-1111-111111111111" diff --git a/products/tasks/backend/tests/test_loops_api.py b/products/tasks/backend/tests/test_loops_api.py index e8823630c24f..bfe3c0a8e2f3 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -306,6 +306,60 @@ def test_replace_rejects_too_many_bundles(self): 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.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_called_once_with([stored_path]) + 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): From 6fff546da01f9f11c8ac0c26c8b8b6808a6b6e0d Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 00:30:32 -0700 Subject: [PATCH 04/17] serialize bundle writes with delete, cap requests --- products/tasks/backend/facade/loops.py | 46 +++++++++++++------ .../tasks/backend/presentation/views/loops.py | 17 +++++-- .../tasks/backend/tests/test_loop_runs.py | 1 + .../tasks/backend/tests/test_loops_api.py | 34 ++++++++++++++ 4 files changed, 81 insertions(+), 17 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index da7c32313e31..e48c45a506a6 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -1097,15 +1097,27 @@ def replace_loop_skill_bundles( # 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 with transaction.atomic(): locked = Loop.objects.unscoped().select_for_update().get(pk=loop.pk) - 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"]) + if locked.deleted: + lost_delete_race = True + 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"]) + + if lost_delete_race: + _delete_skill_bundle_objects(loop.id, written_paths) + return None new_paths = {entry["storage_path"] for entry in entries} _delete_skill_bundle_objects( @@ -1132,14 +1144,20 @@ def soft_delete_loop(loop_id: str | UUID, team_id: int, user: User | None) -> bo # 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. - bundle_paths = [ - entry["storage_path"] - for entry in (loop.skill_bundles or []) - if isinstance(entry, dict) and entry.get("storage_path") - ] + # 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"]) loop.deleted = True - loop.skill_bundles = [] - loop.save(update_fields=["deleted", "skill_bundles", "updated_at"]) _delete_skill_bundle_objects(loop.id, bundle_paths) loop_service.delete_loop_schedules(loop) return True diff --git a/products/tasks/backend/presentation/views/loops.py b/products/tasks/backend/presentation/views/loops.py index f01b21720315..cb2185e0cfdf 100644 --- a/products/tasks/backend/presentation/views/loops.py +++ b/products/tasks/backend/presentation/views/loops.py @@ -34,6 +34,10 @@ ) MAX_LOOP_TRIGGER_PAYLOAD_BYTES = 64 * 1024 +# Whole-request ceiling for the skill_bundles replace. Sits above Django's 20MB +# DATA_UPLOAD_MAX_MEMORY_SIZE so the effective bound stays Django's; this keeps the +# endpoint's own limit explicit and testable if the global setting ever moves. +MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES = 64 * 1024 * 1024 def _loop_limit_response(exc: "loops_facade.LoopLimitError") -> Response: @@ -291,13 +295,20 @@ def destroy(self, request, pk=None, **kwargs): 200: LoopSerializer, 403: OpenApiResponse(description="Not permitted to change this loop's skills"), 404: OpenApiResponse(description="Loop not found"), + 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): - # Request parsing is bounded before this runs: Django's DATA_UPLOAD_MAX_MEMORY_SIZE - # (20MB, posthog/settings/web.py) rejects larger bodies, so oversized base64 payloads - # never reach the serializer. The facade's per-bundle decoded-size cap is the backstop. + # Reject oversized requests from the declared Content-Length, before request.data + # parses (and retains) the body. Django's DATA_UPLOAD_MAX_MEMORY_SIZE (20MB, + # posthog/settings/web.py) bounds the body independently; this explicit gate keeps + # the endpoint's own ceiling visible and returns a structured 413 either way. + if _content_length(request) > 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: diff --git a/products/tasks/backend/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py index 1855b9a27897..623210b30a67 100644 --- a/products/tasks/backend/tests/test_loop_runs.py +++ b/products/tasks/backend/tests/test_loop_runs.py @@ -569,6 +569,7 @@ def test_fire_copies_loop_skill_bundles_into_the_run_manifest(self, mock_copy, m result = fire_loop(loop, trigger, "fire-1", "ctx") 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) diff --git a/products/tasks/backend/tests/test_loops_api.py b/products/tasks/backend/tests/test_loops_api.py index bfe3c0a8e2f3..b7b0c9740c01 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -341,6 +341,40 @@ def test_a_failed_write_deletes_the_bundles_already_written(self, mock_write, mo self.assertIn("first", deleted_paths[0]) self.assertEqual(Loop.objects.unscoped().get(id=loop["id"]).skill_bundles, []) + @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_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") From f0dd93bacfd6cf268e1b18239dff01c96ae49ff3 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 01:22:33 -0700 Subject: [PATCH 05/17] seed skill bundles outside the fire lock --- products/tasks/backend/facade/loops.py | 6 +- .../tasks/backend/logic/services/loop_runs.py | 55 ++++++++++++++++--- .../tasks/backend/presentation/views/loops.py | 18 ++++-- .../tasks/backend/tests/test_loop_runs.py | 54 +++++++++++++++--- .../tasks/backend/tests/test_loops_api.py | 7 +++ 5 files changed, 118 insertions(+), 22 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index e48c45a506a6..523a0e5c57b5 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -991,7 +991,11 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_ MAX_LOOP_SKILL_BUNDLES = 10 -MAX_LOOP_SKILL_BUNDLE_SIZE_BYTES = 30 * 1024 * 1024 +# Decoded per-bundle ceiling. The request is JSON with base64 content and Django's +# DATA_UPLOAD_MAX_MEMORY_SIZE (20MB) bounds the raw body first, 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 def _delete_skill_bundle_objects(loop_id: UUID, paths: list[str]) -> None: diff --git a/products/tasks/backend/logic/services/loop_runs.py b/products/tasks/backend/logic/services/loop_runs.py index 88e97045f1df..323ee5661388 100644 --- a/products/tasks/backend/logic/services/loop_runs.py +++ b/products/tasks/backend/logic/services/loop_runs.py @@ -490,15 +490,55 @@ def _team_rate_capped(loop: Loop) -> bool: return fire_count >= LOOP_TEAM_RATE_CAP_PER_DAY +def _seed_skill_bundles_and_dispatch( + *, + loop: Loop, + 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 loop's 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(loop, task_run) + except Exception as exc: + logger.exception( + "loop_run.skill_bundle_seed_failed", + extra={"loop_id": str(loop.id), "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 _seed_skill_bundle_artifacts(loop: Loop, task_run: TaskRun) -> None: """Copy the loop's stored skill bundles into the new run: 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 — a fire whose skills can't be seeded rolls back and retries - cleanly rather than running with silently missing skills. The database rollback - can't reclaim already-copied objects (and a retried fire mints a new run id, so - nothing would ever reference them), so those are deleted best-effort before - re-raising.""" + Raises on a failed copy; the caller compensates. Already-copied objects are deleted + best-effort before re-raising, since nothing would ever reference them.""" bundles = [entry for entry in (loop.skill_bundles or []) if isinstance(entry, dict) and entry.get("storage_path")] if not bundles: return @@ -620,7 +660,6 @@ def _create_loop_task_and_run(loop: Loop, trigger: LoopTrigger | None, trigger_c "workflow_id_prefix": None, } task_run = task.create_run(mode="background", extra_state=extra_state) - _seed_skill_bundle_artifacts(loop, task_run) team_id = loop.team_id user_id = loop.created_by_id @@ -628,7 +667,9 @@ 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( + loop=loop, + task_run=task_run, team_id=team_id, user_id=user_id, task_id=task_id, diff --git a/products/tasks/backend/presentation/views/loops.py b/products/tasks/backend/presentation/views/loops.py index cb2185e0cfdf..7257ab1fde5e 100644 --- a/products/tasks/backend/presentation/views/loops.py +++ b/products/tasks/backend/presentation/views/loops.py @@ -295,16 +295,24 @@ def destroy(self, request, pk=None, **kwargs): 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): - # Reject oversized requests from the declared Content-Length, before request.data - # parses (and retains) the body. Django's DATA_UPLOAD_MAX_MEMORY_SIZE (20MB, - # posthog/settings/web.py) bounds the body independently; this explicit gate keeps - # the endpoint's own ceiling visible and returns a structured 413 either way. - if _content_length(request) > MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES: + # 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. Django's DATA_UPLOAD_MAX_MEMORY_SIZE (20MB, posthog/settings/web.py) + # bounds the body independently; this keeps the endpoint's ceiling explicit and + # the responses structured. + 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, diff --git a/products/tasks/backend/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py index 623210b30a67..458eee90a7fe 100644 --- a/products/tasks/backend/tests/test_loop_runs.py +++ b/products/tasks/backend/tests/test_loop_runs.py @@ -559,6 +559,14 @@ def _bundle_entry(self, **overrides) -> dict: entry.update(overrides) return entry + 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): @@ -566,7 +574,7 @@ def test_fire_copies_loop_skill_bundles_into_the_run_manifest(self, mock_copy, m loop = self.create_loop(skill_bundles=[entry]) trigger = self.create_trigger(loop) - result = fire_loop(loop, trigger, "fire-1", "ctx") + result, mock_dispatch = self.fire_with_post_commit(loop, trigger) self.assertTrue(result.created) assert result.task_run_id is not None @@ -577,6 +585,7 @@ def test_fire_copies_loop_skill_bundles_into_the_run_manifest(self, mock_copy, m 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() loop.refresh_from_db() self.assertEqual(loop.skill_bundles[0]["storage_path"], entry["storage_path"]) @@ -585,27 +594,53 @@ def test_fire_without_bundles_does_not_touch_storage(self): trigger = self.create_trigger(loop) with patch("posthog.storage.object_storage.copy") as mock_copy: - result = fire_loop(loop, trigger, "fire-1", "ctx") + 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 = self.create_loop(skill_bundles=[self._bundle_entry()]) + 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_fire_fails_when_a_bundle_cannot_be_seeded(self, mock_copy): + def test_a_failed_seed_terminalizes_the_run_and_skips_dispatch(self, mock_copy, mock_delete): loop = self.create_loop(skill_bundles=[self._bundle_entry()]) trigger = self.create_trigger(loop) - with self.assertRaises(RuntimeError): - fire_loop(loop, trigger, "fire-1", "ctx") + 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) + 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): - # The rollback reclaims the run row but not S3, and a retried fire mints a new run - # id — without cleanup every failed attempt would strand another set of objects. + # 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")] first = self._bundle_entry() second = self._bundle_entry( @@ -616,13 +651,14 @@ def test_a_partially_seeded_fire_deletes_the_copies_it_made(self, mock_copy, moc loop = self.create_loop(skill_bundles=[first, second]) trigger = self.create_trigger(loop) - with self.assertRaises(RuntimeError): - fire_loop(loop, trigger, "fire-1", "ctx") + 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) diff --git a/products/tasks/backend/tests/test_loops_api.py b/products/tasks/backend/tests/test_loops_api.py index b7b0c9740c01..983f74557733 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -341,6 +341,13 @@ def test_a_failed_write_deletes_the_bundles_already_written(self, mock_write, mo self.assertIn("first", deleted_paths[0]) self.assertEqual(Loop.objects.unscoped().get(id=loop["id"]).skill_bundles, []) + 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) From 9830f1e6cb4b276c2b1c2d6abb2e35c8f456c0bb Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 02:05:50 -0700 Subject: [PATCH 06/17] narrow error message type in seed failure test --- products/tasks/backend/tests/test_loop_runs.py | 1 + 1 file changed, 1 insertion(+) diff --git a/products/tasks/backend/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py index 458eee90a7fe..65e811e9bea4 100644 --- a/products/tasks/backend/tests/test_loop_runs.py +++ b/products/tasks/backend/tests/test_loop_runs.py @@ -630,6 +630,7 @@ def test_a_failed_seed_terminalizes_the_run_and_skips_dispatch(self, mock_copy, 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) From fab041d17ca90e5b7d960e6c80ad210d39deab46 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 21:53:47 -0700 Subject: [PATCH 07/17] validate zip expansion and cap request parse size --- products/tasks/backend/facade/loops.py | 31 +++++++++-- .../tasks/backend/presentation/views/loops.py | 15 +++--- .../tasks/backend/tests/test_loops_api.py | 53 +++++++++++++++++-- 3 files changed, 84 insertions(+), 15 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index 523a0e5c57b5..532183338a3f 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -991,11 +991,16 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_ MAX_LOOP_SKILL_BUNDLES = 10 -# Decoded per-bundle ceiling. The request is JSON with base64 content and Django's -# DATA_UPLOAD_MAX_MEMORY_SIZE (20MB) bounds the raw body first, 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. +# 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 def _delete_skill_bundle_objects(loop_id: UUID, paths: list[str]) -> None: @@ -1027,7 +1032,9 @@ def replace_loop_skill_bundles( 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 @@ -1056,6 +1063,22 @@ def replace_loop_skill_bundles( ) if hashlib.sha256(content_bytes).hexdigest() != bundle["content_sha256"]: raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' does not match its declared sha256.") + # Central-directory metadata only — nothing is decompressed here. + try: + with zipfile.ZipFile(io.BytesIO(content_bytes)) as archive: + entries = archive.infolist() + except zipfile.BadZipFile: + raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' is not a valid zip archive.") + if len(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 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." + ) decoded.append((bundle, content_bytes)) prefix = loop.get_skill_bundle_s3_prefix() diff --git a/products/tasks/backend/presentation/views/loops.py b/products/tasks/backend/presentation/views/loops.py index 7257ab1fde5e..9d230a320d67 100644 --- a/products/tasks/backend/presentation/views/loops.py +++ b/products/tasks/backend/presentation/views/loops.py @@ -34,10 +34,12 @@ ) MAX_LOOP_TRIGGER_PAYLOAD_BYTES = 64 * 1024 -# Whole-request ceiling for the skill_bundles replace. Sits above Django's 20MB -# DATA_UPLOAD_MAX_MEMORY_SIZE so the effective bound stays Django's; this keeps the -# endpoint's own limit explicit and testable if the global setting ever moves. -MAX_LOOP_SKILL_BUNDLE_REQUEST_BYTES = 64 * 1024 * 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: @@ -303,9 +305,8 @@ def destroy(self, request, pk=None, **kwargs): 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. Django's DATA_UPLOAD_MAX_MEMORY_SIZE (20MB, posthog/settings/web.py) - # bounds the body independently; this keeps the endpoint's ceiling explicit and - # the responses structured. + # 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( diff --git a/products/tasks/backend/tests/test_loops_api.py b/products/tasks/backend/tests/test_loops_api.py index 983f74557733..a8e4be363869 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -1,5 +1,7 @@ +import io import base64 import hashlib +import zipfile from contextlib import nullcontext from datetime import timedelta from uuid import UUID @@ -237,14 +239,22 @@ class LoopSkillBundlesAPITest(LoopsAPITestCase): def _skill_bundles_url(self, loop_id: str) -> str: return f"{self._loop_url(loop_id)}skill_bundles/" - def _bundle_payload(self, name: str = "my-skill", content: bytes = b"zip-bytes") -> dict: + 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).hexdigest(), + "content_sha256": hashlib.sha256(content_bytes).hexdigest(), "bundle_format": "zip", - "content_base64": base64.b64encode(content).decode("ascii"), + "content_base64": base64.b64encode(content_bytes).decode("ascii"), } @patch("posthog.storage.object_storage.delete_objects") @@ -252,7 +262,7 @@ def _bundle_payload(self, name: str = "my-skill", content: bytes = b"zip-bytes") @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 = b"zip-bytes" + content = self._zip_bytes() replaced = self.owner_client.put( self._skill_bundles_url(loop["id"]), @@ -341,6 +351,41 @@ def test_a_failed_write_deletes_the_bundles_already_written(self, mock_write, mo 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() + def test_replace_requires_a_declared_content_length(self): loop = self._create_loop(self.owner_client) From ac2d02e591ce6295e607e5fd23dc8965fd1251ca Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 22:05:23 -0700 Subject: [PATCH 08/17] rename zip entry list to avoid manifest collision --- products/tasks/backend/facade/loops.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index 532183338a3f..293186dfece1 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -1066,14 +1066,14 @@ def replace_loop_skill_bundles( # Central-directory metadata only — nothing is decompressed here. try: with zipfile.ZipFile(io.BytesIO(content_bytes)) as archive: - entries = archive.infolist() + archive_entries = archive.infolist() except zipfile.BadZipFile: raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' is not a valid zip archive.") - if len(entries) > MAX_LOOP_SKILL_BUNDLE_FILES: + 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 entries) + 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 " From d0b3f701b46cbb3f331cca4cf342d676366ab3a8 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 22:18:18 -0700 Subject: [PATCH 09/17] count zip entries from trailer before parsing --- products/tasks/backend/facade/loops.py | 33 +++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index 293186dfece1..6b7f2ecc817b 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -1003,6 +1003,25 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_ MAX_LOOP_SKILL_BUNDLE_UNCOMPRESSED_BYTES = 30 * 1024 * 1024 +def _zip_entry_count(content_bytes: bytes) -> int | None: + """Total entry count 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 count validated here + is the one `ZipFile` will parse. Returns None when no trailer is found. A zip64 + archive reports 0xFFFF here, which already exceeds the bundle cap.""" + eocd_size = 22 + tail = content_bytes[-(eocd_size + 65535) :] + offset = tail.rfind(b"PK\x05\x06") + while offset != -1: + record = tail[offset:] + if len(record) >= eocd_size: + comment_length = int.from_bytes(record[20:22], "little") + if len(record) == eocd_size + comment_length: + return int.from_bytes(record[10:12], "little") + offset = tail.rfind(b"PK\x05\x06", 0, offset) + return None + + 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 @@ -1063,7 +1082,19 @@ def replace_loop_skill_bundles( ) if hashlib.sha256(content_bytes).hexdigest() != bundle["content_sha256"]: raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' does not match its declared sha256.") - # Central-directory metadata only — nothing is decompressed here. + # Entry count comes from the end-of-central-directory record BEFORE ZipFile + # parses the archive: infolist() materializes a ZipInfo per record, so a small + # zip declaring 100k entries would otherwise allocate tens of MB of metadata + # just to be rejected. + entry_count = _zip_entry_count(content_bytes) + if entry_count is None: + raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' is not a valid zip archive.") + if entry_count > MAX_LOOP_SKILL_BUNDLE_FILES: + 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. try: with zipfile.ZipFile(io.BytesIO(content_bytes)) as archive: archive_entries = archive.infolist() From ad48cc60be7076f2e567ec1775c74a366d9cfee2 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 22:21:59 -0700 Subject: [PATCH 10/17] bound zip parsing by central directory size --- products/tasks/backend/facade/loops.py | 50 ++++++++++++------- .../tasks/backend/tests/test_loops_api.py | 29 +++++++++++ 2 files changed, 62 insertions(+), 17 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index 6b7f2ecc817b..a683807b21cb 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -1001,14 +1001,24 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_ # 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 - - -def _zip_entry_count(content_bytes: bytes) -> int | None: - """Total entry count 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 count validated here - is the one `ZipFile` will parse. Returns None when no trailer is found. A zip64 - archive reports 0xFFFF here, which already exceeds the bundle cap.""" +# 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. 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") @@ -1017,7 +1027,9 @@ def _zip_entry_count(content_bytes: bytes) -> int | None: if len(record) >= eocd_size: comment_length = int.from_bytes(record[20:22], "little") if len(record) == eocd_size + comment_length: - return int.from_bytes(record[10:12], "little") + 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 offset = tail.rfind(b"PK\x05\x06", 0, offset) return None @@ -1082,19 +1094,23 @@ def replace_loop_skill_bundles( ) if hashlib.sha256(content_bytes).hexdigest() != bundle["content_sha256"]: raise LoopValidationError(f"Skill bundle for '{bundle['skill_name']}' does not match its declared sha256.") - # Entry count comes from the end-of-central-directory record BEFORE ZipFile - # parses the archive: infolist() materializes a ZipInfo per record, so a small - # zip declaring 100k entries would otherwise allocate tens of MB of metadata - # just to be rejected. - entry_count = _zip_entry_count(content_bytes) - if entry_count is None: + # 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.") - if entry_count > MAX_LOOP_SKILL_BUNDLE_FILES: + 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. + # 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() diff --git a/products/tasks/backend/tests/test_loops_api.py b/products/tasks/backend/tests/test_loops_api.py index a8e4be363869..5d4ac81ae233 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -386,6 +386,35 @@ def test_replace_rejects_a_bundle_with_too_many_entries(self, mock_write): self.assertIn("contains more than", 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) From b6c8c73314c3259d249c57cf5d5613d17e66211e Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 22:30:54 -0700 Subject: [PATCH 11/17] reauthorize bundle swap under the row lock --- products/tasks/backend/facade/loops.py | 28 ++++++++++++++----- .../tasks/backend/tests/test_loops_api.py | 24 ++++++++++++++++ 2 files changed, 45 insertions(+), 7 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index a683807b21cb..455723ecbd93 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -1176,23 +1176,37 @@ def replace_loop_skill_bundles( # of resurrecting bundles on a deleted loop. previous_paths: list[str] = [] lost_delete_race = False + denied: LoopPermissionError | None = None with transaction.atomic(): locked = Loop.objects.unscoped().select_for_update().get(pk=loop.pk) if locked.deleted: lost_delete_race = True 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"]) + 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"]) 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} _delete_skill_bundle_objects( loop.id, [path for path in previous_paths if path.startswith(prefix) and path not in new_paths] diff --git a/products/tasks/backend/tests/test_loops_api.py b/products/tasks/backend/tests/test_loops_api.py index 5d4ac81ae233..3358c81b1197 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -432,6 +432,30 @@ def test_replace_rejects_an_oversized_request_up_front(self): 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") From 09da4fe095896d56ffeaf5083cd2e3dbb0d37900 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 22:53:56 -0700 Subject: [PATCH 12/17] close review gaps in bundle lifecycle and recovery --- products/tasks/backend/facade/loops.py | 79 +++++++++++++------ .../tasks/backend/logic/services/loop_runs.py | 30 +++++++ products/tasks/backend/temporal/client.py | 12 +++ .../tasks/backend/tests/test_loops_api.py | 23 +++++- .../backend/tests/test_temporal_client.py | 66 +++++++++++++++- 5 files changed, 185 insertions(+), 25 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index 455723ecbd93..f686a405ba96 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -595,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)." ) @@ -1001,6 +1001,10 @@ def update_loop(loop_id: str | UUID, team_id: int, user: User | None, validated_ # 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 @@ -1014,30 +1018,36 @@ def _zip_trailer(content_bytes: bytes) -> tuple[int, int] | None: 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. 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.""" + 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") - while offset != -1: - record = tail[offset:] - if len(record) >= eocd_size: - comment_length = int.from_bytes(record[20:22], "little") - if len(record) == eocd_size + comment_length: - 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 - offset = tail.rfind(b"PK\x05\x06", 0, offset) - return None + 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.""" + 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 @@ -1051,6 +1061,25 @@ def _delete_skill_bundle_objects(loop_id: UUID, paths: list[str]) -> None: ) +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: @@ -1082,6 +1111,7 @@ def replace_loop_skill_bundles( _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) @@ -1126,6 +1156,12 @@ def replace_loop_skill_bundles( 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() @@ -1159,7 +1195,7 @@ def replace_loop_skill_bundles( "skill_name": bundle["skill_name"], "skill_source": bundle["skill_source"], "content_sha256": bundle["content_sha256"], - "bundle_format": "zip", + "bundle_format": bundle["bundle_format"], "schema_version": 1, }, } @@ -1208,8 +1244,8 @@ def replace_loop_skill_bundles( raise denied new_paths = {entry["storage_path"] for entry in entries} - _delete_skill_bundle_objects( - loop.id, [path for path in previous_paths if path.startswith(prefix) and path not in new_paths] + _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() @@ -1245,8 +1281,7 @@ def soft_delete_loop(loop_id: str | UUID, team_id: int, user: User | None) -> bo locked.deleted = True locked.skill_bundles = [] locked.save(update_fields=["deleted", "skill_bundles", "updated_at"]) - loop.deleted = True - _delete_skill_bundle_objects(loop.id, bundle_paths) + _expire_skill_bundle_objects(loop.id, loop.team_id, bundle_paths) loop_service.delete_loop_schedules(loop) return True diff --git a/products/tasks/backend/logic/services/loop_runs.py b/products/tasks/backend/logic/services/loop_runs.py index 323ee5661388..8a7b3ca7ca36 100644 --- a/products/tasks/backend/logic/services/loop_runs.py +++ b/products/tasks/backend/logic/services/loop_runs.py @@ -533,6 +533,36 @@ def _seed_skill_bundles_and_dispatch( ) +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. 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 + # No ambient team scope on reconciler/Celery paths, same as Loop._get_before_update. + loop = Loop.objects.unscoped().filter(pk=task.loop_id).first() + if loop is None or not loop.skill_bundles: + 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(loop, task_run) + except Exception: + logger.exception( + "loop_run.skill_bundle_seed_failed", + extra={"loop_id": str(loop.id), "task_run_id": str(task_run.id)}, + ) + return False + return True + + def _seed_skill_bundle_artifacts(loop: Loop, task_run: TaskRun) -> None: """Copy the loop's stored skill bundles into the new run: S3 objects under the run's artifact prefix plus matching ``skill_bundle`` manifest entries, so the sandbox 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_loops_api.py b/products/tasks/backend/tests/test_loops_api.py index 3358c81b1197..18cc02e9bce5 100644 --- a/products/tasks/backend/tests/test_loops_api.py +++ b/products/tasks/backend/tests/test_loops_api.py @@ -294,7 +294,10 @@ def test_owner_replaces_and_clears_skill_bundles(self, mock_write, mock_tag, moc 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, []) - mock_delete.assert_called_once_with([first_storage_path]) + # 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): @@ -386,6 +389,21 @@ def test_replace_rejects_a_bundle_with_too_many_entries(self, mock_write): 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): @@ -494,7 +512,8 @@ def test_deleting_a_loop_releases_its_bundle_objects(self, mock_write, mock_tag, 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_called_once_with([stored_path]) + 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, []) diff --git a/products/tasks/backend/tests/test_temporal_client.py b/products/tasks/backend/tests/test_temporal_client.py index 3f3805d3afec..a91044a6719c 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,70 @@ def _run_reconcile(self, run: TaskRun, start_workflow: Mock) -> str: ): return redispatch_orphaned_task_run(str(run.id)) + def _loop_with_bundle(self) -> Loop: + loop = Loop.objects.create( + team=self.team, + created_by=self.user, + name="Digest", + instructions="/my-skill", + runtime_adapter="claude", + skill_bundles=[ + { + "id": "abcdef0123456789abcdef0123456789", + "name": "my-skill.zip", + "type": "skill_bundle", + "source": "posthog_code_skill", + "size": 9, + "content_type": "application/zip", + "storage_path": "tasks/artifacts/team_1/loop_x/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, + }, + } + ], + ) + self.task.origin_product = Task.OriginProduct.LOOP + self.task.loop = loop + self.task.save(update_fields=["origin_product", "loop"]) + return loop + + @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. + self._loop_with_bundle() + run = self._orphaned_run(pending_dispatch={"create_pr": False, "posthog_mcp_scopes": "read_only"}) + 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", side_effect=RuntimeError("s3 down")) + def test_recovery_aborts_when_seeding_fails(self, mock_copy) -> None: + self._loop_with_bundle() + run = self._orphaned_run(pending_dispatch={"create_pr": False, "posthog_mcp_scopes": "read_only"}) + 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() From a88e37eb9f7ec835c2cfd50141be58c6883af9de Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 23:12:18 -0700 Subject: [PATCH 13/17] seed recovered runs from fire-time snapshot --- .../tasks/backend/logic/services/loop_runs.py | 63 +++++++++++-------- .../tasks/backend/tests/test_loop_runs.py | 3 + .../backend/tests/test_temporal_client.py | 31 ++++----- 3 files changed, 53 insertions(+), 44 deletions(-) diff --git a/products/tasks/backend/logic/services/loop_runs.py b/products/tasks/backend/logic/services/loop_runs.py index 8a7b3ca7ca36..b2c047631569 100644 --- a/products/tasks/backend/logic/services/loop_runs.py +++ b/products/tasks/backend/logic/services/loop_runs.py @@ -492,7 +492,6 @@ def _team_rate_capped(loop: Loop) -> bool: def _seed_skill_bundles_and_dispatch( *, - loop: Loop, task_run: TaskRun, team_id: int, user_id: int | None, @@ -501,20 +500,21 @@ def _seed_skill_bundles_and_dispatch( create_pr: bool, posthog_mcp_scopes: PosthogMcpScopes, ) -> None: - """Post-commit tail of a fire: seed the loop's 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.""" + """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(loop, task_run) + _seed_skill_bundle_artifacts(task_run) except Exception as exc: logger.exception( "loop_run.skill_bundle_seed_failed", - extra={"loop_id": str(loop.id), "task_run_id": run_id, "error": str(exc)}, + 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, @@ -538,38 +538,43 @@ def ensure_loop_skill_bundles_seeded(task_run: TaskRun) -> bool: 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. 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 - # No ambient team scope on reconciler/Celery paths, same as Loop._get_before_update. - loop = Loop.objects.unscoped().filter(pk=task.loop_id).first() - if loop is None or not loop.skill_bundles: - return True + 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).""" 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(loop, task_run) + _seed_skill_bundle_artifacts(task_run) except Exception: logger.exception( "loop_run.skill_bundle_seed_failed", - extra={"loop_id": str(loop.id), "task_run_id": str(task_run.id)}, + extra={"task_run_id": str(task_run.id)}, ) return False return True -def _seed_skill_bundle_artifacts(loop: Loop, task_run: TaskRun) -> None: - """Copy the loop's stored skill bundles into the new run: S3 objects under the run's +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.""" - bundles = [entry for entry in (loop.skill_bundles or []) if isinstance(entry, dict) and entry.get("storage_path")] + 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 @@ -689,6 +694,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 @@ -698,7 +710,6 @@ def _create_loop_task_and_run(loop: Loop, trigger: LoopTrigger | None, trigger_c transaction.on_commit( lambda: _seed_skill_bundles_and_dispatch( - loop=loop, task_run=task_run, team_id=team_id, user_id=user_id, diff --git a/products/tasks/backend/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py index 65e811e9bea4..78b4cf05f9c2 100644 --- a/products/tasks/backend/tests/test_loop_runs.py +++ b/products/tasks/backend/tests/test_loop_runs.py @@ -586,6 +586,9 @@ def test_fire_copies_loop_skill_bundles_into_the_run_manifest(self, mock_copy, m 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"]) diff --git a/products/tasks/backend/tests/test_temporal_client.py b/products/tasks/backend/tests/test_temporal_client.py index a91044a6719c..2515fd9d468d 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 Loop, Task, TaskRun +from products.tasks.backend.models import Task, TaskRun from products.tasks.backend.temporal.client import ( execute_task_processing_workflow, execute_task_processing_workflow_async, @@ -328,14 +328,13 @@ def _run_reconcile(self, run: TaskRun, start_workflow: Mock) -> str: ): return redispatch_orphaned_task_run(str(run.id)) - def _loop_with_bundle(self) -> Loop: - loop = Loop.objects.create( - team=self.team, - created_by=self.user, - name="Digest", - instructions="/my-skill", - runtime_adapter="claude", - skill_bundles=[ + def _orphaned_loop_run(self) -> TaskRun: + # A loop fire freezes its bundle set onto the run's state; seeding reads only + # this snapshot, never the live 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", @@ -354,19 +353,16 @@ def _loop_with_bundle(self) -> Loop: }, } ], - ) - self.task.origin_product = Task.OriginProduct.LOOP - self.task.loop = loop - self.task.save(update_fields=["origin_product", "loop"]) - return loop + } + 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. - self._loop_with_bundle() - run = self._orphaned_run(pending_dispatch={"create_pr": False, "posthog_mcp_scopes": "read_only"}) + run = self._orphaned_loop_run() start_workflow = AsyncMock() outcome = self._run_reconcile(run, start_workflow) @@ -379,8 +375,7 @@ def test_recovery_seeds_loop_skill_bundles_before_dispatch(self, mock_copy, mock @patch("posthog.storage.object_storage.copy", side_effect=RuntimeError("s3 down")) def test_recovery_aborts_when_seeding_fails(self, mock_copy) -> None: - self._loop_with_bundle() - run = self._orphaned_run(pending_dispatch={"create_pr": False, "posthog_mcp_scopes": "read_only"}) + run = self._orphaned_loop_run() start_workflow = AsyncMock() outcome = self._run_reconcile(run, start_workflow) From 632d5cda0024248d08f12ad41eff299b12b4ac2f Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Thu, 23 Jul 2026 23:56:03 -0700 Subject: [PATCH 14/17] validate skill bundle seeds against loop prefix --- products/tasks/backend/facade/api.py | 4 +++ .../tasks/backend/logic/services/loop_runs.py | 22 +++++++++++-- products/tasks/backend/models.py | 11 +++++-- .../tasks/backend/tests/test_loop_runs.py | 28 +++++++++------- .../backend/tests/test_temporal_client.py | 32 ++++++++++++++++--- 5 files changed, 76 insertions(+), 21 deletions(-) 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/logic/services/loop_runs.py b/products/tasks/backend/logic/services/loop_runs.py index b2c047631569..93e2b1d8fa6c 100644 --- a/products/tasks/backend/logic/services/loop_runs.py +++ b/products/tasks/backend/logic/services/loop_runs.py @@ -543,6 +543,9 @@ def ensure_loop_skill_bundles_seeded(task_run: TaskRun) -> bool: 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 []) ) @@ -580,14 +583,29 @@ def _seed_skill_bundle_artifacts(task_run: TaskRun) -> None: 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_prefix = f"{Loop.skill_bundle_s3_prefix_for(task_run.team_id, task_run.task.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) - target_path = f"{run_prefix}/{str(entry['id'])[:8]}_{entry['name']}" - object_storage.copy(entry["storage_path"], target_path) + 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)}) diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index b00837fde535..8c1859bf6184 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -1179,10 +1179,15 @@ class Meta: def __str__(self): return self.name - def get_skill_bundle_s3_prefix(self) -> str: - """Base prefix for this loop's skill bundle objects in S3.""" + @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_{self.team_id}/loop_{self.id}" + 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 diff --git a/products/tasks/backend/tests/test_loop_runs.py b/products/tasks/backend/tests/test_loop_runs.py index 78b4cf05f9c2..64a554103eed 100644 --- a/products/tasks/backend/tests/test_loop_runs.py +++ b/products/tasks/backend/tests/test_loop_runs.py @@ -538,7 +538,8 @@ def test_fire_threads_the_loops_sandbox_environment_into_run_state(self, _name, class TestFireLoopSeedsSkillBundles(LoopRunsTestCase): - def _bundle_entry(self, **overrides) -> dict: + 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", @@ -546,7 +547,7 @@ def _bundle_entry(self, **overrides) -> dict: "source": "posthog_code_skill", "size": 9, "content_type": "application/zip", - "storage_path": "tasks/artifacts/team_1/loop_x/abcdef01_my-skill.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", @@ -559,6 +560,13 @@ def _bundle_entry(self, **overrides) -> dict: 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).""" @@ -570,8 +578,7 @@ def fire_with_post_commit(self, loop: Loop, trigger: LoopTrigger): @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): - entry = self._bundle_entry() - loop = self.create_loop(skill_bundles=[entry]) + loop, (entry,) = self.loop_with_bundles() trigger = self.create_trigger(loop) result, mock_dispatch = self.fire_with_post_commit(loop, trigger) @@ -606,7 +613,7 @@ def test_fire_without_bundles_does_not_touch_storage(self): 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 = self.create_loop(skill_bundles=[self._bundle_entry()]) + loop, _entries = self.loop_with_bundles() trigger = self.create_trigger(loop) with patch("posthog.storage.object_storage.copy") as mock_copy: @@ -623,7 +630,7 @@ def test_seeding_happens_after_commit_not_under_the_fire_lock(self): @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 = self.create_loop(skill_bundles=[self._bundle_entry()]) + loop, _entries = self.loop_with_bundles() trigger = self.create_trigger(loop) result, mock_dispatch = self.fire_with_post_commit(loop, trigger) @@ -646,13 +653,10 @@ def test_a_partially_seeded_fire_deletes_the_copies_it_made(self, mock_copy, moc # 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")] - first = self._bundle_entry() - second = self._bundle_entry( - id="fedcba9876543210fedcba9876543210", - name="other-skill.zip", - storage_path="tasks/artifacts/team_1/loop_x/fedcba98_other-skill.zip", + loop, _entries = self.loop_with_bundles( + {}, + {"id": "fedcba9876543210fedcba9876543210", "name": "other-skill.zip"}, ) - loop = self.create_loop(skill_bundles=[first, second]) trigger = self.create_trigger(loop) result, mock_dispatch = self.fire_with_post_commit(loop, trigger) diff --git a/products/tasks/backend/tests/test_temporal_client.py b/products/tasks/backend/tests/test_temporal_client.py index 2515fd9d468d..f753e02744fd 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,9 +328,20 @@ def _run_reconcile(self, run: TaskRun, start_workflow: Mock) -> str: ): return redispatch_orphaned_task_run(str(run.id)) - def _orphaned_loop_run(self) -> TaskRun: + 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. + # this snapshot, never the live loop, and validates every source path against + # the owning loop's prefix. + loop = Loop.objects.create( + team=self.team, + created_by=self.user, + name="Digest", + instructions="/my-skill", + runtime_adapter="claude", + ) + 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, @@ -342,7 +353,7 @@ def _orphaned_loop_run(self) -> TaskRun: "source": "posthog_code_skill", "size": 9, "content_type": "application/zip", - "storage_path": "tasks/artifacts/team_1/loop_x/abcdef01_my-skill.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", @@ -373,6 +384,19 @@ def test_recovery_seeds_loop_skill_bundles_before_dispatch(self, mock_copy, mock 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() From 225c9a5e7312157ccbfed47223aa3697bca80673 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Fri, 24 Jul 2026 00:11:53 -0700 Subject: [PATCH 15/17] release uploads when the bundle swap fails --- products/tasks/backend/facade/loops.py | 46 +++++++++++++++----------- 1 file changed, 26 insertions(+), 20 deletions(-) diff --git a/products/tasks/backend/facade/loops.py b/products/tasks/backend/facade/loops.py index f686a405ba96..6952d4ed05db 100644 --- a/products/tasks/backend/facade/loops.py +++ b/products/tasks/backend/facade/loops.py @@ -1213,27 +1213,33 @@ def replace_loop_skill_bundles( previous_paths: list[str] = [] lost_delete_race = False denied: LoopPermissionError | None = None - 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 + try: + with transaction.atomic(): + locked = Loop.objects.unscoped().select_for_update().get(pk=loop.pk) + if locked.deleted: + lost_delete_race = True 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"]) + 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) From 525e4dee32c836342997452b5c6b333fb773bd00 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Fri, 24 Jul 2026 00:23:34 -0700 Subject: [PATCH 16/17] narrow loop id before computing seed prefix --- products/tasks/backend/logic/services/loop_runs.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/products/tasks/backend/logic/services/loop_runs.py b/products/tasks/backend/logic/services/loop_runs.py index 93e2b1d8fa6c..c0ed0dcdda11 100644 --- a/products/tasks/backend/logic/services/loop_runs.py +++ b/products/tasks/backend/logic/services/loop_runs.py @@ -589,7 +589,10 @@ def _seed_skill_bundle_artifacts(task_run: TaskRun) -> None: # 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_prefix = f"{Loop.skill_bundle_s3_prefix_for(task_run.team_id, task_run.task.loop_id)}/" + 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 []) From 0ff61e2194201a6751fc2f132669ba9c2ac9b924 Mon Sep 17 00:00:00 2001 From: Charles Vien Date: Fri, 24 Jul 2026 01:05:37 -0700 Subject: [PATCH 17/17] save test loop directly to avoid scope error --- products/tasks/backend/tests/test_temporal_client.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/products/tasks/backend/tests/test_temporal_client.py b/products/tasks/backend/tests/test_temporal_client.py index f753e02744fd..b840f585dbe0 100644 --- a/products/tasks/backend/tests/test_temporal_client.py +++ b/products/tasks/backend/tests/test_temporal_client.py @@ -332,13 +332,16 @@ 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. - loop = Loop.objects.create( + # 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"])