feat(wizard): execute cloud runs

.github/workflows/container-images-cd.yml

@@ -444,7 +444,7 @@ jobs:
             - name: Check for changes that affect Tasks Agent Temporal worker
               id: check_changes_tasks_agent_temporal_worker
               run: |
-                  if git diff --name-only HEAD^ HEAD | grep -qE '^posthog/temporal/common|^posthog/temporal/ai/posthog_code_slack_|^posthog/temporal/ai/slack_app/|^products/slack_app/backend|^products/tasks/backend|^posthog/tasks/|^posthog/settings/temporal.py$|^posthog/management/commands/start_temporal_worker.py$|^pyproject.toml$|^bin/temporal-django-worker$|^posthog/clickhouse/migrations/'; then
+                  if git diff --name-only HEAD^ HEAD | grep -qE '^posthog/temporal/common|^posthog/temporal/ai/posthog_code_slack_|^posthog/temporal/ai/slack_app/|^products/slack_app/backend|^products/tasks/backend|^products/wizard/backend/|^posthog/tasks/|^posthog/settings/temporal.py$|^posthog/management/commands/start_temporal_worker.py$|^pyproject.toml$|^bin/temporal-django-worker$|^posthog/clickhouse/migrations/'; then
                       echo "changed=true" >> $GITHUB_OUTPUT
                   else
                       echo "changed=false" >> $GITHUB_OUTPUT

docs/internal/wizard-registry.md

@@ -1,8 +1,8 @@
 # Wizard registry
 
 Status: design document for the Wizard run stack.
-This layer ships only the data models and the registry parsing described under "Remote configuration".
-The "API" routes and the "Cloud execution" workers describe planned behavior and do not exist yet; they land in later stack layers.
+This layer ships the data models, registry parsing, and cloud execution workers.
+The "API" routes describe planned behavior and land in a later stack layer.
 
 The Wizard Registry lists the programs a person can run in the Wizard.
 Programs are personalized by the signed-in user's distinct ID and the organization that owns the selected project.
@@ -66,6 +66,26 @@ Its command is empty, which keeps the Wizard package's default command.
 
 ## Cloud execution
 
+Wizard analytics events use the process-lifetime background capture client without a Celery hop.
+Delivery is best-effort; an abrupt process exit can lose buffered events.
+
+Wizard execution returns only the last 64 KiB of stdout and stderr from the sandbox, preserving the command's exit code.
+Log streams are drained through bounded tails without storing unbounded log files.
+Repository publishing bounds sandbox reads and rejects staged contents that exceed the cumulative 35 MiB commit payload budget, including base64 expansion.
+
+### Dedicated worker rollout
+
+`WIZARD_TASK_QUEUE` controls both Wizard dispatch and worker registration. It defaults to `general-purpose-task-queue` during rollout.
+`TEMPORAL_TASK_QUEUE` selects the queue a worker process polls; setting it alone does not register Wizard workflows on that queue.
+
+1. Deploy an image containing the Wizard workflow and activity registration before enabling the dedicated worker.
+2. Set both `TEMPORAL_TASK_QUEUE` and `WIZARD_TASK_QUEUE` to `wizard-task-queue` on the dedicated worker.
+3. Verify the worker is healthy and polling before setting `WIZARD_TASK_QUEUE=wizard-task-queue` on dispatching services.
+
+The dedicated worker uses the Tasks Agent image release. Changes to Wizard backend code must trigger that image's deployment.
+
+### Program execution
+
 Cloud workers execute the program snapshot stored when the run was created.
 They do not evaluate the registry again.
 They run the pinned `wizard_version` from that snapshot, never a mutable tag such as `latest`.

posthog/management/commands/start_temporal_worker.py

@@ -275,6 +275,10 @@
     ACTIVITIES as WA_DIGEST_ACTIVITIES,
     WORKFLOWS as WA_DIGEST_WORKFLOWS,
 )
