From 8b3b3c9ec431d4c52116ae4b35a733d4f32722c2 Mon Sep 17 00:00:00 2001 From: Peter Kirkham Date: Wed, 22 Jul 2026 13:34:10 +0200 Subject: [PATCH] chore(tasks): remove code home, workstream grouping, and auto runs Deletes the server side of the PostHog Code Home Tab: the code_home/code_workflow viewsets and routes, the code_workstreams logic and Temporal packages (workflows, activities, 3-minute schedule, on-demand trigger), the CodeWorkflowConfig/CodePrSnapshot/CodeWorkstream models (state-only migration; tables dropped in a follow-up), the home_quick_action run field, and the diagnostic management command. Regenerates OpenAPI-derived frontend/MCP types. Generated-By: PostHog Code Task-Id: 98bddaa7-c8c3-4e21-aa95-9712d59f2404 --- .../security/idor-team-scoped-models.yaml | 6 - posthog/temporal/schedule.py | 2 - products/tasks/backend/facade/api.py | 216 ----------- products/tasks/backend/facade/contracts.py | 96 ----- products/tasks/backend/facade/temporal.py | 7 +- .../logic/code_workstreams/__init__.py | 0 .../logic/code_workstreams/classify.py | 76 ---- .../code_workstreams/default_workflow.py | 51 --- .../logic/code_workstreams/grouping.py | 238 ------------- .../logic/code_workstreams/situations.py | 29 -- .../logic/code_workstreams/test_classify.py | 100 ------ .../logic/code_workstreams/test_grouping.py | 263 -------------- .../logic/code_workstreams/test_validation.py | 63 ---- .../logic/code_workstreams/validation.py | 101 ------ .../commands/evaluate_code_workstreams.py | 128 ------- .../0069_remove_code_home_models.py | 20 ++ .../backend/migrations/max_migration.txt | 2 +- products/tasks/backend/models.py | 115 ------ .../tasks/backend/presentation/serializers.py | 7 - .../presentation/views/code_home_api.py | 142 -------- products/tasks/backend/routes.py | 3 - products/tasks/backend/temporal/__init__.py | 13 - .../temporal/code_workstreams/__init__.py | 0 .../code_workstreams/activities/__init__.py | 0 .../activities/discover_branch_prs.py | 173 --------- .../activities/github_resolution.py | 74 ---- .../activities/list_active_teams.py | 80 ----- .../activities/load_pr_urls.py | 76 ---- .../activities/poll_pull_requests.py | 141 -------- .../activities/rebuild_workstreams.py | 335 ------------------ .../temporal/code_workstreams/client.py | 30 -- .../temporal/code_workstreams/constants.py | 18 - .../temporal/code_workstreams/schedule.py | 37 -- .../test_discover_branch_prs.py | 170 --------- .../test_list_active_teams.py | 152 -------- .../code_workstreams/test_load_pr_urls.py | 66 ---- .../test_poll_pull_requests.py | 41 --- .../test_rebuild_workstreams.py | 310 ---------------- .../code_workstreams/test_workflow.py | 158 --------- .../temporal/code_workstreams/workflow.py | 140 -------- .../backend/temporal/process_task/utils.py | 1 - .../tasks/frontend/generated/api.schemas.ts | 5 - products/tasks/frontend/generated/api.zod.ts | 9 - services/mcp/src/api/generated.ts | 5 - 44 files changed, 23 insertions(+), 3676 deletions(-) delete mode 100644 products/tasks/backend/logic/code_workstreams/__init__.py delete mode 100644 products/tasks/backend/logic/code_workstreams/classify.py delete mode 100644 products/tasks/backend/logic/code_workstreams/default_workflow.py delete mode 100644 products/tasks/backend/logic/code_workstreams/grouping.py delete mode 100644 products/tasks/backend/logic/code_workstreams/situations.py delete mode 100644 products/tasks/backend/logic/code_workstreams/test_classify.py delete mode 100644 products/tasks/backend/logic/code_workstreams/test_grouping.py delete mode 100644 products/tasks/backend/logic/code_workstreams/test_validation.py delete mode 100644 products/tasks/backend/logic/code_workstreams/validation.py delete mode 100644 products/tasks/backend/management/commands/evaluate_code_workstreams.py create mode 100644 products/tasks/backend/migrations/0069_remove_code_home_models.py delete mode 100644 products/tasks/backend/presentation/views/code_home_api.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/__init__.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/activities/__init__.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/activities/discover_branch_prs.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/activities/github_resolution.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/activities/list_active_teams.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/activities/load_pr_urls.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/activities/poll_pull_requests.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/activities/rebuild_workstreams.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/client.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/constants.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/schedule.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/test_discover_branch_prs.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/test_list_active_teams.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/test_load_pr_urls.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/test_poll_pull_requests.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/test_rebuild_workstreams.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/test_workflow.py delete mode 100644 products/tasks/backend/temporal/code_workstreams/workflow.py diff --git a/.semgrep/rules/security/idor-team-scoped-models.yaml b/.semgrep/rules/security/idor-team-scoped-models.yaml index f28f9572e94a..93a0d87d2f3b 100644 --- a/.semgrep/rules/security/idor-team-scoped-models.yaml +++ b/.semgrep/rules/security/idor-team-scoped-models.yaml @@ -84,9 +84,6 @@ rules: |ChangeRequest |ClusteringConfig |ClusteringJob - |CodePrSnapshot - |CodeWorkflowConfig - |CodeWorkstream |Cohort |CohortBackfillChunk |CohortBackfillRun @@ -397,9 +394,6 @@ rules: |ChangeRequest |ClusteringConfig |ClusteringJob - |CodePrSnapshot - |CodeWorkflowConfig - |CodeWorkstream |Cohort |CohortBackfillChunk |CohortBackfillRun diff --git a/posthog/temporal/schedule.py b/posthog/temporal/schedule.py index b48b12dc97c5..683632903968 100644 --- a/posthog/temporal/schedule.py +++ b/posthog/temporal/schedule.py @@ -101,7 +101,6 @@ from products.replay_vision.backend.temporal.reconciler import create_replay_vision_reconciler_schedule from products.signals.backend.emission.conversations_schedule import create_conversations_signals_coordinator_schedule from products.signals.backend.temporal.agentic.schedule import create_signals_scout_coordinator_schedule -from products.tasks.backend.facade.temporal import create_evaluate_code_workstreams_schedule from products.web_analytics.backend.temporal.digest_notification.types import WADigestNotificationInput from products.web_analytics.backend.temporal.weekly_digest.types import WAWeeklyDigestInput @@ -835,7 +834,6 @@ async def create_error_tracking_recommendations_refresh_schedule(client: Client) create_support_reply_coordinator_schedule, create_replay_vision_reconciler_schedule, create_replay_vision_estimates_schedule, - create_evaluate_code_workstreams_schedule, create_github_job_logs_coordinator_schedule, ] diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py index 44bd9da16e48..a047ea44abd9 100644 --- a/products/tasks/backend/facade/api.py +++ b/products/tasks/backend/facade/api.py @@ -42,8 +42,6 @@ is_blocked_sandbox_env_key, ) from products.tasks.backend.error_telemetry import truncate_error_message -from products.tasks.backend.logic.code_workstreams.default_workflow import build_default_bindings -from products.tasks.backend.logic.code_workstreams.validation import validate_bindings from products.tasks.backend.logic.services.image_builder import ( ensure_image_builder_task, is_custom_images_enabled, @@ -55,8 +53,6 @@ ChannelFeedMessage, CodeInvite, CodeInviteRedemption, - CodeWorkflowConfig, - CodeWorkstream, SandboxCustomImage, SandboxEnvironment, SandboxSnapshot, @@ -91,26 +87,12 @@ CODE_INVITE_INVALID_CODE = "invalid_code" CODE_INVITE_NOT_REDEEMABLE = "not_redeemable" -# --- Code-workflow save outcomes --- -# Returned on ``CodeWorkflowSaveResult.outcome``; the presentation layer maps each to an -# HTTP status (saved -> 200, conflict -> 409, invalid -> 422). -CODE_WORKFLOW_SAVED = "saved" -CODE_WORKFLOW_CONFLICT = "conflict" -CODE_WORKFLOW_INVALID = "invalid" - -# --- Code-home tuning --- -# An agent run counts as "active" only if it updated within this window. -CODE_HOME_ACTIVE_AGENT_WINDOW = timedelta(minutes=30) -_CODE_HOME_RUNNING_STATUSES = (TaskRun.Status.QUEUED, TaskRun.Status.IN_PROGRESS) WIZARD_PR_READY_EMAIL_FEATURE_FLAG = "wizard-cloud-run-pr-ready-email-enabled" __all__ = [ "CODE_INVITE_INVALID_CODE", "CODE_INVITE_NOT_REDEEMABLE", "CODE_INVITE_REDEEMED", - "CODE_WORKFLOW_CONFLICT", - "CODE_WORKFLOW_INVALID", - "CODE_WORKFLOW_SAVED", "SandboxNetworkAccessLevel", "SandboxSnapshotStatus", "TaskOriginProduct", @@ -149,8 +131,6 @@ "finalize_task_run_artifact_uploads", "finalize_task_staged_artifacts", "get_active_wizard_cloud_run", - "get_code_home", - "get_code_workflow_config", "get_conversation_task_dtos", "get_latest_pr_url_by_task", "get_merged_pr_task_ids", @@ -192,14 +172,11 @@ "read_task_run_logs", "redeem_code_invite", "redispatch_task_run", - "refresh_team_code_workstreams", "relay_task_run_message", - "reset_code_workflow_bindings", "resolve_slack_thread_context", "resume_task_run_in_cloud", "run_task", "run_task_automation_now", - "save_code_workflow_bindings", "send_cancel", "select_repository_for_message", "set_task_run_output", @@ -3014,7 +2991,6 @@ def bootstrap_task_run( reasoning_effort = validated_data.get("reasoning_effort") github_user_token = validated_data.get("github_user_token") initial_permission_mode = validated_data.get("initial_permission_mode") - home_quick_action = validated_data.get("home_quick_action") imported_mcp_servers = validated_data.get("imported_mcp_servers") relayed_mcp_servers = validated_data.get("relayed_mcp_servers") if run_source == RunSource.SIGNAL_REPORT: @@ -3035,7 +3011,6 @@ def bootstrap_task_run( "provider": provider, "model": model, "reasoning_effort": reasoning_effort, - "home_quick_action": home_quick_action, "rtk_enabled": validated_data.get("rtk_enabled"), }.items(): if value is not None: @@ -4796,197 +4771,6 @@ def resolve_slack_thread_context( return contracts.SlackThreadContextResult(outcome="ok", context=context) -# --- Code workflow config (presentation CRUD) --- -# A user's per-team binding configuration. Reads seed a default config on first access; -# saves are optimistic-locked on ``version`` and validate the bindings before persisting. - - -def _epoch_ms(dt: datetime) -> int: - return int(dt.timestamp() * 1000) - - -def _code_workflow_config_to_dto(config: CodeWorkflowConfig) -> contracts.CodeWorkflowConfigDTO: - return contracts.CodeWorkflowConfigDTO( - id=str(config.id), - version=config.version, - updated_at=config.updated_at, - bindings=config.bindings, - ) - - -def get_code_workflow_config(team_id: int, user_id: int) -> contracts.CodeWorkflowConfigDTO: - """Return the user's config for the team, seeding a default one on first access.""" - config, _ = CodeWorkflowConfig.objects.get_or_create( - team_id=team_id, - user_id=user_id, - defaults={"bindings": build_default_bindings(), "version": 1}, - ) - return _code_workflow_config_to_dto(config) - - -def save_code_workflow_bindings( - team_id: int, user_id: int, *, bindings: dict, expected_version: object -) -> contracts.CodeWorkflowSaveResult: - """Validate and save bindings under optimistic locking. - - Returns a ``conflict`` result when ``expected_version`` is not an int or does not match - the stored version, an ``invalid`` result (with diagnostics) when validation fails, or a - ``saved`` result with the version-bumped config. - """ - with transaction.atomic(): - current, _ = CodeWorkflowConfig.objects.select_for_update().get_or_create( - team_id=team_id, - user_id=user_id, - defaults={"bindings": build_default_bindings(), "version": 1}, - ) - if not isinstance(expected_version, int) or current.version != expected_version: - return contracts.CodeWorkflowSaveResult( - outcome=CODE_WORKFLOW_CONFLICT, - config=_code_workflow_config_to_dto(current), - ) - - result = validate_bindings(bindings) - if not result.can_save: - return contracts.CodeWorkflowSaveResult( - outcome=CODE_WORKFLOW_INVALID, - config=_code_workflow_config_to_dto(current), - diagnostics=[ - contracts.CodeWorkflowDiagnosticDTO( - severity=d.severity, - code=d.code, - message=d.message, - situation_id=d.situation_id, - action_id=d.action_id, - ) - for d in result.diagnostics - ], - ) - - current.bindings = bindings - current.version = current.version + 1 - current.save(update_fields=["bindings", "version", "updated_at"]) - - return contracts.CodeWorkflowSaveResult( - outcome=CODE_WORKFLOW_SAVED, - config=_code_workflow_config_to_dto(current), - ) - - -def reset_code_workflow_bindings(team_id: int, user_id: int) -> contracts.CodeWorkflowConfigDTO: - """Reset the user's bindings back to the defaults and bump the version.""" - with transaction.atomic(): - config, _ = CodeWorkflowConfig.objects.select_for_update().get_or_create( - team_id=team_id, - user_id=user_id, - defaults={"bindings": build_default_bindings(), "version": 1}, - ) - config.bindings = build_default_bindings() - config.version = config.version + 1 - config.save(update_fields=["bindings", "version", "updated_at"]) - return _code_workflow_config_to_dto(config) - - -# --- Code home board --- -# Active agents are computed live off in-flight runs; workstreams are persisted by the -# worker and split into board columns by their stored ``state``. - - -def _code_home_workstream_to_dto(ws: CodeWorkstream) -> contracts.CodeHomeWorkstreamDTO: - return contracts.CodeHomeWorkstreamDTO( - id=ws.key, - repo_name=ws.repo_name, - repo_full_path=ws.repo_full_path, - branch=ws.branch, - pr_url=ws.pr_url, - pr=ws.pr, - primary_situation=ws.primary_situation, - last_activity_at=_epoch_ms(ws.last_activity_at), - tasks=[ - contracts.CodeHomeWorkstreamTaskDTO( - id=t.get("id"), - title=t.get("title"), - status=t.get("status"), - is_generating=False, - needs_permission=False, - quick_action=t.get("quick_action"), - ) - for t in (ws.tasks or []) - ], - situations=ws.situations or [], - ) - - -def _code_home_active_agents(team_id: int, user_id: int) -> list[contracts.CodeHomeActiveAgentDTO]: - cutoff = django_timezone.now() - CODE_HOME_ACTIVE_AGENT_WINDOW - runs = ( - TaskRun.objects.filter( - team_id=team_id, - task__created_by_id=user_id, - task__archived=False, - task__deleted=False, - status__in=_CODE_HOME_RUNNING_STATUSES, - updated_at__gte=cutoff, - ) - .select_related("task") - .order_by("-updated_at") - ) - - seen_tasks: set[str] = set() - agents: list[contracts.CodeHomeActiveAgentDTO] = [] - for run in runs.iterator(): - task = run.task - if str(task.id) in seen_tasks: - continue - if (run.output or {}).get("pr_url"): - continue - seen_tasks.add(str(task.id)) - agents.append( - contracts.CodeHomeActiveAgentDTO( - task_id=str(task.id), - title=task.title, - repo_name=task.repository.split("/")[-1] if task.repository else None, - branch=run.branch, - status=run.status, - last_activity_at=_epoch_ms(run.updated_at), - needs_permission=False, - cloud_pr_url=None, - ) - ) - return agents - - -def get_code_home(team_id: int, user_id: int) -> contracts.CodeHomeDTO: - """Assemble the code-home board: live active agents plus persisted workstreams by column.""" - workstreams = CodeWorkstream.objects.filter(team_id=team_id, user_id=user_id) - needs_attention: list[contracts.CodeHomeWorkstreamDTO] = [] - in_progress: list[contracts.CodeHomeWorkstreamDTO] = [] - for ws in workstreams.iterator(): - dto = _code_home_workstream_to_dto(ws) - if ws.state == CodeWorkstream.WorkstreamState.ATTENTION: - needs_attention.append(dto) - else: - in_progress.append(dto) - - return contracts.CodeHomeDTO( - active_agents=_code_home_active_agents(team_id, user_id), - needs_attention=needs_attention, - in_progress=in_progress, - ) - - -def refresh_team_code_workstreams(team_id: int) -> bool: - """Trigger an on-demand evaluation of the team's code workstreams. - - Returns whether a new evaluation workflow was started (``False`` if one was already - running). - """ - from products.tasks.backend.temporal.code_workstreams.client import ( # noqa: PLC0415 — keep temporalio off the api import path - trigger_team_code_workstreams_evaluation, - ) - - return trigger_team_code_workstreams_evaluation(team_id) - - # --- Id-based bridges to the sandbox/agent-command surface --- # These take a run id (not an ORM TaskRun) so callers never hold a tasks model. The heavy # service modules are imported lazily to keep them off this module's import path. diff --git a/products/tasks/backend/facade/contracts.py b/products/tasks/backend/facade/contracts.py index 1d7ae87431a3..7d6a3ba54c83 100644 --- a/products/tasks/backend/facade/contracts.py +++ b/products/tasks/backend/facade/contracts.py @@ -520,102 +520,6 @@ class TaskAutomationDTO: updated_at: datetime -@dataclass(frozen=True) -class CodeWorkflowConfigDTO: - """A user's per-team code-workflow binding configuration. - - Mirrors exactly the JSON shape the code-workflow endpoints emit: ``id`` and - ``updatedAt`` are stringified, ``version`` powers optimistic locking, and - ``bindings`` is the situation-id → ordered action-list mapping. - """ - - id: str - version: int - updated_at: datetime - bindings: dict - - -@dataclass(frozen=True) -class CodeWorkflowDiagnosticDTO: - """One binding-validation diagnostic. - - Mirrors a ``ValidationDiagnostic``; ``situation_id`` / ``action_id`` are present only - when the diagnostic is scoped to a specific situation or action. - """ - - severity: str - code: str - message: str - situation_id: str | None = None - action_id: str | None = None - - -@dataclass(frozen=True) -class CodeWorkflowSaveResult: - """Outcome of attempting to save code-workflow bindings. - - ``outcome`` is one of ``saved`` (bindings persisted, version bumped), ``conflict`` - (``expected_version`` did not match the stored version), or ``invalid`` (validation - failed). ``config`` is always the resulting/current config; ``diagnostics`` is only - populated on the ``invalid`` outcome. - """ - - outcome: str - config: CodeWorkflowConfigDTO - diagnostics: list[CodeWorkflowDiagnosticDTO] = Field(default_factory=list) - - -@dataclass(frozen=True) -class CodeHomeWorkstreamTaskDTO: - """One grouped task inside a workstream card.""" - - id: str | None - title: str | None - status: str | None - is_generating: bool = False - needs_permission: bool = False - quick_action: str | None = None - - -@dataclass(frozen=True) -class CodeHomeWorkstreamDTO: - """A persisted workstream card for the code-home board.""" - - id: str - repo_name: str | None - repo_full_path: str | None - branch: str | None - pr_url: str | None - pr: dict | None - primary_situation: str | None - last_activity_at: int - tasks: list[CodeHomeWorkstreamTaskDTO] = Field(default_factory=list) - situations: list = Field(default_factory=list) - - -@dataclass(frozen=True) -class CodeHomeActiveAgentDTO: - """A live, in-flight agent run shown on the code-home board.""" - - task_id: str - title: str - repo_name: str | None - branch: str | None - status: str - last_activity_at: int - needs_permission: bool = False - cloud_pr_url: str | None = None - - -@dataclass(frozen=True) -class CodeHomeDTO: - """The full code-home board: live agents plus persisted workstreams by column.""" - - active_agents: list[CodeHomeActiveAgentDTO] = Field(default_factory=list) - needs_attention: list[CodeHomeWorkstreamDTO] = Field(default_factory=list) - in_progress: list[CodeHomeWorkstreamDTO] = Field(default_factory=list) - - @dataclass(frozen=True) class TaskUserBasicInfo: """Lightweight user info for display, mirroring core ``UserBasicSerializer`` output. diff --git a/products/tasks/backend/facade/temporal.py b/products/tasks/backend/facade/temporal.py index 2b94009b8f0b..3caf36972471 100644 --- a/products/tasks/backend/facade/temporal.py +++ b/products/tasks/backend/facade/temporal.py @@ -3,9 +3,8 @@ Wiring that core registers and dispatches on, as objects: the worker registers ``WORKFLOWS``/``ACTIVITIES`` and the workflow class, the common worker reads the histogram -metric config, the schedule bootstrap creates the code-workstreams schedule, and callers -trigger a run via ``execute_task_processing_workflow``. Isolated from ``facade/api.py`` so -``temporalio`` never lands on the light data-surface import path. +metric config, and callers trigger a run via ``execute_task_processing_workflow``. Isolated +from ``facade/api.py`` so ``temporalio`` never lands on the light data-surface import path. """ from products.tasks.backend.temporal import ACTIVITIES, WORKFLOWS @@ -16,7 +15,6 @@ resume_task_in_cloud_workflow, signal_task_followup_message, ) -from products.tasks.backend.temporal.code_workstreams.schedule import create_evaluate_code_workstreams_schedule from products.tasks.backend.temporal.metrics import ( TASKS_LATENCY_HISTOGRAM_BUCKETS, TASKS_LATENCY_HISTOGRAM_METRICS, @@ -38,7 +36,6 @@ "WORKFLOWS", "PostSlackUpdateInput", "ProcessTaskWorkflow", - "create_evaluate_code_workstreams_schedule", "execute_posthog_code_agent_relay_workflow", "execute_task_processing_workflow", "execute_task_processing_workflow_async", diff --git a/products/tasks/backend/logic/code_workstreams/__init__.py b/products/tasks/backend/logic/code_workstreams/__init__.py deleted file mode 100644 index e69de29bb2d1..000000000000 diff --git a/products/tasks/backend/logic/code_workstreams/classify.py b/products/tasks/backend/logic/code_workstreams/classify.py deleted file mode 100644 index cc35f8a5b23b..000000000000 --- a/products/tasks/backend/logic/code_workstreams/classify.py +++ /dev/null @@ -1,76 +0,0 @@ -from collections.abc import Iterable -from dataclasses import dataclass -from typing import Literal, Optional - -from .situations import SITUATION_PRIORITY, SituationId - -STALE_THRESHOLD_MS = 7 * 24 * 60 * 60 * 1000 - -PrState = Literal["open", "draft", "merged", "closed"] -CiStatus = Literal["passing", "failing", "pending", "none"] -ReviewDecision = Literal["approved", "changes_requested", "review_required"] - - -@dataclass -class ClassifyPr: - state: PrState - ci_status: CiStatus - review_decision: Optional[ReviewDecision] - unresolved_threads: int - is_current_user_author: bool - mergeable: Optional[bool] = None - - -@dataclass -class ClassifyInput: - has_pr_url: bool - pr: Optional[ClassifyPr] - branch: Optional[str] - last_activity_at: int - now: int - commits_ahead: Optional[int] = None - - -def classify(input: ClassifyInput) -> set[SituationId]: - out: set[SituationId] = set() - pr = input.pr - - if pr is not None: - if pr.state in ("merged", "closed"): - out.add("done") - return out - - if pr.ci_status == "failing": - out.add("ci_failing") - if pr.review_decision == "changes_requested": - out.add("changes_requested") - if pr.unresolved_threads > 0 and pr.is_current_user_author: - out.add("comments_waiting") - if ( - pr.state == "open" - and pr.ci_status == "passing" - and pr.review_decision == "approved" - and pr.mergeable is not False - ): - out.add("ready_to_merge") - if pr.state in ("open", "draft"): - out.add("in_review") - elif input.has_pr_url: - out.add("in_review") - elif input.branch: - ahead = input.commits_ahead - if ahead is None or ahead > 0: - out.add("working") - - if input.now - input.last_activity_at > STALE_THRESHOLD_MS: - out.add("stale") - - return out - - -def pick_primary_situation(situations: Iterable[SituationId]) -> Optional[SituationId]: - present = set(situations) - for sid in SITUATION_PRIORITY: - if sid in present: - return sid - return None diff --git a/products/tasks/backend/logic/code_workstreams/default_workflow.py b/products/tasks/backend/logic/code_workstreams/default_workflow.py deleted file mode 100644 index 677d8c507b56..000000000000 --- a/products/tasks/backend/logic/code_workstreams/default_workflow.py +++ /dev/null @@ -1,51 +0,0 @@ -from .situations import SITUATION_IDS - -_DEFAULT_BINDINGS: dict[str, list[dict]] = { - "working": [ - { - "id": "create_pr", - "label": "Create PR", - "skillId": "", - "prompt": "Open a PR for the current branch. Use the task history to write a concise description.", - }, - ], - "in_review": [], - "ci_failing": [ - { - "id": "fix_ci", - "label": "Fix CI", - "skillId": "", - "prompt": "CI is failing on this PR. Investigate the failing checks and push a fix.", - }, - ], - "changes_requested": [ - { - "id": "address_comments", - "label": "Address review", - "skillId": "", - "prompt": "Address the change requests on this PR — read the latest review and respond with code.", - }, - ], - "comments_waiting": [ - { - "id": "address_threads", - "label": "Address comments", - "skillId": "", - "prompt": "Address the unresolved review comments on this PR.", - }, - ], - "ready_to_merge": [ - { - "id": "final_check", - "label": "Final check", - "skillId": "", - "prompt": "Do a last-pass review of this PR. Call out anything risky before I merge.", - }, - ], - "stale": [], - "done": [], -} - - -def build_default_bindings() -> dict[str, list[dict]]: - return {sid: [dict(action) for action in _DEFAULT_BINDINGS.get(sid, [])] for sid in SITUATION_IDS} diff --git a/products/tasks/backend/logic/code_workstreams/grouping.py b/products/tasks/backend/logic/code_workstreams/grouping.py deleted file mode 100644 index cec5a34a0e68..000000000000 --- a/products/tasks/backend/logic/code_workstreams/grouping.py +++ /dev/null @@ -1,238 +0,0 @@ -from collections.abc import Mapping -from dataclasses import dataclass, field -from typing import Optional - -from .classify import ClassifyInput, ClassifyPr, classify -from .situations import ATTENTION_SITUATIONS, SituationId - -RUNNING_STATUSES = frozenset({"queued", "in_progress"}) - -RUNNING_STALE_THRESHOLD_MS = 30 * 60 * 1000 - -# A normal PR is never headed from a repo's default branch, so these are always treated as base -# branches when resolving a run to a PR by branch (see build_workstreams). If this ever needs to -# cover non-standard defaults, derive the branch from the GitHub integration rather than growing -# this set. -DEFAULT_BASE_BRANCHES = frozenset({"master", "main"}) - - -@dataclass -class TaskInput: - id: str - title: str - status: Optional[str] - last_activity_at: int - repo_name: Optional[str] = None - repo_full_path: Optional[str] = None - branch: Optional[str] = None - base_branch: Optional[str] = None - cloud_pr_url: Optional[str] = None - cloud_pr_merged: bool = False - folder_path: Optional[str] = None - quick_action: Optional[str] = None - - -@dataclass -class PrInput: - url: str - number: int - title: str - state: str - ci_status: str - review_decision: Optional[str] - unresolved_threads: int - mergeable: Optional[bool] - is_current_user_requested_reviewer: bool - is_current_user_author: bool - author: Optional[str] - last_updated_at: int - head_branch: Optional[str] = None - - -@dataclass -class WorkstreamTask: - id: str - title: str - status: Optional[str] - quick_action: Optional[str] = None - - -@dataclass -class Workstream: - id: str - repo_name: Optional[str] - repo_full_path: Optional[str] - branch: Optional[str] - pr_url: Optional[str] - pr: Optional[PrInput] - tasks: list[WorkstreamTask] - situations: list[SituationId] - last_activity_at: int - - -@dataclass -class WorkstreamsResult: - needs_attention: list[Workstream] = field(default_factory=list) - in_progress: list[Workstream] = field(default_factory=list) - - -def _is_running(status: Optional[str]) -> bool: - return bool(status) and status in RUNNING_STATUSES - - -def _is_actively_running(task: TaskInput, now: int, has_pr: bool) -> bool: - if not _is_running(task.status): - return False - if has_pr: - return False - return now - task.last_activity_at <= RUNNING_STALE_THRESHOLD_MS - - -def _feature_branch(task: TaskInput) -> Optional[str]: - # `TaskRun.branch` starts as the run's base branch (e.g. "master") and is only synced to a - # real feature branch once the agent pushes one. Grouping on the base branch would collapse - # every no-PR run that never pushed into one blob, so only treat a distinct branch as real. - branch = task.branch - if branch and branch != task.base_branch: - return branch - return None - - -def workstream_key(task: TaskInput, pr_url: Optional[str]) -> Optional[str]: - if pr_url: - return f"pr:{pr_url}" - repo = task.repo_full_path or task.repo_name - branch = _feature_branch(task) - if repo and branch: - return f"branch:{repo}#{branch}" - if task.folder_path: - return f"path:{task.folder_path}" - return None - - -def branch_lookup_key(repo: Optional[str], branch: Optional[str]) -> Optional[tuple[str, str]]: - if repo and branch: - return (repo.casefold(), branch) - return None - - -def build_workstreams( - tasks: list[TaskInput], - pr_by_task: Mapping[str, PrInput], - now: int, - pr_by_branch: Optional[Mapping[tuple[str, str], PrInput]] = None, -) -> WorkstreamsResult: - # A run sitting on a base branch must not resolve to whatever PR is headed from it. A branch is - # a base branch if some sibling task uses it as a base while on a *different* branch, or if it's - # a conventional default (folded in so a *lone* base-branch run, with no sibling to mark it, is - # still caught). A feature-branch follow-up has branch == base_branch, so it is never flagged - # and still resolves — real feature branches are never named like the defaults. - base_branches = { - t.base_branch for t in tasks if t.base_branch and t.base_branch != t.branch - } | DEFAULT_BASE_BRANCHES - - def pr_of(task: TaskInput) -> Optional[PrInput]: - # A task's own PR wins; otherwise resolve via the branch it ran on so a - # follow-up run (e.g. "Fix CI") that pushed to an open PR's branch but - # never recorded a pr_url still groups under that PR's workstream. - own = pr_by_task.get(task.id) - if own is not None: - return own - if pr_by_branch and task.branch not in base_branches: - key = branch_lookup_key(task.repo_full_path or task.repo_name, task.branch) - if key is not None: - return pr_by_branch.get(key) - return None - - def pr_url_of(task: TaskInput) -> Optional[str]: - snap = pr_of(task) - if snap is not None: - return snap.url - return task.cloud_pr_url - - groups: dict[str, list[TaskInput]] = {} - - for task in tasks: - pr_url = pr_url_of(task) - if _is_actively_running(task, now, bool(pr_url)): - continue - key = workstream_key(task, pr_url) - if not key: - continue - groups.setdefault(key, []).append(task) - - needs_attention: list[Workstream] = [] - in_progress: list[Workstream] = [] - - for key, group_tasks in groups.items(): - group_tasks.sort(key=lambda t: t.last_activity_at, reverse=True) - head = group_tasks[0] - - pr: Optional[PrInput] = None - pr_url = None - for t in group_tasks: - snap = pr_of(t) - url = snap.url if snap is not None else t.cloud_pr_url - if url: - pr = snap - pr_url = url - break - - if any(t.cloud_pr_merged and t.cloud_pr_url == pr_url for t in group_tasks): - continue - - branch = _feature_branch(head) - last_activity_at = head.last_activity_at - - situations = sorted( - classify( - ClassifyInput( - has_pr_url=bool(pr_url), - pr=_to_classify_pr(pr), - branch=branch, - last_activity_at=last_activity_at, - now=now, - ) - ) - ) - - if "done" in situations: - continue - - workstream = Workstream( - id=key, - repo_name=head.repo_name, - repo_full_path=head.repo_full_path, - branch=branch, - pr_url=pr_url, - pr=pr, - tasks=[ - WorkstreamTask(id=t.id, title=t.title, status=t.status, quick_action=t.quick_action) - for t in group_tasks - ], - situations=situations, - last_activity_at=last_activity_at, - ) - - if any(s in ATTENTION_SITUATIONS for s in situations): - needs_attention.append(workstream) - else: - in_progress.append(workstream) - - needs_attention.sort(key=lambda w: w.last_activity_at, reverse=True) - in_progress.sort(key=lambda w: w.last_activity_at, reverse=True) - - return WorkstreamsResult(needs_attention=needs_attention, in_progress=in_progress) - - -def _to_classify_pr(pr: Optional[PrInput]) -> Optional[ClassifyPr]: - if pr is None: - return None - return ClassifyPr( - state=pr.state, # type: ignore[arg-type] - ci_status=pr.ci_status, # type: ignore[arg-type] - review_decision=pr.review_decision, # type: ignore[arg-type] - unresolved_threads=pr.unresolved_threads, - is_current_user_author=pr.is_current_user_author, - mergeable=pr.mergeable, - ) diff --git a/products/tasks/backend/logic/code_workstreams/situations.py b/products/tasks/backend/logic/code_workstreams/situations.py deleted file mode 100644 index fc97048c522b..000000000000 --- a/products/tasks/backend/logic/code_workstreams/situations.py +++ /dev/null @@ -1,29 +0,0 @@ -from typing import Literal, get_args - -SituationId = Literal[ - "working", - "in_review", - "ci_failing", - "changes_requested", - "comments_waiting", - "ready_to_merge", - "stale", - "done", -] - -SITUATION_IDS: tuple[SituationId, ...] = get_args(SituationId) - -SITUATION_PRIORITY: tuple[SituationId, ...] = ( - "done", - "ready_to_merge", - "ci_failing", - "changes_requested", - "comments_waiting", - "in_review", - "working", - "stale", -) - -ATTENTION_SITUATIONS: frozenset[SituationId] = frozenset( - {"ci_failing", "changes_requested", "comments_waiting", "stale"} -) diff --git a/products/tasks/backend/logic/code_workstreams/test_classify.py b/products/tasks/backend/logic/code_workstreams/test_classify.py deleted file mode 100644 index 75243ae6856d..000000000000 --- a/products/tasks/backend/logic/code_workstreams/test_classify.py +++ /dev/null @@ -1,100 +0,0 @@ -from dataclasses import replace - -import pytest - -from products.tasks.backend.logic.code_workstreams.classify import ( - STALE_THRESHOLD_MS, - ClassifyInput, - ClassifyPr, - classify, - pick_primary_situation, -) - -NOW = 1_700_000_000_000 - - -def _pr(**overrides) -> ClassifyPr: - return replace( - ClassifyPr( - state="open", - ci_status="passing", - review_decision=None, - unresolved_threads=0, - is_current_user_author=True, - mergeable=True, - ), - **overrides, - ) - - -def _input(**overrides) -> ClassifyInput: - return replace( - ClassifyInput( - has_pr_url=False, - pr=None, - branch=None, - last_activity_at=NOW, - now=NOW, - ), - **overrides, - ) - - -@pytest.mark.parametrize("state", ["merged", "closed"]) -def test_terminal_pr_state_is_done_and_exclusive(state): - assert classify(_input(pr=_pr(state=state, ci_status="failing"))) == {"done"} - - -def test_failing_ci_open_pr(): - assert classify(_input(pr=_pr(ci_status="failing"))) == {"ci_failing", "in_review"} - - -def test_changes_requested(): - result = classify(_input(pr=_pr(review_decision="changes_requested"))) - assert result == {"changes_requested", "in_review"} - - -@pytest.mark.parametrize("is_current_user_author,expected", [(True, True), (False, False)]) -def test_comments_waiting_only_for_author(is_current_user_author, expected): - result = classify(_input(pr=_pr(unresolved_threads=2, is_current_user_author=is_current_user_author))) - assert ("comments_waiting" in result) is expected - - -@pytest.mark.parametrize("mergeable,expected", [(True, True), (False, False)]) -def test_ready_to_merge_requires_mergeable(mergeable, expected): - result = classify(_input(pr=_pr(review_decision="approved", ci_status="passing", mergeable=mergeable))) - assert ("ready_to_merge" in result) is expected - - -def test_pr_url_without_data_is_in_review(): - assert classify(_input(has_pr_url=True, pr=None)) == {"in_review"} - - -@pytest.mark.parametrize("commits_ahead,expected", [(3, {"working"}), (None, {"working"}), (0, set())]) -def test_branch_with_commits_is_working(commits_ahead, expected): - assert classify(_input(branch="feat/x", commits_ahead=commits_ahead)) == expected - - -def test_stale_stacks_on_top(): - old = NOW - STALE_THRESHOLD_MS - 1 - result = classify(_input(pr=_pr(ci_status="failing"), last_activity_at=old)) - assert "stale" in result - assert "ci_failing" in result - - -def test_stale_never_stacks_on_done(): - old = NOW - STALE_THRESHOLD_MS - 1 - result = classify(_input(pr=_pr(state="merged"), last_activity_at=old)) - assert result == {"done"} - - -@pytest.mark.parametrize( - "situations,expected", - [ - ({"working", "ci_failing", "stale"}, "ci_failing"), - ({"stale", "working"}, "working"), - (set(), None), - ], -) -def test_pick_primary_situation_priority(situations, expected): - assert pick_primary_situation(situations) == expected diff --git a/products/tasks/backend/logic/code_workstreams/test_grouping.py b/products/tasks/backend/logic/code_workstreams/test_grouping.py deleted file mode 100644 index d5af79a8d384..000000000000 --- a/products/tasks/backend/logic/code_workstreams/test_grouping.py +++ /dev/null @@ -1,263 +0,0 @@ -from dataclasses import replace - -import pytest - -from products.tasks.backend.logic.code_workstreams.classify import STALE_THRESHOLD_MS -from products.tasks.backend.logic.code_workstreams.grouping import ( - RUNNING_STALE_THRESHOLD_MS, - PrInput, - TaskInput, - build_workstreams, - workstream_key, -) - -NOW = 1_700_000_000_000 - - -def _task(**overrides) -> TaskInput: - return replace( - TaskInput( - id="t1", - title="Task", - status="completed", - last_activity_at=NOW, - repo_name="posthog", - repo_full_path="posthog/posthog", - branch="feat/x", - cloud_pr_url=None, - folder_path=None, - ), - **overrides, - ) - - -def _pr(**overrides) -> PrInput: - return replace( - PrInput( - url="https://github.com/posthog/posthog/pull/1", - number=1, - title="PR", - state="open", - ci_status="passing", - review_decision=None, - unresolved_threads=0, - mergeable=True, - is_current_user_requested_reviewer=False, - is_current_user_author=True, - author="me", - last_updated_at=NOW, - ), - **overrides, - ) - - -@pytest.mark.parametrize( - "task,pr_url,expected", - [ - ( - _task(branch="b", repo_name="r", repo_full_path="org/r", folder_path="/p"), - "https://x/pull/9", - "pr:https://x/pull/9", - ), - (_task(branch="b", repo_name="r", repo_full_path="org/r", folder_path="/p"), None, "branch:org/r#b"), - (_task(branch="b", repo_name="r", repo_full_path=None, folder_path="/p"), None, "branch:r#b"), - (_task(branch=None, repo_name=None, repo_full_path=None, folder_path="/p"), None, "path:/p"), - (_task(branch=None, repo_name=None, repo_full_path=None, folder_path=None), None, None), - # Branch equal to the base branch is not a real feature branch: don't group on it. - (_task(branch="master", base_branch="master", folder_path=None), None, None), - (_task(branch="master", base_branch="master", folder_path="/p"), None, "path:/p"), - (_task(branch="master", base_branch="master"), "https://x/pull/9", "pr:https://x/pull/9"), - # A branch distinct from the base branch still groups. - (_task(branch="feat/x", base_branch="master", repo_full_path="org/r"), None, "branch:org/r#feat/x"), - ], -) -def test_workstream_key_precedence(task, pr_url, expected): - assert workstream_key(task, pr_url) == expected - - -def test_running_task_no_pr_is_excluded_from_workstreams(): - result = build_workstreams([_task(status="in_progress", branch=None, folder_path=None)], {}, NOW) - assert not result.needs_attention - assert not result.in_progress - - -def test_running_task_with_pr_is_grouped(): - task = _task(status="in_progress") - result = build_workstreams([task], {task.id: _pr(ci_status="failing")}, NOW) - assert len(result.needs_attention) == 1 - - -def test_idle_running_task_falls_through_to_grouping(): - old = NOW - RUNNING_STALE_THRESHOLD_MS - 1 - task = _task(status="in_progress", last_activity_at=old) - result = build_workstreams([task], {}, NOW) - assert len(result.in_progress) == 1 - - -def test_failing_ci_lands_in_needs_attention(): - task = _task() - result = build_workstreams([task], {task.id: _pr(ci_status="failing")}, NOW) - assert len(result.needs_attention) == 1 - ws = result.needs_attention[0] - assert ws.pr_url == _pr().url - assert "ci_failing" in ws.situations - - -def test_healthy_open_pr_lands_in_in_progress(): - task = _task() - result = build_workstreams([task], {task.id: _pr()}, NOW) - assert len(result.in_progress) == 1 - assert result.in_progress[0].situations == ["in_review"] - - -def test_tasks_group_by_shared_pr_url(): - url = "https://github.com/posthog/posthog/pull/42" - t1 = _task(id="a", branch="x", last_activity_at=NOW) - t2 = _task(id="b", branch="y", last_activity_at=NOW - 1000) - pr_by_task = {"a": _pr(url=url), "b": _pr(url=url)} - result = build_workstreams([t1, t2], pr_by_task, NOW) - assert len(result.needs_attention) + len(result.in_progress) == 1 - ws = (result.needs_attention + result.in_progress)[0] - assert {t.id for t in ws.tasks} == {"a", "b"} - - -def test_stale_branch_no_pr_is_attention(): - old = NOW - STALE_THRESHOLD_MS - 1 - task = _task(status="completed", last_activity_at=old, cloud_pr_url=None) - result = build_workstreams([task], {}, NOW) - assert len(result.needs_attention) == 1 - assert "stale" in result.needs_attention[0].situations - - -def test_task_without_grouping_key_is_skipped(): - task = _task(status="completed", branch=None, repo_name=None, folder_path=None, cloud_pr_url=None) - result = build_workstreams([task], {}, NOW) - assert not result.needs_attention - assert not result.in_progress - - -def test_base_branch_no_pr_tasks_do_not_collapse_and_are_dropped(): - # Tasks that ran on the base branch and never pushed a feature branch or PR left no real - # work to track, so they must neither collapse into one blob nor surface as workstreams. - t1 = _task(id="a", branch="master", base_branch="master", cloud_pr_url=None) - t2 = _task(id="b", branch="master", base_branch="master", cloud_pr_url=None) - result = build_workstreams([t1, t2], {}, NOW) - assert not result.needs_attention - assert not result.in_progress - - -def test_tasks_group_by_shared_feature_branch(): - t1 = _task(id="a", branch="feat/x", base_branch="master", last_activity_at=NOW) - t2 = _task(id="b", branch="feat/x", base_branch="master", last_activity_at=NOW - 1000) - result = build_workstreams([t1, t2], {}, NOW) - workstreams = result.needs_attention + result.in_progress - assert len(workstreams) == 1 - assert {t.id for t in workstreams[0].tasks} == {"a", "b"} - assert workstreams[0].branch == "feat/x" - - -def test_feature_branch_and_base_branch_tasks_do_not_merge(): - feature = _task(id="a", branch="feat/x", base_branch="master", cloud_pr_url=None) - base = _task(id="b", branch="master", base_branch="master", cloud_pr_url=None) - result = build_workstreams([feature, base], {}, NOW) - workstreams = result.needs_attention + result.in_progress - assert len(workstreams) == 1 - assert {t.id for t in workstreams[0].tasks} == {"a"} - - -def test_follow_up_task_resolves_to_pr_by_branch(): - # A follow-up run pushed to the PR's branch (branch == base, no own pr_url); it should still - # group under the PR workstream via the branch→PR map and inherit its situation. - url = "https://github.com/posthog/posthog/pull/7" - follow_up = _task( - id="b", - repo_full_path="posthog/posthog", - branch="feat/x", - base_branch="feat/x", - cloud_pr_url=None, - ) - pr_by_branch = {("posthog/posthog", "feat/x"): _pr(url=url, ci_status="failing", head_branch="feat/x")} - result = build_workstreams([follow_up], {}, NOW, pr_by_branch) - assert len(result.needs_attention) == 1 - ws = result.needs_attention[0] - assert ws.id == f"pr:{url}" - assert "ci_failing" in ws.situations - - -def test_follow_up_task_groups_with_original_pr_task(): - url = "https://github.com/posthog/posthog/pull/7" - original = _task(id="a", repo_full_path="posthog/posthog", branch="feat/x", cloud_pr_url=url) - follow_up = _task( - id="b", - repo_full_path="posthog/posthog", - branch="feat/x", - base_branch="feat/x", - cloud_pr_url=None, - ) - pr_by_branch = {("posthog/posthog", "feat/x"): _pr(url=url, head_branch="feat/x")} - result = build_workstreams([original, follow_up], {original.id: _pr(url=url)}, NOW, pr_by_branch) - workstreams = result.needs_attention + result.in_progress - assert len(workstreams) == 1 - assert {t.id for t in workstreams[0].tasks} == {"a", "b"} - - -def test_branch_resolution_is_repo_scoped(): - # A branch named "main" in one repo must not pull in a PR for "main" in another repo. - url = "https://github.com/posthog/other/pull/1" - task = _task(id="a", repo_full_path="posthog/posthog", branch="main", base_branch="main", cloud_pr_url=None) - pr_by_branch = {("posthog/other", "main"): _pr(url=url, head_branch="main")} - result = build_workstreams([task], {}, NOW, pr_by_branch) - workstreams = result.needs_attention + result.in_progress - assert not workstreams - - -def test_base_branch_task_does_not_resolve_to_base_headed_pr(): - # A no-op run on the base branch must not collapse into a PR that happens to be headed from - # that base branch, while a real feature-branch follow-up still resolves. - feature = _task(id="a", repo_full_path="posthog/posthog", branch="feat/x", base_branch="master", cloud_pr_url=None) - base = _task(id="b", repo_full_path="posthog/posthog", branch="master", base_branch="master", cloud_pr_url=None) - pr_by_branch = { - ("posthog/posthog", "feat/x"): _pr(url="https://github.com/posthog/posthog/pull/1", head_branch="feat/x"), - ("posthog/posthog", "master"): _pr(url="https://github.com/posthog/posthog/pull/2", head_branch="master"), - } - result = build_workstreams([feature, base], {}, NOW, pr_by_branch) - workstreams = result.needs_attention + result.in_progress - assert len(workstreams) == 1 - assert workstreams[0].id == "pr:https://github.com/posthog/posthog/pull/1" - assert {t.id for t in workstreams[0].tasks} == {"a"} - - -def test_lone_base_branch_task_does_not_resolve_to_base_headed_pr(): - # Even with no sibling feature task to mark "master" a base branch, a lone run on the default - # branch must not collapse into a PR that happens to be headed from "master". - task = _task(id="a", repo_full_path="posthog/posthog", branch="master", base_branch="master", cloud_pr_url=None) - pr_by_branch = { - ("posthog/posthog", "master"): _pr(url="https://github.com/posthog/posthog/pull/2", head_branch="master") - } - result = build_workstreams([task], {}, NOW, pr_by_branch) - workstreams = result.needs_attention + result.in_progress - assert not workstreams - - -def test_quick_action_is_carried_onto_workstream_task(): - task = _task(id="a", quick_action="Fix CI") - result = build_workstreams([task], {task.id: _pr(ci_status="failing")}, NOW) - ws = result.needs_attention[0] - assert ws.tasks[0].quick_action == "Fix CI" - - -@pytest.mark.parametrize( - "task,pr_by_task", - [ - (_task(), {"t1": _pr(state="merged")}), - ( - _task(cloud_pr_url="https://github.com/posthog/posthog/pull/1", cloud_pr_merged=True), - {}, - ), - ], -) -def test_merged_pr_is_omitted_from_active_workstreams(task, pr_by_task): - result = build_workstreams([task], pr_by_task, NOW) - - assert not result.needs_attention - assert not result.in_progress diff --git a/products/tasks/backend/logic/code_workstreams/test_validation.py b/products/tasks/backend/logic/code_workstreams/test_validation.py deleted file mode 100644 index abec6abceff8..000000000000 --- a/products/tasks/backend/logic/code_workstreams/test_validation.py +++ /dev/null @@ -1,63 +0,0 @@ -from products.tasks.backend.logic.code_workstreams.default_workflow import build_default_bindings -from products.tasks.backend.logic.code_workstreams.situations import SITUATION_IDS -from products.tasks.backend.logic.code_workstreams.validation import validate_bindings - - -def test_default_bindings_cover_every_situation(): - bindings = build_default_bindings() - assert set(bindings.keys()) == set(SITUATION_IDS) - - -def test_default_bindings_are_valid(): - result = validate_bindings(build_default_bindings()) - assert result.can_save - assert result.diagnostics == [] - - -def test_duplicate_action_id_is_error(): - bindings = { - "working": [ - {"id": "a", "label": "A", "skillId": "s", "prompt": "p"}, - {"id": "a", "label": "B", "skillId": "s", "prompt": "p"}, - ] - } - result = validate_bindings(bindings) - assert not result.can_save - assert any(d.code == "duplicate_action_id" for d in result.diagnostics) - - -def test_empty_fields_are_errors(): - bindings = {"working": [{"id": "a", "label": " ", "skillId": "", "prompt": ""}]} - result = validate_bindings(bindings) - codes = {d.code for d in result.diagnostics} - assert codes == {"action_empty_label", "action_empty_prompt"} - assert not result.can_save - - -def test_missing_skill_is_allowed(): - bindings = {"working": [{"id": "a", "label": "A", "skillId": "", "prompt": "p"}]} - result = validate_bindings(bindings) - assert result.can_save - assert result.diagnostics == [] - - -def test_empty_bindings_are_valid(): - assert validate_bindings({}).can_save - - -def test_non_dict_bindings_is_rejected(): - result = validate_bindings([{"id": "a"}]) # type: ignore[arg-type] - assert not result.can_save - assert {d.code for d in result.diagnostics} == {"bindings_not_object"} - - -def test_non_list_situation_value_is_rejected(): - result = validate_bindings({"working": "not a list"}) - assert not result.can_save - assert any(d.code == "situation_not_list" for d in result.diagnostics) - - -def test_non_dict_action_is_rejected(): - result = validate_bindings({"working": ["not an object"]}) - assert not result.can_save - assert any(d.code == "action_not_object" for d in result.diagnostics) diff --git a/products/tasks/backend/logic/code_workstreams/validation.py b/products/tasks/backend/logic/code_workstreams/validation.py deleted file mode 100644 index 9c8ea5b21be7..000000000000 --- a/products/tasks/backend/logic/code_workstreams/validation.py +++ /dev/null @@ -1,101 +0,0 @@ -from collections.abc import Mapping -from dataclasses import dataclass -from typing import Any - -from .situations import SITUATION_IDS - - -@dataclass -class ValidationDiagnostic: - severity: str - code: str - message: str - situation_id: str | None = None - action_id: str | None = None - - -@dataclass -class ValidationResult: - diagnostics: list[ValidationDiagnostic] - can_save: bool - - -def validate_bindings(bindings: Mapping[str, Any]) -> ValidationResult: - diagnostics: list[ValidationDiagnostic] = [] - - if not isinstance(bindings, dict): - return ValidationResult( - diagnostics=[ - ValidationDiagnostic( - severity="error", - code="bindings_not_object", - message="bindings must be an object", - ) - ], - can_save=False, - ) - - for sid in SITUATION_IDS: - actions = bindings.get(sid) or [] - if not isinstance(actions, list): - diagnostics.append( - ValidationDiagnostic( - severity="error", - code="situation_not_list", - message=f"{sid} must be a list of actions", - situation_id=sid, - ) - ) - continue - seen_ids: set[str] = set() - for action in actions: - if not isinstance(action, dict): - diagnostics.append( - ValidationDiagnostic( - severity="error", - code="action_not_object", - message=f"An action in {sid} must be an object", - situation_id=sid, - ) - ) - continue - action_id = str(action.get("id", "")) - if action_id in seen_ids: - diagnostics.append( - ValidationDiagnostic( - severity="error", - code="duplicate_action_id", - message=f'Duplicate action id "{action_id}" in {sid}', - situation_id=sid, - action_id=action_id, - ) - ) - continue - seen_ids.add(action_id) - - label = str(action.get("label", "")) - prompt = str(action.get("prompt", "")) - - if label.strip() == "": - diagnostics.append( - ValidationDiagnostic( - severity="error", - code="action_empty_label", - message=f"An action in {sid} has no label", - situation_id=sid, - action_id=action_id, - ) - ) - if prompt.strip() == "": - diagnostics.append( - ValidationDiagnostic( - severity="error", - code="action_empty_prompt", - message=f'Action "{label}" in {sid} has an empty prompt', - situation_id=sid, - action_id=action_id, - ) - ) - - can_save = not any(d.severity == "error" for d in diagnostics) - return ValidationResult(diagnostics=diagnostics, can_save=can_save) diff --git a/products/tasks/backend/management/commands/evaluate_code_workstreams.py b/products/tasks/backend/management/commands/evaluate_code_workstreams.py deleted file mode 100644 index 278c6ac8a594..000000000000 --- a/products/tasks/backend/management/commands/evaluate_code_workstreams.py +++ /dev/null @@ -1,128 +0,0 @@ -import json - -from django.core.management.base import BaseCommand, CommandError - -from posthog.egress.limiter.policies import Priority -from posthog.models.scoping import team_scope - -from products.tasks.backend.models import CodePrSnapshot, CodeWorkstream -from products.tasks.backend.temporal.code_workstreams.activities.github_resolution import resolve_github_integration -from products.tasks.backend.temporal.code_workstreams.activities.list_active_teams import list_active_code_teams -from products.tasks.backend.temporal.code_workstreams.activities.load_pr_urls import ( - LoadTeamPrUrlsInput, - load_team_pr_urls, -) -from products.tasks.backend.temporal.code_workstreams.activities.poll_pull_requests import poll_pull_requests_for_team -from products.tasks.backend.temporal.code_workstreams.activities.rebuild_workstreams import ( - RebuildTeamWorkstreamsInput, - rebuild_team_workstreams, -) - - -class Command(BaseCommand): - help = "Run one code-workstreams evaluation cycle synchronously (no Temporal worker/schedule). Diagnostic + local testing." - - def add_arguments(self, parser): - parser.add_argument("--team-id", type=int, help="Team to evaluate. Omit to list active teams and exit.") - parser.add_argument( - "--skip-poll", action="store_true", help="Skip the GitHub PR poll (grouping/classify only)." - ) - parser.add_argument( - "--pr-url", - type=str, - help="Probe a single PR: print exactly what get_pull_request_snapshot returns (needs --team-id).", - ) - - def handle(self, *args, **options): - team_id = options.get("team_id") - - if team_id is None: - active = list_active_code_teams() - self.stdout.write(f"Active code teams (recent task activity): {active.team_ids}") - self.stdout.write("Re-run with --team-id to evaluate one.") - return - - if options.get("pr_url"): - self._probe_pr(team_id, options["pr_url"]) - return - - self.stdout.write(f"== Evaluating team {team_id} ==") - - prs = load_team_pr_urls(LoadTeamPrUrlsInput(team_id=team_id)) - self.stdout.write(f"PR URLs found on recent task runs: {len(prs.prs)}") - for ref in prs.prs[:10]: - self.stdout.write(f" - {ref.pr_url} (integration={ref.github_integration_id})") - - if options["skip_poll"]: - self.stdout.write("Skipping GitHub poll (--skip-poll).") - elif prs.prs: - try: - result = poll_pull_requests_for_team(team_id, prs.prs) - self.stdout.write( - f"Polled {result.polled}, updated {result.updated}, rate_limited={result.rate_limited}" - ) - except Exception as e: - raise CommandError(f"PR poll failed: {type(e).__name__}: {e}") from e - - out = rebuild_team_workstreams(RebuildTeamWorkstreamsInput(team_id=team_id)) - self.stdout.write(f"Rebuilt: users={out.users}, workstreams={out.workstreams}, pruned={out.pruned}") - - with team_scope(team_id): - snapshots = list(CodePrSnapshot.objects.filter(team_id=team_id)) - workstreams = list(CodeWorkstream.objects.filter(team_id=team_id)) - - self.stdout.write(f"\nCodePrSnapshot rows: {len(snapshots)}") - for s in snapshots[:25]: - self.stdout.write( - f" {s.pr_url} state={s.state} ci={s.ci_status} review={s.review_decision} " - f"threads={s.unresolved_threads} author={s.author_login}" - ) - - self.stdout.write(f"\nCodeWorkstream rows: {len(workstreams)}") - for ws in workstreams[:25]: - self.stdout.write( - f" [{ws.state}] {ws.key} → situations={ws.situations} (user={ws.user_id}, tasks={len(ws.tasks)})" - ) - - if not workstreams: - self.stdout.write( - "\nNo workstreams written. Likely causes: tasks have no PR URL and no (repository + run branch) " - "grouping key, all of the team's recent tasks are still actively-running agents, or there are no " - "task runs in the last 30 days for this team." - ) - elif snapshots and all( - s.ci_status == "none" and s.review_decision is None and s.unresolved_threads == 0 for s in snapshots - ): - self.stdout.write( - "\nAll PR snapshots are empty (ci=none, no review, 0 threads). The GitHub poll reached GitHub but " - "the data came back blank — likely the GitHub App installation lacks 'Checks: read' / 'Pull " - "requests: read'. Run with --pr-url to see the raw result + any GraphQL errors." - ) - - def _probe_pr(self, team_id: int, pr_url: str) -> None: - prs = load_team_pr_urls(LoadTeamPrUrlsInput(team_id=team_id)) - ref = next((r for r in prs.prs if r.pr_url == pr_url), None) - if ref is None: - raise CommandError( - f"{pr_url} is not among the team's recent task-run PR URLs ({len(prs.prs)} found). " - "Is there a task run with output.pr_url == this URL in the last 30 days?" - ) - self.stdout.write( - f"Resolved PrRef: team_integration={ref.github_integration_id}, user_integration={ref.github_user_integration_id}" - ) - # One interactive diagnostic call is not deferrable bulk — don't let the BATCH lane shed - # the very probe that's investigating degraded polling. - integration = resolve_github_integration( - ref.github_integration_id, ref.github_user_integration_id, priority=Priority.CRITICAL - ) - if integration is None: - raise CommandError( - "No GitHub integration resolved for this PR: the task isn't linked to one, the team has no GitHub " - "integration, and the task creator has no user GitHub integration. Connect GitHub (team or user)." - ) - self.stdout.write(f"Resolved integration: {type(integration).__name__}") - try: - snapshot = integration.get_pull_request_snapshot(pr_url) - except Exception as e: - raise CommandError(f"get_pull_request_snapshot raised: {type(e).__name__}: {e}") from e - self.stdout.write(json.dumps(snapshot, indent=2, default=str)) diff --git a/products/tasks/backend/migrations/0069_remove_code_home_models.py b/products/tasks/backend/migrations/0069_remove_code_home_models.py new file mode 100644 index 000000000000..9dd53481998c --- /dev/null +++ b/products/tasks/backend/migrations/0069_remove_code_home_models.py @@ -0,0 +1,20 @@ +from django.db import migrations + + +class Migration(migrations.Migration): + dependencies = [ + ("tasks", "0068_loop_creator_backfill"), + ] + + # State-only removal: the Code Home / workstreams feature is deleted, but the tables + # (posthog_code_workstream, posthog_code_pr_snapshot, posthog_code_workflow_config) + # stay in place and get dropped in a follow-up RunSQL migration after a deploy cycle. + operations = [ + migrations.SeparateDatabaseAndState( + state_operations=[ + migrations.DeleteModel(name="CodeWorkstream"), + migrations.DeleteModel(name="CodePrSnapshot"), + migrations.DeleteModel(name="CodeWorkflowConfig"), + ], + ), + ] diff --git a/products/tasks/backend/migrations/max_migration.txt b/products/tasks/backend/migrations/max_migration.txt index 38ed8b426b0e..d4a712f57d71 100644 --- a/products/tasks/backend/migrations/max_migration.txt +++ b/products/tasks/backend/migrations/max_migration.txt @@ -1 +1 @@ -0068_loop_creator_backfill +0069_remove_code_home_models diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index da1de46c1568..01b8f37e3c8b 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -2440,121 +2440,6 @@ def __str__(self): return f"Presence: user {self.user_id} on task {self.task_id} via device {self.push_token_id}" -class CodeWorkflowConfig(TeamScopedRootMixin): - # nosemgrep: prefer-uuid7-django-pk -- mirrors sibling task models in this app - id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) - team = models.ForeignKey("posthog.Team", on_delete=models.CASCADE, related_name="+") - user = models.ForeignKey("posthog.User", on_delete=models.CASCADE, related_name="+") - version = models.PositiveIntegerField(default=1) - bindings = models.JSONField(default=dict, help_text="Situation id → ordered WorkflowAction list") - created_at = models.DateTimeField(default=django_timezone.now) - updated_at = models.DateTimeField(auto_now=True) - - class Meta: - db_table = "posthog_code_workflow_config" - constraints = [ - models.UniqueConstraint(fields=["team", "user"], name="code_workflow_config_team_user_unique"), - ] - - def __str__(self): - return f"CodeWorkflowConfig(team={self.team_id}, user={self.user_id}, v{self.version})" - - -class CodePrSnapshot(TeamScopedRootMixin): - class State(models.TextChoices): - OPEN = "open", "Open" - DRAFT = "draft", "Draft" - MERGED = "merged", "Merged" - CLOSED = "closed", "Closed" - - class CiStatus(models.TextChoices): - PASSING = "passing", "Passing" - FAILING = "failing", "Failing" - PENDING = "pending", "Pending" - NONE = "none", "None" - - class ReviewDecision(models.TextChoices): - APPROVED = "approved", "Approved" - CHANGES_REQUESTED = "changes_requested", "Changes requested" - REVIEW_REQUIRED = "review_required", "Review required" - - # nosemgrep: prefer-uuid7-django-pk -- mirrors sibling task models in this app - id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) - team = models.ForeignKey("posthog.Team", on_delete=models.CASCADE, related_name="+") - github_integration = models.ForeignKey( - "posthog.Integration", on_delete=models.SET_NULL, null=True, blank=True, related_name="+" - ) - pr_url = models.CharField(max_length=500) - number = models.PositiveIntegerField() - title = models.TextField(blank=True, default="") - state = models.CharField(max_length=10, choices=State.choices) - ci_status = models.CharField(max_length=10, choices=CiStatus.choices, default=CiStatus.NONE) - review_decision = models.CharField(max_length=20, choices=ReviewDecision.choices, null=True, blank=True) - unresolved_threads = models.PositiveIntegerField(default=0) - mergeable = models.BooleanField(null=True, blank=True) - author_login = models.CharField(max_length=255, null=True, blank=True) - head_branch = models.CharField( - max_length=255, - null=True, - blank=True, - help_text="PR head (source) branch, used to group follow-up task runs under this PR's workstream", - ) - requested_reviewer_logins = models.JSONField(default=list, help_text="GitHub logins requested as reviewers") - pr_updated_at = models.DateTimeField(null=True, blank=True, help_text="PR's last-updated time on GitHub") - fingerprint = models.CharField(max_length=64, blank=True, default="", help_text="Change-detection hash") - fetched_at = models.DateTimeField(default=django_timezone.now, help_text="When this snapshot was last polled") - created_at = models.DateTimeField(default=django_timezone.now) - updated_at = models.DateTimeField(auto_now=True) - - class Meta: - db_table = "posthog_code_pr_snapshot" - constraints = [ - models.UniqueConstraint(fields=["team", "pr_url"], name="code_pr_snapshot_team_url_unique"), - ] - - def __str__(self): - return f"CodePrSnapshot({self.pr_url} {self.state})" - - -class CodeWorkstream(TeamScopedRootMixin): - class WorkstreamState(models.TextChoices): - ATTENTION = "attention", "Needs attention" - IN_PROGRESS = "in_progress", "In progress" - - # nosemgrep: prefer-uuid7-django-pk -- mirrors sibling task models in this app - id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) - team = models.ForeignKey("posthog.Team", on_delete=models.CASCADE, related_name="+") - user = models.ForeignKey("posthog.User", on_delete=models.CASCADE, related_name="+") - key = models.CharField(max_length=600, help_text="Grouping key: pr: | branch:# | path:") - repo_name = models.CharField(max_length=255, null=True, blank=True) - repo_full_path = models.CharField(max_length=512, null=True, blank=True) - branch = models.CharField(max_length=255, null=True, blank=True) - pr_url = models.CharField(max_length=500, null=True, blank=True) - pr_snapshot = models.ForeignKey(CodePrSnapshot, on_delete=models.SET_NULL, null=True, blank=True, related_name="+") - pr = models.JSONField(null=True, blank=True, help_text="Per-user-resolved PrSnapshot wire shape") - situations = models.JSONField(default=list, help_text="List of situation ids this workstream is in") - primary_situation = models.CharField(max_length=20, null=True, blank=True, help_text="Board column placement") - state = models.CharField(max_length=20, choices=WorkstreamState.choices) - tasks = models.JSONField(default=list, help_text="List of {id, title, status} for grouped tasks") - last_activity_at = models.DateTimeField() - generated_at = models.DateTimeField(default=django_timezone.now, help_text="When this row was last rebuilt") - created_at = models.DateTimeField(default=django_timezone.now) - updated_at = models.DateTimeField(auto_now=True) - - class Meta: - db_table = "posthog_code_workstream" - ordering = ["-last_activity_at"] - constraints = [ - models.UniqueConstraint(fields=["team", "user", "key"], name="code_workstream_team_user_key_unique"), - ] - indexes = [ - models.Index(fields=["team", "user", "state"], name="code_workstream_state_idx"), - ] - - def __str__(self): - return f"CodeWorkstream({self.key} {self.state})" - - @receiver(post_save, sender=TaskRun) def track_task_run_completion(sender, instance: TaskRun, created: bool, **kwargs): try: diff --git a/products/tasks/backend/presentation/serializers.py b/products/tasks/backend/presentation/serializers.py index 5c97215dec6d..eb579e18e9ee 100644 --- a/products/tasks/backend/presentation/serializers.py +++ b/products/tasks/backend/presentation/serializers.py @@ -1973,13 +1973,6 @@ class TaskRunBootstrapCreateRequestSerializer( "follows the server-side default (enabled); false opts this run out." ), ) - home_quick_action = serializers.CharField( - required=False, - default=None, - allow_blank=False, - max_length=120, - help_text="Label of the Home-tab quick action that started this run (e.g. 'Fix CI'), surfaced on the workstream.", - ) def validate(self, attrs): errors: dict[str, str] = {} diff --git a/products/tasks/backend/presentation/views/code_home_api.py b/products/tasks/backend/presentation/views/code_home_api.py deleted file mode 100644 index 731ccc5ca745..000000000000 --- a/products/tasks/backend/presentation/views/code_home_api.py +++ /dev/null @@ -1,142 +0,0 @@ -from rest_framework import status, viewsets -from rest_framework.authentication import SessionAuthentication -from rest_framework.decorators import action -from rest_framework.permissions import IsAuthenticated -from rest_framework.response import Response - -from posthog.api.routing import TeamAndOrgViewSetMixin -from posthog.auth import OAuthAccessTokenAuthentication, PersonalAPIKeyAuthentication -from posthog.permissions import APIScopePermission - -from products.tasks.backend.facade import api as tasks_facade -from products.tasks.backend.facade.contracts import ( - CodeHomeDTO, - CodeHomeWorkstreamDTO, - CodeWorkflowConfigDTO, - CodeWorkflowDiagnosticDTO, -) - -_AUTH_CLASSES = [SessionAuthentication, PersonalAPIKeyAuthentication, OAuthAccessTokenAuthentication] - - -def _serialize_diagnostic(d: CodeWorkflowDiagnosticDTO) -> dict: - out = {"severity": d.severity, "code": d.code, "message": d.message} - if d.situation_id is not None: - out["situationId"] = d.situation_id - if d.action_id is not None: - out["actionId"] = d.action_id - return out - - -def _serialize_config(config: CodeWorkflowConfigDTO) -> dict: - return { - "id": config.id, - "version": config.version, - "updatedAt": config.updated_at.isoformat(), - "bindings": config.bindings, - } - - -def _serialize_home(home: CodeHomeDTO) -> dict: - return { - "activeAgents": [ - { - "taskId": agent.task_id, - "title": agent.title, - "repoName": agent.repo_name, - "branch": agent.branch, - "status": agent.status, - "lastActivityAt": agent.last_activity_at, - "needsPermission": agent.needs_permission, - "cloudPrUrl": agent.cloud_pr_url, - } - for agent in home.active_agents - ], - "needsAttention": [_serialize_workstream(ws) for ws in home.needs_attention], - "inProgress": [_serialize_workstream(ws) for ws in home.in_progress], - } - - -def _serialize_workstream(ws: CodeHomeWorkstreamDTO) -> dict: - return { - "id": ws.id, - "repoName": ws.repo_name, - "repoFullPath": ws.repo_full_path, - "branch": ws.branch, - "prUrl": ws.pr_url, - "pr": ws.pr, - "tasks": [ - { - "id": t.id, - "title": t.title, - "status": t.status, - "isGenerating": t.is_generating, - "needsPermission": t.needs_permission, - "quickAction": t.quick_action, - } - for t in ws.tasks - ], - "situations": ws.situations, - "primarySituation": ws.primary_situation, - "lastActivityAt": ws.last_activity_at, - } - - -class CodeWorkflowViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet): - scope_object = "task" - authentication_classes = _AUTH_CLASSES - permission_classes = [IsAuthenticated, APIScopePermission] - hide_api_docs = True - - def list(self, request, *args, **kwargs): - config = tasks_facade.get_code_workflow_config(self.team_id, request.user.id) - return Response(_serialize_config(config)) - - @action(detail=False, methods=["post"], url_path="save", required_scopes=["task:write"]) - def save(self, request, **kwargs): - config_in = request.data.get("config") or {} - expected_version = request.data.get("expectedVersion") - bindings = config_in.get("bindings") or {} - - result = tasks_facade.save_code_workflow_bindings( - self.team_id, - request.user.id, - bindings=bindings, - expected_version=expected_version, - ) - - if result.outcome == tasks_facade.CODE_WORKFLOW_CONFLICT: - return Response( - {"status": "conflict", "config": _serialize_config(result.config)}, - status=status.HTTP_409_CONFLICT, - ) - if result.outcome == tasks_facade.CODE_WORKFLOW_INVALID: - return Response( - { - "status": "invalid", - "config": _serialize_config(result.config), - "diagnostics": [_serialize_diagnostic(d) for d in result.diagnostics], - }, - status=status.HTTP_422_UNPROCESSABLE_ENTITY, - ) - return Response({"status": "saved", "config": _serialize_config(result.config)}) - - @action(detail=False, methods=["post"], url_path="reset", required_scopes=["task:write"]) - def reset(self, request, **kwargs): - config = tasks_facade.reset_code_workflow_bindings(self.team_id, request.user.id) - return Response(_serialize_config(config)) - - -class CodeHomeViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet): - scope_object = "task" - authentication_classes = _AUTH_CLASSES - permission_classes = [IsAuthenticated, APIScopePermission] - hide_api_docs = True - - def list(self, request, *args, **kwargs): - return Response(_serialize_home(tasks_facade.get_code_home(self.team_id, request.user.id))) - - @action(detail=False, methods=["post"], url_path="refresh", required_scopes=["task:write"]) - def refresh(self, request, **kwargs): - started = tasks_facade.refresh_team_code_workstreams(self.team_id) - return Response({"started": started}, status=status.HTTP_202_ACCEPTED) diff --git a/products/tasks/backend/routes.py b/products/tasks/backend/routes.py index 94b256ebfa4c..6bfbbccf691c 100644 --- a/products/tasks/backend/routes.py +++ b/products/tasks/backend/routes.py @@ -4,7 +4,6 @@ import products.tasks.backend.presentation.views.loops as loops import products.tasks.backend.presentation.views.seat_api as seats import products.tasks.backend.presentation.views.channels_api as channels -import products.tasks.backend.presentation.views.code_home_api as code_home def register_routes(routers: RouterRegistry) -> None: @@ -39,7 +38,5 @@ def register_routes(routers: RouterRegistry) -> None: routers.projects.register( r"sandbox_custom_images", tasks.SandboxCustomImageViewSet, "project_sandbox_custom_images", ["team_id"] ) - routers.projects.register(r"code_workflow", code_home.CodeWorkflowViewSet, "project_code_workflow", ["team_id"]) - routers.projects.register(r"code_home", code_home.CodeHomeViewSet, "project_code_home", ["team_id"]) routers.root.register(r"code/invites", tasks.CodeInviteViewSet, "code_invites") routers.root.register(r"seats", seats.SeatViewSet, "seats") diff --git a/products/tasks/backend/temporal/__init__.py b/products/tasks/backend/temporal/__init__.py index b8273cc23977..4f62417f7f64 100644 --- a/products/tasks/backend/temporal/__init__.py +++ b/products/tasks/backend/temporal/__init__.py @@ -1,12 +1,6 @@ from .automation import RunTaskAutomationWorkflow, run_task_automation_activity from .build_image.activities import build_and_publish_image, mark_image_build_failed, scan_image_spec from .build_image.workflow import BuildSandboxImageWorkflow -from .code_workstreams.activities.discover_branch_prs import discover_branch_prs -from .code_workstreams.activities.list_active_teams import list_active_code_teams -from .code_workstreams.activities.load_pr_urls import load_team_pr_urls -from .code_workstreams.activities.poll_pull_requests import poll_team_pull_requests -from .code_workstreams.activities.rebuild_workstreams import rebuild_team_workstreams -from .code_workstreams.workflow import EvaluateCodeWorkstreamsWorkflow, EvaluateTeamCodeWorkstreamsWorkflow from .create_snapshot.activities import ( cleanup_sandbox as snapshot_cleanup_sandbox, clone_repository as snapshot_clone_repository, @@ -68,8 +62,6 @@ PostHogCodeAgentRelayWorkflow, RunTaskAutomationWorkflow, RunLoopWorkflow, - EvaluateCodeWorkstreamsWorkflow, - EvaluateTeamCodeWorkstreamsWorkflow, BuildSandboxImageWorkflow, ] @@ -125,9 +117,4 @@ scan_image_spec, build_and_publish_image, mark_image_build_failed, - list_active_code_teams, - load_team_pr_urls, - discover_branch_prs, - poll_team_pull_requests, - rebuild_team_workstreams, ] diff --git a/products/tasks/backend/temporal/code_workstreams/__init__.py b/products/tasks/backend/temporal/code_workstreams/__init__.py deleted file mode 100644 index e69de29bb2d1..000000000000 diff --git a/products/tasks/backend/temporal/code_workstreams/activities/__init__.py b/products/tasks/backend/temporal/code_workstreams/activities/__init__.py deleted file mode 100644 index e69de29bb2d1..000000000000 diff --git a/products/tasks/backend/temporal/code_workstreams/activities/discover_branch_prs.py b/products/tasks/backend/temporal/code_workstreams/activities/discover_branch_prs.py deleted file mode 100644 index 25c478d07336..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/activities/discover_branch_prs.py +++ /dev/null @@ -1,173 +0,0 @@ -import random -from collections.abc import Callable -from dataclasses import dataclass -from typing import Optional - -from django.core.exceptions import ObjectDoesNotExist -from django.utils import timezone - -import requests -from temporalio import activity - -from posthog.egress.github.transport import GitHubEgressBudgetExhausted, GitHubRateLimitError -from posthog.models.github_integration_base import GitHubIntegrationBase, GitHubIntegrationError -from posthog.temporal.common.utils import close_db_connections - -from products.tasks.backend.logic.code_workstreams.grouping import DEFAULT_BASE_BRANCHES -from products.tasks.backend.models import TaskRun -from products.tasks.backend.temporal.code_workstreams.activities.github_resolution import ( - TeamIntegrationResolver, - resolve_github_integration, -) -from products.tasks.backend.temporal.code_workstreams.activities.load_pr_urls import PrRef -from products.tasks.backend.temporal.code_workstreams.constants import ( - ACTIVITY_WINDOW, - MAX_BRANCH_QUERIES_PER_TEAM_PER_CYCLE, -) - - -@dataclass -class _BranchCandidate: - repository: str - branch: str - github_integration_id: Optional[int] - github_user_integration_id: Optional[str] - - -@dataclass -class DiscoverBranchPrsInput: - team_id: int - # PR urls already found this cycle; discovery skips these. - known_pr_urls: list[str] - # Remaining capacity before hitting MAX_PRS_PER_TEAM_PER_CYCLE. - budget: int - - -@dataclass -class DiscoverBranchPrsOutput: - prs: list[PrRef] - - -@activity.defn -@close_db_connections -def discover_branch_prs(input: DiscoverBranchPrsInput) -> DiscoverBranchPrsOutput: - return discover_branch_prs_for_team( - input.team_id, - set(input.known_pr_urls), - input.budget, - heartbeat=activity.heartbeat, - ) - - -def _collect_branch_candidates(team_id: int) -> list[_BranchCandidate]: - """Recent runs with a branch and a repository, deduped by (repo, branch), capped per cycle. - - The branch analogue of load_pr_urls' output.pr_url harvest: a run's branch is durable even - when no pr_url was recorded. Base branches are excluded — they never head a task's own PR. - """ - cutoff = timezone.now() - ACTIVITY_WINDOW - runs = ( - TaskRun.objects.filter(team_id=team_id, updated_at__gte=cutoff) - .exclude(branch__isnull=True) - .exclude(branch="") - .exclude(branch__in=list(DEFAULT_BASE_BRANCHES)) - .select_related("task") - .order_by("-updated_at") - ) - resolver = TeamIntegrationResolver(team_id) - seen: set[tuple[str, str]] = set() - candidates: list[_BranchCandidate] = [] - for run in runs.iterator(): - task = run.task - repository = task.repository - branch = run.branch - if not repository or not branch: - continue - key = (repository.casefold(), branch) - if key in seen: - continue - seen.add(key) - team_int, user_int = resolver.resolve(task) - if team_int is None and user_int is None: - continue - candidates.append(_BranchCandidate(repository, branch, team_int, user_int)) - if len(candidates) >= MAX_BRANCH_QUERIES_PER_TEAM_PER_CYCLE: - break - return candidates - - -def discover_branch_prs_for_team( - team_id: int, - known_pr_urls: set[str], - budget: int, - *, - heartbeat: Callable[[int], None] | None = None, -) -> DiscoverBranchPrsOutput: - if budget <= 0: - return DiscoverBranchPrsOutput(prs=[]) - - candidates = _collect_branch_candidates(team_id) - # A shed sweep breaks mid-list; shuffling gives the tail equal coverage across cycles instead - # of the same recency-ordered prefix consuming the budget every run. - random.shuffle(candidates) - integrations: dict[str, GitHubIntegrationBase | None] = {} - found: dict[str, PrRef] = {} - - for index, candidate in enumerate(candidates): - if heartbeat is not None: - heartbeat(index) - if len(found) >= budget: - break - - cache_key = ( - f"i:{candidate.github_integration_id}" - if candidate.github_integration_id is not None - else f"u:{candidate.github_user_integration_id}" - ) - try: - if cache_key not in integrations: - integrations[cache_key] = resolve_github_integration( - candidate.github_integration_id, candidate.github_user_integration_id - ) - integration = integrations[cache_key] - except ObjectDoesNotExist: - # Cache None so other candidates sharing this integration don't re-run the failing lookup. - activity.logger.warning("code_workstreams_discover_integration_missing", repository=candidate.repository) - integrations[cache_key] = None - continue - except (GitHubIntegrationError, GitHubRateLimitError, requests.RequestException) as e: - # Token-refresh failure; skip this candidate rather than aborting the activity. - activity.logger.warning( - "code_workstreams_discover_integration_unavailable", repository=candidate.repository, error=str(e) - ) - continue - if integration is None: - continue - - try: - urls = integration.find_pull_request_urls_for_branch(candidate.repository, candidate.branch) - except GitHubEgressBudgetExhausted: - # Our own limiter shed the sweep — stop for this cycle; the next scheduled run resumes. - activity.logger.warning("code_workstreams_discover_budget_exhausted", repository=candidate.repository) - break - except Exception as e: - activity.logger.warning( - "code_workstreams_discover_branch_failed", - repository=candidate.repository, - branch=candidate.branch, - error=str(e), - ) - continue - - for url in urls: - if url in known_pr_urls or url in found: - continue - found[url] = PrRef( - pr_url=url, - github_integration_id=candidate.github_integration_id, - github_user_integration_id=candidate.github_user_integration_id, - ) - if len(found) >= budget: - break - - return DiscoverBranchPrsOutput(prs=list(found.values())) diff --git a/products/tasks/backend/temporal/code_workstreams/activities/github_resolution.py b/products/tasks/backend/temporal/code_workstreams/activities/github_resolution.py deleted file mode 100644 index c000c28f3398..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/activities/github_resolution.py +++ /dev/null @@ -1,74 +0,0 @@ -from typing import Optional - -from posthog.egress.limiter.policies import Priority -from posthog.models import Integration -from posthog.models.github_integration_base import GitHubIntegrationBase -from posthog.models.integration import GitHubIntegration -from posthog.models.user_integration import UserGitHubIntegration, UserIntegration - -from products.tasks.backend.models import Task - - -class TeamIntegrationResolver: - """Resolve the GitHub integration ids to use for a task's PR lookups. - - Order: the task's own integration, else the team default, else the task creator's personal - integration. Shared by ``load_pr_urls`` and ``discover_branch_prs`` to keep resolution in sync. - """ - - def __init__(self, team_id: int) -> None: - self._team_github_id: Optional[int] = ( - Integration.objects.filter(team_id=team_id, kind="github").values_list("id", flat=True).first() - ) - self._user_github_by_creator: dict[int, Optional[str]] = {} - - def _creator_user_github(self, creator_id: Optional[int]) -> Optional[str]: - if creator_id is None: - return None - if creator_id not in self._user_github_by_creator: - uid = UserIntegration.objects.filter(user_id=creator_id, kind="github").values_list("id", flat=True).first() - self._user_github_by_creator[creator_id] = str(uid) if uid else None - return self._user_github_by_creator[creator_id] - - def resolve(self, task: Task) -> tuple[Optional[int], Optional[str]]: - team_int = task.github_integration_id - user_int = str(task.github_user_integration_id) if task.github_user_integration_id else None - if team_int is None and user_int is None: - team_int = self._team_github_id - if team_int is None: - user_int = self._creator_user_github(task.created_by_id) - return team_int, user_int - - -def resolve_github_integration( - github_integration_id: Optional[int], - github_user_integration_id: Optional[str], - *, - priority: Priority = Priority.BATCH, -) -> GitHubIntegrationBase | None: - """Instantiate the GitHub integration for the given ids, refreshing an expired token. - - Defaults to the sheddable BATCH lane: code-workstreams sweeps are deferrable bulk, so the - egress limiter sheds them before user-facing traffic when an installation's budget runs hot. - Interactive callers (the diagnostic management command) override to CRITICAL. - - Raises ``ObjectDoesNotExist`` if the id no longer resolves; may also raise on token-refresh failure. - """ - integration: GitHubIntegrationBase - if github_integration_id is not None: - integration = GitHubIntegration( - Integration.objects.get(id=github_integration_id), - source="code_workstreams", - priority=priority, - ) - elif github_user_integration_id is not None: - integration = UserGitHubIntegration( - UserIntegration.objects.get(id=github_user_integration_id), - source="code_workstreams", - priority=priority, - ) - else: - return None - if integration.access_token_expired(): - integration.refresh_access_token() - return integration diff --git a/products/tasks/backend/temporal/code_workstreams/activities/list_active_teams.py b/products/tasks/backend/temporal/code_workstreams/activities/list_active_teams.py deleted file mode 100644 index 8682d35444e6..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/activities/list_active_teams.py +++ /dev/null @@ -1,80 +0,0 @@ -from dataclasses import dataclass - -from django.utils import timezone - -import posthoganalytics -from temporalio import activity - -from posthog.models import Team -from posthog.temporal.common.utils import close_db_connections - -from products.tasks.backend.models import Task, TaskRun -from products.tasks.backend.temporal.code_workstreams.constants import ( - ACTIVITY_WINDOW, - HOME_TAB_FLAG, - MAX_TEAMS_PER_CYCLE, -) - -CODE_ORIGIN_PRODUCTS = ( - Task.OriginProduct.USER_CREATED, - Task.OriginProduct.SLACK, - Task.OriginProduct.AUTOMATION, -) - - -@dataclass -class ListActiveCodeTeamsOutput: - team_ids: list[int] - truncated: bool - - -def _org_home_tab_enabled(organization_id: str) -> bool: - # Evaluate locally and fail closed so a flag-service blip doesn't fan out work. - try: - return bool( - posthoganalytics.feature_enabled( - HOME_TAB_FLAG, - distinct_id=organization_id, - groups={"organization": organization_id}, - group_properties={"organization": {"id": organization_id}}, - only_evaluate_locally=True, - send_feature_flag_events=False, - ) - ) - except Exception as e: - activity.logger.warning( - "code_workstreams_home_tab_flag_check_failed", organization_id=organization_id, error=str(e) - ) - return False - - -@activity.defn -@close_db_connections -def list_active_code_teams(_: None = None) -> ListActiveCodeTeamsOutput: - cutoff = timezone.now() - ACTIVITY_WINDOW - team_ids = list( - TaskRun.objects.filter(updated_at__gte=cutoff, task__origin_product__in=CODE_ORIGIN_PRODUCTS) - .order_by("team_id") - .values_list("team_id", flat=True) - .distinct() - ) - - org_enabled: dict[str, bool] = {} - enabled_team_ids: list[int] = [] - for team_id, organization_id in Team.objects.filter(id__in=team_ids).values_list("id", "organization_id"): - org_id = str(organization_id) - if org_id not in org_enabled: - org_enabled[org_id] = _org_home_tab_enabled(org_id) - if org_enabled[org_id]: - enabled_team_ids.append(team_id) - enabled_team_ids.sort() - - truncated = len(enabled_team_ids) > MAX_TEAMS_PER_CYCLE - if truncated: - activity.logger.warning( - "code_workstreams_active_teams_truncated", - total=len(enabled_team_ids), - cap=MAX_TEAMS_PER_CYCLE, - ) - enabled_team_ids = enabled_team_ids[:MAX_TEAMS_PER_CYCLE] - return ListActiveCodeTeamsOutput(team_ids=enabled_team_ids, truncated=truncated) diff --git a/products/tasks/backend/temporal/code_workstreams/activities/load_pr_urls.py b/products/tasks/backend/temporal/code_workstreams/activities/load_pr_urls.py deleted file mode 100644 index 455673666d5f..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/activities/load_pr_urls.py +++ /dev/null @@ -1,76 +0,0 @@ -from dataclasses import dataclass -from typing import Optional - -from django.utils import timezone - -from temporalio import activity - -from posthog.models.github_integration_base import GitHubIntegrationBase -from posthog.temporal.common.utils import close_db_connections - -from products.tasks.backend.models import TaskRun -from products.tasks.backend.temporal.code_workstreams.activities.github_resolution import TeamIntegrationResolver -from products.tasks.backend.temporal.code_workstreams.constants import ACTIVITY_WINDOW, MAX_PRS_PER_TEAM_PER_CYCLE - - -@dataclass -class PrRef: - pr_url: str - github_integration_id: Optional[int] - github_user_integration_id: Optional[str] - - -def _pr_url_belongs_to_task_repo(pr_url: str, repository: Optional[str]) -> bool: - """Whether ``pr_url`` points at the task's own ``repository`` (``owner/repo``). - - ``output.pr_url`` is user-writable, and a team's GitHub App installation can - usually reach far more repos than any one task targeted. Without this check a - user could point ``pr_url`` at an arbitrary PR the installation can see and - have Code Home surface its metadata. The agent always opens the PR against - ``task.repository``, so legitimate runs match; runs without a configured - repository fail closed. - """ - if not repository: - return False - parsed = GitHubIntegrationBase.parse_pull_request_url(pr_url) - if parsed is None: - return False - owner, repo, _ = parsed - return f"{owner}/{repo}".casefold() == repository.casefold() - - -@dataclass -class LoadTeamPrUrlsInput: - team_id: int - - -@dataclass -class LoadTeamPrUrlsOutput: - prs: list[PrRef] - - -@activity.defn -@close_db_connections -def load_team_pr_urls(input: LoadTeamPrUrlsInput) -> LoadTeamPrUrlsOutput: - cutoff = timezone.now() - ACTIVITY_WINDOW - runs = ( - TaskRun.objects.filter(team_id=input.team_id, updated_at__gte=cutoff, output__pr_url__isnull=False) - .select_related("task") - .order_by("-updated_at") - ) - - resolver = TeamIntegrationResolver(input.team_id) - seen: dict[str, PrRef] = {} - for run in runs.iterator(): - url = (run.output or {}).get("pr_url") - if not url or url in seen: - continue - task = run.task - if not _pr_url_belongs_to_task_repo(url, task.repository): - continue - team_int, user_int = resolver.resolve(task) - seen[url] = PrRef(pr_url=url, github_integration_id=team_int, github_user_integration_id=user_int) - if len(seen) >= MAX_PRS_PER_TEAM_PER_CYCLE: - break - - return LoadTeamPrUrlsOutput(prs=list(seen.values())) diff --git a/products/tasks/backend/temporal/code_workstreams/activities/poll_pull_requests.py b/products/tasks/backend/temporal/code_workstreams/activities/poll_pull_requests.py deleted file mode 100644 index 013b3eeb261e..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/activities/poll_pull_requests.py +++ /dev/null @@ -1,141 +0,0 @@ -from __future__ import annotations - -import random -import hashlib -from collections.abc import Callable -from dataclasses import dataclass - -from django.core.exceptions import ObjectDoesNotExist -from django.utils import timezone -from django.utils.dateparse import parse_datetime - -import requests -from temporalio import activity - -from posthog.egress.github.transport import GitHubEgressBudgetExhausted, GitHubRateLimitError -from posthog.models.github_integration_base import GitHubIntegrationBase, GitHubIntegrationError -from posthog.models.scoping import team_scope -from posthog.temporal.common.utils import close_db_connections - -from products.tasks.backend.models import CodePrSnapshot -from products.tasks.backend.temporal.code_workstreams.activities.github_resolution import resolve_github_integration -from products.tasks.backend.temporal.code_workstreams.activities.load_pr_urls import PrRef - - -@dataclass -class PollTeamPullRequestsInput: - team_id: int - prs: list[PrRef] - - -@dataclass -class PollTeamPullRequestsOutput: - polled: int - updated: int - rate_limited: bool - - -def _fingerprint(url: str, updated_at: str | None) -> str: - return hashlib.sha256(f"{url}|{updated_at or ''}".encode()).hexdigest() - - -def _resolve_integration(ref: PrRef) -> GitHubIntegrationBase | None: - return resolve_github_integration(ref.github_integration_id, ref.github_user_integration_id) - - -@activity.defn -@close_db_connections -def poll_team_pull_requests(input: PollTeamPullRequestsInput) -> PollTeamPullRequestsOutput: - return poll_pull_requests_for_team(input.team_id, input.prs, heartbeat=activity.heartbeat) - - -def poll_pull_requests_for_team( - team_id: int, - prs: list[PrRef], - *, - heartbeat: Callable[[int], None] | None = None, -) -> PollTeamPullRequestsOutput: - integrations: dict[str, GitHubIntegrationBase | None] = {} - polled = 0 - updated = 0 - rate_limited = False - - # A shed sweep breaks mid-list; shuffling gives the tail equal coverage across cycles instead - # of permanently starving whatever sits past the shed point of a large team's list. - prs = list(prs) - random.shuffle(prs) - - for index, ref in enumerate(prs): - if heartbeat is not None: - heartbeat(index) - - cache_key = ( - f"i:{ref.github_integration_id}" - if ref.github_integration_id is not None - else f"u:{ref.github_user_integration_id}" - ) - try: - if cache_key not in integrations: - integrations[cache_key] = _resolve_integration(ref) - integration = integrations[cache_key] - except ObjectDoesNotExist: - activity.logger.warning("code_workstreams_pr_integration_missing", pr_url=ref.pr_url) - continue - except (GitHubIntegrationError, GitHubRateLimitError, requests.RequestException) as e: - # A token-refresh failure for one PR must not abort the whole activity (which would - # block the team's rebuild this cycle); skip this PR and move on. - activity.logger.warning("code_workstreams_pr_integration_unavailable", pr_url=ref.pr_url, error=str(e)) - continue - if integration is None: - continue - - try: - snap = integration.get_pull_request_snapshot(ref.pr_url) - except GitHubEgressBudgetExhausted: - # Our own limiter shed the sweep — stop for this cycle; the next scheduled run resumes. - activity.logger.warning("code_workstreams_pr_budget_exhausted", team_id=team_id, polled=polled) - rate_limited = True - break - except GitHubRateLimitError: - activity.logger.warning("code_workstreams_pr_rate_limited", team_id=team_id, polled=polled) - rate_limited = True - break - except GitHubIntegrationError as e: - activity.logger.warning("code_workstreams_pr_fetch_failed", pr_url=ref.pr_url, error=str(e)) - continue - except Exception as e: - activity.logger.warning("code_workstreams_pr_fetch_error", pr_url=ref.pr_url, error=str(e)) - continue - - polled += 1 - if not snap.get("success"): - continue - - fingerprint = _fingerprint(ref.pr_url, snap.get("updated_at")) - with team_scope(team_id): - existing = CodePrSnapshot.objects.filter(team_id=team_id, pr_url=ref.pr_url).first() - if existing is not None and existing.fingerprint == fingerprint: - continue - CodePrSnapshot.objects.update_or_create( - team_id=team_id, - pr_url=ref.pr_url, - defaults={ - "github_integration_id": ref.github_integration_id, - "number": snap.get("number") or 0, - "title": snap.get("title") or "", - "state": snap["state"], - "ci_status": snap["ci_status"], - "review_decision": snap.get("review_decision"), - "unresolved_threads": snap.get("unresolved_threads") or 0, - "mergeable": snap.get("mergeable"), - "author_login": snap.get("author_login"), - "head_branch": snap.get("head_branch"), - "requested_reviewer_logins": snap.get("requested_reviewer_logins") or [], - "pr_updated_at": parse_datetime(snap["updated_at"]) if snap.get("updated_at") else None, - "fingerprint": fingerprint, - "fetched_at": timezone.now(), - }, - ) - updated += 1 - - return PollTeamPullRequestsOutput(polled=polled, updated=updated, rate_limited=rate_limited) diff --git a/products/tasks/backend/temporal/code_workstreams/activities/rebuild_workstreams.py b/products/tasks/backend/temporal/code_workstreams/activities/rebuild_workstreams.py deleted file mode 100644 index f05ece255fec..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/activities/rebuild_workstreams.py +++ /dev/null @@ -1,335 +0,0 @@ -import uuid -from collections import defaultdict -from dataclasses import dataclass -from datetime import UTC, datetime -from typing import Optional - -from django.db.models import F, Max, Q, Window -from django.db.models.functions import RowNumber -from django.utils import timezone - -from temporalio import activity - -from posthog.models.scoping import team_scope -from posthog.models.user_integration import UserIntegration -from posthog.temporal.common.utils import close_db_connections - -from products.tasks.backend.logic.code_workstreams.classify import pick_primary_situation -from products.tasks.backend.logic.code_workstreams.grouping import ( - DEFAULT_BASE_BRANCHES, - PrInput, - TaskInput, - Workstream, - branch_lookup_key, - build_workstreams, -) -from products.tasks.backend.models import CodePrSnapshot, CodeWorkstream, Task, TaskRun -from products.tasks.backend.temporal.code_workstreams.constants import ( - ACTIVITY_WINDOW, - MAX_TASKS_PER_TEAM, - MAX_TASKS_PER_USER, -) -from products.tasks.backend.temporal.process_task.utils import parse_run_state - - -@dataclass -class RebuildTeamWorkstreamsInput: - team_id: int - - -@dataclass -class RebuildTeamWorkstreamsOutput: - users: int - workstreams: int - pruned: int - - -def _epoch_ms(dt: datetime) -> int: - return int(dt.timestamp() * 1000) - - -def _from_epoch_ms(ms: int) -> datetime: - return datetime.fromtimestamp(ms / 1000, tz=UTC) - - -def _repo_name(full_path: Optional[str]) -> Optional[str]: - if not full_path: - return None - return full_path.split("/")[-1] - - -def _pr_url_from_run(run: Optional[TaskRun]) -> Optional[str]: - return (run.output or {}).get("pr_url") if run and run.output else None - - -def _task_pr_is_merged(task: Task, pr_url: Optional[str]) -> bool: - if not pr_url: - return False - for run in task.runs.all(): - output = run.output - if ( - run.team_id == task.team_id - and isinstance(output, dict) - and output.get("pr_url") == pr_url - and output.get("pr_merged") is True - ): - return True - return False - - -def _task_to_input(task: Task) -> tuple[TaskInput, Optional[str]]: - run: Optional[TaskRun] = task.latest_run - last_activity = run.updated_at if run else task.updated_at - cloud_pr_url = _pr_url_from_run(run) - state = parse_run_state(run.state if run else None) - return ( - TaskInput( - id=str(task.id), - title=task.title, - status=run.status if run else None, - last_activity_at=_epoch_ms(last_activity), - repo_name=_repo_name(task.repository), - repo_full_path=task.repository, - branch=run.branch if run else None, - base_branch=state.pr_base_branch, - cloud_pr_url=cloud_pr_url, - cloud_pr_merged=_task_pr_is_merged(task, cloud_pr_url), - folder_path=None, - quick_action=state.home_quick_action, - ), - cloud_pr_url, - ) - - -def _pr_wire(pr: PrInput) -> dict: - return { - "url": pr.url, - "number": pr.number, - "title": pr.title, - "state": pr.state, - "ciStatus": pr.ci_status, - "reviewDecision": pr.review_decision, - "unresolvedThreads": pr.unresolved_threads, - "mergeable": pr.mergeable, - "isCurrentUserRequestedReviewer": pr.is_current_user_requested_reviewer, - "isCurrentUserAuthor": pr.is_current_user_author, - "author": pr.author, - "lastUpdatedAt": pr.last_updated_at, - } - - -def _build_pr_input(snapshot: CodePrSnapshot, user_github_logins: set[str]) -> PrInput: - author = snapshot.author_login - reviewer_logins = snapshot.requested_reviewer_logins or [] - return PrInput( - url=snapshot.pr_url, - number=snapshot.number, - title=snapshot.title, - state=snapshot.state, - ci_status=snapshot.ci_status, - review_decision=snapshot.review_decision, - unresolved_threads=snapshot.unresolved_threads, - mergeable=snapshot.mergeable, - is_current_user_requested_reviewer=bool(user_github_logins.intersection(reviewer_logins)), - is_current_user_author=bool(author and author in user_github_logins), - author=author, - last_updated_at=_epoch_ms(snapshot.pr_updated_at) if snapshot.pr_updated_at else 0, - head_branch=snapshot.head_branch, - ) - - -def _branch_resolution_pref(snapshot: CodePrSnapshot) -> tuple[int, float, str]: - # Order so the best snapshot for a (repo, head_branch) collision is written last (last wins): - # prefer still-open PRs over merged/closed (a reused branch's new PR beats the stale one), - # then the most recently updated, with pr_url as a stable final tiebreaker. - open_score = 1 if snapshot.state in ("open", "draft") else 0 - updated = snapshot.pr_updated_at.timestamp() if snapshot.pr_updated_at else 0.0 - return (open_score, updated, snapshot.pr_url) - - -def _repo_from_pr_url(pr_url: str) -> Optional[str]: - # https://///pull/ -> "/" (host-agnostic, covers enterprise). - idx = pr_url.find("/pull/") - if idx == -1: - return None - segments = [s for s in pr_url[:idx].split("/") if s] - if len(segments) < 2: - return None - owner, repo = segments[-2], segments[-1] - return f"{owner}/{repo}" - - -def _github_logins_by_user(user_ids: list[int]) -> dict[int, set[str]]: - logins: dict[int, set[str]] = defaultdict(set) - for user_id, config in UserIntegration.objects.filter(user_id__in=user_ids, kind="github").values_list( - "user_id", "config" - ): - login = (config or {}).get("github_user", {}).get("login") - if login: - logins[user_id].add(login) - return logins - - -def _select_recent_task_ids(team_id: int, cutoff: datetime) -> list[uuid.UUID]: - # Rank each user's tasks by most-recent run activity and keep only their freshest - # MAX_TASKS_PER_USER, so one high-volume user can't evict another user's tasks. The - # MAX_TASKS_PER_TEAM cap then bounds the whole team's set, still favouring recency. - # Candidates come from runs inside the activity window so the query's working set - # scales with recent activity, not the team's all-time task count. Only runs that - # could form a workstream (a PR, or a branch that isn't a default base — see - # workstream_key) consume cap slots: tasks that can never produce a lane must not - # evict ones that can. task_id tie-breakers keep ranking stable across rebuilds - # when activity timestamps collide, so the cap cutoff can't flicker and prune an - # unchanged task's workstream. - return list( - TaskRun.objects.filter( - team_id=team_id, - updated_at__gte=cutoff, - task__archived=False, - task__deleted=False, - task__created_by__isnull=False, - ) - .filter( - Q(output__pr_url__isnull=False) - | (Q(branch__isnull=False) & ~Q(branch="") & ~Q(branch__in=DEFAULT_BASE_BRANCHES)) - ) - .values("task_id", "task__created_by_id") - .annotate(last_activity=Max("updated_at")) - .annotate( - user_rank=Window( - expression=RowNumber(), - partition_by=F("task__created_by_id"), - order_by=(F("last_activity").desc(), F("task_id").asc()), - ) - ) - .filter(user_rank__lte=MAX_TASKS_PER_USER) - .order_by("-last_activity", "task_id") - .values_list("task_id", flat=True)[:MAX_TASKS_PER_TEAM] - ) - - -@activity.defn -@close_db_connections -def rebuild_team_workstreams(input: RebuildTeamWorkstreamsInput) -> RebuildTeamWorkstreamsOutput: - now = timezone.now() - cutoff = now - ACTIVITY_WINDOW - now_ms = _epoch_ms(now) - - recent_task_ids = _select_recent_task_ids(input.team_id, cutoff) - tasks = list( - Task.objects.filter(id__in=recent_task_ids, team_id=input.team_id, archived=False, deleted=False) - .select_related("created_by") - .prefetch_related("runs") - ) - - by_user: dict[int, list[Task]] = defaultdict(list) - needed_pr_urls: set[str] = set() - needed_branches: set[str] = set() - for task in tasks: - if task.created_by_id is None: - continue - by_user[task.created_by_id].append(task) - run = task.latest_run - pr_url = _pr_url_from_run(run) - if pr_url: - needed_pr_urls.add(pr_url) - # The branch a run actually worked on lets us link follow-up runs (no pr_url of - # their own) to the open PR for that branch. - if run and run.branch: - needed_branches.add(run.branch) - - github_logins_by_user = _github_logins_by_user(list(by_user.keys())) - - total_workstreams = 0 - total_pruned = 0 - - with team_scope(input.team_id): - # Only load snapshots we will actually look up, so memory stays bounded by recent tasks - # rather than the team's all-time snapshot count. We also pull snapshots whose head - # branch matches a run's branch so branch-resolved grouping can find them. - snapshots = list( - CodePrSnapshot.objects.filter(team_id=input.team_id).filter( - Q(pr_url__in=needed_pr_urls) | Q(head_branch__in=needed_branches) - ) - ) - snapshots_by_url = {s.pr_url: s for s in snapshots} - # Sorted so a (repo, head_branch) collision resolves deterministically (best wins last). - snapshots_by_branch = sorted((s for s in snapshots if s.head_branch), key=_branch_resolution_pref) - - for user_id, user_tasks in by_user.items(): - user_github_logins = github_logins_by_user.get(user_id, set()) - task_inputs: list[TaskInput] = [] - pr_by_task: dict[str, PrInput] = {} - pr_by_branch: dict[tuple[str, str], PrInput] = {} - snapshot_id_by_url: dict[str, str] = {} - for task in user_tasks: - task_input, pr_url = _task_to_input(task) - task_inputs.append(task_input) - if pr_url and pr_url in snapshots_by_url: - snapshot = snapshots_by_url[pr_url] - pr_by_task[task_input.id] = _build_pr_input(snapshot, user_github_logins) - snapshot_id_by_url[pr_url] = str(snapshot.id) - - # is_current_user_author depends on the user, so the branch map is built per user. - for snapshot in snapshots_by_branch: - repo = _repo_from_pr_url(snapshot.pr_url) - key = branch_lookup_key(repo, snapshot.head_branch) - if key is None: - continue - pr_by_branch[key] = _build_pr_input(snapshot, user_github_logins) - snapshot_id_by_url[snapshot.pr_url] = str(snapshot.id) - - result = build_workstreams(task_inputs, pr_by_task, now_ms, pr_by_branch) - live_keys: set[str] = set() - for state, workstreams in ( - (CodeWorkstream.WorkstreamState.ATTENTION, result.needs_attention), - (CodeWorkstream.WorkstreamState.IN_PROGRESS, result.in_progress), - ): - for ws in workstreams: - live_keys.add(ws.id) - _persist_workstream(input.team_id, user_id, state, ws, snapshot_id_by_url, now) - total_workstreams += 1 - - pruned, _ = ( - CodeWorkstream.objects.filter(team_id=input.team_id, user_id=user_id) - .exclude(key__in=live_keys) - .delete() - ) - total_pruned += pruned - - return RebuildTeamWorkstreamsOutput( - users=len(by_user), - workstreams=total_workstreams, - pruned=total_pruned, - ) - - -def _persist_workstream( - team_id: int, - user_id: int, - state: str, - ws: Workstream, - snapshot_id_by_url: dict[str, str], - generated_at: datetime, -) -> None: - CodeWorkstream.objects.update_or_create( - team_id=team_id, - user_id=user_id, - key=ws.id, - defaults={ - "repo_name": ws.repo_name, - "repo_full_path": ws.repo_full_path, - "branch": ws.branch, - "pr_url": ws.pr_url, - "pr_snapshot_id": snapshot_id_by_url.get(ws.pr_url) if ws.pr_url else None, - "pr": _pr_wire(ws.pr) if ws.pr else None, - "situations": list(ws.situations), - "primary_situation": pick_primary_situation(ws.situations), - "state": state, - "tasks": [ - {"id": t.id, "title": t.title, "status": t.status, "quick_action": t.quick_action} for t in ws.tasks - ], - "last_activity_at": _from_epoch_ms(ws.last_activity_at), - "generated_at": generated_at, - }, - ) diff --git a/products/tasks/backend/temporal/code_workstreams/client.py b/products/tasks/backend/temporal/code_workstreams/client.py deleted file mode 100644 index a22ca90bc69b..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/client.py +++ /dev/null @@ -1,30 +0,0 @@ -import logging - -from django.conf import settings - -from asgiref.sync import async_to_sync -from temporalio.common import WorkflowIDReusePolicy -from temporalio.exceptions import WorkflowAlreadyStartedError - -from posthog.temporal.common.client import sync_connect - -from products.tasks.backend.temporal.code_workstreams.workflow import EvaluateTeamCodeWorkstreamsInput - -logger = logging.getLogger(__name__) - - -def trigger_team_code_workstreams_evaluation(team_id: int) -> bool: - client = sync_connect() - workflow_id = f"evaluate-team-code-workstreams-ondemand-{team_id}" - try: - async_to_sync(client.start_workflow)( # type: ignore[misc] - "evaluate-team-code-workstreams", # type: ignore[arg-type] - EvaluateTeamCodeWorkstreamsInput(team_id=team_id), # type: ignore[arg-type] - id=workflow_id, - id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE, - task_queue=settings.TASKS_TASK_QUEUE, - ) - return True - except WorkflowAlreadyStartedError: - logger.info("code_workstreams_refresh_already_running", extra={"team_id": team_id}) - return False diff --git a/products/tasks/backend/temporal/code_workstreams/constants.py b/products/tasks/backend/temporal/code_workstreams/constants.py deleted file mode 100644 index e93d06dbb919..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/constants.py +++ /dev/null @@ -1,18 +0,0 @@ -from datetime import timedelta - -HOME_TAB_FLAG = "posthog-code-home-tab" - -ACTIVITY_WINDOW = timedelta(days=30) - -MAX_PRS_PER_TEAM_PER_CYCLE = 50 -# Per-user cap keeps one high-volume user from evicting other users' tasks from the Home tab; -# the per-team cap bounds the total work each rebuild cycle does for a team. -MAX_TASKS_PER_USER = 50 -MAX_TASKS_PER_TEAM = 3000 -MAX_TEAMS_PER_CYCLE = 2000 - -# Caps GitHub API calls per cycle (one per unique repo+branch); separate from -# MAX_PRS_PER_TEAM_PER_CYCLE because one branch can yield 0..n PRs. -MAX_BRANCH_QUERIES_PER_TEAM_PER_CYCLE = 30 - -TEAM_FANOUT_CONCURRENCY = 20 diff --git a/products/tasks/backend/temporal/code_workstreams/schedule.py b/products/tasks/backend/temporal/code_workstreams/schedule.py deleted file mode 100644 index d7a6f4c5f183..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/schedule.py +++ /dev/null @@ -1,37 +0,0 @@ -from dataclasses import asdict -from datetime import timedelta - -from django.conf import settings - -from temporalio.client import ( - Client, - Schedule, - ScheduleActionStartWorkflow, - ScheduleIntervalSpec, - ScheduleOverlapPolicy, - SchedulePolicy, - ScheduleSpec, -) - -from posthog.temporal.common.schedule import a_create_schedule, a_schedule_exists, a_update_schedule - -from products.tasks.backend.temporal.code_workstreams.workflow import EvaluateCodeWorkstreamsInput - -SCHEDULE_ID = "evaluate-code-workstreams-schedule" - - -async def create_evaluate_code_workstreams_schedule(client: Client): - schedule = Schedule( - action=ScheduleActionStartWorkflow( - "evaluate-code-workstreams", - asdict(EvaluateCodeWorkstreamsInput()), - id="evaluate-code-workstreams", - task_queue=settings.TASKS_TASK_QUEUE, - ), - spec=ScheduleSpec(intervals=[ScheduleIntervalSpec(every=timedelta(minutes=3))]), - policy=SchedulePolicy(overlap=ScheduleOverlapPolicy.SKIP), - ) - if await a_schedule_exists(client, SCHEDULE_ID): - await a_update_schedule(client, SCHEDULE_ID, schedule) - else: - await a_create_schedule(client, SCHEDULE_ID, schedule, trigger_immediately=False) diff --git a/products/tasks/backend/temporal/code_workstreams/test_discover_branch_prs.py b/products/tasks/backend/temporal/code_workstreams/test_discover_branch_prs.py deleted file mode 100644 index 811b4dc235a3..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/test_discover_branch_prs.py +++ /dev/null @@ -1,170 +0,0 @@ -import random - -import pytest -from unittest.mock import MagicMock, patch - -from posthog.egress.github.transport import GitHubEgressBudgetExhausted -from posthog.models import Integration, Organization, Team - -from products.tasks.backend.models import Task, TaskRun -from products.tasks.backend.temporal.code_workstreams.activities.discover_branch_prs import ( - DiscoverBranchPrsInput, - _collect_branch_candidates, - discover_branch_prs, -) - -_RESOLVE = "products.tasks.backend.temporal.code_workstreams.activities.discover_branch_prs.resolve_github_integration" - - -def _make_team(): - org = Organization.objects.create(name=f"DiscoverOrg-{random.randint(1, 99999)}") - team = Team.objects.create(organization=org, name=f"DiscoverTeam-{random.randint(1, 99999)}") - # A team-level GitHub integration so TeamIntegrationResolver resolves a usable id. - Integration.objects.create(team=team, kind="github", config={}, sensitive_config={}, errors="") - return team - - -def _run_on_branch(team, repository, branch, *, pr_url=None, status=TaskRun.Status.COMPLETED): - task = Task.objects.create( - team=team, - title="t", - description="d", - origin_product=Task.OriginProduct.USER_CREATED, - repository=repository, - ) - return TaskRun.objects.create( - task=task, - team=team, - status=status, - branch=branch, - output={"pr_url": pr_url} if pr_url else None, - ) - - -def _fake_integration(urls_by_branch): - fake = MagicMock() - fake.find_pull_request_urls_for_branch.side_effect = lambda repo, branch: urls_by_branch.get(branch, []) - return fake - - -@pytest.mark.django_db(transaction=True) -def test_discovers_pr_for_branch_without_recorded_pr_url(activity_environment): - team = _make_team() - _run_on_branch(team, "acme/widgets", "posthog-code/feature-x") - - fake = _fake_integration({"posthog-code/feature-x": ["https://github.com/acme/widgets/pull/7"]}) - with patch(_RESOLVE, return_value=fake): - result = activity_environment.run( - discover_branch_prs, - DiscoverBranchPrsInput(team_id=team.id, known_pr_urls=[], budget=50), - ) - - assert [ref.pr_url for ref in result.prs] == ["https://github.com/acme/widgets/pull/7"] - fake.find_pull_request_urls_for_branch.assert_called_once_with("acme/widgets", "posthog-code/feature-x") - - -@pytest.mark.django_db(transaction=True) -def test_skips_pr_urls_already_known(activity_environment): - team = _make_team() - _run_on_branch(team, "acme/widgets", "posthog-code/feature-x") - - fake = _fake_integration({"posthog-code/feature-x": ["https://github.com/acme/widgets/pull/7"]}) - with patch(_RESOLVE, return_value=fake): - result = activity_environment.run( - discover_branch_prs, - DiscoverBranchPrsInput( - team_id=team.id, - known_pr_urls=["https://github.com/acme/widgets/pull/7"], - budget=50, - ), - ) - - assert result.prs == [] - - -@pytest.mark.django_db(transaction=True) -def test_respects_budget(activity_environment): - team = _make_team() - _run_on_branch(team, "acme/widgets", "branch-a") - _run_on_branch(team, "acme/widgets", "branch-b") - - fake = _fake_integration( - { - "branch-a": ["https://github.com/acme/widgets/pull/1"], - "branch-b": ["https://github.com/acme/widgets/pull/2"], - } - ) - with patch(_RESOLVE, return_value=fake): - result = activity_environment.run( - discover_branch_prs, - DiscoverBranchPrsInput(team_id=team.id, known_pr_urls=[], budget=1), - ) - - assert len(result.prs) == 1 - - -@pytest.mark.django_db(transaction=True) -def test_zero_budget_makes_no_github_calls(activity_environment): - team = _make_team() - _run_on_branch(team, "acme/widgets", "posthog-code/feature-x") - - fake = _fake_integration({"posthog-code/feature-x": ["https://github.com/acme/widgets/pull/7"]}) - with patch(_RESOLVE, return_value=fake) as resolve: - result = activity_environment.run( - discover_branch_prs, - DiscoverBranchPrsInput(team_id=team.id, known_pr_urls=[], budget=0), - ) - - assert result.prs == [] - resolve.assert_not_called() - - -@pytest.mark.django_db(transaction=True) -def test_collect_candidates_dedupes_repo_branch_and_skips_base_and_repoless(): - team = _make_team() - # Same (repo, branch) twice, plus a case-only repo variant — all one candidate. - _run_on_branch(team, "acme/widgets", "posthog-code/feature-x") - _run_on_branch(team, "acme/widgets", "posthog-code/feature-x") - _run_on_branch(team, "AcMe/Widgets", "posthog-code/feature-x") - _run_on_branch(team, "acme/widgets", "main") # base branch: skipped - _run_on_branch(team, None, "posthog-code/feature-y") # no repo: skipped - _run_on_branch(team, "acme/widgets", "posthog-code/feature-z") - - candidates = _collect_branch_candidates(team.id) - - pairs = sorted((c.repository.casefold(), c.branch) for c in candidates) - assert pairs == [ - ("acme/widgets", "posthog-code/feature-x"), - ("acme/widgets", "posthog-code/feature-z"), - ] - - -@pytest.mark.django_db(transaction=True) -def test_collect_candidates_requires_an_integration(): - org = Organization.objects.create(name=f"NoIntOrg-{random.randint(1, 99999)}") - team = Team.objects.create(organization=org, name=f"NoIntTeam-{random.randint(1, 99999)}") - # No integration to authenticate with ⇒ no candidate. - _run_on_branch(team, "acme/widgets", "posthog-code/feature-x") - - assert _collect_branch_candidates(team.id) == [] - - -@pytest.mark.django_db(transaction=True) -def test_discover_stops_the_cycle_when_budget_shed(activity_environment): - # Guards the break-on-shed seam: if GitHubEgressBudgetExhausted stops being caught here it - # falls into the outer except-and-continue, and a shed sweep keeps grinding through denied - # gate checks instead of yielding the cycle. - team = _make_team() - _run_on_branch(team, "acme/widgets", "posthog-code/feature-x") - _run_on_branch(team, "acme/widgets", "posthog-code/feature-y") - - fake = MagicMock() - fake.find_pull_request_urls_for_branch.side_effect = GitHubEgressBudgetExhausted("shed") - with patch(_RESOLVE, return_value=fake): - result = activity_environment.run( - discover_branch_prs, - DiscoverBranchPrsInput(team_id=team.id, known_pr_urls=[], budget=50), - ) - - assert result.prs == [] - fake.find_pull_request_urls_for_branch.assert_called_once() diff --git a/products/tasks/backend/temporal/code_workstreams/test_list_active_teams.py b/products/tasks/backend/temporal/code_workstreams/test_list_active_teams.py deleted file mode 100644 index 8984dc689ad4..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/test_list_active_teams.py +++ /dev/null @@ -1,152 +0,0 @@ -import random - -import pytest -from unittest.mock import patch - -from posthog.models import Organization, Team - -from products.tasks.backend.models import Task, TaskRun -from products.tasks.backend.temporal.code_workstreams.activities.list_active_teams import list_active_code_teams - -FLAG_PATH = ( - "products.tasks.backend.temporal.code_workstreams.activities.list_active_teams.posthoganalytics.feature_enabled" -) - - -def _org() -> Organization: - return Organization.objects.create(name=f"WorkstreamsOrg-{random.randint(1, 99999)}") - - -def _team(org: Organization) -> Team: - return Team.objects.create(organization=org, name=f"WorkstreamsTeam-{random.randint(1, 99999)}") - - -def _active_team( - org: Organization, - origin_product: Task.OriginProduct = Task.OriginProduct.USER_CREATED, -) -> Team: - team = _team(org) - task = Task.objects.create( - team=team, - title="t", - description="d", - origin_product=origin_product, - ) - TaskRun.objects.create(task=task, team=team, status=TaskRun.Status.COMPLETED) - return team - - -@pytest.mark.django_db(transaction=True) -def test_only_returns_teams_whose_org_has_flag(activity_environment): - enabled_org = _org() - disabled_org = _org() - enabled_team = _active_team(enabled_org) - _active_team(disabled_org) - - def _flag(key, distinct_id=None, **kwargs): - return distinct_id == str(enabled_org.id) - - with patch(FLAG_PATH, side_effect=_flag): - result = activity_environment.run(list_active_code_teams, None) - - assert result.team_ids == [enabled_team.id] - assert result.truncated is False - - -@pytest.mark.django_db(transaction=True) -def test_flag_checked_once_per_org_not_per_team(activity_environment): - org = _org() - team_a = _active_team(org) - team_b = _active_team(org) - - with patch(FLAG_PATH, return_value=True) as flag: - result = activity_environment.run(list_active_code_teams, None) - - assert sorted(result.team_ids) == sorted([team_a.id, team_b.id]) - assert flag.call_count == 1 - - -@pytest.mark.django_db(transaction=True) -def test_excludes_org_when_flag_check_raises(activity_environment): - org = _org() - _active_team(org) - - with patch(FLAG_PATH, side_effect=RuntimeError("flag service down")): - result = activity_environment.run(list_active_code_teams, None) - - assert result.team_ids == [] - - -@pytest.mark.django_db(transaction=True) -def test_returns_empty_when_no_orgs_enabled(activity_environment): - org = _org() - _active_team(org) - - with patch(FLAG_PATH, return_value=False): - result = activity_environment.run(list_active_code_teams, None) - - assert result.team_ids == [] - assert result.truncated is False - - -@pytest.mark.django_db(transaction=True) -@pytest.mark.parametrize( - "origin_product", - [ - Task.OriginProduct.USER_CREATED, - Task.OriginProduct.SLACK, - Task.OriginProduct.AUTOMATION, - ], -) -def test_includes_teams_with_posthog_code_origin(activity_environment, origin_product): - org = _org() - team = _active_team(org, origin_product) - - with patch(FLAG_PATH, return_value=True): - result = activity_environment.run(list_active_code_teams, None) - - assert result.team_ids == [team.id] - - -@pytest.mark.django_db(transaction=True) -@pytest.mark.parametrize( - "origin_product", - [ - Task.OriginProduct.ERROR_TRACKING, - Task.OriginProduct.EVAL_CLUSTERS, - Task.OriginProduct.SUPPORT_QUEUE, - Task.OriginProduct.SESSION_SUMMARIES, - Task.OriginProduct.SIGNAL_REPORT, - Task.OriginProduct.SIGNALS_SCOUT, - ], -) -def test_excludes_teams_with_only_non_code_origin(activity_environment, origin_product): - org = _org() - _active_team(org, origin_product) - - # Org has the flag on, but its only recent run came from another product on - # the shared tasks infra — it must not be pulled into workstream evaluation. - with patch(FLAG_PATH, return_value=True): - result = activity_environment.run(list_active_code_teams, None) - - assert result.team_ids == [] - - -@pytest.mark.django_db(transaction=True) -def test_includes_team_with_mix_of_code_and_non_code_runs(activity_environment): - org = _org() - team = _team(org) - code_task = Task.objects.create( - team=team, title="t", description="d", origin_product=Task.OriginProduct.USER_CREATED - ) - other_task = Task.objects.create( - team=team, title="t", description="d", origin_product=Task.OriginProduct.ERROR_TRACKING - ) - TaskRun.objects.create(task=code_task, team=team, status=TaskRun.Status.COMPLETED) - TaskRun.objects.create(task=other_task, team=team, status=TaskRun.Status.COMPLETED) - - # A single PostHog Code run qualifies the team even amid other-product runs. - with patch(FLAG_PATH, return_value=True): - result = activity_environment.run(list_active_code_teams, None) - - assert result.team_ids == [team.id] diff --git a/products/tasks/backend/temporal/code_workstreams/test_load_pr_urls.py b/products/tasks/backend/temporal/code_workstreams/test_load_pr_urls.py deleted file mode 100644 index 19f6c24e817e..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/test_load_pr_urls.py +++ /dev/null @@ -1,66 +0,0 @@ -import random - -import pytest - -from posthog.models import Organization, Team - -from products.tasks.backend.models import Task, TaskRun -from products.tasks.backend.temporal.code_workstreams.activities.load_pr_urls import ( - LoadTeamPrUrlsInput, - _pr_url_belongs_to_task_repo, - load_team_pr_urls, -) - - -@pytest.mark.parametrize( - "pr_url,repository,expected", - [ - ("https://github.com/acme/widgets/pull/12", "acme/widgets", True), - # Repo names are case-insensitive on GitHub; the stored value is lowercased. - ("https://github.com/AcMe/Widgets/pull/12", "acme/widgets", True), - # Same repo name under a different owner must not match. - ("https://github.com/evil/widgets/pull/12", "acme/widgets", False), - # Same owner, different repo. - ("https://github.com/acme/secrets/pull/12", "acme/widgets", False), - # No configured repository ⇒ fail closed. - ("https://github.com/acme/widgets/pull/12", None, False), - ("https://github.com/acme/widgets/pull/12", "", False), - # Not a parseable GitHub PR URL. - ("https://github.com/acme/widgets", "acme/widgets", False), - ("https://example.com/acme/widgets/pull/12", "acme/widgets", False), - ("not-a-url", "acme/widgets", False), - ], -) -def test_pr_url_belongs_to_task_repo(pr_url, repository, expected): - assert _pr_url_belongs_to_task_repo(pr_url, repository) is expected - - -@pytest.mark.django_db(transaction=True) -def test_load_team_pr_urls_drops_cross_repo_pr_urls(activity_environment): - org = Organization.objects.create(name=f"PrUrlsOrg-{random.randint(1, 99999)}") - team = Team.objects.create(organization=org, name=f"PrUrlsTeam-{random.randint(1, 99999)}") - - def _run_with_pr(repository, pr_url): - task = Task.objects.create( - team=team, - title="t", - description="d", - origin_product=Task.OriginProduct.USER_CREATED, - repository=repository, - ) - TaskRun.objects.create( - task=task, - team=team, - status=TaskRun.Status.COMPLETED, - output={"pr_url": pr_url}, - ) - - _run_with_pr("acme/widgets", "https://github.com/acme/widgets/pull/1") - # User-writable output.pr_url pointed at a repo the task never targeted. - _run_with_pr("acme/widgets", "https://github.com/acme/secrets/pull/9") - # Run without a configured repository must not leak its PR either. - _run_with_pr(None, "https://github.com/acme/widgets/pull/2") - - result = activity_environment.run(load_team_pr_urls, LoadTeamPrUrlsInput(team_id=team.id)) - - assert [ref.pr_url for ref in result.prs] == ["https://github.com/acme/widgets/pull/1"] diff --git a/products/tasks/backend/temporal/code_workstreams/test_poll_pull_requests.py b/products/tasks/backend/temporal/code_workstreams/test_poll_pull_requests.py deleted file mode 100644 index 3fcfbad96d14..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/test_poll_pull_requests.py +++ /dev/null @@ -1,41 +0,0 @@ -from unittest.mock import MagicMock, patch - -from parameterized import parameterized - -from posthog.egress.github.transport import GitHubEgressBudgetExhausted, GitHubRateLimitError - -from products.tasks.backend.temporal.code_workstreams.activities.load_pr_urls import PrRef -from products.tasks.backend.temporal.code_workstreams.activities.poll_pull_requests import poll_pull_requests_for_team - -_RESOLVE = "products.tasks.backend.temporal.code_workstreams.activities.poll_pull_requests._resolve_integration" - - -def _refs(count: int) -> list[PrRef]: - return [ - PrRef( - pr_url=f"https://github.com/acme/widgets/pull/{n}", github_integration_id=1, github_user_integration_id=None - ) - for n in range(1, count + 1) - ] - - -# Guards the shed/backoff seam: if either exception stops being caught here, the whole team sweep -# activity fails and Temporal re-runs it immediately — hammering the budget it was just shed from — -# instead of yielding until the next scheduled cycle. -@parameterized.expand( - [ - ("egress_budget_shed", GitHubEgressBudgetExhausted("shed")), - ("github_rate_limited", GitHubRateLimitError("429", retry_after=60)), - ] -) -def test_poll_stops_the_cycle_when_shed_or_rate_limited(_name: str, exc: Exception) -> None: - integration = MagicMock() - integration.get_pull_request_snapshot.side_effect = exc - - with patch(_RESOLVE, return_value=integration): - result = poll_pull_requests_for_team(1, _refs(3)) - - assert result.rate_limited is True - assert result.polled == 0 - # break, not continue: one attempt, then yield the remaining PRs to the next cycle - integration.get_pull_request_snapshot.assert_called_once() diff --git a/products/tasks/backend/temporal/code_workstreams/test_rebuild_workstreams.py b/products/tasks/backend/temporal/code_workstreams/test_rebuild_workstreams.py deleted file mode 100644 index 878b6b5357df..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/test_rebuild_workstreams.py +++ /dev/null @@ -1,310 +0,0 @@ -import random -from collections import Counter -from datetime import UTC, datetime, timedelta -from types import SimpleNamespace -from typing import cast - -import pytest -from freezegun import freeze_time -from unittest.mock import patch - -from django.utils import timezone - -from posthog.models import Organization, Team, User - -from products.tasks.backend.models import CodePrSnapshot, CodeWorkstream, Task, TaskRun -from products.tasks.backend.temporal.code_workstreams.activities import rebuild_workstreams -from products.tasks.backend.temporal.code_workstreams.activities.rebuild_workstreams import ( - RebuildTeamWorkstreamsInput, - _branch_resolution_pref, - _build_pr_input, - _repo_from_pr_url, - _select_recent_task_ids, - _task_to_input, - rebuild_team_workstreams, -) -from products.tasks.backend.temporal.code_workstreams.constants import ACTIVITY_WINDOW -from products.tasks.backend.temporal.process_task.utils import parse_run_state - - -def _snapshot(**overrides) -> CodePrSnapshot: - defaults = { - "pr_url": "https://github.com/org/repo/pull/1", - "number": 1, - "title": "PR", - "state": "open", - "ci_status": "passing", - "review_decision": None, - "unresolved_threads": 2, - "mergeable": True, - "author_login": "octocat", - "requested_reviewer_logins": ["reviewer1"], - "pr_updated_at": None, - } - defaults.update(overrides) - return CodePrSnapshot(**defaults) - - -@pytest.mark.parametrize( - "state,expected", - [ - ({"pr_base_branch": "master"}, "master"), - ({"pr_base_branch": "main", "mode": "background"}, "main"), - ({"mode": "background"}, None), - ({}, None), - (None, None), - ], -) -def test_parse_run_state_reads_pr_base_branch(state, expected): - assert parse_run_state(state).pr_base_branch == expected - - -@pytest.mark.parametrize( - "pr_url,expected", - [ - ("https://github.com/posthog/posthog/pull/123", "posthog/posthog"), - ("https://github.com/owner/repo/pull/1", "owner/repo"), - ("https://github.enterprise.com/posthog/posthog/pull/9", "posthog/posthog"), - ("not-a-url", None), - ("https://github.com/onlyowner", None), - ], -) -def test_repo_from_pr_url(pr_url, expected): - assert _repo_from_pr_url(pr_url) == expected - - -def test_task_to_input_preserves_merge_signal_from_older_run(): - pr_url = "https://github.com/org/repo/pull/1" - merged_run = SimpleNamespace( - id="merged-run", - team_id=1, - created_at=datetime(2026, 5, 30, tzinfo=UTC), - updated_at=datetime(2026, 5, 30, tzinfo=UTC), - status=TaskRun.Status.COMPLETED, - branch=None, - state={}, - output={"pr_url": pr_url, "pr_merged": True}, - ) - latest_run = SimpleNamespace( - id="latest-run", - team_id=1, - created_at=datetime(2026, 5, 31, tzinfo=UTC), - updated_at=datetime(2026, 5, 31, tzinfo=UTC), - status=TaskRun.Status.COMPLETED, - branch=None, - state={}, - output={"pr_url": pr_url}, - ) - task = cast( - Task, - SimpleNamespace( - id="task-id", - team_id=1, - title="Task", - updated_at=datetime(2026, 5, 31, tzinfo=UTC), - repository=None, - latest_run=latest_run, - runs=SimpleNamespace(all=lambda: [merged_run, latest_run]), - ), - ) - - task_input, _ = _task_to_input(task) - - assert task_input.cloud_pr_merged is True - - -def test_build_pr_input_carries_head_branch(): - pr = _build_pr_input(_snapshot(head_branch="feat/x"), set()) - assert pr.head_branch == "feat/x" - - -def test_branch_resolution_pref_prefers_open_then_recent(): - old = datetime(2026, 1, 1, tzinfo=UTC) - new = datetime(2026, 6, 1, tzinfo=UTC) - closed_new = _snapshot(pr_url="c", state="closed", pr_updated_at=new) - open_old = _snapshot(pr_url="a", state="open", pr_updated_at=old) - open_new = _snapshot(pr_url="b", state="open", pr_updated_at=new) - # Sorting ascending puts the winner last (last-wins when building the map). - winner = sorted([closed_new, open_old, open_new], key=_branch_resolution_pref)[-1] - assert winner.pr_url == "b" - - -@pytest.mark.parametrize( - "state,expected", - [ - ({"home_quick_action": "Fix CI"}, "Fix CI"), - ({"mode": "background"}, None), - (None, None), - ], -) -def test_parse_run_state_reads_home_quick_action(state, expected): - assert parse_run_state(state).home_quick_action == expected - - -@pytest.mark.parametrize( - "author_login,user_github_logins,expected", - [ - ("octocat", {"octocat"}, True), - ("octocat", {"octocat", "alt"}, True), - ("octocat", {"someone-else"}, False), - ("octocat", set(), False), - (None, {"octocat"}, False), - ], -) -def test_build_pr_input_is_author_requires_identity_match(author_login, user_github_logins, expected): - pr = _build_pr_input(_snapshot(author_login=author_login), user_github_logins) - assert pr.is_current_user_author is expected - - -@pytest.mark.parametrize( - "requested_reviewer_logins,user_github_logins,expected", - [ - (["alice", "bob"], {"bob"}, True), - (["alice", "bob"], {"carol"}, False), - ([], {"bob"}, False), - (["alice"], set(), False), - ], -) -def test_build_pr_input_is_requested_reviewer_requires_identity_match( - requested_reviewer_logins, user_github_logins, expected -): - pr = _build_pr_input(_snapshot(requested_reviewer_logins=requested_reviewer_logins), user_github_logins) - assert pr.is_current_user_requested_reviewer is expected - - -def _org() -> Organization: - return Organization.objects.create(name=f"WsOrg-{random.randint(1, 10**9)}") - - -def _team(org: Organization) -> Team: - return Team.objects.create(organization=org, name=f"WsTeam-{random.randint(1, 10**9)}") - - -def _user(org: Organization) -> User: - return User.objects.create_and_join(org, f"u{random.randint(1, 10**9)}@example.com", None) - - -def _task_with_run_at( - team: Team, - user: User, - activity_at: datetime, - branch: str | None = "feat/home", - output: dict | None = None, -) -> Task: - task = Task.objects.create( - team=team, created_by=user, title="t", description="d", origin_product=Task.OriginProduct.USER_CREATED - ) - run = TaskRun.objects.create(task=task, team=team, status=TaskRun.Status.COMPLETED, branch=branch, output=output) - # updated_at is auto_now, so set the activity timestamp with a bulk update to bypass it. - TaskRun.objects.filter(id=run.id).update(updated_at=activity_at) - return task - - -@pytest.mark.django_db -@freeze_time("2026-06-01") -def test_select_recent_task_ids_caps_per_user_and_keeps_low_volume_user(): - org = _org() - team = _team(org) - heavy = _user(org) - light = _user(org) - now = timezone.now() - - # Heavy user floods the team with 55 very recent tasks; only its freshest 50 should survive. - heavy_tasks = [_task_with_run_at(team, heavy, now - timedelta(minutes=i)) for i in range(1, 56)] - # Light user's older tasks must not be evicted by the heavy user's firehose. - light_tasks = [_task_with_run_at(team, light, now - timedelta(hours=5, minutes=i)) for i in range(1, 4)] - - selected = set(_select_recent_task_ids(team.id, now - ACTIVITY_WINDOW)) - - by_user = Counter(t.created_by_id for t in heavy_tasks + light_tasks if t.id in selected) - assert by_user[heavy.id] == 50 - assert by_user[light.id] == 3 - - # The dropped heavy-user tasks are its five oldest. - dropped = {t.id for t in heavy_tasks if t.id not in selected} - assert dropped == {t.id for t in heavy_tasks[-5:]} - - -@pytest.mark.django_db -@freeze_time("2026-06-01") -def test_select_recent_task_ids_applies_team_cap_across_users(): - org = _org() - team = _team(org) - now = timezone.now() - for _ in range(3): - user = _user(org) - for i in range(1, 5): - _task_with_run_at(team, user, now - timedelta(minutes=i)) - - # Per-user cap admits all 12 tasks; the team cap is the ceiling that bounds the set. - with patch.object(rebuild_workstreams, "MAX_TASKS_PER_TEAM", 7): - selected = _select_recent_task_ids(team.id, now - ACTIVITY_WINDOW) - - assert len(selected) == 7 - - -@pytest.mark.django_db -@freeze_time("2026-06-01") -def test_select_recent_task_ids_breaks_activity_ties_deterministically(): - org = _org() - team = _team(org) - user = _user(org) - now = timezone.now() - - # More tied tasks than the cap: without a stable tie-breaker the rank-50 cutoff - # shuffles between rebuilds and the prune deletes an unchanged task's workstream. - tied = [_task_with_run_at(team, user, now - timedelta(hours=1)) for _ in range(55)] - - selected = _select_recent_task_ids(team.id, now - ACTIVITY_WINDOW) - - assert selected == sorted(t.id for t in tied)[:50] - - -@pytest.mark.django_db -@freeze_time("2026-06-01") -def test_select_recent_task_ids_ignores_tasks_that_cannot_form_workstreams(): - org = _org() - team = _team(org) - user = _user(org) - now = timezone.now() - - # A fresh burst of lane-ineligible tasks (no PR, no feature branch) exceeds the cap... - junk = [ - _task_with_run_at(team, user, now - timedelta(minutes=i), branch=branch) - for i, branch in enumerate([None, "", "master", "main"] * 13, start=1) - ] - # ...but must not consume slots and evict the user's older lane-bearing tasks. - branch_task = _task_with_run_at(team, user, now - timedelta(hours=2)) - pr_task = _task_with_run_at( - team, - user, - now - timedelta(hours=3), - branch=None, - output={"pr_url": "https://github.com/org/repo/pull/1"}, - ) - - selected = set(_select_recent_task_ids(team.id, now - ACTIVITY_WINDOW)) - - assert branch_task.id in selected - assert pr_task.id in selected - assert selected.isdisjoint({t.id for t in junk}) - - -@pytest.mark.django_db -@freeze_time("2026-06-01") -def test_rebuild_omits_run_with_merged_pr_without_snapshot(): - org = _org() - team = _team(org) - user = _user(org) - _task_with_run_at( - team, - user, - timezone.now() - timedelta(days=10), - branch=None, - output={"pr_url": "https://github.com/org/repo/pull/1", "pr_merged": True}, - ) - - result = rebuild_team_workstreams(RebuildTeamWorkstreamsInput(team_id=team.id)) - - assert result.workstreams == 0 - assert not CodeWorkstream.objects.for_team(team.id).filter(user=user).exists() diff --git a/products/tasks/backend/temporal/code_workstreams/test_workflow.py b/products/tasks/backend/temporal/code_workstreams/test_workflow.py deleted file mode 100644 index 4c74fecd5d04..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/test_workflow.py +++ /dev/null @@ -1,158 +0,0 @@ -import uuid - -import pytest - -import temporalio.worker -from temporalio import activity -from temporalio.testing import WorkflowEnvironment -from temporalio.worker import Worker - -from products.tasks.backend.temporal.code_workstreams.activities.discover_branch_prs import DiscoverBranchPrsOutput -from products.tasks.backend.temporal.code_workstreams.activities.list_active_teams import ListActiveCodeTeamsOutput -from products.tasks.backend.temporal.code_workstreams.activities.load_pr_urls import ( - LoadTeamPrUrlsInput, - LoadTeamPrUrlsOutput, - PrRef, -) -from products.tasks.backend.temporal.code_workstreams.activities.poll_pull_requests import PollTeamPullRequestsOutput -from products.tasks.backend.temporal.code_workstreams.activities.rebuild_workstreams import RebuildTeamWorkstreamsOutput -from products.tasks.backend.temporal.code_workstreams.workflow import ( - EvaluateCodeWorkstreamsInput, - EvaluateCodeWorkstreamsWorkflow, - EvaluateTeamCodeWorkstreamsInput, - EvaluateTeamCodeWorkstreamsWorkflow, -) - - -def _team_pipeline_activities(calls: list[str]): - @activity.defn(name="load_team_pr_urls") - async def load(input) -> LoadTeamPrUrlsOutput: - calls.append("load") - return LoadTeamPrUrlsOutput( - prs=[ - PrRef(pr_url="https://github.com/o/r/pull/1", github_integration_id=1, github_user_integration_id=None) - ] - ) - - @activity.defn(name="discover_branch_prs") - async def discover(input) -> DiscoverBranchPrsOutput: - calls.append("discover") - return DiscoverBranchPrsOutput(prs=[]) - - @activity.defn(name="poll_team_pull_requests") - async def poll(input) -> PollTeamPullRequestsOutput: - calls.append("poll") - return PollTeamPullRequestsOutput(polled=1, updated=1, rate_limited=False) - - @activity.defn(name="rebuild_team_workstreams") - async def rebuild(input) -> RebuildTeamWorkstreamsOutput: - calls.append("rebuild") - return RebuildTeamWorkstreamsOutput(users=1, workstreams=2, pruned=0) - - return [load, discover, poll, rebuild] - - -@pytest.mark.asyncio -async def test_team_workflow_runs_pipeline_in_order(): - calls: list[str] = [] - task_queue = str(uuid.uuid4()) - async with await WorkflowEnvironment.start_time_skipping() as env: - async with Worker( - env.client, - task_queue=task_queue, - workflows=[EvaluateTeamCodeWorkstreamsWorkflow], - activities=_team_pipeline_activities(calls), - workflow_runner=temporalio.worker.UnsandboxedWorkflowRunner(), - ): - await env.client.execute_workflow( - EvaluateTeamCodeWorkstreamsWorkflow.run, - EvaluateTeamCodeWorkstreamsInput(team_id=1), - id=str(uuid.uuid4()), - task_queue=task_queue, - ) - assert calls == ["load", "discover", "poll", "rebuild"] - - -@pytest.mark.asyncio -async def test_team_workflow_skips_poll_when_no_prs(): - calls: list[str] = [] - - @activity.defn(name="load_team_pr_urls") - async def load(input) -> LoadTeamPrUrlsOutput: - calls.append("load") - return LoadTeamPrUrlsOutput(prs=[]) - - @activity.defn(name="discover_branch_prs") - async def discover(input) -> DiscoverBranchPrsOutput: - calls.append("discover") - return DiscoverBranchPrsOutput(prs=[]) - - @activity.defn(name="poll_team_pull_requests") - async def poll(input) -> PollTeamPullRequestsOutput: - calls.append("poll") - return PollTeamPullRequestsOutput(polled=0, updated=0, rate_limited=False) - - @activity.defn(name="rebuild_team_workstreams") - async def rebuild(input) -> RebuildTeamWorkstreamsOutput: - calls.append("rebuild") - return RebuildTeamWorkstreamsOutput(users=0, workstreams=0, pruned=0) - - task_queue = str(uuid.uuid4()) - async with await WorkflowEnvironment.start_time_skipping() as env: - async with Worker( - env.client, - task_queue=task_queue, - workflows=[EvaluateTeamCodeWorkstreamsWorkflow], - activities=[load, discover, poll, rebuild], - workflow_runner=temporalio.worker.UnsandboxedWorkflowRunner(), - ): - await env.client.execute_workflow( - EvaluateTeamCodeWorkstreamsWorkflow.run, - EvaluateTeamCodeWorkstreamsInput(team_id=1), - id=str(uuid.uuid4()), - task_queue=task_queue, - ) - assert calls == ["load", "discover", "rebuild"] - - -@pytest.mark.asyncio -async def test_dispatcher_fans_out_per_team(): - started_teams: list[int] = [] - - @activity.defn(name="list_active_code_teams") - async def list_teams(input=None) -> ListActiveCodeTeamsOutput: - return ListActiveCodeTeamsOutput(team_ids=[1, 2, 3], truncated=False) - - @activity.defn(name="load_team_pr_urls") - async def load(input: LoadTeamPrUrlsInput) -> LoadTeamPrUrlsOutput: - started_teams.append(input.team_id) - return LoadTeamPrUrlsOutput(prs=[]) - - @activity.defn(name="discover_branch_prs") - async def discover(input) -> DiscoverBranchPrsOutput: - return DiscoverBranchPrsOutput(prs=[]) - - @activity.defn(name="poll_team_pull_requests") - async def poll(input) -> PollTeamPullRequestsOutput: - return PollTeamPullRequestsOutput(polled=0, updated=0, rate_limited=False) - - @activity.defn(name="rebuild_team_workstreams") - async def rebuild(input) -> RebuildTeamWorkstreamsOutput: - return RebuildTeamWorkstreamsOutput(users=0, workstreams=0, pruned=0) - - task_queue = str(uuid.uuid4()) - async with await WorkflowEnvironment.start_time_skipping() as env: - async with Worker( - env.client, - task_queue=task_queue, - workflows=[EvaluateCodeWorkstreamsWorkflow, EvaluateTeamCodeWorkstreamsWorkflow], - activities=[list_teams, load, discover, poll, rebuild], - workflow_runner=temporalio.worker.UnsandboxedWorkflowRunner(), - ): - await env.client.execute_workflow( - EvaluateCodeWorkstreamsWorkflow.run, - EvaluateCodeWorkstreamsInput(), - id=str(uuid.uuid4()), - task_queue=task_queue, - ) - assert sorted(started_teams) == [1, 2, 3] diff --git a/products/tasks/backend/temporal/code_workstreams/workflow.py b/products/tasks/backend/temporal/code_workstreams/workflow.py deleted file mode 100644 index c108f3fe48fc..000000000000 --- a/products/tasks/backend/temporal/code_workstreams/workflow.py +++ /dev/null @@ -1,140 +0,0 @@ -import json -import asyncio -from dataclasses import dataclass -from datetime import timedelta - -from temporalio import workflow -from temporalio.common import RetryPolicy, WorkflowIDReusePolicy - -from posthog.temporal.common.base import PostHogWorkflow - -from products.tasks.backend.temporal.code_workstreams.activities.discover_branch_prs import ( - DiscoverBranchPrsInput, - DiscoverBranchPrsOutput, - discover_branch_prs, -) -from products.tasks.backend.temporal.code_workstreams.activities.list_active_teams import ( - ListActiveCodeTeamsOutput, - list_active_code_teams, -) -from products.tasks.backend.temporal.code_workstreams.activities.load_pr_urls import ( - LoadTeamPrUrlsInput, - LoadTeamPrUrlsOutput, - load_team_pr_urls, -) -from products.tasks.backend.temporal.code_workstreams.activities.poll_pull_requests import ( - PollTeamPullRequestsInput, - poll_team_pull_requests, -) -from products.tasks.backend.temporal.code_workstreams.activities.rebuild_workstreams import ( - RebuildTeamWorkstreamsInput, - rebuild_team_workstreams, -) -from products.tasks.backend.temporal.code_workstreams.constants import ( - MAX_PRS_PER_TEAM_PER_CYCLE, - TEAM_FANOUT_CONCURRENCY, -) - - -@dataclass -class EvaluateTeamCodeWorkstreamsInput: - team_id: int - - -@dataclass -class EvaluateCodeWorkstreamsInput: - pass - - -@workflow.defn(name="evaluate-team-code-workstreams") -class EvaluateTeamCodeWorkstreamsWorkflow(PostHogWorkflow): - @staticmethod - def parse_inputs(inputs: list[str]) -> EvaluateTeamCodeWorkstreamsInput: - loaded = json.loads(inputs[0]) - return EvaluateTeamCodeWorkstreamsInput(team_id=loaded["team_id"]) - - @workflow.run - async def run(self, input: EvaluateTeamCodeWorkstreamsInput) -> None: - pr_urls: LoadTeamPrUrlsOutput = await workflow.execute_activity( - load_team_pr_urls, - LoadTeamPrUrlsInput(team_id=input.team_id), - start_to_close_timeout=timedelta(minutes=1), - retry_policy=RetryPolicy(maximum_attempts=3), - ) - - prs = list(pr_urls.prs) - - # Branch discovery surfaces PRs whose run never wrote output.pr_url. It calls GitHub, so it - # runs as its own heartbeated activity rather than blocking the DB-only load step. - budget = MAX_PRS_PER_TEAM_PER_CYCLE - len(prs) - if budget > 0: - discovered: DiscoverBranchPrsOutput = await workflow.execute_activity( - discover_branch_prs, - DiscoverBranchPrsInput( - team_id=input.team_id, - known_pr_urls=[pr.pr_url for pr in prs], - budget=budget, - ), - start_to_close_timeout=timedelta(minutes=5), - heartbeat_timeout=timedelta(minutes=2), - retry_policy=RetryPolicy(maximum_attempts=2), - ) - prs.extend(discovered.prs) - - if prs: - await workflow.execute_activity( - poll_team_pull_requests, - PollTeamPullRequestsInput(team_id=input.team_id, prs=prs), - start_to_close_timeout=timedelta(minutes=10), - heartbeat_timeout=timedelta(minutes=2), - retry_policy=RetryPolicy(maximum_attempts=3), - ) - - await workflow.execute_activity( - rebuild_team_workstreams, - RebuildTeamWorkstreamsInput(team_id=input.team_id), - start_to_close_timeout=timedelta(minutes=5), - retry_policy=RetryPolicy(maximum_attempts=3), - ) - - -@workflow.defn(name="evaluate-code-workstreams") -class EvaluateCodeWorkstreamsWorkflow(PostHogWorkflow): - @staticmethod - def parse_inputs(inputs: list[str]) -> EvaluateCodeWorkstreamsInput: - return EvaluateCodeWorkstreamsInput() - - @workflow.run - async def run(self, input: EvaluateCodeWorkstreamsInput) -> None: - active: ListActiveCodeTeamsOutput = await workflow.execute_activity( - list_active_code_teams, - start_to_close_timeout=timedelta(minutes=2), - retry_policy=RetryPolicy(maximum_attempts=3), - ) - if not active.team_ids: - return - - parent_id = workflow.info().workflow_id - semaphore = asyncio.Semaphore(TEAM_FANOUT_CONCURRENCY) - - async def evaluate_team(team_id: int) -> None: - async with semaphore: - await workflow.execute_child_workflow( - EvaluateTeamCodeWorkstreamsWorkflow.run, - EvaluateTeamCodeWorkstreamsInput(team_id=team_id), - id=f"{parent_id}-team-{team_id}", - id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE, - retry_policy=RetryPolicy(maximum_attempts=1), - ) - - results = await asyncio.gather( - *(evaluate_team(team_id) for team_id in active.team_ids), - return_exceptions=True, - ) - failures = [r for r in results if isinstance(r, Exception)] - if failures: - workflow.logger.warning( - "code_workstreams_dispatch_partial_failures", - total=len(active.team_ids), - failures=len(failures), - ) diff --git a/products/tasks/backend/temporal/process_task/utils.py b/products/tasks/backend/temporal/process_task/utils.py index d5cc23d3690a..242f9e67a667 100644 --- a/products/tasks/backend/temporal/process_task/utils.py +++ b/products/tasks/backend/temporal/process_task/utils.py @@ -295,7 +295,6 @@ class RunState(BaseModel, extra="allow"): auto_publish: bool | None = None github_credential_source: GitHubCredentialSource | None = None pr_base_branch: str | None = None - home_quick_action: str | None = None run_source: RunSource | None = None signal_report_id: str | None = None runtime_adapter: RuntimeAdapter | None = None diff --git a/products/tasks/frontend/generated/api.schemas.ts b/products/tasks/frontend/generated/api.schemas.ts index f7b059a1a767..522bf387ac2a 100644 --- a/products/tasks/frontend/generated/api.schemas.ts +++ b/products/tasks/frontend/generated/api.schemas.ts @@ -2339,11 +2339,6 @@ export interface TaskRunBootstrapCreateRequestApi { * @nullable */ rtk_enabled?: boolean | null - /** - * Label of the Home-tab quick action that started this run (e.g. 'Fix CI'), surfaced on the workstream. - * @maxLength 120 - */ - home_quick_action?: string } /** diff --git a/products/tasks/frontend/generated/api.zod.ts b/products/tasks/frontend/generated/api.zod.ts index 34d95706e919..93b85b99f591 100644 --- a/products/tasks/frontend/generated/api.zod.ts +++ b/products/tasks/frontend/generated/api.zod.ts @@ -2013,8 +2013,6 @@ export const tasksRunsCreateBodyEnvironmentDefault = `local` export const tasksRunsCreateBodyModeDefault = `background` export const tasksRunsCreateBodyBranchMax = 255 -export const tasksRunsCreateBodyHomeQuickActionMax = 120 - export const TasksRunsCreateBody = /* @__PURE__ */ zod .object({ imported_mcp_servers: zod @@ -2140,13 +2138,6 @@ export const TasksRunsCreateBody = /* @__PURE__ */ zod .describe( 'Whether rtk command-output compression is enabled for this run. Omitted or null follows the server-side default (enabled); false opts this run out.' ), - home_quick_action: zod - .string() - .max(tasksRunsCreateBodyHomeQuickActionMax) - .optional() - .describe( - "Label of the Home-tab quick action that started this run (e.g. 'Fix CI'), surfaced on the workstream." - ), }) .describe('Request body for creating a task run without starting execution yet.') diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index 79a9bd9fee95..858145e9030f 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -66922,11 +66922,6 @@ export namespace Schemas { * @nullable */ rtk_enabled?: boolean | null; - /** - * Label of the Home-tab quick action that started this run (e.g. 'Fix CI'), surfaced on the workstream. - * @maxLength 120 - */ - home_quick_action?: string; } export interface TaskRunCancelRequest {