feat(array): temporal workflow for sandbox environment

posthog/temporal/common/utils.py

@@ -1,7 +1,40 @@
+import inspect
+from collections.abc import Callable
 from datetime import datetime
+from functools import wraps
+from typing import ParamSpec, TypeVar
 
+from asgiref.sync import sync_to_async
 from temporalio import workflow
 
+P = ParamSpec("P")
+T = TypeVar("T")
+
+
+def asyncify(fn: Callable[P, T]) -> Callable[P, T]:
+    """Decorator to convert a sync function using sync_to_async - this preserves type hints for Temporal's serialization while allowing sync Django ORM code.
+
+    This preserves type hints for Temporal's serialization while allowing
+    sync Django ORM code.
+
+    Usage:
+        @activity.defn
+        @asyncify
+        def my_activity(task_id: str) -> TaskDetails:
+            task = Task.objects.get(id=task_id)
+            return TaskDetails(...)
+    """
+    if inspect.iscoroutinefunction(fn):
+        raise TypeError(
+            f"@asyncify should only be used on sync functions. " f"'{fn.__name__}' is already async. Remove @asyncify."
+        )
+
+    @wraps(fn)
+    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+        return await sync_to_async(fn)(*args, **kwargs)
+
+    return wrapper
+
 
 def get_scheduled_start_time():
     """Return the start time of a workflow.

products/tasks/backend/api.py

@@ -594,6 +594,7 @@ class AgentDefinitionViewSet(TeamAndOrgViewSetMixin, viewsets.ReadOnlyModelViewS
     serializer_class = AgentDefinitionSerializer
     authentication_classes = [SessionAuthentication, PersonalAPIKeyAuthentication]
     queryset = None  # No model queryset since we're using hardcoded agents
+    scope_object = "task"
     posthog_feature_flag = {"tasks": ["list", "retrieve"]}
 
     @extend_schema(

products/tasks/backend/lib/constants.py

@@ -0,0 +1,19 @@
+SETUP_REPOSITORY_PROMPT = """
+Your goal is to setup the repository in the current environment.
+
+You are operating in a sandbox environment. You must install all dependencies necessary and setup the enviornment such that it is ready for executing code tasks.
+
+CONTEXT:
+
+CWD: {cwd}
+
+REPOSITORY: {repository}
+
+INSTRUCTIONS:
+
+1. Install all dependencies necessary to run the repository
+2. Run any setup scripts that are available
+3. Verify the setup by running tests or build if available
+
+DO NOT make any code changes to the repository. The final state of the disk of this sandbox is what will be used for subsequent tasks, so do not leave any cruft behind, and make sure the repository is in a ready to use state.
+"""

products/tasks/backend/migrations/0007_sandboxsnapshot.py

@@ -0,0 +1,68 @@
+# Generated by Django 4.2.22 on 2025-09-30 13:58
+
+import uuid
+
+import django.db.models.deletion
+import django.contrib.postgres.fields
+import django.contrib.postgres.indexes
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("posthog", "0867_add_updated_at_to_feature_flags"),
+        ("tasks", "0006_remove_workflowstage_agent_alter_task_workflow_and_more"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="SandboxSnapshot",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                (
+                    "external_id",
+                    models.CharField(blank=True, help_text="Snapshot ID from external provider.", max_length=255),
+                ),
+                (
+                    "repos",
+                    django.contrib.postgres.fields.ArrayField(
+                        base_field=models.CharField(max_length=255),
+                        default=list,
+                        help_text="List of repositories in format 'org/repo'",
+                        size=None,
+                    ),
+                ),
+                (
+                    "metadata",
+                    models.JSONField(blank=True, default=dict, help_text="Additional metadata for the snapshot."),
+                ),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[("in_progress", "In Progress"), ("complete", "Complete"), ("error", "Error")],
+                        default="in_progress",
+                        max_length=20,
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                (
+                    "integration",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, related_name="snapshots", to="posthog.integration"
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "posthog_sandbox_snapshot",
+                "indexes": [
+                    models.Index(
+                        fields=["integration", "status", "-created_at"], name="posthog_san_integra_93cdea_idx"
+                    ),
+                    models.Index(fields=["integration", "-created_at"], name="posthog_san_integra_50465d_idx"),
+                    models.Index(fields=["status"], name="posthog_san_status_a49734_idx"),
+                    django.contrib.postgres.indexes.GinIndex(fields=["repos"], name="posthog_san_repos_9cb7b7_gin"),
+                ],
+            },
+        ),
+    ]

products/tasks/backend/migrations/0008_remove_sandboxsnapshot_posthog_san_integra_50465d_idx_and_more.py

@@ -0,0 +1,37 @@
+# Generated by Django 4.2.22 on 2025-09-30 14:15
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("posthog", "0867_add_updated_at_to_feature_flags"),
+        ("tasks", "0007_sandboxsnapshot"),
+    ]
+
+    operations = [
+        migrations.RemoveIndex(
+            model_name="sandboxsnapshot",
+            name="posthog_san_integra_50465d_idx",
+        ),
+        migrations.RemoveIndex(
+            model_name="sandboxsnapshot",
+            name="posthog_san_status_a49734_idx",
+        ),
+        migrations.RemoveIndex(
+            model_name="sandboxsnapshot",
+            name="posthog_san_repos_9cb7b7_gin",
+        ),
+        migrations.AlterField(
+            model_name="sandboxsnapshot",
+            name="integration",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="snapshots",
+                to="posthog.integration",
+            ),
+        ),
+    ]

products/tasks/backend/migrations/0009_task_created_by.py

@@ -0,0 +1,22 @@
+# Generated by Django 4.2.22 on 2025-10-02 16:50
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("tasks", "0008_remove_sandboxsnapshot_posthog_san_integra_50465d_idx_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="task",
+            name="created_by",
+            field=models.ForeignKey(
+                blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL
+            ),
+        ),
+    ]

products/tasks/backend/migrations/max_migration.txt

@@ -1 +1 @@
-0006_remove_workflowstage_agent_alter_task_workflow_and_more
+0009_task_created_by

products/tasks/backend/models.py

@@ -1,6 +1,7 @@
 import uuid
 from typing import Optional, cast
 
+from django.contrib.postgres.fields import ArrayField
 from django.db import models, transaction
 from django.utils import timezone
 
@@ -229,6 +230,7 @@ class OriginProduct(models.TextChoices):
 
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
     team = models.ForeignKey("posthog.Team", on_delete=models.CASCADE)
+    created_by = models.ForeignKey("posthog.User", on_delete=models.SET_NULL, null=True, blank=True)
     title = models.CharField(max_length=255)
     description = models.TextField()
     origin_product = models.CharField(max_length=20, choices=OriginProduct.choices)
@@ -455,3 +457,92 @@ def progress_percentage(self):
         if self.total_steps and self.total_steps > 0:
             return min(100, (self.completed_steps / self.total_steps) * 100)
         return 0
+
+
+class SandboxSnapshot(models.Model):
+    """Tracks sandbox snapshots used for sandbox environments in tasks."""
+
+    class Status(models.TextChoices):
+        IN_PROGRESS = "in_progress", "In Progress"
+        COMPLETE = "complete", "Complete"
+        ERROR = "error", "Error"
+
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+
+    integration = models.ForeignKey(
+        Integration,
+        on_delete=models.SET_NULL,
+        related_name="snapshots",
+        null=True,
+        blank=True,
+    )
+
+    external_id = models.CharField(max_length=255, blank=True, help_text="Snapshot ID from external provider.")
+
+    repos = ArrayField(
+        models.CharField(max_length=255),
+        default=list,
+        help_text="List of repositories in format 'org/repo'",
+    )
+
+    metadata = models.JSONField(default=dict, blank=True, help_text="Additional metadata for the snapshot.")
+
+    status = models.CharField(
+        max_length=20,
+        choices=Status.choices,
+        default=Status.IN_PROGRESS,
+    )
+
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        db_table = "posthog_sandbox_snapshot"
+        indexes = [
+            models.Index(fields=["integration", "status", "-created_at"]),
+        ]
+
+    def __str__(self):
+        repo_count = len(self.repos)
+        return f"Snapshot {self.external_id} ({self.get_status_display()}, {repo_count} repos)"
+
+    def is_complete(self) -> bool:
+        return self.status == self.Status.COMPLETE
+
+    def has_repo(self, repo: str) -> bool:
+        repo_lower = repo.lower()
+        return any(r.lower() == repo_lower for r in self.repos)
+
+    def has_repos(self, repos: list[str]) -> bool:
+        return all(self.has_repo(repo) for repo in repos)
+
+    def update_status(self, status: Status):
+        self.status = status
+        self.save(update_fields=["status"])
+
+    @classmethod
+    def get_latest_snapshot_for_integration(
+        cls, integration_id: int, status: Status = Status.COMPLETE
+    ) -> Optional["SandboxSnapshot"]:
+        return (
+            cls.objects.filter(
+                integration_id=integration_id,
+                status=status,
+            )
+            .order_by("-created_at")
+            .first()
+        )
+
+    @classmethod
+    def get_latest_snapshot_with_repos(
+        cls, integration_id: int, required_repos: list[str], status: Status = Status.COMPLETE
+    ) -> Optional["SandboxSnapshot"]:
+        snapshots = cls.objects.filter(
+            integration_id=integration_id,
+            status=status,
+        ).order_by("-created_at")
+
+        for snapshot in snapshots:
+            if snapshot.has_repos(required_repos):
+                return snapshot
+        return None