+from products.wizard.backend.facade.temporal import (
+    ACTIVITIES as WIZARD_ACTIVITIES,
+    WORKFLOWS as WIZARD_WORKFLOWS,
+)
 
 # When adding modules to a queue, also update the corresponding CI trigger
 # in .github/workflows/container-images-cd.yml (check_changes_*_temporal_worker)
@@ -414,6 +418,11 @@
         TASKS_WORKFLOWS + POSTHOG_CODE_SLACK_WORKFLOWS,
         TASKS_ACTIVITIES + POSTHOG_CODE_SLACK_ACTIVITIES,
     ),
+    (
+        settings.WIZARD_TASK_QUEUE,
+        WIZARD_WORKFLOWS,
+        WIZARD_ACTIVITIES,
+    ),
     (
         settings.MAX_AI_TASK_QUEUE,
         AI_WORKFLOWS,
@@ -659,7 +668,7 @@
             help="Disable the combined metrics server (useful for workers with GIL contention issues)",
         )
 
     def handle(self, *args, **options):
         temporal_host = options["temporal_host"]
         temporal_port = options["temporal_port"]
         namespace = options["namespace"]

posthog/management/commands/test/test_start_temporal_worker.py

@@ -1,5 +1,7 @@
 import pytest
 
+from django.conf import settings
+
 from posthog.management.commands.start_temporal_worker import (
     DATA_SYNC_WORKFLOWS,
     WA_DIGEST_ACTIVITIES,
@@ -9,11 +11,30 @@
     workflows_include_data_import_syncs,
 )
 
+from products.wizard.backend.facade.temporal import (
+    ACTIVITIES as WIZARD_ACTIVITIES,
+    WORKFLOWS as WIZARD_WORKFLOWS,
+)
+
 
 class _NotADataSyncWorkflow:
     pass
 
 
+def test_wizard_queue_registers_workflows_and_activities() -> None:
+    entries = [
+        (workflows, activities)
+        for queue, workflows, activities in _task_queue_specs
+        if queue == settings.WIZARD_TASK_QUEUE
+    ]
+    workflows = {workflow for registered_workflows, _ in entries for workflow in registered_workflows}
+    activities = {activity for _, registered_activities in entries for activity in registered_activities}
+    assert WIZARD_WORKFLOWS
+    assert WIZARD_ACTIVITIES
+    assert set(WIZARD_WORKFLOWS) <= workflows
+    assert set(WIZARD_ACTIVITIES) <= activities
+
+
 # Data-import sources import vendor SDKs (google-ads, etc.) that register protobuf descriptors into a
 # process-global pool exactly once. The worker eagerly loads them at boot only for queues that run
 # data syncs; everything else stays lazy to keep startup fast. Queue settings collapse to a single

posthog/settings/temporal.py

@@ -245,6 +245,11 @@ def _set_temporal_task_queue(task_queue: str) -> str:
 HEALTH_CHECK_TASK_QUEUE = _set_temporal_task_queue("health-check-task-queue")
 DUCKLAKE_TASK_QUEUE = _set_temporal_task_queue("ducklake-task-queue")
 TASKS_TASK_QUEUE = _set_temporal_task_queue("tasks-task-queue")
+# Defaults to the general-purpose fleet so dispatch always has a live worker. Routing Wizard runs
+# to a dedicated, separately-scalable worker takes two steps in order: deploy a worker fleet
+# polling "wizard-task-queue", then set this env on the dispatching services. Setting it first
+# strands runs on a pollerless queue until the workflow execution timeout closes them.
+WIZARD_TASK_QUEUE = _set_temporal_task_queue(os.getenv("WIZARD_TASK_QUEUE", "general-purpose-task-queue"))
 WIZARD_RUN_ARTIFACTS_S3_BUCKET = os.getenv("WIZARD_RUN_ARTIFACTS_S3_BUCKET", "")
 TASKS_DISPATCHER_BATCH_SIZE = get_from_env("TASKS_DISPATCHER_BATCH_SIZE", 50, type_cast=int)
 TASKS_DISPATCHER_CONCURRENCY = get_from_env("TASKS_DISPATCHER_CONCURRENCY", 20, type_cast=int)

products/tasks/backend/facade/repo_selection.py

@@ -6,6 +6,7 @@
 handle the typed outcomes.
 """
 
+from products.tasks.backend.github_repository_access import inaccessible_repositories_via_integration
 from products.tasks.backend.logic.repo_selection import (
     REPO_SELECTION_DUMMY_REPOSITORY,
     RepoSelectionRejectedError,
@@ -15,13 +16,27 @@
     resolve_team_github_integration,
     select_repository,
 )
+from products.tasks.backend.temporal.process_task.utils import get_github_token
+
+
+def repository_accessible_via_integration(team_id: int, integration_id: int, repository: str) -> bool:
+    return not inaccessible_repositories_via_integration(team_id, integration_id, [repository])
+
+
+def resolve_team_github_integration_id(team_id: int) -> int | None:
+    integration = resolve_team_github_integration(team_id, team_only=True)
+    return integration.integration.id if integration is not None else None
+
 
 __all__ = [
     "REPO_SELECTION_DUMMY_REPOSITORY",
     "RepoSelectionRejectedError",
     "RepoSelectionResult",
     "RepoSelectionUnavailableError",
+    "get_github_token",
     "list_team_connected_repositories",
+    "repository_accessible_via_integration",
     "resolve_team_github_integration",
+    "resolve_team_github_integration_id",
     "select_repository",
 ]

products/tasks/backend/facade/sandbox.py

@@ -26,6 +26,7 @@
     get_sandbox_class,
     get_sandbox_class_for_backend,
     is_public_sandbox_repo,
+    sandbox_repo_path,
 )
 
 __all__ = [
@@ -45,4 +46,5 @@
     "get_sandbox_class",
     "get_sandbox_class_for_backend",
     "is_public_sandbox_repo",
+    "sandbox_repo_path",
 ]

products/tasks/backend/logic/services/sandbox.py

@@ -231,12 +231,18 @@ def effective_memory_request_mb(self) -> int:
 """Repos the sandbox is allowed to clone unauthenticated, even when the team has no GitHub integration"""
 # TODO: Remove `posthog/.github` when we switch repo discovery to repo-less agent (now it works as a lightweight dummy)
 
-SENSITIVE_AGENT_RUNTIME_ENV_NAMES: frozenset[str] = frozenset(
-    {"POSTHOG_TASK_RUN_EVENT_INGEST_TOKEN", "POSTHOG_TASK_RUN_SESSION_TOKEN"}
+# this helps redact sensitive environment variables for logging
+SENSITIVE_SANDBOX_ENV_NAMES: frozenset[str] = frozenset(
+    {
+        "GITHUB_TOKEN",
+        "POSTHOG_TASK_RUN_EVENT_INGEST_TOKEN",
+        "POSTHOG_TASK_RUN_SESSION_TOKEN",
+        "POSTHOG_WIZARD_API_KEY",
+    }
 )
 SHELL_ARGUMENT_VALUE_PATTERN = r"'(?:[^']|'\"'\"')*'|\"(?:\\.|[^\"])*\"|\S+"
-SENSITIVE_AGENT_RUNTIME_ENV_PATTERN = re.compile(
-    r"(?P<name>" + "|".join(re.escape(name) for name in SENSITIVE_AGENT_RUNTIME_ENV_NAMES) + r")="
+SENSITIVE_SANDBOX_ENV_PATTERN = re.compile(
+    r"(?P<name>" + "|".join(re.escape(name) for name in SENSITIVE_SANDBOX_ENV_NAMES) + r")="
     rf"(?P<value>{SHELL_ARGUMENT_VALUE_PATTERN})"
 )
 SENSITIVE_AGENT_RUNTIME_ARGUMENT_PATTERN = re.compile(
@@ -245,6 +251,7 @@ def effective_memory_request_mb(self) -> int:
 SENSITIVE_FILE_HEREDOC_PATTERN = re.compile(
     r"(?P<prefix><<'POSTHOG_FILE_EOF'\n).*?(?P<suffix>\nPOSTHOG_FILE_EOF)", re.DOTALL
 )
+GITHUB_CLONE_TOKEN_PATTERN = re.compile(r"(?P<prefix>https://x-access-token:)[^@\s]+(?P<suffix>@github\.com/)")
 
 
 def is_public_sandbox_repo(repository: str | None) -> bool:
@@ -258,9 +265,15 @@ def sandbox_repo_path(repository: str) -> str:
 
 
 def redact_sandbox_command(command: str) -> str:
-    redacted = SENSITIVE_AGENT_RUNTIME_ENV_PATTERN.sub(r"\g<name>=<redacted>", command)
-    redacted = SENSITIVE_AGENT_RUNTIME_ARGUMENT_PATTERN.sub(r"\g<name> <redacted>", redacted)
-    return SENSITIVE_FILE_HEREDOC_PATTERN.sub(r"\g<prefix><redacted>\g<suffix>", redacted)
+    redacted = command
+    for pattern, substitution in (
+        (SENSITIVE_SANDBOX_ENV_PATTERN, r"\g<name>=<redacted>"),
+        (SENSITIVE_AGENT_RUNTIME_ARGUMENT_PATTERN, r"\g<name> <redacted>"),
+        (SENSITIVE_FILE_HEREDOC_PATTERN, r"\g<prefix><redacted>\g<suffix>"),
+        (GITHUB_CLONE_TOKEN_PATTERN, r"\g<prefix><redacted>\g<suffix>"),
+    ):
+        redacted = pattern.sub(substitution, redacted)
+    return redacted
 
 
 def build_agent_runtime_env_prefix(

products/tasks/backend/logic/services/test_docker_sandbox.py

@@ -126,7 +126,7 @@ def test_redact_sandbox_command_hides_event_ingest_token(self, command):
         assert "secret token" not in redacted
         assert "POSTHOG_TASK_RUN_EVENT_INGEST_TOKEN=<redacted>" in redacted
 
-    def test_redact_sandbox_command_hides_mcp_credentials(self):
+    def test_redact_sandbox_command_hides_mcp_credentials(self) -> None:
         command = 'agent-server --mcpServers \'[{"headers":{"Authorization":"Bearer secret-token"}}]\' --port 8080'
 
         redacted = redact_sandbox_command(command)
@@ -141,6 +141,21 @@ def test_redact_sandbox_command_hides_file_heredoc(self):
         assert "c2VjcmV0" not in redacted
         assert "<redacted>" in redacted
 
+    def test_redact_sandbox_command_hides_github_clone_token(self) -> None:
+        command = "git clone https://x-access-token:github-secret@github.com/PostHog/posthog.git"
+
+        redacted = redact_sandbox_command(command)
+
+        assert "github-secret" not in redacted
+        assert "https://x-access-token:<redacted>@github.com/PostHog/posthog.git" in redacted
+
+    @pytest.mark.parametrize("name", ("GITHUB_TOKEN", "POSTHOG_WIZARD_API_KEY"))
+    def test_redact_sandbox_command_hides_worker_credentials(self, name: str) -> None:
+        redacted = redact_sandbox_command(f"docker run -e {name}=worker-secret sandbox")
+
+        assert "worker-secret" not in redacted
+        assert f"{name}=<redacted>" in redacted
+
     @pytest.mark.parametrize(
         "input_url,expected_url",
         [

products/wizard/backend/facade/api.py

@@ -16,19 +16,27 @@
 from products.wizard.backend import metrics
 from products.wizard.backend.facade.contracts import (
     CreatePullRequestArtifactInput,
+    CreateWizardRunInput,
+    ListWizardRunsInput,
     UpsertWizardSessionInput,
     WizardProgram,
     WizardRunArtifactDTO,
+    WizardRunCreationResult,
+    WizardRunDTO,
     WizardRunGitDiffArtifactDTO,
+    WizardRunPage,
     WizardRunPullRequestArtifactDTO,
     WizardSessionDTO,
 )
+from products.wizard.backend.facade.enums import WizardRunStage, WizardRunStatus
+from products.wizard.backend.facade.validation import validate_wizard_version as validate_wizard_version_value
 from products.wizard.backend.logic import (
-    pubsub,
     registry as registry_service,
+    runs as run_service,
     sessions,
 )
 from products.wizard.backend.logic.artifacts import service as artifacts
+from products.wizard.backend.logic.sessions import pubsub
 
 
 def upsert(params: UpsertWizardSessionInput) -> tuple[WizardSessionDTO, bool]:
@@ -91,6 +99,40 @@ def get_registry(*, distinct_id: str, organization_id: str) -> tuple[WizardProgr
     return registry_service.get_registry(distinct_id=distinct_id, organization_id=organization_id)
 
 
+def create_run(params: CreateWizardRunInput) -> WizardRunDTO:
+    return run_service.create_run(params)
+
+
+def create_run_with_result(params: CreateWizardRunInput) -> WizardRunCreationResult:
+    return run_service.create_run_with_result(params)
+
+
+def get_run(team_id: int, run_id: UUID) -> WizardRunDTO:
+    return run_service.get_run(team_id, run_id)
+
+
+def list_runs(params: ListWizardRunsInput) -> WizardRunPage:
+    return run_service.list_runs(params)
+
+
+def update_run_stage(team_id: int, run_id: UUID, stage: WizardRunStage) -> WizardRunDTO:
+    return run_service.update_run_stage(team_id, run_id, stage)
+
+
+def cancel_run(team_id: int, run_id: UUID) -> WizardRunDTO:
+    return run_service.cancel_run(team_id, run_id)
+
+
+def update_run_status(
+    team_id: int,
+    run_id: UUID,
+    status: WizardRunStatus,
+    *,
+    error_code: str | None = None,
+) -> WizardRunDTO:
+    return run_service.transition_run(team_id, run_id, status, error_code=error_code)
+
+
 def create_git_diff_artifact(team_id: int, run_id: UUID, content: bytes) -> WizardRunGitDiffArtifactDTO | None:
     """
     Uploads a git diff generated in a Wizard run.
@@ -117,3 +159,11 @@ def get_git_diff_artifact_content(team_id: int, run_id: UUID, artifact_id: UUID)
     Returns the content of a git diff artifact generated in a Wizard run.
     """
     return artifacts.get_git_diff_artifact_content(team_id, run_id, artifact_id)
+
+
+def validate_git_repository(repository: str) -> None:
+    run_service.validate_git_repository_name(repository)
+
+
+def validate_wizard_version(wizard_version: object) -> str:
+    return validate_wizard_version_value(wizard_version)

products/wizard/backend/facade/temporal.py

@@ -0,0 +1,3 @@
+from products.wizard.backend.temporal import ACTIVITIES, WORKFLOWS
+
+__all__ = ["ACTIVITIES", "WORKFLOWS"]

products/wizard/backend/logic/runs/__init__.py

@@ -0,0 +1,27 @@
+from products.wizard.backend.logic.runs.lifecycle import (
+    cancel_run,
+    complete_run,
+    create_run,
+    create_run_with_result,
+    fail_run,
+    get_run,
+    list_runs,
+    start_run,
+    transition_run,
+    update_run_stage,
+)
+from products.wizard.backend.logic.runs.validation import validate_git_repository_name
+
+__all__ = [
+    "cancel_run",
+    "complete_run",
+    "create_run",
+    "create_run_with_result",
+    "fail_run",
+    "get_run",
+    "list_runs",
+    "start_run",
+    "transition_run",
+    "update_run_stage",
+    "validate_git_repository_name",
+]