insights/products/tasks/backend/facade/api.py

"""
Facade API for the tasks product — the data surface other apps may import.

Responsibilities:
- Accept ids / DTOs as input.
- Call into the product's models and logic.
- Convert Django models to DTOs before returning — never return ORM instances.
- Stay thin and stable.

This module is deliberately light: it imports the models and small helpers only. The
heavy behavioral surfaces (sandbox provisioning, warming, the multi-turn agent machinery,
temporal workflows, max tools) live in sibling facade submodules (``sandbox``, ``warm``,
``agents``, ``temporal``, ``max_tools``, ``webhooks``, ``streams``, ``repo_selection``) so a
config-only importer never drags docker/temporalio onto the ``django.setup()`` path.
Functions that bridge to those heavy surfaces import them lazily inside the function body.
"""

import re
import hashlib
import logging
from collections.abc import Collection, Iterable, Sequence
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Literal
from urllib.parse import urlparse
from uuid import UUID, uuid4

from django.conf import settings
from django.core.exceptions import ValidationError as DjangoValidationError
from django.db import IntegrityError, transaction
from django.db.models import CharField, Count, Exists, F, Func, Min, OuterRef, Q, QuerySet, Subquery
from django.db.models.fields.json import KeyTextTransform
from django.utils import timezone as django_timezone

import hanzo_insights

from insights.event_usage import groups
from insights.models import Team, User
from insights.models.integration import Integration

from products.tasks.backend.constants import (
    AGENT_OTEL_TELEMETRY_STATE_KEY,
    MAX_CUSTOM_IMAGES_PER_TEAM,
    MAX_CUSTOM_IMAGES_PER_USER,
    PI_CLOUD_RUNTIME_FEATURE_FLAG,
    RESERVED_SANDBOX_ENVIRONMENT_VARIABLE_KEYS,
    TASK_SESSION_MAX_SIZE_BYTES,
    is_blocked_sandbox_env_key,
)
from products.tasks.backend.error_telemetry import truncate_error_message
from products.tasks.backend.github_repository_access import (
    inaccessible_repositories_via_integration as _inaccessible_repositories_via_integration,
)
from products.tasks.backend.logic.services.image_builder import (
    ensure_image_builder_task,
    is_custom_images_enabled,
    read_spec_from_builder_sandbox,
)
from products.tasks.backend.mentions import resolve_mentioned_user_ids
from products.tasks.backend.models import (
    Channel,
    ChannelContextGeneration,
    ChannelFeedMessage,
    ChannelInstructions,
    ChannelStar,
    CodeInvite,
    CodeInviteRedemption,
    MCPBuiltInAgentKey,
    SandboxCustomImage,
    SandboxEnvironment,
    SandboxSession,
    SandboxSnapshot,
    Task,
    TaskActivity,
    TaskAutomation,
    TaskClientProvenance,
    TaskPin,
    TaskRun,
    TaskSession,
    TaskThreadMessage,
    TaskThreadMessageMention,
)
from products.tasks.backend.pr_urls import merge_pr_output
from products.tasks.backend.prompts import build_wizard_pr_agent_prompt, generate_wizard_head_branch
from products.tasks.backend.visibility import task_control_q, task_run_visibility_q, task_visibility_q

from . import contracts

logger = logging.getLogger(__name__)

_TASK_LOG_READ_EXECUTOR = ThreadPoolExecutor(max_workers=8, thread_name_prefix="task-log-read")

# --- Enum re-exports ---
# Value types (not ORM models), safe to expose. External callers compare against the
# string-valued ``.status`` / ``.environment`` / ``.origin_product`` fields on the DTOs.
TaskRunStatus = TaskRun.Status
TaskRunEnvironment = TaskRun.Environment
TaskOriginProduct = Task.OriginProduct
TaskRuntime = Task.Runtime
SandboxNetworkAccessLevel = SandboxEnvironment.NetworkAccessLevel
SandboxSnapshotStatus = SandboxSnapshot.Status

# --- Code-invite redeem outcomes ---
# Returned on ``CodeInviteRedeemResult.outcome``; the presentation layer maps each to an
# HTTP response. ``REDEEMED`` covers both a fresh redemption and the idempotent no-op when
# the user already redeemed this code (both surface as success).
CODE_INVITE_REDEEMED = "redeemed"
CODE_INVITE_INVALID_CODE = "invalid_code"
CODE_INVITE_NOT_REDEEMABLE = "not_redeemable"

WIZARD_PR_READY_EMAIL_FEATURE_FLAG = "wizard-cloud-run-pr-ready-email-enabled"

# Runtime posture for a setup-wizard cloud run, applied in create_wizard_cloud_run. The model is
# pinned because these runs route to the unbilled `onboarding` gateway product, which allowlists a
# narrow model set; the string form avoids pulling the temporal RuntimeAdapter enum onto this path.
WIZARD_CLOUD_RUN_RUNTIME_ADAPTER = "claude"
WIZARD_CLOUD_RUN_MODEL = "claude-sonnet-5"
WIZARD_CLOUD_RUN_AI_STAGE = "wizard_pr_agent"

__all__ = [
    "CODE_INVITE_INVALID_CODE",
    "CODE_INVITE_NOT_REDEEMABLE",
    "CODE_INVITE_REDEEMED",
    "SandboxNetworkAccessLevel",
    "SandboxSnapshotStatus",
    "TaskOriginProduct",
    "TaskRuntime",
    "TaskRunEnvironment",
    "TaskRunStatus",
    "append_task_run_log",
    "ensure_task_run_session",
    "beacon_task_presence",
    "bootstrap_task_run",
    "can_mint_readonly_github_token",
    "check_task_run_startable",
    "collect_task_run_state_metrics",
    "compute_repository_readiness",
    "create_and_run_task",
    "create_completed_sandbox_snapshot",
    "create_run",
    "create_sandbox_connection_token",
    "build_sandbox_custom_image",
    "create_sandbox_custom_image",
    "create_sandbox_environment",
    "create_task",
    "create_task_automation",
    "create_task_without_run",
    "create_task_run_connection_token",
    "create_task_run_living_artifact",
    "create_task_run_stream_read_token",
    "resolve_stream_base_url",
    "claim_and_fail_stale_run",
    "delete_sandbox_custom_image",
    "delete_sandbox_environment",
    "ensure_personal_channel_id",
    "ensure_sandbox_custom_image_builder_task",
    "delete_task_automation",
    "edit_task_run_living_artifact",
    "complete_idle_local_task_run",
    "fail_task_run",
    "finalize_task_run_artifact_uploads",
    "finalize_task_staged_artifacts",
    "get_active_wizard_cloud_run",
    "get_conversation_task_dtos",
    "get_latest_pr_url_by_task",
    "get_merged_pr_task_ids",
    "get_latest_run_by_task",
    "get_resume_snapshot_carry_state",
    "get_sandbox_custom_image",
    "get_sandbox_environment",
    "get_sandbox_snapshot",
    "get_stale_prewarmed_queued_task_run_ids",
    "get_stale_queued_task_run_ids",
    "get_task_automation",
    "get_task_detail",
    "get_task_id_for_run",
    "get_task_run",
    "get_task_run_session",
    "sync_task_run_session",
    "get_task_run_detail",
    "get_task_run_sandbox_connection",
    "get_task_run_living_artifact",
    "capture_relay_command_telemetry",
    "get_task_run_stream_info",
    "get_task_summaries",
    "is_internal_debug_team",
    "is_task_controllable_by_user",
    "is_valid_sandbox_env_var_key",
    "latest_task_run_pr_merged_subquery",
    "latest_task_run_pr_url_subquery",
    "leave_task_presence",
    "list_sandbox_custom_images",
    "list_sandbox_environments",
    "sandbox_custom_images_enabled",
    "list_task_automations",
    "list_task_run_living_artifacts",
    "list_task_repositories",
    "list_task_runs",
    "list_tasks",
    "pi_cloud_runtime_enabled",
    "prepare_task_run_artifact_uploads",
    "prepare_task_staged_artifacts",
    "presign_task_run_artifact",
    "read_task_run_artifact",
    "read_task_run_logs",
    "record_task_run_user_activity",
    "redeem_code_invite",
    "redispatch_task_run",
    "relay_task_run_message",
    "resolve_slack_thread_context",
    "resume_task_run_in_cloud",
    "run_task",
    "run_task_automation_now",
    "send_cancel",
    "select_repository_for_message",
    "set_task_run_output",
    "set_task_title",
    "slack_actor_state_updates",
    "signal_report_queryset",
    "signal_task_run_user_message",
    "signal_workflow_completion",
    "soft_delete_task",
    "start_task_run",
    "task_accessible_for_run_view",
    "task_exists",
    "task_ids_with_pr_url_subquery",
    "task_run_has_slack_mapping",
    "task_run_is_terminal",
    "task_runtime",
    "task_visible",
    "update_sandbox_environment",
    "update_task",
    "update_task_automation",
    "update_task_run",
    "update_task_run_state",
    "upsert_internal_sandbox_env",
    "validate_set_output",
    "validate_task_run_sandbox_token",
    "validate_task_run_artifact_ids",
    "warm_task_sandbox",
]


# --- Mappers ---


def _task_to_dto(task: Task) -> contracts.TaskDTO:
    return contracts.TaskDTO(
        id=task.id,
        team_id=task.team_id,
        title=task.title,
        description=task.description,
        origin_product=task.origin_product,
        repository=task.repository,
        internal=task.internal,
        archived=task.archived,
        created_at=task.created_at,
        updated_at=task.updated_at,
        created_by_id=task.created_by_id,
        task_number=task.task_number,
        slug=task.slug,
    )


def _task_run_to_dto(run: TaskRun, *, task: Task | None = None) -> contracts.TaskRunDTO:
    """Map a TaskRun to its DTO.

    Pass ``task`` (or rely on a ``select_related("task", "task__created_by")`` queryset)
    so the denormalised parent-task fields can be populated without an extra query.
    """
    parent = task if task is not None else getattr(run, "task", None)
    created_by = getattr(parent, "created_by", None) if parent is not None else None
    return contracts.TaskRunDTO(
        id=run.id,
        task_id=run.task_id,
        team_id=run.team_id,
        status=run.status,
        environment=run.environment,
        stage=run.stage,
        branch=run.branch,
        error_message=run.error_message,
        output=run.output,
        state=run.state or {},
        artifacts=run.artifacts or [],
        created_at=run.created_at,
        updated_at=run.updated_at,
        completed_at=run.completed_at,
        is_terminal=run.is_terminal,
        workflow_id=run.workflow_id,
        mode=run.mode,
        task_origin_product=parent.origin_product if parent is not None else None,
        created_by_id=parent.created_by_id if parent is not None else None,
        created_by_distinct_id=str(created_by.distinct_id) if created_by is not None else None,
        pr_url=(run.output or {}).get("pr_url"),
    )


def _mascot_config(user: "User") -> dict | None:
    """Mirror core ``UserBasicSerializer.get_mascot_config`` so ``created_by`` output is identical."""
    config = user.mascot_config
    if not config:
        return None
    if config.get("version") == 2:
        actor_options = config.get("actor_options", {})
        return {
            "use_as_profile": config.get("use_as_profile"),
            "color": actor_options.get("color"),
            "accessories": actor_options.get("accessories"),
            "skin": actor_options.get("skin"),
        }
    return {
        "use_as_profile": config.get("use_as_profile"),
        "color": config.get("color"),
        "accessories": config.get("accessories"),
        "skin": config.get("skin"),
    }


def _user_basic_info(user: "User | None") -> contracts.TaskUserBasicInfo | None:
    """Map a core ``User`` to the display DTO (matches ``UserBasicSerializer`` fields)."""
    if user is None:
        return None
    return contracts.TaskUserBasicInfo(
        id=user.id,
        uuid=user.uuid,
        distinct_id=str(user.distinct_id),
        first_name=user.first_name,
        last_name=user.last_name,
        email=user.email,
        is_email_verified=user.is_email_verified,
        mascot_config=_mascot_config(user),
        role_at_organization=user.role_at_organization,
    )


# Presigned log URLs are cached just under their 1-hour S3 expiry to avoid regeneration.
_TASK_RUN_LOG_URL_CACHE_TTL = 55 * 60


def _task_run_log_url(run: TaskRun) -> str | None:
    """Presigned S3 URL for a run's log, cached. Mirrors ``TaskRunDetailSerializer.get_log_url``."""
    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    from products.tasks.backend.redis import get_tasks_cache  # noqa: PLC0415 — keep redis off the api import path

    cache_key = f"task_run_log_url:{run.id}"
    cached_url = get_tasks_cache().get(cache_key)
    if cached_url:
        return cached_url

    presigned_url = object_storage.get_presigned_url(run.log_url, expiration=3600)
    if presigned_url:
        get_tasks_cache().set(cache_key, presigned_url, timeout=_TASK_RUN_LOG_URL_CACHE_TTL)
    return presigned_url


def _task_run_detail_to_dto(run: TaskRun) -> contracts.TaskRunDetailDTO:
    """Map a ``TaskRun`` to its HTTP detail DTO.

    Reproduces the SMF-derived fields ``TaskRunDetailSerializer`` computed: ``log_url`` does
    presigned-URL I/O (with caching), and ``runtime_adapter`` / ``provider`` / ``model`` /
    ``reasoning_effort`` are parsed off the run ``state``.
    """
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        parse_run_state,
    )

    state = parse_run_state(run.state)
    return contracts.TaskRunDetailDTO(
        id=run.id,
        task=run.task_id,
        stage=run.stage,
        branch=run.branch,
        status=run.status,
        environment=run.environment,
        runtime_adapter=state.runtime_adapter.value if state.runtime_adapter is not None else None,
        provider=state.provider.value if state.provider is not None else None,
        model=state.model,
        reasoning_effort=state.reasoning_effort.value if state.reasoning_effort is not None else None,
        log_url=_task_run_log_url(run),
        error_message=run.error_message,
        output=run.output,
        state=run.state or {},
        artifacts=run.artifacts or [],
        created_at=run.created_at,
        updated_at=run.updated_at,
        completed_at=run.completed_at,
    )


class _LatestRunUnset:
    pass


_LATEST_RUN_UNSET = _LatestRunUnset()


def _task_detail_to_dto(
    task: Task,
    *,
    include_latest_run: bool = True,
    latest_run: TaskRun | None | _LatestRunUnset = _LATEST_RUN_UNSET,
) -> contracts.TaskDetailDTO:
    """Map a ``Task`` to its HTTP detail DTO."""
    if not include_latest_run:
        resolved_latest_run = None
    elif isinstance(latest_run, _LatestRunUnset):
        resolved_latest_run = task.latest_run
    else:
        resolved_latest_run = latest_run
    latest_run_id = getattr(task, "_latest_run_id", None)
    if latest_run_id is None and resolved_latest_run is not None:
        latest_run_id = resolved_latest_run.id
    return contracts.TaskDetailDTO(
        id=task.id,
        task_number=task.task_number,
        slug=task.slug,
        title=task.title,
        title_manually_set=task.title_manually_set,
        description=task.description,
        origin_product=task.origin_product,
        runtime=task.runtime,
        repository=task.repository,
        repositories=task.repositories or ([task.repository] if task.repository else []),
        github_integration=task.github_integration_id,
        github_user_integration=task.github_user_integration_id,
        signal_report=task.signal_report_id,
        json_schema=task.json_schema,
        internal=task.internal,
        archived=task.archived,
        archived_at=task.archived_at,
        ci_prompt=task.ci_prompt,
        latest_run=_task_run_detail_to_dto(resolved_latest_run) if resolved_latest_run is not None else None,
        created_at=task.created_at,
        updated_at=task.updated_at,
        created_by=_user_basic_info(task.created_by if task.created_by_id else None),
        latest_run_id=latest_run_id,
        channel=task.channel_id,
    )


def _sandbox_env_to_dto(env: SandboxEnvironment) -> contracts.SandboxEnvironmentDTO:
    return contracts.SandboxEnvironmentDTO(
        id=env.id,
        team_id=env.team_id,
        name=env.name,
        network_access_level=env.network_access_level,
        private=env.private,
        internal=env.internal,
        include_default_domains=env.include_default_domains,
        allowed_domains=list(env.allowed_domains or []),
        repositories=list(env.repositories or []),
        effective_domains=env.get_effective_domains(),
        has_environment_variables=bool(env.environment_variables),
        created_by=_user_basic_info(env.created_by if env.created_by_id else None),
        created_at=env.created_at,
        updated_at=env.updated_at,
        custom_image_id=env.custom_image_id,
        custom_image_name=env.custom_image.name if env.custom_image else None,
        custom_image_status=env.custom_image.status if env.custom_image else None,
    )


def _sandbox_snapshot_to_dto(snapshot: SandboxSnapshot) -> contracts.SandboxSnapshotDTO:
    return contracts.SandboxSnapshotDTO(
        id=snapshot.id,
        external_id=snapshot.external_id,
        status=snapshot.status,
        repos=list(snapshot.repos or []),
    )


# --- Reads ---


def get_resume_snapshot_carry_state(run_state: dict[str, Any] | None) -> dict[str, Any]:
    """State keys a successor run must merge (whole dict, never ``snapshot_external_id`` alone)
    to resume from a prior run's sandbox snapshot; empty when there is no usable snapshot."""
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        parse_run_state,
    )

    return parse_run_state(run_state).resume_snapshot_carry_state()


def get_task_run(run_id: str | UUID, team_id: int | None = None) -> contracts.TaskRunDTO | None:
    """Fetch a single task run as a DTO, optionally scoped to a team."""
    qs = TaskRun.objects.select_related("task", "task__created_by")
    if team_id is not None:
        qs = qs.filter(team_id=team_id)
    run = qs.filter(id=run_id).first()
    if run is None:
        return None
    return _task_run_to_dto(run)


def find_signal_implementation_run(
    *, team_id: int, repository: str, head_branch: str | None
) -> contracts.SignalImplementationRunDTO | None:
    """The signals-origin implementation run that produced this PR, if any.

    Matches the PR's head branch against ``state.self_driving_head_branch``, the branch name the
    server generated at run creation (signals' auto_start) and stamped into PATCH-protected run
    state. That stamp is the only end of the run->PR link no caller can write: ``output.pr_url``
    and ``output.head_branch`` are settable by any team member with task access, so matching on
    them would let one member aim the approve-first review carve-out at an App-authored PR whose
    contents they chose. The head branch itself comes from GitHub (webhook payload or REST fetch),
    so both ends of the join are attested.

    Callers (stamp's inbox carve-out) pass the repository the PR event came from and own fork
    safety: pass ``head_branch`` only for a repo-native head, never a fork's (a fork's head ref is
    attacker-controlled). Dropping failed and cancelled runs and soft-deleted tasks stops a dead
    or disowned run from keeping the carve-out alive on later pushes. A COMPLETED run still
    matches: success flips the run to COMPLETED right after it opens the PR, so excluding it
    would end re-reviews the moment the implementation finishes.
    """
    # TODO(security): the run->PR link is only as strong as the branch NAME, and the name is not a
    # secret. state.self_driving_head_branch is unforgeable, but the name it holds is readable by any
    # team member (auto_start writes it into the task description, and TaskRunDetailSerializer exposes
    # `state`) and low-entropy. A run that finishes WITHOUT opening a PR keeps its stamp yet leaves its
    # branch unclaimed, so a member can read the name, have their own task's agent push an App-authored
    # repo-native PR from that exact branch, then set_output the original run's pr_url to fire the
    # carve-out: the head ref genuinely belongs to this run, so an approve-first review lands on a PR
    # whose contents they chose. The real fix is to bind on the head SHA the sandbox actually pushed
    # (recorded server-side into protected state) rather than the branch name, because a run that never
    # pushed has no SHA to bind, which removes the unclaimed-branch surface entirely; re-reviews then
    # pin to the PR identity once the first attested SHA establishes it. It is a heavy change (sandbox
    # has to report the pushed SHA back, and the re-review path needs the PR-identity pin), so it is
    # deferred. The exposure is intra-tenant and gated behind an opt-in toggle that defaults off, which
    # makes it an accepted residual for the current internal rollout; close it before any external one.
    if not head_branch:
        return None
    run = (
        TaskRun.objects.filter(
            team_id=team_id,
            state__self_driving_head_branch=head_branch,
            task__repository__iexact=repository.strip(),
            task__deleted=False,
        )
        .exclude(status__in=(TaskRun.Status.FAILED, TaskRun.Status.CANCELLED))
        .order_by("-created_at")
        .select_related("task")
        .first()
    )
    if run is None or run.team_id != team_id:
        return None
    task = run.task
    # Belt and braces: only signals' auto_start stamps the branch key today, but the ai_stage and
    # signal-report checks keep a future writer of the key from silently widening the carve-out.
    if task.signal_report_id is None or (run.state or {}).get("ai_stage") != "implementation":
        return None
    return contracts.SignalImplementationRunDTO(
        run_id=run.id,
        task_id=task.id,
        team_id=run.team_id,
        signal_report_id=task.signal_report_id,
        task_created_by_id=task.created_by_id,
    )


def get_wizard_pr_ready_email_context(run_id: str | UUID) -> contracts.WizardPrReadyEmailContextDTO | None:
    """Data ``send_wizard_pr_ready_email`` needs for a run, or ``None`` if the run has no PR URL yet."""
    run = TaskRun.objects.select_related("task").filter(id=run_id).first()
    if run is None:
        return None
    pr_url = (run.output or {}).get("pr_url") if isinstance(run.output, dict) else None
    if not pr_url:
        return None
    task = run.task
    return contracts.WizardPrReadyEmailContextDTO(
        task_id=task.id,
        run_id=run.id,
        team_id=run.team_id,
        origin_product=task.origin_product,
        pr_url=pr_url,
        repository=task.repository,
        branch=run.branch,
        created_by_id=task.created_by_id,
        already_sent=task.pr_ready_email_sent_at is not None,
    )


def mark_task_pr_ready_email_sent(task_id: str | UUID, pr_url: str) -> None:
    """Record confirmed PR-ready email delivery for a task, if it still exists."""
    task = Task.objects.filter(id=task_id).first()
    if task is not None:
        task.mark_pr_ready_email_sent(pr_url)


def get_task_id_for_run(run_id: str | UUID, team_id: int) -> UUID | None:
    """The parent task id for a run, team-scoped. ``None`` if the run isn't found for the team.

    A lightweight ``task_run_id -> task_id`` resolution (no DTO build) for callers that only
    need to deep-link a run to its task.
    """
    return TaskRun.objects.filter(id=run_id, team_id=team_id).values_list("task_id", flat=True).first()


def task_exists(task_id: str | UUID, team_id: int) -> bool:
    """Whether a (non-deleted) task exists for the team."""
    return Task.objects.filter(id=task_id, team_id=team_id).exists()


def task_owned_by_user(task_id: str | UUID, team_id: int, user_id: int) -> bool:
    return Task.objects.filter(id=task_id, team_id=team_id, created_by_id=user_id).exists()


def count_in_progress_runs_for_github_integration(team_id: int, integration_id: int) -> int:
    """In-progress runs whose task uses this team GitHub integration.

    Used by core's integration API to block disconnecting a GitHub integration while
    live runs still depend on it for credential refresh — deleting the row SET_NULLs
    ``Task.github_integration`` and permanently orphans every live sandbox's token.
    """
    return TaskRun.objects.filter(
        team_id=team_id,
        status=TaskRun.Status.IN_PROGRESS,
        task__github_integration_id=integration_id,
    ).count()


def is_task_controllable_by_user(task_id: str | UUID, user_id: int | None) -> bool:
    """Whether the user may mutate the task under the task control rules.

    Tasks belong to their creator, plus team-wide signal-pipeline tasks and legacy unowned
    tasks. Used by core's file-system flow to gate delete/restore on a filed task; public-channel
    read visibility deliberately does not qualify.
    """
    return Task.objects.filter(task_control_q(user_id), pk=task_id).exists()


def get_sandbox_snapshot(snapshot_id: str | UUID) -> contracts.SandboxSnapshotDTO | None:
    """Fetch a sandbox snapshot as a DTO."""
    snapshot = SandboxSnapshot.objects.filter(id=snapshot_id).first()
    return _sandbox_snapshot_to_dto(snapshot) if snapshot is not None else None


def get_tasks_by_ids(task_ids: Iterable[str | UUID], team_ids: Iterable[int]) -> list[contracts.TaskDTO]:
    """Tasks matching the supplied ids, restricted to ``team_ids``.

    For multi-team callers (e.g. the Slack App Home Tasks card) that already resolved the
    set of accessible teams upstream and need a bulk DTO fetch in one query.
    """
    ids = [str(t) for t in task_ids]
    teams = list(team_ids)
    if not ids or not teams:
        return []
    return [_task_to_dto(task) for task in Task.objects.filter(id__in=ids, team_id__in=teams)]


def get_latest_pr_url_by_task(task_ids: Iterable[str | UUID]) -> dict[str, str]:
    """Latest non-empty ``output.pr_url`` per task, for the supplied task ids."""
    ids = [str(t) for t in task_ids]
    if not ids:
        return {}
    rows = (
        TaskRun.objects.filter(task_id__in=ids, output__pr_url__isnull=False)
        .exclude(output__pr_url="")
        .order_by("task_id", "-created_at", "-id")
        .annotate(output_pr_url_text=KeyTextTransform("pr_url", "output"))
        .values("task_id", "output_pr_url_text")
        .distinct("task_id")
    )
    return {str(row["task_id"]): row["output_pr_url_text"] for row in rows if row["output_pr_url_text"]}


def task_ids_with_pr_url_subquery(team_id: int) -> QuerySet[TaskRun, Any]:
    """A ``values('task_id')`` queryset of ``team_id``'s tasks that produced a non-empty ``output.pr_url``.

    For embedding in a caller's ``task_id__in=...`` lookup so the report→PR correlation can be
    *decorrelated*: instead of a per-report ``Exists`` over runs, the caller drives off this small,
    index-backed set (served by the partial ``task_run_output_pr_url_idx``) and joins outward to its
    own report-association tables. Returns a query expression — no ORM instances cross the boundary.

    Scoped to ``team_id`` so the set stays bounded to the request's tenant rather than scanning every
    team's PR-bearing runs — associated runs are always same-team, so this drops no valid matches.
    """
    return (
        TaskRun.objects.filter(team_id=team_id, output__pr_url__isnull=False)
        .exclude(output__pr_url="")
        .values("task_id")
    )


def latest_task_run_pr_url_subquery(*conditions: Q, **task_run_filter) -> Subquery:
    """``Subquery`` of the latest non-empty output.pr_url for runs matching the supplied correlation
    (keyword lookups and/or positional ``Q`` objects). Returns a query expression to embed in the
    caller's queryset — no ORM instances cross the boundary, and the tasks facade stays free of the
    caller's domain."""
    return Subquery(
        TaskRun.objects.filter(*conditions, output__pr_url__isnull=False, **task_run_filter)
        .exclude(output__pr_url="")
        .order_by("-created_at")
        .annotate(output_pr_url_text=KeyTextTransform("pr_url", "output"))
        .values("output_pr_url_text")[:1],
        output_field=CharField(),
    )


def latest_task_run_pr_merged_subquery(*conditions: Q, **task_run_filter) -> Subquery:
    """``Subquery`` of the webhook-attested merge flag on the same run ``latest_task_run_pr_url_subquery``
    resolves for the same correlation — so a caller displaying that PR can say whether it merged rather
    than inferring it. Same filter and ordering, so the two always describe one run. NULL when no
    PR-bearing run matches; treat that as "not merged"."""
    return Subquery(
        TaskRun.objects.filter(*conditions, output__pr_url__isnull=False, **task_run_filter)
        .exclude(output__pr_url="")
        .order_by("-created_at")
        .annotate(output_pr_merged_flag=KeyTextTransform("pr_merged", "output"))
        .values("output_pr_merged_flag")[:1],
        output_field=CharField(),
    )


def get_merged_pr_task_ids(task_ids: Iterable[str | UUID]) -> set[str]:
    """Of the supplied tasks, those whose latest PR-bearing run has a webhook-attested merged PR.

    Batched counterpart to ``latest_task_run_pr_merged_subquery``, matching ``get_latest_pr_url_by_task``
    run-for-run so the merge flag describes the PR URL that helper returns.
    """
    ids = [str(t) for t in task_ids]
    if not ids:
        return set()
    rows = (
        TaskRun.objects.filter(task_id__in=ids, output__pr_url__isnull=False)
        .exclude(output__pr_url="")
        .order_by("task_id", "-created_at", "-id")
        .annotate(output_pr_merged_flag=KeyTextTransform("pr_merged", "output"))
        .values("task_id", "output_pr_merged_flag")
        .distinct("task_id")
    )
    return {str(row["task_id"]) for row in rows if row["output_pr_merged_flag"] in ("true", "True")}


def get_latest_run_by_task(task_ids: Iterable[str | UUID]) -> dict[str, contracts.TaskRunDTO]:
    """Most-recent run per task (by ``created_at`` then ``id``), for the supplied task ids."""
    ids = [str(t) for t in task_ids]
    if not ids:
        return {}
    runs = (
        TaskRun.objects.filter(task_id__in=ids)
        .select_related("task", "task__created_by")
        .order_by("task_id", "-created_at", "-id")
        .distinct("task_id")
    )
    return {str(run.task_id): _task_run_to_dto(run) for run in runs}


def get_active_wizard_cloud_run(team_id: int) -> contracts.WizardCloudRunDTO | None:
    """The team's active onboarding wizard cloud run, for rehydrating the setup FAB.

    The drop flow starts the wizard cloud run server-side (``create_wizard_cloud_run``),
    so a freshly-signed-in user has no client-side handle. Returns the most recent run
    across the team's onboarding (``ORIGIN_PRODUCT == ONBOARDING``) tasks that's still
    running, or completed within the last day (so we can show "Insights is wired up" +
    the PR); otherwise ``None``. Team-scoped.
    """
    onboarding_task_ids = Task.objects.filter(
        team_id=team_id, origin_product=Task.OriginProduct.ONBOARDING, archived=False
    ).values_list("id", flat=True)
    fresh_after = django_timezone.now() - timedelta(days=1)
    # Scan runs newest-first and surface the first that qualifies: picking the newest task up front
    # would let a newer onboarding task with no live run hide an older task's still-running one.
    # Both the task set and the run are scoped by team_id so a mismatched/legacy run row can't leak
    # another team's handle back to the requester.
    #
    # ``origin_product == ONBOARDING`` is caller-settable, so it alone can't tell a genuine
    # server-started wizard run from one a project member planted through the normal task APIs.
    # Also require the immutable markers ``create_wizard_cloud_run`` stamps: a cloud environment
    # and the ``wizard_config`` state key (a protected key callers cannot set, see the run PATCH
    # allowlist), so we never hand a provisioned user someone else's attacker-controlled handle.
    runs = TaskRun.objects.filter(
        task_id__in=onboarding_task_ids,
        team_id=team_id,
        environment=TaskRun.Environment.CLOUD,
        state__has_key="wizard_config",
    ).order_by("-created_at", "-id")
    for run in runs:
        # Non-terminal runs always surface; terminal ones only while the result is still
        # fresh enough to be worth showing on first landing.
        if run.is_terminal:
            anchor = run.updated_at or run.created_at
            if anchor is None or anchor < fresh_after:
                continue
        return contracts.WizardCloudRunDTO(
            task_id=run.task_id,
            run_id=run.id,
            status=run.status,
            started_at=run.created_at,
        )
    return None


def get_stale_queued_task_run_ids(
    older_than: timedelta,
    limit: int,
    *,
    created_hard_cap: timedelta | None = None,
    hard_cap_min_queued: timedelta = timedelta(hours=1),
    environment: str | None = None,
) -> list[UUID]:
    """Ids of runs stuck in QUEUED, by ``updated_at`` age or an optional ``created_at`` backstop.

    ``environment`` restricts the sweep to runs of that environment. A QUEUED cloud run is
    awaiting a workflow that should have started, but a local (desktop) run sits in QUEUED by
    design while the desktop agent drives it — so sweep callers must scope themselves and act
    per environment: dispatch recovery must only touch cloud runs (cloud-dispatching a local
    run hijacks the user's live local session), and the janitor fails stale cloud runs but
    quietly completes stale local ones.

    Intentionally cross-team — the janitor sweep runs without a team context.
    """
    now = django_timezone.now()
    stale = Q(updated_at__lt=now - older_than)
    if created_hard_cap is not None:
        stale |= Q(created_at__lt=now - created_hard_cap, updated_at__lt=now - hard_cap_min_queued)
    queryset = TaskRun.objects.filter(status=TaskRun.Status.QUEUED)  # nosemgrep: celery-task-team-scope-audit
    if environment is not None:
        queryset = queryset.filter(environment=environment)
    return list(queryset.filter(stale).order_by("updated_at").values_list("id", flat=True)[:limit])


def get_stale_prewarmed_queued_task_run_ids(older_than: timedelta, limit: int) -> list[UUID]:
    """Ids of prewarmed runs orphaned in QUEUED — their processing workflow never started, so the
    in-workflow ``WARM_IDLE_TIMEOUT`` (10m) never armed to finalize them.

    A live warm run idles in QUEUED awaiting its first message and self-terminates at
    ``WARM_IDLE_TIMEOUT``, so a prewarmed run still QUEUED well past that window has no workflow
    behind it (dispatch lost — e.g. an ``on_commit`` callback that never ran) and can be reaped
    immediately rather than lingering until the 24h stale sweep. ``older_than`` should sit safely
    above ``WARM_IDLE_TIMEOUT`` so a still-idling warm run is never killed early.

    Intentionally cross-team — the janitor sweep runs without a team context.
    """
    now = django_timezone.now()
    return list(
        TaskRun.objects.filter(  # nosemgrep: celery-task-team-scope-audit
            status=TaskRun.Status.QUEUED,
            state__prewarmed=True,
            updated_at__lt=now - older_than,
        )
        .order_by("updated_at")
        .values_list("id", flat=True)[:limit]
    )


def _gauge_rows(values_qs, value_key: str, *, with_status: bool, now=None) -> list[contracts.TaskRunGaugeRow]:
    rows = []
    for row in values_qs:
        raw = row[value_key]
        value = (now - raw).total_seconds() if now is not None else raw
        rows.append(
            contracts.TaskRunGaugeRow(
                environment=row["environment"],
                origin_product=row["task__origin_product"] or "unknown",
                value=value,
                status=row["status"] if with_status else None,
            )
        )
    return rows


def collect_task_run_state_metrics(
    *,
    open_statuses: Sequence[str],
    age_statuses: Sequence[str],
    terminal_statuses: Sequence[str],
    window_seconds: int,
) -> contracts.TaskRunStateMetricsDTO:
    """Aggregate TaskRun state for monitoring gauges.

    The caller (a core celery task) owns which statuses count as open/age/terminal and the
    recency window; this returns the raw numbers grouped by (status, environment,
    parent origin_product) so no ORM leaks across the boundary.
    """
    now = django_timezone.now()
    window_start = now - timedelta(seconds=window_seconds)
    return contracts.TaskRunStateMetricsDTO(
        runs_in_status=_gauge_rows(
            TaskRun.objects.filter(status__in=open_statuses)
            .values("status", "environment", "task__origin_product")
            .annotate(count=Count("id")),
            "count",
            with_status=True,
        ),
        oldest_open_age_seconds=_gauge_rows(
            TaskRun.objects.filter(status__in=age_statuses)
            .values("status", "environment", "task__origin_product")
            .annotate(oldest_created_at=Min("created_at")),
            "oldest_created_at",
            with_status=True,
            now=now,
        ),
        created_recently=_gauge_rows(
            TaskRun.objects.filter(created_at__gte=window_start)
            .values("environment", "task__origin_product")
            .annotate(count=Count("id")),
            "count",
            with_status=False,
        ),
        terminal_recently=_gauge_rows(
            TaskRun.objects.filter(status__in=terminal_statuses, updated_at__gte=window_start)
            .values("status", "environment", "task__origin_product")
            .annotate(count=Count("id")),
            "count",
            with_status=True,
        ),
    )


# --- Writes ---


def create_and_run_task(
    *,
    team,
    title: str,
    description: str,
    origin_product: "Task.OriginProduct",
    user_id: int,
    repository: str | None = None,
    create_pr: bool = True,
    mode: str = "background",
    start_workflow: bool = True,
    branch: str | None = None,
    signal_report_id: str | None = None,
    internal: bool = False,
    sandbox_environment_id: str | None = None,
    channel_id: str | UUID | None = None,
    **extra,
) -> contracts.CreatedTaskDTO:
    """Create a task and (optionally) kick off its processing workflow.

    Thin wrapper over ``Task.create_and_run`` that returns ids + the created run as a DTO
    instead of leaking the ORM ``Task``. ``team`` is a core ``insights.Team`` (not a tasks
    model). Less-common keyword arguments are forwarded verbatim via ``**extra``.

    ``channel_id`` files the task into a channel's feed; left NULL for non-channel surfaces.
    An id the creator can't file into (see ``_visible_channel``) is ignored rather than
    raising — feed placement must never break task creation.
    """
    # create_pr=False sessions (research, repo selection, custom agents) can never open the
    # billable PR, so the quota gate must not block them.
    if origin_product == Task.OriginProduct.SIGNAL_REPORT and create_pr:
        # Distinct stage from create_task's `manual_create`: the main caller here is the
        # auto-start pipeline, whose over-quota hits must not pollute the manual-path
        # dark-launch bucket.
        enforce_self_driving_pr_quota(team, report_id=signal_report_id, stage="task_create")
    channel = _visible_channel(channel_id, team.id, user_id) if channel_id is not None else None
    task = Task.create_and_run(
        team=team,
        title=title,
        description=description,
        origin_product=origin_product,
        user_id=user_id,
        repository=repository,
        channel=channel,
        create_pr=create_pr,
        mode=mode,
        start_workflow=start_workflow,
        branch=branch,
        signal_report_id=signal_report_id,
        internal=internal,
        sandbox_environment_id=sandbox_environment_id,
        **extra,
    )
    latest = task.latest_run
    return contracts.CreatedTaskDTO(
        task_id=task.id,
        team_id=task.team_id,
        latest_run=_task_run_to_dto(latest, task=task) if latest is not None else None,
    )


def create_wizard_cloud_run(
    *,
    team,
    user_id: int,
    repository: str,
    branch: str | None = None,
) -> contracts.CreatedTaskDTO:
    """Create + run a cloud setup-wizard task.

    The workflow runs the published wizard in the sandbox (it integrates Insights), then the agent
    commits the changes, opens a PR on the user's repo, and keeps it green — it never implements
    Insights itself (see the wizard PR agent prompt). The wizard authenticates with its own scoped
    token (see ``create_wizard_oauth_access_token``), independent of the agent's sandbox token, so
    the agent runs with read-only Insights scopes.``wizard_config`` marks the run so the workflow runs the wizard pre-agent step.

    ``user_id`` is the person going through onboarding; it becomes the task's ``created_by`` so the
    run is explicitly attributed to them.

    The PR head branch is generated here (not by the agent) so the GitHub PR webhook can bind the
    opened PR back to this run by branch + repository — wizard PRs are bot-authored, which the
    agent-side PR attribution cannot match.

    The model is pinned rather than left to the agent's default because these runs bill to nobody:
    they route to the unbilled ``onboarding`` gateway product, whose model allowlist is narrow, and
    Insights absorbs the cost. Keep the pin inside that allowlist or the run fails at the gateway.
    """
    head_branch = generate_wizard_head_branch()
    prompt = build_wizard_pr_agent_prompt(head_branch)
    return create_and_run_task(
        team=team,
        title="Set up Insights",
        description=prompt,
        origin_product=Task.OriginProduct.ONBOARDING,
        user_id=user_id,
        repository=repository,
        create_pr=True,
        mode="background",
        branch=branch,
        wizard_config={},
        wizard_head_branch=head_branch,
        insights_mcp_scopes="read_only",
        runtime_adapter=WIZARD_CLOUD_RUN_RUNTIME_ADAPTER,
        model=WIZARD_CLOUD_RUN_MODEL,
        ai_stage=WIZARD_CLOUD_RUN_AI_STAGE,
        # The agent server boots idle; this is the message that actually kicks it off once ready
        # (delivered by forward_pending_user_message). Without it the run stalls after "Started agent".
        pending_user_message=prompt,
    )


def recent_wizard_cloud_run_times(user_id: int, since: datetime) -> list[datetime]:
    """Creation times of a user's recent wizard cloud runs that still count against their quota.

    Backs the outcome-aware cloud_run throttles: failed and cancelled runs are excluded so a
    user whose run broke (or who cancelled a stuck one) can start another without waiting out
    the window. The hard ceiling on total attempts lives in the cloud_run view as an atomic
    cache reservation, not here.

    The filter trusts only PATCH-immutable markers: ``created_by`` (set at creation) and the
    protected ``wizard_config`` state key that only ``create_wizard_cloud_run`` stamps (see
    ``_PROTECTED_RUN_STATE_KEYS``). Mutable fields like the run's ``environment`` are
    deliberately NOT filtered — a run PATCHed from cloud to local must keep consuming quota,
    or flipping it would launder sandbox boots out of the limits.

    Deliberately user-scoped across teams: the throttle is per user, and a user can run the
    wizard on projects in different teams. Returns only timestamps, no run data.
    """
    return list(
        TaskRun.objects.filter(
            task__created_by_id=user_id,
            state__has_key="wizard_config",
            created_at__gte=since,
        )
        .exclude(status__in=[TaskRun.Status.FAILED, TaskRun.Status.CANCELLED])
        .order_by("created_at")
        .values_list("created_at", flat=True)
    )


def create_task_without_run(
    *,
    team,
    user_id: int,
    origin_product: "Task.OriginProduct",
    title: str = "",
    description: str = "",
    repository: str | None = None,
    mcp_builtin_agent_key: MCPBuiltInAgentKey | None = None,
) -> UUID:
    """Create a Task row with no initial run, returning its id.

    For callers that own run creation themselves — e.g. the sandbox warm path, which boots the first
    run via the warming facade. ``team`` is a core ``insights.Team`` (not a tasks model).
    """
    task = Task.create_without_run(
        team=team,
        title=title,
        description=description,
        origin_product=origin_product,
        user_id=user_id,
        repository=repository,
        mcp_builtin_agent_key=mcp_builtin_agent_key,
    )
    return task.id


def create_run(
    task_id: str | UUID,
    *,
    mode: str = "background",
    extra_state: dict | None = None,
    branch: str | None = None,
) -> contracts.TaskRunDTO:
    """Create a new run for an existing task (e.g. resuming an interactive sandbox session)."""
    task = Task.objects.get(id=task_id)
    run = task.create_run(mode=mode, extra_state=extra_state, branch=branch)
    return _task_run_to_dto(run, task=task)


def update_task_run_state(
    run_id: str | UUID,
    *,
    updates: dict | None = None,
    remove_keys: Iterable[str] | None = None,
) -> dict:
    """Atomically merge state updates into a run's ``state`` and return the new state."""
    return TaskRun.update_state_atomic(run_id, updates=updates, remove_keys=remove_keys)


def slack_actor_state_updates(*, user_id: int, slack_user_id: str | None = None) -> dict[str, Any]:
    """Run-state updates recording the Slack user currently steering a run.

    Credential resolution and reply tagging read the keys this builds, so every
    writer must go through here rather than assembling the dict inline.
    """
    from products.tasks.backend.logic.services.run_actor import (  # noqa: PLC0415 — keep tasks internals off the api import path
        slack_actor_state_updates as _slack_actor_state_updates,
    )

    return _slack_actor_state_updates(user_id=user_id, slack_user_id=slack_user_id)


def set_task_run_created_at_for_seeding(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, created_at: datetime
) -> None:
    """Backdate a run's ``created_at`` — DEBUG-only escape hatch for dev seeding.

    Signals' billing/refund seeding needs runs whose billable moment falls on an earlier UTC
    day; ``created_at`` is deliberately not writable through the PATCH surface, so the hatch
    lives here rather than callers reaching into the ORM.
    """
    if not settings.DEBUG:
        raise RuntimeError("set_task_run_created_at_for_seeding is DEBUG-only")
    TaskRun.objects.filter(pk=run_id, task_id=task_id, team_id=team_id).update(created_at=created_at)


def fail_task_run(run_id: str | UUID, error: str, error_type: str | None = None) -> bool:
    """Mark a QUEUED run as failed. Returns whether a run was acted on.

    Refetches filtered on ``status=QUEUED`` so a run that left the queue between the
    candidate scan and this call is skipped. Intentionally cross-team (janitor sweep).
    """
    run = TaskRun.objects.filter(
        pk=run_id, status=TaskRun.Status.QUEUED
    ).first()  # nosemgrep: celery-task-team-scope-audit
    if run is None:
        return False
    run.mark_failed(error, error_type=error_type)
    return True


def complete_idle_local_task_run(run_id: str | UUID) -> bool:
    """Quietly finalize a local (desktop-driven) run left idling in QUEUED. Returns whether
    a run was acted on.

    Local runs never get a cloud workflow, so QUEUED is their steady state while the desktop
    drives the session — once the desktop goes away, nothing else ever terminalizes the row.
    An idle session that ended is the run's normal end state, so it finalizes as COMPLETED,
    and without a push notification: pinging a user a day after they closed their session is
    noise, not signal.

    Compare-and-set claim (like ``claim_and_fail_stale_run``): the conditional update flips the
    run only while it is still QUEUED *and* local, so a run that left the queue — or was handed
    off to cloud (handoff keeps status QUEUED) — between the candidate scan and this call is
    skipped rather than terminalized under its just-dispatched workflow. The winner finalizes
    via ``mark_completed`` (``completed_at``, stream + analytics). Intentionally cross-team
    (janitor sweep).
    """
    claimed = TaskRun.objects.filter(
        pk=run_id, status=TaskRun.Status.QUEUED, environment=TaskRun.Environment.LOCAL
    ).update(status=TaskRun.Status.COMPLETED)  # nosemgrep: celery-task-team-scope-audit
    if not claimed:
        return False
    run = TaskRun.objects.filter(pk=run_id).first()  # nosemgrep: celery-task-team-scope-audit
    if run is not None:
        run.mark_completed(notify=False, analytics_properties={"finalized_by": "stale_local_queued_sweep"})
    return True


def claim_and_fail_stale_run(run_id: str | UUID, error: str, error_type: str | None = None) -> bool:
    """Compare-and-set reap of a stranded run. Returns whether this caller won the claim.

    Atomically flips a run still in ``QUEUED``/``IN_PROGRESS`` to ``FAILED`` via a conditional
    update, so concurrent reapers of the same row resolve to exactly one winner (the losers match
    zero rows). The winner finalizes via ``mark_failed`` (error message, ``completed_at``, stream +
    analytics). Intentionally cross-team (janitor sweep).
    """
    claimed = TaskRun.objects.filter(
        id=run_id,
        status__in=(TaskRun.Status.QUEUED, TaskRun.Status.IN_PROGRESS),
    ).update(status=TaskRun.Status.FAILED)  # nosemgrep: celery-task-team-scope-audit
    if not claimed:
        return False
    run = TaskRun.objects.filter(pk=run_id).first()  # nosemgrep: celery-task-team-scope-audit
    if run is not None:
        run.mark_failed(error, error_type=error_type)
    return True


def redispatch_task_run(run_id: str | UUID) -> str:
    """Re-dispatch a QUEUED run whose create-time workflow dispatch was lost. Cross-team janitor call.

    Idempotent recover-only wrapper over the temporal client — never fails the run. Returns the
    outcome (``recovered`` / ``already_running`` / ``left_queue`` / ``error``).
    """
    from products.tasks.backend.temporal.client import (  # noqa: PLC0415 — keep temporalio off the api import path
        redispatch_orphaned_task_run,
    )

    return redispatch_orphaned_task_run(str(run_id))


def upsert_internal_sandbox_env(
    team_id: int,
    name: str,
    network_access_level: "SandboxEnvironment.NetworkAccessLevel",
    *,
    private: bool = False,
    internal: bool = True,
    allowed_domains: list[str] | None = None,
    include_default_domains: bool = False,
) -> UUID:
    """Get-or-create an internal sandbox environment, reasserting policy on every call.

    Only rows already carrying the requested ``internal`` flag are matched for reuse: users
    can create environments with arbitrary names through the sandbox environment API, and a
    same-named user row must never be adopted (or deleted) by internal provisioning — the
    internal env is created alongside it instead. Reasserted policy covers the whole
    execution surface, not just network: the user-controllable ``custom_image``,
    ``environment_variables``, and ``repositories`` fields are cleared on every call, so
    nothing a person set on a row (before or after it became internal) can ride into an
    internally provisioned run.

    ``SandboxEnvironment`` has no unique constraint on ``(team_id, name)``, so concurrent
    callers can both INSERT. We dedupe on ``MultipleObjectsReturned`` by keeping the oldest
    matching row and deleting the rest.
    """
    defaults: dict = {
        "network_access_level": network_access_level,
        "private": private,
        "custom_image": None,
        "environment_variables": {},
        "repositories": [],
    }
    if allowed_domains is not None:
        defaults["allowed_domains"] = allowed_domains
        defaults["include_default_domains"] = include_default_domains
    try:
        env, _ = SandboxEnvironment.objects.update_or_create(
            team_id=team_id, name=name, internal=internal, defaults=defaults
        )
        return env.id
    except SandboxEnvironment.MultipleObjectsReturned:
        with transaction.atomic():
            dupes = list(
                SandboxEnvironment.objects.filter(team_id=team_id, name=name, internal=internal).order_by("created_at")
            )
            keeper = dupes[0]
            SandboxEnvironment.objects.filter(id__in=[d.id for d in dupes[1:]]).delete()
        for key, value in defaults.items():
            setattr(keeper, key, value)
        keeper.save(update_fields=list(defaults.keys()))
        return keeper.id


def create_completed_sandbox_snapshot(external_id: str) -> UUID:
    """Record a completed sandbox snapshot for an externally-built image; return its id."""
    snapshot = SandboxSnapshot.objects.create(external_id=external_id, status=SandboxSnapshot.Status.COMPLETE)
    return snapshot.id


# --- Desktop invites ---


def redeem_code_invite(code: str, user_id: int) -> contracts.CodeInviteRedeemResult:
    """Redeem a Insights Desktop invite for a user.

    Idempotent: a user who already redeemed this code gets ``REDEEMED`` without a second
    redemption row. A fresh redemption takes a row lock on the invite, re-checks
    redeemability under the lock, records the redemption, bumps ``redemption_count``, and
    captures the activation analytics — all in one transaction, mirroring the original view.
    """
    code_str = code.strip()

    try:
        invite_code = CodeInvite.objects.get(code__iexact=code_str)
    except CodeInvite.DoesNotExist:
        return contracts.CodeInviteRedeemResult(outcome=CODE_INVITE_INVALID_CODE)

    user = User.objects.get(pk=user_id)

    if CodeInviteRedemption.objects.filter(invite_code=invite_code, user=user).exists():
        return contracts.CodeInviteRedeemResult(outcome=CODE_INVITE_REDEEMED)

    with transaction.atomic():
        invite_code = CodeInvite.objects.select_for_update().get(id=invite_code.id)

        if not invite_code.is_redeemable:
            return contracts.CodeInviteRedeemResult(outcome=CODE_INVITE_NOT_REDEEMABLE)

        organization = user.organization if hasattr(user, "organization") else None

        CodeInviteRedemption.objects.create(
            invite_code=invite_code,
            user=user,
            organization=organization,
        )

        CodeInvite.objects.filter(id=invite_code.id).update(redemption_count=F("redemption_count") + 1)

        hanzo_insights.capture(
            distinct_id=str(user.distinct_id),
            event="code_invite_redeemed",
            groups=groups(organization=organization),
        )

    return contracts.CodeInviteRedeemResult(outcome=CODE_INVITE_REDEEMED)


# --- Sandbox environments (presentation CRUD) ---
# Visibility: an environment is reachable by a team member if it is non-private, or it is
# theirs (``created_by``). ``list`` additionally hides ``internal`` environments.


def is_valid_sandbox_env_var_key(key: str) -> bool:
    """Whether ``key`` is a valid environment-variable name (``[A-Za-z_][A-Za-z0-9_]*``)."""
    return SandboxEnvironment.is_valid_env_var_key(key)


def is_blocked_sandbox_env_var_key(key: str) -> bool:
    return is_blocked_sandbox_env_key(key)


def is_reserved_sandbox_env_var_key(key: str) -> bool:
    return key in RESERVED_SANDBOX_ENVIRONMENT_VARIABLE_KEYS


def _validate_user_sandbox_env_vars(environment_variables: dict | None) -> None:
    for key in environment_variables or {}:
        if not SandboxEnvironment.is_valid_env_var_key(key):
            raise ValueError(f"Invalid environment variable key: {key!r}")
        if is_blocked_sandbox_env_key(key) or key in RESERVED_SANDBOX_ENVIRONMENT_VARIABLE_KEYS:
            raise ValueError(f"Environment variable key {key!r} is not allowed")


def _accessible_sandbox_envs(team_id: int, user_id: int):
    return (
        SandboxEnvironment.objects.filter(team_id=team_id)
        .filter(Q(private=False) | Q(created_by_id=user_id))
        .select_related("created_by", "custom_image")
    )


def list_sandbox_environments(team_id: int, user_id: int) -> list[contracts.SandboxEnvironmentDTO]:
    """Non-internal environments visible to the user, for the list view."""
    return [_sandbox_env_to_dto(env) for env in _accessible_sandbox_envs(team_id, user_id).filter(internal=False)]


def get_sandbox_environment(env_id: str | UUID, team_id: int, user_id: int) -> contracts.SandboxEnvironmentDTO | None:
    """A single environment visible to the user (internal ones are retrievable by id)."""
    env = _accessible_sandbox_envs(team_id, user_id).filter(pk=env_id).first()
    return _sandbox_env_to_dto(env) if env is not None else None


def _validate_custom_image_id(team_id: int, user_id: int, custom_image_id: str | UUID | None) -> None:
    if custom_image_id is None:
        return
    if not sandbox_custom_images_enabled(team_id, user_id):
        raise ValueError("Custom sandbox images require the Modal VM runtime, which is not enabled")
    image = SandboxCustomImage.get_accessible_for_task(
        image_id=custom_image_id, team_id=team_id, task_created_by_id=user_id
    )
    if image is None:
        raise ValueError(f"Invalid custom_image_id: {custom_image_id}")


def create_sandbox_environment(
    team_id: int,
    user_id: int,
    *,
    name: str,
    network_access_level: str,
    allowed_domains: list[str],
    include_default_domains: bool,
    repositories: list[str],
    environment_variables: dict,
    private: bool,
    custom_image_id: str | None = None,
) -> contracts.SandboxEnvironmentDTO:
    """Create a team environment owned by the user and return it as a DTO."""
    _validate_user_sandbox_env_vars(environment_variables)
    _validate_custom_image_id(team_id, user_id, custom_image_id)
    env = SandboxEnvironment.objects.create(
        team_id=team_id,
        created_by_id=user_id,
        name=name,
        network_access_level=network_access_level,
        allowed_domains=allowed_domains,
        include_default_domains=include_default_domains,
        repositories=repositories,
        environment_variables=environment_variables,
        private=private,
        custom_image_id=custom_image_id,
    )
    return _sandbox_env_to_dto(SandboxEnvironment.objects.select_related("created_by", "custom_image").get(pk=env.pk))


def update_sandbox_environment(
    env_id: str | UUID, team_id: int, user_id: int, **fields
) -> contracts.SandboxEnvironmentDTO | None:
    """Partially update a visible environment. Returns ``None`` if not found/visible."""
    env = _accessible_sandbox_envs(team_id, user_id).filter(pk=env_id).first()
    if env is None:
        return None
    if "environment_variables" in fields:
        _validate_user_sandbox_env_vars(fields["environment_variables"])
    if "custom_image_id" in fields:
        _validate_custom_image_id(team_id, user_id, fields["custom_image_id"])
    for key, value in fields.items():
        setattr(env, key, value)
    env.save()
    return _sandbox_env_to_dto(SandboxEnvironment.objects.select_related("created_by", "custom_image").get(pk=env.pk))


def delete_sandbox_environment(env_id: str | UUID, team_id: int, user_id: int) -> bool:
    """Delete a visible environment. Returns whether a row was deleted."""
    env = _accessible_sandbox_envs(team_id, user_id).filter(pk=env_id).first()
    if env is None:
        return False
    env.delete()
    return True


# --- Sandbox custom images (presentation CRUD + builder/build flows) ---


def sandbox_custom_images_enabled(team_id: int, user_id: int) -> bool:
    """Whether custom base images are available for this team (Modal VM runtime flag gate)."""
    team = Team.objects.only("id", "organization_id").get(id=team_id)
    user = User.objects.only("id", "distinct_id").get(id=user_id)
    return is_custom_images_enabled(
        distinct_id=user.distinct_id or f"user-{user_id}",
        organization_id=str(team.organization_id),
    )


def _custom_image_to_dto(
    image: SandboxCustomImage, *, include_build_log: bool = False
) -> contracts.SandboxCustomImageDTO:
    from products.tasks.backend.logic.services.image_spec import spec_json_to_yaml  # noqa: PLC0415

    return contracts.SandboxCustomImageDTO(
        id=image.id,
        team_id=image.team_id,
        name=image.name,
        description=image.description,
        repository=image.repository,
        private=image.private,
        status=image.status,
        version=image.version,
        modal_image_name=image.modal_image_name,
        error=image.error,
        spec=image.spec or {},
        spec_yaml=spec_json_to_yaml(image.spec or {}),
        scan_result=image.scan_result or {},
        build_log=image.build_log if include_build_log else "",
        builder_task_id=image.builder_task_id,
        created_by=_user_basic_info(image.created_by if image.created_by_id else None),
        created_at=image.created_at,
        updated_at=image.updated_at,
    )


def _accessible_custom_images(team_id: int, user_id: int):
    return (
        SandboxCustomImage.objects.filter(team_id=team_id)
        .filter(Q(private=False) | Q(created_by_id=user_id))
        .select_related("created_by")
    )


def _reload_image_dto(image_pk: UUID) -> contracts.SandboxCustomImageDTO:
    return _custom_image_to_dto(SandboxCustomImage.objects.select_related("created_by").get(pk=image_pk))


def list_sandbox_custom_images(team_id: int, user_id: int) -> list[contracts.SandboxCustomImageDTO]:
    """Non-archived custom images visible to the user, newest first."""
    images = (
        _accessible_custom_images(team_id, user_id)
        .exclude(status=SandboxCustomImage.Status.ARCHIVED)
        .order_by("-created_at")
    )
    return [_custom_image_to_dto(image) for image in images]


def get_sandbox_custom_image(
    image_id: str | UUID, team_id: int, user_id: int
) -> contracts.SandboxCustomImageDTO | None:
    """Single-image detail; the only read that includes the (potentially large) build log."""
    image = _accessible_custom_images(team_id, user_id).filter(id=image_id).first()
    return _custom_image_to_dto(image, include_build_log=True) if image is not None else None


def create_sandbox_custom_image(
    team_id: int,
    user_id: int,
    *,
    name: str,
    description: str = "",
    repository: str | None = None,
    private: bool = False,
) -> contracts.SandboxCustomImageDTO:
    """Create a draft custom image and dispatch its interactive image-builder agent task."""
    from products.tasks.backend.logic.services.image_spec import validate_image_repository  # noqa: PLC0415

    if repository:
        validate_image_repository(repository)

    counts = (
        SandboxCustomImage.objects.filter(team_id=team_id)
        .exclude(status=SandboxCustomImage.Status.ARCHIVED)
        .aggregate(team=Count("id"), user=Count("id", filter=Q(created_by_id=user_id)))
    )
    if counts["team"] >= MAX_CUSTOM_IMAGES_PER_TEAM:
        raise ValueError(f"This team already has {MAX_CUSTOM_IMAGES_PER_TEAM} custom images; delete one first")
    if counts["user"] >= MAX_CUSTOM_IMAGES_PER_USER:
        raise ValueError(f"You already have {MAX_CUSTOM_IMAGES_PER_USER} custom images; delete one first")

    image = SandboxCustomImage.objects.create(
        team_id=team_id,
        created_by_id=user_id,
        name=name,
        description=description,
        repository=repository or "",
        private=private,
    )
    ensure_image_builder_task(image, user_id)
    return _reload_image_dto(image.pk)


def ensure_sandbox_custom_image_builder_task(
    image_id: str | UUID, team_id: int, user_id: int
) -> contracts.SandboxCustomImageDTO | None:
    """Revive (or reuse) the image's builder session; new sessions are seeded with the stored spec."""
    image = _accessible_custom_images(team_id, user_id).filter(id=image_id).first()
    if image is None:
        return None
    ensure_image_builder_task(image, user_id)
    return _reload_image_dto(image.pk)


def build_sandbox_custom_image(
    image_id: str | UUID, team_id: int, user_id: int, *, spec_yaml: str | None = None
) -> contracts.SandboxCustomImageDTO | None:
    """Persist the image spec and kick off the scan → build → publish workflow.

    The spec comes from ``spec_yaml`` when provided, otherwise it is read from the
    builder task's live sandbox. Raises ``ValueError`` on an invalid or empty spec.
    """
    from products.tasks.backend.logic.services.image_spec import (  # noqa: PLC0415
        SandboxImageSpecError,
        parse_image_spec_json,
        parse_image_spec_yaml,
        validate_spec_buildable,
    )
    from products.tasks.backend.metrics import (
        observe_custom_image_build,  # noqa: PLC0415 — keep prometheus deps off the api import path
    )
    from products.tasks.backend.temporal.client import execute_build_sandbox_image_workflow  # noqa: PLC0415

    image = _accessible_custom_images(team_id, user_id).filter(id=image_id).first()
    if image is None:
        return None
    if image.status in (SandboxCustomImage.Status.SCANNING, SandboxCustomImage.Status.BUILDING):
        raise ValueError("A build is already in progress for this image")

    try:
        spec = parse_image_spec_yaml(spec_yaml) if spec_yaml is not None else read_spec_from_builder_sandbox(image)
    except SandboxImageSpecError as e:
        # Builder sandbox gone → the stored spec is the only correct rebuild source.
        if spec_yaml is None and image.spec:
            spec = parse_image_spec_json(image.spec)
        else:
            raise ValueError(str(e))
    if spec.is_empty:
        raise ValueError("The image spec is empty; add packages, commands, or env vars before building")
    validate_spec_buildable(spec, image.repository)

    updated = (
        _accessible_custom_images(team_id, user_id)
        .filter(id=image.id)
        .exclude(status__in=(SandboxCustomImage.Status.SCANNING, SandboxCustomImage.Status.BUILDING))
        .update(
            spec=spec.model_dump(),
            status=SandboxCustomImage.Status.SCANNING,
            error="",
            updated_at=django_timezone.now(),
        )
    )
    if not updated:
        raise ValueError("A build is already in progress for this image")

    observe_custom_image_build("started")
    execute_build_sandbox_image_workflow(str(image.id), team_id)
    return _reload_image_dto(image.pk)


def update_sandbox_custom_image(
    image_id: str | UUID,
    team_id: int,
    user_id: int,
    *,
    name: str | None = None,
    description: str | None = None,
) -> contracts.SandboxCustomImageDTO | None:
    """Rename (and optionally re-describe) a visible custom image.

    Only mutable metadata (`name`, `description`) is editable here — the build spec,
    status, and published image reference are managed by the build/scan flow. Returns
    ``None`` when the image is not visible to the caller.
    """
    updates: dict[str, str] = {}
    if name is not None:
        updates["name"] = name
    if description is not None:
        updates["description"] = description
    if not updates:
        return get_sandbox_custom_image(image_id, team_id, user_id)

    updated = (
        _accessible_custom_images(team_id, user_id)
        .filter(id=image_id)
        .update(**updates, updated_at=django_timezone.now())
    )
    if not updated:
        return None
    # Re-read through get_sandbox_custom_image rather than _reload_image_dto: if a
    # concurrent delete removes the row between the update above and this read,
    # `.first()` returns None (→ 404) instead of .get() raising DoesNotExist (→ 500).
    return get_sandbox_custom_image(image_id, team_id, user_id)


def delete_sandbox_custom_image(image_id: str | UUID, team_id: int, user_id: int) -> bool:
    """Delete a visible custom image. Environments referencing it fall back to the default base (SET_NULL)."""
    image = _accessible_custom_images(team_id, user_id).filter(id=image_id).first()
    if image is None:
        return False
    image.delete()
    return True


# --- Task automations (presentation CRUD) ---
# Visibility mirrors the task-run visibility filter traversed via the automation's ``task`` FK
# (creator, legacy unowned tasks, and signals-pipeline tasks). Most read fields proxy off the
# linked ``Task``; the schedule (Temporal) is kept in sync from inside the write functions.


def _task_automation_to_dto(automation: TaskAutomation) -> contracts.TaskAutomationDTO:
    return contracts.TaskAutomationDTO(
        id=automation.id,
        name=automation.name,
        prompt=automation.prompt,
        repository=automation.repository,
        github_integration=automation.github_integration_id,
        cron_expression=automation.cron_expression,
        timezone=automation.timezone,
        template_id=automation.template_id,
        enabled=automation.enabled,
        last_run_at=automation.last_run_at,
        last_run_status=automation.last_run_status,
        last_task_id=str(automation.task_id),
        last_task_run_id=str(automation.last_task_run_id) if automation.last_task_run_id else None,
        last_error=automation.last_error,
        created_at=automation.created_at,
        updated_at=automation.updated_at,
    )


def _visible_task_automations(team_id: int, user_id: int | None):
    return TaskAutomation.objects.filter(task__team_id=team_id).filter(task_run_visibility_q(user_id))


def list_task_automations(team_id: int, user_id: int | None) -> list[contracts.TaskAutomationDTO]:
    """Automations for the team visible to the user, ordered by task title then newest first."""
    automations = _visible_task_automations(team_id, user_id).order_by("task__title", "-created_at")
    return [_task_automation_to_dto(automation) for automation in automations]


def get_task_automation(
    automation_id: str | UUID, team_id: int, user_id: int | None
) -> contracts.TaskAutomationDTO | None:
    """A single automation visible to the user. Returns ``None`` if not found/visible."""
    automation = _visible_task_automations(team_id, user_id).filter(pk=automation_id).first()
    return _task_automation_to_dto(automation) if automation is not None else None


def create_task_automation(
    team_id: int,
    user_id: int | None,
    *,
    name: str,
    prompt: str,
    repository: str,
    github_integration_id: int | None = None,
    cron_expression: str,
    timezone: str = "UTC",
    template_id: str | None = None,
    enabled: bool = True,
) -> contracts.TaskAutomationDTO:
    """Create an automation (and its backing task) and sync its Temporal schedule.

    Falls back to the team's default GitHub integration when none is supplied, mirroring the
    original serializer behavior.
    """
    if github_integration_id is None:
        default_integration = Integration.objects.filter(team_id=team_id, kind="github").first()
        github_integration_id = default_integration.id if default_integration else None

    with transaction.atomic():
        task = Task.objects.create(
            team_id=team_id,
            created_by_id=user_id,
            title=name,
            description=prompt,
            origin_product=Task.OriginProduct.AUTOMATION,
            repository=repository,
            github_integration_id=github_integration_id,
        )
        automation = TaskAutomation.objects.create(
            task=task,
            cron_expression=cron_expression,
            timezone=timezone,
            template_id=template_id,
            enabled=enabled,
        )

    _sync_automation_schedule(automation)
    return _task_automation_to_dto(automation)


def update_task_automation(
    automation_id: str | UUID, team_id: int, user_id: int | None, **fields
) -> contracts.TaskAutomationDTO | None:
    """Partially update a visible automation (and its backing task) and re-sync its schedule.

    Returns ``None`` if the automation is not found/visible. The ``github_integration_id``
    key (when present) updates the backing task's GitHub integration.
    """
    automation = _visible_task_automations(team_id, user_id).filter(pk=automation_id).first()
    if automation is None:
        return None

    task_field_map = {
        "name": "title",
        "prompt": "description",
        "repository": "repository",
        "github_integration_id": "github_integration_id",
    }
    task_updates = {task_field_map[key]: fields.pop(key) for key in list(fields) if key in task_field_map}

    with transaction.atomic():
        for key, value in fields.items():
            setattr(automation, key, value)
        automation.save()

        if task_updates:
            task = automation.task
            fields_to_update = []
            for field, value in task_updates.items():
                if getattr(task, field) != value:
                    setattr(task, field, value)
                    fields_to_update.append(field)
            if fields_to_update:
                fields_to_update.append("updated_at")
                task.save(update_fields=fields_to_update)

    _sync_automation_schedule(automation)
    return _task_automation_to_dto(automation)


def delete_task_automation(automation_id: str | UUID, team_id: int, user_id: int | None) -> bool:
    """Delete a visible automation and its Temporal schedule. Returns whether a row was deleted."""
    automation = _visible_task_automations(team_id, user_id).filter(pk=automation_id).first()
    if automation is None:
        return False

    from products.tasks.backend.automation_service import (  # noqa: PLC0415 — keep temporalio off the api import path
        delete_automation_schedule,
    )

    delete_automation_schedule(automation)
    automation.delete()
    return True


def run_task_automation_now(
    automation_id: str | UUID, team_id: int, user_id: int | None
) -> contracts.TaskAutomationDTO | None:
    """Trigger an automation run immediately and return the refreshed automation DTO.

    Returns ``None`` if the automation is not found/visible.
    """
    automation = _visible_task_automations(team_id, user_id).filter(pk=automation_id).first()
    if automation is None:
        return None

    from products.tasks.backend.automation_service import (  # noqa: PLC0415 — keep temporalio off the api import path
        run_task_automation,
    )

    run_task_automation(str(automation.id))
    automation.refresh_from_db()
    return _task_automation_to_dto(automation)


def _sync_automation_schedule(automation: TaskAutomation) -> None:
    from products.tasks.backend.automation_service import (  # noqa: PLC0415 — keep temporalio off the api import path
        sync_automation_schedule,
    )

    sync_automation_schedule(automation)


# --- Task runs (presentation lifecycle) ---
# Every function takes ids/primitives and returns a TaskRunDetailDTO (or a small result),
# moving all ORM access and Temporal/Slack/S3 orchestration behind the facade. Visibility on
# the parent task is enforced via ``task_visibility_q``; runs are always team-scoped.

# Run-state keys that are server-owned and must never be mutable through the PATCH endpoint:
#   - github_credential_source / pr_authorship_mode fix the run's GitHub identity at creation;
#     a caller could otherwise flip a caller-token run to ``server_integration`` and have the
#     task creator's server-side token injected into their sandbox.
#   - sandbox_id is the credential-propagation target; a caller could otherwise repoint a visible
#     run at a sandbox they control and capture the run owner's token on the next rotation.
#   - sandbox_cpu_cores / sandbox_memory_gb / sandbox_ttl_seconds / inactivity_timeout_seconds set
#     the run's compute and lifetime at creation; a caller could otherwise PATCH a queued run to
#     provision an oversized or long-lived sandbox beyond what they're entitled to.
#   - use_modal_directory_resume_snapshots is the server-side directory snapshot rollout decision;
#     a caller could otherwise force directory snapshot creation while the feature flag is off.
#   - use_modal_vm_sandbox is reserved for trusted server-created runs such as image builders;
#     a caller could otherwise force the VM runtime while the feature flag or custom-image gate is off.
#   - snapshot_external_id / snapshot_kind / snapshot_mount_path control which Modal image is
#     restored on resume and where directory snapshots are mounted.
#   - workflow_id is the run's Temporal workflow address (``TaskRun.workflow_id`` prefers it over
#     the derived id); a caller could otherwise repoint their run at another team's workflow and
#     signal or terminate-and-restart it.
#   - pending_external_followups / pending_external_followups_generation are the workflow-owned
#     durable queue; a caller could otherwise inject actor identity into a restored follow-up.
# These keys are reserved for server-owned run state, never PATCH input.
_PROTECTED_RUN_STATE_KEYS = frozenset(
    {
        "github_credential_source",
        "pr_authorship_mode",
        "repositories",
        "verified_pr_urls",
        "sandbox_id",
        "sandbox_cpu_cores",
        "sandbox_memory_gb",
        "sandbox_ttl_seconds",
        "inactivity_timeout_seconds",
        "wizard_config",
        "wizard_head_branch",
        "use_modal_directory_resume_snapshots",
        "use_modal_vm_sandbox",
        # Rollout stamps written once at dispatch by _capture_run_feature_flags; a PATCHable
        # value would let a task controller bypass the org feature flags (for telemetry, that
        # means injecting the internal OTLP capture token into their sandbox and re-enabling
        # the run-log mirror with the rollout off).
        AGENT_OTEL_TELEMETRY_STATE_KEY,
        "sandbox_event_ingest_enabled",
        "snapshot_external_id",
        "snapshot_kind",
        "snapshot_mount_path",
        "workflow_id",
        "pending_dispatch",
        # Written once at loop fire time; seeding copies these storage paths into the
        # run's artifact prefix, so a PATCHable value would be an arbitrary
        # object-storage read (and write-location) primitive.
        "skill_bundle_seeds",
        "cancel_requested_at",
        "cancel_requested_by_user_id",
        "cancel_source",
        "cancel_fallback_cleanup_complete",
        "pending_external_followups",
        "pending_external_followups_generation",
        # Credential grant decided at Task.create_and_run time by server-owned callers (the scout
        # runner); a PATCHable key would let any task controller mint a GitHub token onto a
        # queued repo-less run.
        "github_read_access",
        # Loop provenance is stamped once at run creation (see loop_runs._create_loop_task_and_run)
        # and drives loop bookkeeping in handle_loop_run_terminal. A run update must never be able
        # to forge or repoint it, or a caller could steer terminal side effects at another loop.
        "loop_id",
        "loop_trigger_id",
        "trigger_context",
        "config_snapshot",
        # Stamped once at run creation. The review carve-outs read ai_stage="implementation" as proof
        # a run is self-driving, so a PATCHable value would forge that and unlock the bot/draft bypass.
        "ai_stage",
        # The server-generated head branch the run->PR link is keyed on (find_signal_implementation_run).
        # A PATCHable value would let a caller re-aim the approve-first carve-out at any App-authored
        # PR, which is the exact forgery the stamp exists to prevent.
        "self_driving_head_branch",
        # The run's model posture, chosen at creation by the server-owned caller and read back out
        # of state when the run dispatches. It decides what the run costs, and for a run routed to
        # an unbilled gateway product (create_wizard_cloud_run pins claude-sonnet-5 for the
        # `onboarding` product) it is the only thing keeping the run off the more expensive models
        # that product still allowlists. Every writer is server-side, so nothing legitimate PATCHes
        # these.
        "runtime_adapter",
        "provider",
        "model",
        "reasoning_effort",
    }
)

_TERMINAL_TASK_RUN_STATUSES = (TaskRun.Status.COMPLETED, TaskRun.Status.FAILED, TaskRun.Status.CANCELLED)

# `output.pr_merged` is GitHub's word, recorded by the PR webhook (`_record_run_pr_merged`) — never
# the caller's. Signals reads it to decide refund finality (billing.report_pr_is_merged): a report
# whose PR merged keeps its resolved status through a refund instead of being suppressed and having
# its PR closed. Any caller-writable path to this flag would let a `task:write` holder mark an open
# PR merged, resolve the report, and refund it while keeping the work — so every output writer has to
# go through `_merge_caller_output`, not merge caller output directly.
_WEBHOOK_ATTESTED_RUN_OUTPUT_KEYS = frozenset({"pr_merged"})


def _apply_caller_output(stored: object, incoming: dict, merged: dict) -> dict:
    """Enforce the webhook-attested keys on a caller's output write.

    `merged` is whatever the writer built from `incoming` (a wholesale replacement for
    `set_task_run_output`, a merge over stored output for the PATCH path); this drops any attested
    key the caller supplied and restores the stored one. The attestation is bound to the PR it was
    made about, so when the caller points the run at a different `pr_url` the stored flag described
    the old PR and is dropped rather than silently transferred onto the new one.
    """
    existing = stored if isinstance(stored, dict) else {}
    same_pr = not incoming.get("pr_url") or incoming["pr_url"] == existing.get("pr_url")
    for key in _WEBHOOK_ATTESTED_RUN_OUTPUT_KEYS:
        merged.pop(key, None)
        if same_pr and existing.get(key):
            merged[key] = existing[key]
    return merged


def _task_run_queryset():
    return TaskRun.objects.select_related(
        "task", "task__created_by", "task__github_integration", "task__github_user_integration"
    )


def _get_task_for_run_control(task_id: str | UUID, team_id: int, user_id: int | None) -> Task | None:
    """The task, only if the user may drive runs on it (``task_control_q``, not mere visibility)."""
    return Task.objects.filter(id=task_id, team_id=team_id).filter(task_control_q(user_id)).first()


def _get_visible_run(run_id: str | UUID, task_id: str | UUID, team_id: int) -> TaskRun | None:
    """A run scoped to its parent task + team. Caller is responsible for task visibility."""
    return _task_run_queryset().filter(pk=run_id, team_id=team_id, task_id=task_id).first()


def task_run_exists(run_id: str | UUID, task_id: str | UUID, team_id: int) -> bool:
    """Precheck so callers can 404 before doing expensive work (e.g. a render)."""
    try:
        return TaskRun.objects.filter(pk=run_id, team_id=team_id, task_id=task_id).exists()
    except (ValueError, TypeError, DjangoValidationError):
        return False


def task_accessible_for_run_view(
    task_id: str | UUID,
    team_id: int,
    user_id: int | None,
    *,
    bypass_visibility: bool = False,
    for_control: bool = False,
) -> bool:
    """Whether the parent task exists and (unless bypassed) is visible to the user.

    Mirrors the parent-task gate in ``TaskRunViewSet.safely_get_queryset``: runs are always scoped
    to a task, and access to that task is gated by ``task_visibility_q`` except for internal-debug
    read actions, which the caller signals via ``bypass_visibility``. Run-mutating actions pass
    ``for_control`` to use the narrower ``task_control_q`` — public-channel visibility lets
    teammates watch a run, not drive it.

    Slack-originated tasks are the exception: those threads are multiplayer, so any same-team
    user can already steer the run from Slack (follow-ups record them as the run's actor, with
    sandbox credentials minted for them). The runs API mirrors that and lets team members drive
    and watch Slack tasks' runs — otherwise a non-creator actor's sandbox 404s on every callback
    (reply relay, log-append heartbeat, completion PATCH) and the thread dies silently.
    """
    task_filter = Task.objects.filter(id=task_id, team_id=team_id)
    if not bypass_visibility:
        scope_q = task_control_q(user_id) if for_control else task_visibility_q(user_id)
        task_filter = task_filter.filter(scope_q | Q(origin_product=Task.OriginProduct.SLACK))
    return task_filter.exists()


def list_task_runs(task_id: str | UUID, team_id: int) -> list[contracts.TaskRunDetailDTO]:
    """All runs for a task, team-scoped. Caller enforces task visibility."""
    runs = _task_run_queryset().filter(team_id=team_id, task_id=task_id)
    return [_task_run_detail_to_dto(run) for run in runs]


def get_task_run_detail(run_id: str | UUID, task_id: str | UUID, team_id: int) -> contracts.TaskRunDetailDTO | None:
    """A single run as a detail DTO, scoped to its task + team."""
    run = _get_visible_run(run_id, task_id, team_id)
    return _task_run_detail_to_dto(run) if run is not None else None


def get_task_run_stream_info(
    run_id: str | UUID, task_id: str | UUID, team_id: int
) -> contracts.TaskRunStreamInfoDTO | None:
    """The minimal run facts the SSE stream view needs. ``None`` if the run isn't found."""
    from products.tasks.backend.metrics import (  # noqa: PLC0415 — keep prometheus deps off the api import path
        origin_product_label,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    return contracts.TaskRunStreamInfoDTO(
        id=run.id,
        state=run.state or {},
        origin_product=origin_product_label(run),
    )


def signal_workflow_completion(run_id: str | UUID, status: str, error_message: str | None) -> None:
    """Send a completion signal to a run's Temporal workflow (best-effort)."""
    import asyncio  # noqa: PLC0415 — only needed when signalling

    from insights.temporal.common.client import sync_connect  # noqa: PLC0415 — keep temporalio off the api import path

    from products.tasks.backend.temporal.process_task.workflow import (  # noqa: PLC0415 — keep temporalio off the api import path
        ProcessTaskWorkflow,
    )

    run = TaskRun.objects.filter(pk=run_id).first()
    if run is None:
        return
    try:
        client = sync_connect()
        handle = client.get_workflow_handle(run.workflow_id)
        asyncio.run(handle.signal(ProcessTaskWorkflow.complete_task, args=[status, error_message]))
        logger.info("Signaled workflow completion for task run %s with status %s", run.id, status)
    except Exception as e:
        logger.warning("Failed to signal workflow completion for task run %s: %s", run.id, e)


def _post_slack_update_for_pr(run: TaskRun) -> None:
    pr_url = (run.output or {}).get("pr_url") if isinstance(run.output, dict) else None
    if not pr_url:
        return

    from products.slack_app.backend.models import (  # noqa: PLC0415 — cross-product import kept off the api import path
        SlackThreadTaskMapping,
    )
    from products.tasks.backend.temporal.process_task.activities.post_slack_update import (  # noqa: PLC0415 — keep temporalio off the api import path
        PostSlackUpdateInput,
        post_slack_update,
    )

    try:
        mapping = (
            SlackThreadTaskMapping.objects.filter(task_run=run)
            .order_by("-updated_at")
            .values("integration_id", "channel", "thread_ts", "mentioning_slack_user_id")
            .first()
        )
        if not mapping:
            return
        post_slack_update(
            PostSlackUpdateInput(
                run_id=str(run.id),
                slack_thread_context={
                    "integration_id": mapping["integration_id"],
                    "channel": mapping["channel"],
                    "thread_ts": mapping["thread_ts"],
                    "mentioning_slack_user_id": mapping["mentioning_slack_user_id"],
                },
            )
        )
    except Exception:
        logger.exception("task_run_slack_update_for_pr_failed for run %s", run.id)


def _is_wizard_pr_ready_email_enabled(run: TaskRun) -> bool:
    user = run.task.created_by
    if user is None or not user.distinct_id:
        return False
    try:
        team = Team.objects.only("id", "uuid", "organization_id").get(id=run.team_id)
        organization_id = str(team.organization_id)
        return bool(
            hanzo_insights.feature_enabled(
                WIZARD_PR_READY_EMAIL_FEATURE_FLAG,
                user.distinct_id,
                groups={"organization": organization_id},
                group_properties={"organization": {"id": organization_id}},
                only_evaluate_locally=False,
                send_feature_flag_events=False,
            )
        )
    except Exception:
        logger.exception("wizard_pr_ready_email_feature_flag_check_failed", extra={"run_id": str(run.id)})
        return False


def _is_github_pull_request_url_for_repository(pr_url: str, repository: str | None) -> bool:
    if not repository:
        return False
    try:
        parsed_url = urlparse(pr_url)
    except ValueError:
        return False

    if parsed_url.scheme != "https" or parsed_url.netloc != "github.com":
        return False
    if parsed_url.params or parsed_url.query or parsed_url.fragment:
        return False

    path_parts = parsed_url.path.strip("/").split("/")
    repository_parts = repository.strip("/").split("/")
    if len(path_parts) != 4 or len(repository_parts) != 2:
        return False

    return (
        path_parts[0].lower() == repository_parts[0].lower()
        and path_parts[1].lower() == repository_parts[1].lower()
        and path_parts[2] == "pull"
        and path_parts[3].isdigit()
    )


def _send_wizard_pr_ready_email_for_pr(run: TaskRun) -> None:
    pr_url = (run.output or {}).get("pr_url") if isinstance(run.output, dict) else None
    if not pr_url or run.task.origin_product != Task.OriginProduct.ONBOARDING:
        return
    if run.task.created_by_id is None:
        return
    if not _is_github_pull_request_url_for_repository(pr_url, run.task.repository):
        logger.warning(
            "wizard_pr_ready_email_invalid_pr_url",
            extra={"run_id": str(run.id), "task_id": str(run.task_id), "repository": run.task.repository},
        )
        return
    if not _is_wizard_pr_ready_email_enabled(run):
        return

    if not run.task.mark_pr_ready_email_queued(pr_url):
        return

    from insights.tasks.email import send_wizard_pr_ready_email  # noqa: PLC0415 - keep email task import lazy

    transaction.on_commit(lambda: send_wizard_pr_ready_email.delay(str(run.id)))


def enforce_self_driving_pr_quota(team: Team, *, report_id: str | None = None, stage: str = "manual_create") -> None:
    """Refuse to create a PR-opening self-driving task while the team's org is over its self-driving
    credits quota with enforcement on. The implementation task is the step that leads to the
    billable PR, so the manual create-from-report path must respect the same limit as the pipeline
    auto-start gate (products/signals/backend/auto_start.py). Emits `signal_report_quota_paused`
    at ``stage`` whenever the org is limited, so each caller's gate stays measurable during the
    dark launch like every other gate. Raises ``QuotaLimitExceeded`` (402).
    """
    from insights.exceptions import QuotaLimitExceeded  # noqa: PLC0415 — keep billing deps off the api import path

    from products.signals.backend.quota import (  # noqa: PLC0415 — cross-product read kept off the api import path
        capture_signal_report_quota_paused,
        self_driving_quota_gate,
    )

    gate = self_driving_quota_gate(team)
    if gate.limited:
        capture_signal_report_quota_paused(team, report_id=report_id, stage=stage, enforced=gate.enforced)
    if gate.enforced:
        raise QuotaLimitExceeded(
            "Your organization reached its self-driving pull request limit. "
            "Increase the limit from the Inbox usage widget, or ask an org admin to do so."
        )


def update_task_run(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    validated_data: dict,
    only_if_non_terminal: bool = False,
) -> contracts.TaskRunDetailDTO | None:
    """Apply a PATCH to a run: merge output/state, set completion, then dispatch side effects.

    Mirrors ``TaskRunViewSet.partial_update`` byte-for-byte: protected state keys are stripped,
    output/state merges take a row lock, terminal transitions signal Temporal + dispatch
    push/Slack updates after commit, and a cloud→local transition cancels the workflow.
    """
    from products.tasks.backend.automation_service import (  # noqa: PLC0415 — keep temporalio off the api import path
        update_automation_run_result,
    )
    from products.tasks.backend.logic.services.loop_runs import (  # noqa: PLC0415 (keep temporalio off the api import path)
        handle_loop_run_terminal,
    )
    from products.tasks.backend.metrics import (  # noqa: PLC0415 — keep prometheus deps off the api import path
        observe_agent_turn_failed,
        observe_wizard_run_unbound,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None

    validated_data = dict(validated_data)
    has_output_merge = "output" in validated_data and isinstance(validated_data["output"], dict)
    has_state_merge = "state" in validated_data and isinstance(validated_data["state"], dict)
    if has_state_merge:
        validated_data["state"] = {
            k: v for k, v in validated_data["state"].items() if k not in _PROTECTED_RUN_STATE_KEYS
        }
    state_remove_keys = [
        k for k in (validated_data.get("state_remove_keys") or []) if k not in _PROTECTED_RUN_STATE_KEYS
    ]
    has_state_mutation = has_state_merge or bool(state_remove_keys)
    update_fields: set[str] = set()

    with transaction.atomic():
        if has_output_merge or has_state_mutation or only_if_non_terminal:
            run = TaskRun.objects.select_for_update().get(pk=run.pk)
        if only_if_non_terminal and run.is_terminal:
            return _task_run_detail_to_dto(run)

        old_status = run.status
        old_environment = run.environment
        old_pr_url = (run.output or {}).get("pr_url") if isinstance(run.output, dict) else None

        for key, value in validated_data.items():
            if key == "output" and isinstance(value, dict):
                existing_output = run.output if isinstance(run.output, dict) else {}
                # Same attested-key policy as set_task_run_output — this PATCH surface is
                # caller-controlled too, so it can't be a second path to output.pr_merged.
                merged_output = merge_pr_output(existing_output, value)
                setattr(run, key, _apply_caller_output(existing_output, value, merged_output))
                update_fields.add(key)
                continue
            if key == "state_remove_keys":
                continue
            if key == "state" and has_state_merge:
                existing_state = run.state if isinstance(run.state, dict) else {}
                next_state = dict(existing_state)
                for remove_key in state_remove_keys:
                    next_state.pop(remove_key, None)
                next_state.update(value)
                setattr(run, key, next_state)
                update_fields.add(key)
                continue
            setattr(run, key, value)
            update_fields.add(key)

        if state_remove_keys and not has_state_merge:
            existing_state = run.state if isinstance(run.state, dict) else {}
            next_state = dict(existing_state)
            for remove_key in state_remove_keys:
                next_state.pop(remove_key, None)
            run.state = next_state
            update_fields.add("state")

        new_status = validated_data.get("status")
        if new_status in _TERMINAL_TASK_RUN_STATUSES:
            if not run.completed_at:
                run.completed_at = django_timezone.now()
                update_fields.add("completed_at")

        update_fields.add("updated_at")
        run.save(update_fields=list(update_fields))
        run.publish_stream_state_event()

    update_automation_run_result(run)
    # Only on the actual transition: a repeat PATCH with the same terminal status, or an
    # output-only PATCH on an already-terminal run, must not re-run loop bookkeeping
    # (consecutive_failures would double-count). The workflow's status-update activity
    # applies the same guard on its side.
    if new_status in _TERMINAL_TASK_RUN_STATUSES and old_status != new_status:
        handle_loop_run_terminal(run)

    if new_status in _TERMINAL_TASK_RUN_STATUSES and old_status != new_status:
        if new_status == TaskRun.Status.FAILED:
            observe_agent_turn_failed(run)
            # This PATCH performed the DB transition, so it owns the task_run_failed
            # capture. The workflow's status-update activity sees the row already
            # FAILED and skips its own capture, keeping the event single-emitted.
            run.capture_event(
                "task_run_failed",
                {
                    "error_message": truncate_error_message(run.error_message),
                    "error_type": "agent_reported",
                    "duration_seconds": run._duration_seconds(),
                },
            )
        observe_wizard_run_unbound(run)
        signal_workflow_completion(run.id, new_status, validated_data.get("error_message"))
        if new_status == TaskRun.Status.CANCELLED:
            from products.tasks.backend.push_dispatcher import (  # noqa: PLC0415 — keep push deps off the api import path
                notify_task_run_cancelled,
            )

            notify_task_run_cancelled(run)
    new_environment = validated_data.get("environment")
    if new_environment == "local" and old_environment == TaskRun.Environment.CLOUD:
        signal_workflow_completion(run.id, "cancelled", "handoff")

    new_pr_url = (run.output or {}).get("pr_url") if isinstance(run.output, dict) else None
    if new_pr_url and new_pr_url != old_pr_url:
        _post_slack_update_for_pr(run)
        _send_wizard_pr_ready_email_for_pr(run)
        post_pr_created_thread_update(run, new_pr_url)
        # Surface the PR in the run's progress timeline the moment the agent reports it, so the install
        # UI advances past "Started agent" instead of waiting on the 15-min CI follow-up loop to emit
        # these. Steps coalesce by id with the workflow's own pr/ci emissions (frontend mergeProgressStep),
        # so the double-emit is harmless. Tolerant: a logging/stream hiccup must not fail the PATCH.
        try:
            run.emit_progress_event("pr", "completed", "Opened pull request", "setup", detail=new_pr_url)
            run.emit_progress_event("ci", "in_progress", "Keeping CI green", "setup")
        except Exception:
            logger.warning("task_run.pr_progress_emit_failed", extra={"run_id": str(run.id)}, exc_info=True)

    return _task_run_detail_to_dto(run)


def validate_set_output(run_id: str | UUID, task_id: str | UUID, team_id: int, *, output: dict) -> str | None:
    """Validate output against the task's json_schema. Returns an error message or ``None``."""
    import jsonschema  # noqa: PLC0415 — only needed when a json_schema is set

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    task = run.task
    if task.json_schema:
        try:
            jsonschema.validate(instance=output, schema=task.json_schema)
        except jsonschema.ValidationError as e:
            return f"Output validation error: {e.message}"
    return None


def set_task_run_output(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, output: dict
) -> contracts.TaskRunDetailDTO | None:
    """Persist a run's output. Completes the run for structured-output tasks; posts Slack PR update."""
    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    task = run.task
    # Preserve PR facts a webhook may have written concurrently: this assignment is wholesale,
    # so a bare `= output` would drop output.pr_url recorded out of band.
    existing = run.output if isinstance(run.output, dict) else {}
    merged = merge_pr_output(existing, output)
    run.output = _apply_caller_output(existing, output, merged)
    run.save(update_fields=["output", "updated_at"])
    if task.json_schema:
        signal_workflow_completion(run.id, TaskRun.Status.COMPLETED, None)
    run.publish_stream_state_event()
    _post_slack_update_for_pr(run)
    _send_wizard_pr_ready_email_for_pr(run)
    if merged.get("pr_url"):
        post_pr_created_thread_update(run, merged["pr_url"])
    return _task_run_detail_to_dto(run)


def append_task_run_log(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, entries: list[dict]
) -> contracts.TaskRunDetailDTO | None:
    """Append log entries to a run's S3 log and heartbeat its workflow."""
    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    run.append_log(entries)
    run.heartbeat_workflow(agent_active=True)
    return _task_run_detail_to_dto(run)


def ensure_task_run_session(run_id: str | UUID) -> UUID:
    with transaction.atomic():
        run = TaskRun.objects.select_for_update(of=("self",)).select_related("task__team").get(id=run_id)
        if run.active_task_session_id is not None:
            return run.active_task_session_id

        task_session = TaskSession.create_for_task(run.task)
        run.active_task_session = task_session
        run.save(update_fields=["active_task_session", "updated_at"])
        return task_session.id


def get_task_run_session(
    run_id: str | UUID, task_id: str | UUID, team_id: int
) -> tuple[UUID, str | None, str | None] | None:
    from insights.storage import object_storage  # noqa: PLC0415

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None or run.active_task_session_id is None:
        return None
    task_session = TaskSession.objects.unscoped().get(id=run.active_task_session_id)
    if task_session.object_storage_key is None:
        return task_session.id, None, None
    download_url = object_storage.get_presigned_url(task_session.object_storage_key, expiration=3600)
    if not download_url:
        raise RuntimeError("Unable to prepare task session download")
    return task_session.id, download_url, task_session.content_sha256


def _validate_task_session_content(content: bytes) -> None:
    if not content or len(content) > TASK_SESSION_MAX_SIZE_BYTES:
        raise ValueError("The task session content size is invalid")


def _get_open_sandbox_session(run_id: UUID, sandbox_id: str) -> SandboxSession | None:
    return (
        SandboxSession.objects.unscoped()
        .filter(
            task_run_id=run_id,
            sandbox_id=sandbox_id,
            ended_at__isnull=True,
        )
        .first()
    )


def _delete_task_session_object(task_session_id: UUID, object_storage_key: str) -> None:
    from insights.storage import object_storage  # noqa: PLC0415

    try:
        object_storage.delete(object_storage_key)
    except Exception as error:
        logger.warning(
            "task_session.failed_to_delete_object",
            extra={
                "task_session_id": str(task_session_id),
                "object_storage_key": object_storage_key,
                "error": str(error),
            },
        )


def validate_task_run_sandbox_token(
    token: str,
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    sandbox_id: str,
) -> bool:
    from jwt import InvalidTokenError  # noqa: PLC0415

    from products.tasks.backend.logic.services.connection_token import (  # noqa: PLC0415
        validate_sandbox_event_ingest_token,
    )

    try:
        claims = validate_sandbox_event_ingest_token(token)
    except (InvalidTokenError, ValueError):
        return False
    return (
        claims.run_id == str(run_id)
        and claims.task_id == str(task_id)
        and claims.team_id == team_id
        and claims.sandbox_id == sandbox_id
    )


def sync_task_run_session(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    sandbox_id: str,
    expected_content_sha256: str | None,
    content: bytes,
) -> tuple[UUID, str] | None:
    from insights.storage import object_storage  # noqa: PLC0415

    visible_run = _get_visible_run(run_id, task_id, team_id)
    if visible_run is None or visible_run.active_task_session_id is None:
        return None
    _validate_task_session_content(content)
    if _get_open_sandbox_session(visible_run.id, sandbox_id) is None:
        raise ValueError("The task session writer is not the active sandbox")

    content_sha256 = hashlib.sha256(content).hexdigest()
    object_storage_key = (
        f"task-sessions/{visible_run.task.team.organization_id}/{visible_run.task_id}/"
        f"{visible_run.active_task_session_id}/{uuid4()}.jsonl"
    )
    object_storage.write(object_storage_key, content)

    previous_object_storage_key: str | None = None
    try:
        with transaction.atomic():
            locked_run = TaskRun.objects.select_for_update(of=("self",)).get(id=visible_run.id)
            if locked_run.active_task_session_id != visible_run.active_task_session_id:
                raise ValueError("The task session sync is stale")
            if _get_open_sandbox_session(locked_run.id, sandbox_id) is None:
                raise ValueError("The task session writer is not the active sandbox")

            task_session = TaskSession.objects.unscoped().select_for_update().get(id=locked_run.active_task_session_id)
            if task_session.content_sha256 == content_sha256:
                transaction.on_commit(lambda: _delete_task_session_object(task_session.id, object_storage_key))
                return task_session.id, content_sha256
            if task_session.content_sha256 != expected_content_sha256:
                raise ValueError("The task session content is stale")

            previous_object_storage_key = task_session.object_storage_key
            task_session.object_storage_key = object_storage_key
            task_session.content_sha256 = content_sha256
            task_session.size = len(content)
            task_session.save(update_fields=["object_storage_key", "content_sha256", "size", "updated_at"])
            if previous_object_storage_key is not None:
                transaction.on_commit(lambda: _delete_task_session_object(task_session.id, previous_object_storage_key))
        task_session.tag_object()
        return task_session.id, content_sha256
    except Exception:
        _delete_task_session_object(visible_run.active_task_session_id, object_storage_key)
        raise


def task_run_has_slack_mapping(run_id: str | UUID, task_id: str | UUID, team_id: int) -> bool | None:
    """Whether a run is mapped to a Slack thread. ``None`` if the run isn't found."""
    from products.slack_app.backend.models import (  # noqa: PLC0415 — cross-product import kept off the api import path
        SlackThreadTaskMapping,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    return SlackThreadTaskMapping.objects.filter(task_run=run).exists()


def task_run_is_terminal(run_id: str | UUID, task_id: str | UUID, team_id: int) -> bool | None:
    """Whether a run is in a terminal state. ``None`` if the run isn't found."""
    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    return run.is_terminal


# --- Task run artifacts (S3 + manifest) ---


def _build_artifact_storage_path(run: TaskRun, artifact_id: str, name: str) -> tuple[str, str]:
    from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        get_safe_artifact_name,
    )

    safe_name = get_safe_artifact_name(name)
    prefix = run.get_artifact_s3_prefix()
    return safe_name, f"{prefix}/{artifact_id[:8]}_{safe_name}"


def _tag_artifact_object(run: TaskRun, storage_path: str) -> None:
    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    try:
        object_storage.tag(storage_path, {"ttl_days": "30", "team_id": str(run.team_id)})
    except Exception as exc:
        logger.warning(
            "task_run.artifact_tag_failed",
            extra={"task_run_id": str(run.id), "storage_path": storage_path, "error": str(exc)},
        )


def _build_artifact_manifest_entry(
    *,
    artifact_id: str,
    name: str,
    artifact_type: str,
    source: str,
    size: int,
    content_type: str,
    storage_path: str,
    uploaded_at: str,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    entry: dict[str, Any] = {
        "id": artifact_id,
        "name": name,
        "type": artifact_type,
        "source": source,
        "size": size,
        "content_type": content_type,
        "storage_path": storage_path,
        "uploaded_at": uploaded_at,
    }
    if metadata:
        entry["metadata"] = metadata
    return entry


def _find_artifact_manifest_entry(manifest: list[dict], artifact_id: str, storage_path: str) -> dict | None:
    return next(
        (e for e in manifest if e.get("id") == artifact_id or e.get("storage_path") == storage_path),
        None,
    )


def _save_artifact_manifest(run: TaskRun, manifest: list[dict]) -> None:
    run.artifacts = manifest
    run.save(update_fields=["artifacts", "updated_at"])


def upload_task_run_artifacts(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, artifacts: list[dict]
) -> tuple[list[dict], list[dict]] | None:
    """Write artifact bytes to S3 and append them to the run manifest.

    Returns ``(uploaded, manifest)`` — the entries created for ``artifacts`` and the full
    manifest including them — or ``None`` when the run isn't visible.

    An artifact may carry an explicit ``id``; entries with that id are upserted into the
    manifest (same-id S3 writes overwrite the same key), so callers that derive ids
    deterministically get idempotent uploads under retries. Without an ``id`` each call
    appends a fresh entry.
    """
    import uuid as uuid_module  # noqa: PLC0415

    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None

    uploaded: list[dict] = []
    for artifact in artifacts:
        artifact_id = str(artifact.get("id") or uuid_module.uuid4().hex)
        safe_name, storage_path = _build_artifact_storage_path(run, artifact_id, artifact["name"])

        content_bytes = artifact["content_bytes"]
        extras: dict[str, str] = {}
        content_type = artifact.get("content_type")
        if content_type:
            extras["ContentType"] = content_type

        object_storage.write(storage_path, content_bytes, extras or None)
        _tag_artifact_object(run, storage_path)

        uploaded.append(
            _build_artifact_manifest_entry(
                artifact_id=artifact_id,
                name=safe_name,
                artifact_type=artifact["type"],
                source=artifact.get("source") or "",
                size=len(content_bytes),
                content_type=content_type or "",
                storage_path=storage_path,
                uploaded_at=django_timezone.now().isoformat(),
                metadata=artifact.get("metadata"),
            )
        )
        logger.info(
            "task_run.artifact_uploaded",
            extra={
                "task_run_id": str(run.id),
                "storage_path": storage_path,
                "artifact_type": artifact["type"],
                "size": len(content_bytes),
            },
        )

    with transaction.atomic():
        run = TaskRun.objects.select_for_update().get(pk=run.pk)
        uploaded_ids = {entry["id"] for entry in uploaded}
        manifest = [entry for entry in (run.artifacts or []) if entry.get("id") not in uploaded_ids]
        manifest.extend(uploaded)
        _save_artifact_manifest(run, manifest)

    return uploaded, manifest


def prepare_task_run_artifact_uploads(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    artifacts: list[dict],
    upload_expiration_seconds: int,
    form_overhead_bytes: int,
) -> tuple[list[dict] | None, bool]:
    """Reserve S3 keys and presigned POST forms for direct artifact uploads.

    Returns ``(prepared, ok)``: ``(None, _)`` when the run isn't found, ``(None, False)`` when a
    presigned POST could not be generated, else ``(prepared, True)``.
    """
    import uuid as uuid_module  # noqa: PLC0415

    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None, True

    prepared: list[dict] = []
    for artifact in artifacts:
        artifact_id = uuid_module.uuid4().hex
        safe_name, storage_path = _build_artifact_storage_path(run, artifact_id, artifact["name"])
        content_type = artifact.get("content_type") or ""
        conditions: list[list[str | int]] = [["content-length-range", 0, artifact["size"] + form_overhead_bytes]]

        presigned_post = object_storage.get_presigned_post(
            storage_path, conditions=conditions, expiration=upload_expiration_seconds
        )
        if not presigned_post:
            return None, False

        prepared_artifact = {
            "id": artifact_id,
            "name": safe_name,
            "type": artifact["type"],
            "source": artifact.get("source") or "",
            "size": artifact["size"],
            "content_type": content_type,
            "storage_path": storage_path,
            "expires_in": upload_expiration_seconds,
            "presigned_post": presigned_post,
        }
        if metadata := artifact.get("metadata"):
            prepared_artifact["metadata"] = metadata
        prepared.append(prepared_artifact)
    return prepared, True


def finalize_task_run_artifact_uploads(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, artifacts: list[dict]
) -> tuple[list[dict] | None, str | None]:
    """Verify directly-uploaded S3 objects and attach them to the run manifest.

    Returns ``(finalized_entries, error)``: ``(None, None)`` when the run isn't found,
    ``(None, error_message)`` on a validation failure, else ``(finalized_entries, None)``.
    """
    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        get_safe_artifact_name,
    )
    from products.tasks.backend.presentation.serializers import (  # noqa: PLC0415 — pure size helpers, kept off the api import path
        build_task_run_artifact_size_error,
        get_task_run_artifact_max_size_bytes,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None, None

    manifest = list(run.artifacts or [])
    artifact_prefix = f"{run.get_artifact_s3_prefix()}/"
    finalized_entries: list[dict] = []
    new_storage_paths: list[str] = []

    for artifact in artifacts:
        artifact_id = artifact["id"]
        storage_path = artifact["storage_path"]

        if not storage_path.startswith(artifact_prefix) or f"/{artifact_id[:8]}_" not in storage_path:
            return None, "Artifact storage path is invalid for this run"

        existing_entry = _find_artifact_manifest_entry(manifest, artifact_id, storage_path)
        if existing_entry is not None:
            finalized_entries.append(existing_entry)
            continue

        s3_object = object_storage.head_object(storage_path)
        if not s3_object:
            return None, "Artifact upload not found in object storage"

        safe_name = get_safe_artifact_name(artifact["name"])
        content_type = artifact.get("content_type") or s3_object.get("ContentType") or ""
        content_length = s3_object.get("ContentLength")
        if not isinstance(content_length, int):
            return None, "Artifact upload metadata is unavailable"

        max_size_bytes = get_task_run_artifact_max_size_bytes(safe_name, content_type, artifact.get("type"))
        if content_length > max_size_bytes:
            return None, build_task_run_artifact_size_error(safe_name, max_size_bytes)

        entry = _build_artifact_manifest_entry(
            artifact_id=artifact_id,
            name=safe_name,
            artifact_type=artifact["type"],
            source=artifact.get("source") or "",
            size=content_length,
            content_type=content_type,
            storage_path=storage_path,
            uploaded_at=django_timezone.now().isoformat(),
            metadata=artifact.get("metadata"),
        )
        manifest.append(entry)
        finalized_entries.append(entry)
        new_storage_paths.append(storage_path)

    _save_artifact_manifest(run, manifest)
    for storage_path in new_storage_paths:
        _tag_artifact_object(run, storage_path)

    # Mint a fresh presigned download URL per response entry so the caller (e.g. the
    # upload_artifact tool) can surface a link to the file. Presigned URLs expire, so
    # they are attached to the response only and never written back to the manifest.
    response_entries: list[dict] = []
    for entry in finalized_entries:
        presigned_url = object_storage.get_presigned_url(entry["storage_path"])
        response_entries.append({**entry, "url": presigned_url} if presigned_url else dict(entry))

    return response_entries, None


def list_task_run_living_artifacts(run_id: str | UUID, task_id: str | UUID, team_id: int) -> list[dict] | None:
    from products.tasks.backend.logic.services.living_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        get_task_artifacts_for_run,
        serialize_task_artifact,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    return [serialize_task_artifact(artifact) for artifact in get_task_artifacts_for_run(run)]


def get_task_run_living_artifact(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, artifact_id: str | UUID
) -> dict | None:
    from products.tasks.backend.logic.services.living_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        get_task_artifact_for_run,
        open_task_artifact,
        serialize_task_artifact,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    artifact = get_task_artifact_for_run(run, artifact_id)
    if artifact is None:
        return None
    serialized = serialize_task_artifact(artifact)
    serialized["content"] = open_task_artifact(artifact)
    return serialized


def create_task_run_living_artifact(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    artifact: dict,
) -> tuple[dict | None, str | None]:
    from products.tasks.backend.logic.services.living_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        create_living_artifact,
        serialize_task_artifact,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None, None
    try:
        created = create_living_artifact(run=run, **artifact)
    except Exception as exc:
        logger.warning("Failed to create living artifact for task run %s: %s", run.id, exc)
        return None, str(exc)
    return serialize_task_artifact(created), None


def edit_task_run_living_artifact(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    artifact_id: str | UUID,
    content: str | None = None,
    content_bytes: bytes | None = None,
    content_type: str | None = None,
    source_artifact_id: str | None = None,
    source_storage_path: str | None = None,
    name: str | None = None,
    metadata: dict | None = None,
) -> tuple[dict | None, str | None]:
    from products.tasks.backend.logic.services.living_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        edit_living_artifact,
        get_task_artifact_for_run,
        serialize_task_artifact,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None, None
    artifact = get_task_artifact_for_run(run, artifact_id)
    if artifact is None:
        return None, "not_found"
    try:
        updated = edit_living_artifact(
            artifact=artifact,
            run=run,
            content=content,
            content_bytes=content_bytes,
            content_type=content_type,
            source_artifact_id=source_artifact_id,
            source_storage_path=source_storage_path,
            name=name,
            metadata=metadata,
        )
    except Exception as exc:
        logger.warning("Failed to edit living artifact %s for task run %s: %s", artifact_id, run.id, exc)
        return None, str(exc)
    return serialize_task_artifact(updated), None


def presign_task_run_artifact(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, storage_path: str
) -> tuple[str | None, str | None]:
    """Presign a download URL for an artifact on the run.

    Returns ``(url, error)``: ``(None, None)`` if the run isn't found, ``(None, "not_found")`` if
    the artifact isn't on the run, ``(None, "unavailable")`` if presigning fails, else ``(url, None)``.
    """
    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None, None

    artifacts = run.artifacts or []
    if not any(artifact.get("storage_path") == storage_path for artifact in artifacts):
        return None, "not_found"

    url = object_storage.get_presigned_url(storage_path)
    if not url:
        return None, "unavailable"
    return url, None


def read_task_run_artifact(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, storage_path: str
) -> tuple[bytes | None, dict | None, str | None]:
    """Read artifact bytes for download, walking the resume chain.

    Returns ``(content, artifact_entry, error)``. ``error`` is one of ``None`` (run not found),
    ``"not_found"`` (artifact not on the run/chain), ``"read_failed"`` (storage read raised), or
    ``"content_missing"`` (object absent). On success returns ``(content, artifact_entry, None)``.
    """
    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None, None, None

    artifact = run.find_artifact_in_resume_chain(storage_path)
    if artifact is None:
        return None, None, "not_found"

    try:
        content = object_storage.read_bytes(storage_path, missing_ok=True)
    except Exception:
        logger.exception(
            "task_run.artifact_download_failed",
            extra={"task_run_id": str(run.id), "storage_path": storage_path},
        )
        return None, artifact, "read_failed"

    if content is None:
        return None, artifact, "content_missing"
    return content, artifact, None


def read_task_run_logs(run_id: str | UUID, task_id: str | UUID, team_id: int) -> str | None:
    """Concatenated JSONL logs across the run's resume chain (oldest ancestor first)."""
    from insights.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None

    resume_chain = run.get_resume_chain()
    chunks: Iterable[str]
    if len(resume_chain) == 1:
        chunks = [object_storage.read(resume_chain[0].log_url, missing_ok=True) or ""]
    else:
        chunks = _TASK_LOG_READ_EXECUTOR.map(
            lambda ancestor: object_storage.read(ancestor.log_url, missing_ok=True) or "", resume_chain
        )

    parts: list[str] = []
    for chunk in chunks:
        if chunk:
            if not chunk.endswith("\n"):
                chunk = chunk + "\n"
            parts.append(chunk)
    return "".join(parts)


def create_task_run_connection_token(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, user_id: int, distinct_id: str
) -> str | None:
    """Mint a sandbox connection token for a run. ``None`` if the run isn't found."""
    from products.tasks.backend.logic.services.connection_token import (  # noqa: PLC0415 — keep sandbox deps off the api import path
        create_sandbox_connection_token as _create,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    return _create(task_run=run, user_id=user_id, distinct_id=distinct_id)


def create_task_run_stream_read_token(run_id: str | UUID, task_id: str | UUID, team_id: int) -> str | None:
    """Mint a run-scoped token for reading a run's live event stream. ``None`` if the run isn't found."""
    from products.tasks.backend.logic.services.connection_token import (  # noqa: PLC0415 — keep sandbox deps off the api import path
        create_stream_read_token as _create,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    return _create(task_run=run)


def task_uses_pi_runtime(task_id: str | UUID, team_id: int) -> bool:
    return Task.objects.filter(id=task_id, team_id=team_id, runtime=Task.Runtime.PI).exists()


def resolve_stream_base_url(*, distinct_id: str, organization_id: str | UUID, force_proxy: bool = False) -> str | None:
    """Agent-proxy base URL for the read leg, or ``None`` to read from Django directly.

    Returns the configured agent-proxy URL only when it is set for this environment AND the
    read-via-proxy flag is enabled for the user, so rollout stays gradual and reversible. The
    server owns this decision; clients just connect to whatever URL comes back.
    """
    from products.tasks.backend.constants import STREAM_VIA_PROXY_FEATURE_FLAG  # noqa: PLC0415

    proxy_url = settings.TASKS_AGENT_PROXY_PUBLIC_URL
    if not proxy_url:
        return None
    # Local dev disables the analytics SDK, so the rollout flag never evaluates; the URL setting
    # is the opt-in there. Prod (DEBUG off) still gates on the flag below.
    if settings.DEBUG or force_proxy:
        return proxy_url
    try:
        enabled = bool(
            hanzo_insights.feature_enabled(
                STREAM_VIA_PROXY_FEATURE_FLAG,
                distinct_id=distinct_id,
                groups={"organization": str(organization_id)},
                group_properties={"organization": {"id": str(organization_id)}},
                only_evaluate_locally=False,
                send_feature_flag_events=False,
            )
        )
    except Exception:
        return None
    return proxy_url if enabled else None


# --- Task run commands (user_message signal + sandbox proxy) ---


def validate_task_run_artifact_ids(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, artifact_ids: list[str]
) -> tuple[list[str], bool]:
    """Resolve artifact ids for a run. Returns ``(missing_ids, found)``; ``found=False`` if the run isn't found."""
    from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        get_task_run_artifacts_by_id,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return [], False
    _, missing_artifact_ids = get_task_run_artifacts_by_id(run, artifact_ids)
    return missing_artifact_ids, True


def signal_task_run_user_message(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    content: str | None,
    artifact_ids: list[str],
    actor_user_id: int | None = None,
    message_id: str | None = None,
    actor_slack_user_id: str | None = None,
    steer: bool = False,
) -> bool | None:
    """Queue a user_message follow-up signal on the run's workflow.

    Returns ``True`` on success, ``False`` when the target workflow is gone
    (completed or evicted — a terminal outcome), ``None`` when the run isn't
    found. Transient signalling failures propagate so a calling Temporal
    activity retries rather than reporting a dead end to the user.
    """
    from temporalio.service import RPCError, RPCStatusCode  # noqa: PLC0415 — keep temporalio off the api import path

    from products.tasks.backend.temporal.client import (  # noqa: PLC0415 — keep temporalio off the api import path
        signal_task_followup_message,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None
    try:
        context = {"actor_slack_user_id": actor_slack_user_id} if actor_slack_user_id else None
        signal_task_followup_message(
            run.workflow_id,
            content,
            artifact_ids,
            message_id,
            actor_user_id,
            context,
            steer=steer,
        )
    except RPCError as e:
        if e.status == RPCStatusCode.NOT_FOUND:
            logger.warning("Follow-up signal target workflow gone for task run %s", run.id)
            return False
        raise
    record_task_run_user_activity(run.id, team_id)
    return True


def record_task_run_user_activity(run_id: str | UUID, team_id: int) -> None:
    """Stamp a user message against the run's open sandbox usage sessions.

    Best-effort (the ledger swallows its own failures): records last-activity on
    every message and starts the user-attributable billing window on the first one.
    Usage ledger only — no workflow side effects.
    """
    from products.tasks.backend.logic.services.sandbox_usage import (  # noqa: PLC0415 — keep sandbox deps off the api import path
        record_task_run_user_activity as _record_user_activity,
    )

    _record_user_activity(run_id, team_id)


def get_task_run_sandbox_connection(
    run_id: str | UUID, task_id: str | UUID, team_id: int, *, user_id: int, distinct_id: str
) -> contracts.TaskRunSandboxConnectionDTO | None:
    """Sandbox connection details for proxying a command. ``None`` if the run isn't found.

    ``sandbox_url`` is ``None`` when the run has no active sandbox (no connection token is minted
    in that case).
    """
    from products.tasks.backend.logic.services.connection_token import (  # noqa: PLC0415 — keep sandbox deps off the api import path
        create_sandbox_connection_token as _create,
    )
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        parse_run_state,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return None

    run_state = parse_run_state(run.state)
    if not run_state.sandbox_url:
        return contracts.TaskRunSandboxConnectionDTO(sandbox_url=None, sandbox_connect_token=None)

    connection_token = _create(task_run=run, user_id=user_id, distinct_id=distinct_id)
    return contracts.TaskRunSandboxConnectionDTO(
        sandbox_url=run_state.sandbox_url,
        sandbox_connect_token=run_state.sandbox_connect_token,
        connection_token=connection_token,
    )


# Relay control verbs whose outcome Insights AI funnels track. Captured here (gated on
# origin_product) so the generic relay stays product-agnostic while the conversation layer stops
# firing them as the renderer drives permission/cancel through `runs/{run}/command/`.
_INSIGHTS_AI_RELAY_TELEMETRY_METHODS: frozenset[str] = frozenset({"cancel", "permission_response"})


def capture_relay_command_telemetry(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    method: str,
    params: dict | None,
    success: bool,
) -> None:
    """Emit Insights AI control-verb telemetry for a relayed agent command.

    Preserves the ``task_run_cancelled`` / ``permission_responded`` funnels once the renderer moves
    permission/cancel onto the generic relay. ``conversation_id`` is intentionally null (the relay
    has no conversation); ``TaskRun.capture_event`` stamps ``origin_product``/``run_id`` so
    generic-task usage stays out of the Insights AI funnels. Mirrors the old conversation-layer
    semantics: a cancel is recorded only when it actually reached the agent, while a permission
    response is recorded with its forward ``success`` either way.
    """
    if method not in _INSIGHTS_AI_RELAY_TELEMETRY_METHODS:
        return
    run = _get_visible_run(run_id, task_id, team_id)
    if run is None or run.task.origin_product != Task.OriginProduct.INSIGHTS_AI:
        return

    params = params or {}
    if method == "cancel":
        if not success:
            return
        run.capture_event(
            "task_run_cancelled",
            {
                "execution_type": "sandbox",
                "surface": "relay",
                "conversation_id": None,
                "cancel_source": "user",
            },
        )
        return

    run.capture_event(
        "permission_responded",
        {
            "execution_type": "sandbox",
            "surface": "relay",
            "conversation_id": None,
            "request_id": params.get("requestId"),
            "option_id": params.get("optionId"),
            "success": success,
        },
    )


# --- Task run relay (Slack) ---


def _pick_relay_text(*, text: str, text_parts: list[str] | None) -> str:
    """Pick the text to post. If ``text_parts`` has any non-empty entries,
    the last one wins (that's the post-last-tool-use answer). Otherwise fall
    back to the joined ``text`` field."""
    if text_parts:
        for part in reversed(text_parts):
            if isinstance(part, str) and part.strip():
                return part
    return text


def relay_task_run_message(
    run_id: str | UUID,
    task_id: str | UUID,
    team_id: int,
    *,
    text: str,
    text_parts: list[str] | None = None,
    message_id: str | None = None,
) -> tuple[str, str | None]:
    """Queue a Slack relay workflow for a run message, or under the agent-design
    flag signal the running task workflow to stream the text inline.

    Returns ``(status, relay_id)`` where status is ``"accepted"`` (relay_id set), ``"skipped"``
    (run not found / terminal / no Slack mapping / empty text / streamed inline under the
    agent-design flag), or ``"failed"``.

    When ``text_parts`` is provided the last non-empty entry is used — it's the
    post-last-tool-use answer, and posting only that keeps the interim narration
    ("Let me check…") out of the Slack thread. Older callers still send just
    ``text`` and get the previous behavior unchanged.
    """
    from products.slack_app.backend.models import (  # noqa: PLC0415 — cross-product import kept off the api import path
        SlackThreadTaskMapping,
    )
    from products.tasks.backend.temporal.client import (  # noqa: PLC0415 — keep temporalio off the api import path
        execute_insights_code_agent_relay_workflow,
        signal_agent_text_delta,
    )
    from products.tasks.backend.temporal.process_task.activities.feature_flags import (  # noqa: PLC0415 — keep temporal off the api import path
        AGENT_DESIGN_STATE_KEY,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None or run.is_terminal:
        return "skipped", None
    if not SlackThreadTaskMapping.objects.filter(task_run=run).exists():
        return "skipped", None

    posted_text = _pick_relay_text(text=text, text_parts=text_parts)
    trimmed = posted_text.strip()
    if not trimmed:
        return "skipped", None

    if bool((run.state or {}).get(AGENT_DESIGN_STATE_KEY)):
        try:
            signal_agent_text_delta(run.workflow_id, trimmed)
        except Exception:
            logger.exception("task_run_relay_text_signal_failed", extra={"run_id": str(run.id)})
        return "skipped", None

    try:
        relay_id = execute_insights_code_agent_relay_workflow(
            run_id=str(run.id),
            text=trimmed,
            delete_progress=True,
            message_id=message_id,
        )
    except Exception:
        logger.exception("task_run_relay_message_enqueue_failed", extra={"run_id": str(run.id)})
        return "failed", None
    return "accepted", relay_id


# --- Task run creation / start / cloud resume ---


def user_can_author_repository(user_id: int, repository: str) -> bool:
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        get_user_github_integration,
        user_github_integration_is_usable,
    )

    user = User.objects.filter(id=user_id).first()
    integration = get_user_github_integration(user, repository=repository, allow_refresh=False)
    return user_github_integration_is_usable(integration)


def _ensure_task_team_github_integration(task: Task) -> bool:
    if task.github_integration_id is not None:
        return True
    github_integration = Integration.objects.filter(team_id=task.team_id, kind="github").first()
    if github_integration is None:
        return False
    task.github_integration = github_integration
    task.save(update_fields=["github_integration", "updated_at"])
    return True


def _resolve_cloud_pr_authorship_mode(
    task: Task,
    *,
    pr_authorship_mode,
    request_user_id: int | None,
    github_user_token: str | None,
):
    """Resolve the effective PR-authorship mode for a cloud run.

    Returns ``(mode, error)``: ``error`` is a ``TaskRunValidationError`` when authorship can't be
    established (mode is then ``None``); otherwise ``error`` is ``None`` and ``mode`` is the
    resolved value. Mirrors the original view helper exactly.
    """
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        PrAuthorshipMode,
        resolve_user_github_integration_for_task,
        user_github_integration_is_usable,
    )

    if pr_authorship_mode != PrAuthorshipMode.USER or github_user_token:
        return pr_authorship_mode, None

    if task.created_by_id != request_user_id:
        return None, contracts.TaskRunValidationError(
            kind="validation_error",
            code="github_authorization_required",
            detail="User-authored runs must be started by the task creator, or provide github_user_token.",
            attr="pr_authorship_mode",
        )

    user_github_integration = resolve_user_github_integration_for_task(task, allow_refresh=False)
    if user_github_integration is not None and user_github_integration_is_usable(user_github_integration):
        if task.github_user_integration_id != user_github_integration.integration.id:
            task.github_user_integration = user_github_integration.integration
            task.save(update_fields=["github_user_integration", "updated_at"])
        return PrAuthorshipMode.USER, None

    if _ensure_task_team_github_integration(task):
        return PrAuthorshipMode.BOT, None

    return None, contracts.TaskRunValidationError(
        kind="validation_error",
        code="github_authorization_required",
        detail="Link a GitHub account with repo access before running user-authored cloud tasks.",
        attr="pr_authorship_mode",
    )


def _github_credential_source_extra_state(pr_authorship_mode, github_user_token: str | None) -> dict[str, str]:
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        GitHubCredentialSource,
        PrAuthorshipMode,
    )

    if pr_authorship_mode != PrAuthorshipMode.USER:
        return {}
    source = GitHubCredentialSource.CALLER_TOKEN if github_user_token else GitHubCredentialSource.SERVER_INTEGRATION
    return {"github_credential_source": source.value}


def bootstrap_task_run(
    task_id: str | UUID, team_id: int, user_id: int | None, *, validated_data: dict
) -> contracts.TaskRunCreateResult | None:
    """Create a task run (without starting execution) from validated bootstrap data.

    Returns ``None`` if the task isn't found/visible (the view raises 404). Otherwise returns a
    ``TaskRunCreateResult`` carrying either the created run DTO or a structured validation error.
    Mirrors ``TaskRunViewSet.create`` byte-for-byte (minus the usage gate, which the view applies).
    """
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        PrAuthorshipMode,
        RunSource,
        cache_github_user_token,
        get_provider_for_runtime_adapter,
        get_reasoning_effort_error,
    )

    task = _get_task_for_run_control(task_id, team_id, user_id)
    if task is None:
        return None

    mode = validated_data.get("mode", "background")
    environment = validated_data.get("environment", TaskRun.Environment.LOCAL)
    branch = validated_data.get("branch")
    sandbox_environment_id = validated_data.get("sandbox_environment_id")
    pr_authorship_mode = validated_data.get("pr_authorship_mode")
    auto_publish = validated_data.get("auto_publish")
    run_source = validated_data.get("run_source")
    signal_report_id = validated_data.get("signal_report_id")
    runtime_adapter = validated_data.get("runtime_adapter")
    model = validated_data.get("model")
    reasoning_effort = validated_data.get("reasoning_effort")
    context_window = validated_data.get("context_window")
    fast_mode = validated_data.get("fast_mode")
    github_user_token = validated_data.get("github_user_token")
    initial_permission_mode = validated_data.get("initial_permission_mode")
    imported_mcp_servers = validated_data.get("imported_mcp_servers")
    relayed_mcp_servers = validated_data.get("relayed_mcp_servers")
    if run_source == RunSource.SIGNAL_REPORT:
        pr_authorship_mode = PrAuthorshipMode.BOT

    extra_state: dict | None = None
    if initial_permission_mode is not None:
        extra_state = {"initial_permission_mode": initial_permission_mode}

    provider = get_provider_for_runtime_adapter(runtime_adapter)
    for key, value in {
        "pr_base_branch": branch,
        "pr_authorship_mode": pr_authorship_mode,
        "auto_publish": auto_publish,
        "run_source": run_source,
        "signal_report_id": signal_report_id,
        "runtime_adapter": runtime_adapter,
        "provider": provider,
        "model": model,
        "reasoning_effort": reasoning_effort,
        "context_window": context_window,
        "fast_mode": fast_mode,
        "rtk_enabled": validated_data.get("rtk_enabled"),
    }.items():
        if value is not None:
            extra_state = extra_state or {}
            extra_state[key] = value.value if hasattr(value, "value") else value

    reasoning_effort_error = get_reasoning_effort_error(
        runtime_adapter=runtime_adapter, model=model, reasoning_effort=reasoning_effort
    )
    if reasoning_effort_error is not None:
        return contracts.TaskRunCreateResult(
            error=contracts.TaskRunValidationError(
                kind="validation_error", code="invalid_input", detail=reasoning_effort_error, attr="reasoning_effort"
            )
        )

    pr_authorship_mode, validation_error = _resolve_cloud_pr_authorship_mode(
        task,
        pr_authorship_mode=pr_authorship_mode,
        request_user_id=user_id,
        github_user_token=github_user_token,
    )
    if validation_error is not None:
        return contracts.TaskRunCreateResult(error=validation_error)
    if pr_authorship_mode is not None:
        extra_state = extra_state or {}
        extra_state["pr_authorship_mode"] = (
            pr_authorship_mode.value if hasattr(pr_authorship_mode, "value") else pr_authorship_mode
        )

    if credential_source := _github_credential_source_extra_state(pr_authorship_mode, github_user_token):
        extra_state = extra_state or {}
        extra_state.update(credential_source)

    custom_image_id = validated_data.get("custom_image_id")
    if custom_image_id is not None:
        custom_image = SandboxCustomImage.get_accessible_for_task(
            image_id=custom_image_id, team_id=task.team_id, task_created_by_id=task.created_by_id
        )
        if custom_image is None:
            return contracts.TaskRunCreateResult(
                error=contracts.TaskRunValidationError(kind="detail", detail="Invalid custom_image_id")
            )
        if not custom_image.is_ready:
            return contracts.TaskRunCreateResult(
                error=contracts.TaskRunValidationError(
                    kind="detail", detail=f"Custom image is not ready (status: {custom_image.status})"
                )
            )
        extra_state = extra_state or {}
        extra_state["custom_image_id"] = str(custom_image.id)

    if sandbox_environment_id is not None:
        sandbox_environment = SandboxEnvironment.get_accessible_for_task(
            environment_id=sandbox_environment_id,
            team_id=task.team_id,
            task_created_by_id=task.created_by_id,
        )
        if sandbox_environment is None:
            return contracts.TaskRunCreateResult(
                error=contracts.TaskRunValidationError(kind="detail", detail="Invalid sandbox_environment_id")
            )
        extra_state = extra_state or {}
        extra_state["sandbox_environment_id"] = str(sandbox_environment.id)
        logger.info(
            "Applying sandbox environment to task run",
            extra={
                "task_id": str(task.id),
                "sandbox_environment_id": str(sandbox_environment.id),
                "sandbox_environment_name": sandbox_environment.name,
                "network_access_level": sandbox_environment.network_access_level,
            },
        )

    logger.info(
        "Creating task run for task %s with mode=%s, branch=%s, environment=%s", task.id, mode, branch, environment
    )
    run = task.create_run(environment=environment, mode=mode, branch=branch, extra_state=extra_state)

    if imported_mcp_servers or relayed_mcp_servers:
        update_fields = ["updated_at"]
        if imported_mcp_servers:
            # Kept out of `state` (a plain JSONField) because header values carry credentials.
            run.imported_mcp_servers = imported_mcp_servers
            update_fields.append("imported_mcp_servers")
        if relayed_mcp_servers:
            run.relayed_mcp_servers = relayed_mcp_servers
            update_fields.append("relayed_mcp_servers")
        run.save(update_fields=update_fields)

    if github_user_token and pr_authorship_mode == PrAuthorshipMode.USER:
        cache_github_user_token(str(run.id), github_user_token)

    return contracts.TaskRunCreateResult(run=_task_run_detail_to_dto(_task_run_queryset().get(pk=run.pk)))


def _trigger_task_processing_workflow(
    task: Task,
    run: TaskRun,
    user_id: int | None,
    *,
    initial_message: str | None = None,
    initial_artifact_ids: list[str] | None = None,
    raise_on_error: bool = False,
) -> None:
    from products.tasks.backend.temporal.client import (  # noqa: PLC0415 — keep temporalio off the api import path
        execute_task_processing_workflow,
    )
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        RunSource,
        parse_run_state,
    )
    from products.tasks.backend.temporal.process_task.workflow import PendingFollowup  # noqa: PLC0415

    # SIGNAL_REPORT: implementation runs log their work on the report (notes, code references)
    # via the task:write artefact tools.
    full_mcp_run_sources = frozenset({None, RunSource.MANUAL, RunSource.SIGNAL_REPORT})
    run_source = parse_run_state(run.state).run_source
    insights_mcp_scopes: Literal["read_only", "full"] = "full" if run_source in full_mcp_run_sources else "read_only"
    try:
        logger.info("Attempting to trigger task processing workflow for task %s, run %s", task.id, run.id)
        if initial_message or initial_artifact_ids:
            execute_task_processing_workflow(
                task_id=str(task.id),
                run_id=str(run.id),
                team_id=task.team.id,
                user_id=user_id,
                insights_mcp_scopes=insights_mcp_scopes,
                initial_message=PendingFollowup(
                    message=initial_message,
                    artifact_ids=initial_artifact_ids or [],
                    actor_user_id=user_id,
                    message_id=str(uuid4()),
                ),
            )
        else:
            execute_task_processing_workflow(
                task_id=str(task.id),
                run_id=str(run.id),
                team_id=task.team.id,
                user_id=user_id,
                insights_mcp_scopes=insights_mcp_scopes,
            )
        logger.info("Workflow trigger completed for task %s, run %s", task.id, run.id)
    except Exception as e:
        logger.exception("Failed to trigger task processing workflow for task %s, run %s: %s", task.id, run.id, e)
        if raise_on_error:
            raise


# Statuses from which a cloud run may be started via the start endpoint.
_STARTABLE_TASK_RUN_STATUSES = (TaskRun.Status.NOT_STARTED, TaskRun.Status.QUEUED)


def check_task_run_startable(run_id: str | UUID, task_id: str | UUID, team_id: int) -> str:
    """Whether a run can be started via the start endpoint.

    Returns ``"not_found"`` (run missing), ``"not_cloud"``, ``"bad_status:<current>"``, or
    ``"ok"``. The usage gate (429) is applied by the view between this check and ``start_task_run``.
    """
    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return "not_found"
    if run.environment != TaskRun.Environment.CLOUD:
        return "not_cloud"
    if run.status not in _STARTABLE_TASK_RUN_STATUSES:
        return f"bad_status:{run.status}"
    return "ok"


def start_task_run(
    run_id: str | UUID, task_id: str | UUID, team_id: int, user_id: int | None, *, validated_data: dict
) -> tuple[str, UUID | None]:
    """Apply run-scoped attachments and trigger the cloud workflow for a startable run.

    Caller must have already verified startability and applied the usage gate. Returns
    ``(outcome, task_id)``: ``"not_found"``, ``"missing_artifacts:<csv>"``, or ``"started"``
    (``task_id`` set). Rolls back any pending-state writes on failure (re-raising), mirroring
    the original view.
    """
    from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        get_task_run_artifacts_by_id,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return "not_found", None
    task = run.task

    pending_user_message = validated_data.get("pending_user_message")
    pending_user_artifact_ids = validated_data.get("pending_user_artifact_ids") or []

    if pending_user_artifact_ids:
        _, missing_artifact_ids = get_task_run_artifacts_by_id(run, pending_user_artifact_ids)
        if missing_artifact_ids:
            return "missing_artifacts:" + ",".join(missing_artifact_ids), None

    state_updates: dict = {}
    if task.runtime != Task.Runtime.PI:
        if pending_user_message is not None:
            state_updates["pending_user_message"] = pending_user_message
        if pending_user_artifact_ids:
            state_updates["pending_user_artifact_ids"] = pending_user_artifact_ids

    previous_state = dict(run.state or {})
    try:
        if state_updates:
            TaskRun.update_state_atomic(run.id, updates=state_updates)
            run.refresh_from_db()
        logger.info("Triggering workflow for task %s, existing run %s", task.id, run.id)
        _trigger_task_processing_workflow(
            task,
            run,
            user_id,
            initial_message=(pending_user_message or task.description or None)
            if task.runtime == Task.Runtime.PI
            else None,
            initial_artifact_ids=pending_user_artifact_ids if task.runtime == Task.Runtime.PI else None,
            raise_on_error=True,
        )
    except Exception:
        if state_updates:
            rollback_updates = {
                key: previous_state[key] for key in state_updates.keys() if key in previous_state
            } or None
            rollback_remove_keys = [key for key in state_updates.keys() if key not in previous_state] or None
            TaskRun.update_state_atomic(run.id, updates=rollback_updates, remove_keys=rollback_remove_keys)
        raise

    return "started", task.id


def resume_task_run_in_cloud(
    run_id: str | UUID, task_id: str | UUID, team_id: int, user_id: int | None
) -> tuple[str, contracts.TaskRunDetailDTO | None, str | None]:
    """Resume a run in a cloud sandbox, terminating any prior workflow.

    Returns ``(outcome, run_dto, debug_use_modal)``. ``outcome`` is one of: ``"not_found"``,
    ``"already_active"`` (400), ``"auth_error:<detail>"`` (400, github auth), ``"workflow_failed"``
    (502), or ``"resumed"`` (run_dto set). Mirrors ``TaskRunViewSet.resume_in_cloud``.
    """
    from products.tasks.backend.facade.streams import reset_task_run_stream  # noqa: PLC0415
    from products.tasks.backend.redis import run_uses_dedicated_stream  # noqa: PLC0415
    from products.tasks.backend.temporal.client import (  # noqa: PLC0415 — keep temporalio off the api import path
        resume_task_in_cloud_workflow,
    )
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        PrAuthorshipMode,
        get_pr_authorship_mode,
    )

    run = _get_visible_run(run_id, task_id, team_id)
    if run is None:
        return "not_found", None, None
    logger.info(
        "resume_in_cloud_called",
        extra={
            "task_run_id": str(run.id),
            "task_id": str(run.task_id),
            "prior_status": run.status,
            "prior_environment": run.environment,
            "prior_state_keys": sorted((run.state or {}).keys()),
            "prior_snapshot_external_id": (run.state or {}).get("snapshot_external_id"),
            "prior_snapshot_kind": (run.state or {}).get("snapshot_kind"),
            "prior_snapshot_mount_path": (run.state or {}).get("snapshot_mount_path"),
        },
    )

    with transaction.atomic():
        run = (
            TaskRun.objects.select_for_update(of=("self",))
            .select_related("task", "task__created_by", "task__github_integration", "task__github_user_integration")
            .get(pk=run.pk)
        )

        is_cloud_active = run.environment == TaskRun.Environment.CLOUD and run.status in (
            TaskRun.Status.QUEUED,
            TaskRun.Status.IN_PROGRESS,
        )
        if is_cloud_active:
            return "already_active", None, None

        if get_pr_authorship_mode(run.task, run.state) == PrAuthorshipMode.USER:
            pr_authorship_mode, validation_error = _resolve_cloud_pr_authorship_mode(
                run.task,
                pr_authorship_mode=PrAuthorshipMode.USER,
                request_user_id=user_id,
                github_user_token=None,
            )
            if validation_error is not None:
                return f"auth_error:{validation_error.detail}", None, None
            if pr_authorship_mode is not None:
                run.state = {
                    **(run.state or {}),
                    "pr_authorship_mode": (
                        pr_authorship_mode.value if hasattr(pr_authorship_mode, "value") else pr_authorship_mode
                    ),
                }

        prior_status = run.status
        prior_environment = run.environment
        prior_completed_at = run.completed_at
        prior_state = dict(run.state or {})
        run.prepare_for_cloud_handoff()

    logger.info("Resuming task run in cloud", extra={"task_run_id": str(run.id), "task_id": str(run.task_id)})

    try:
        if not reset_task_run_stream(
            str(run.id),
            use_dedicated=run_uses_dedicated_stream(run.state),
        ):
            raise RuntimeError("Failed to reset task run event stream")
        resume_task_in_cloud_workflow(str(run.id), run.workflow_id)
    except Exception as e:
        logger.exception("Failed to trigger handoff workflow", extra={"task_run_id": str(run.id), "error": str(e)})
        with transaction.atomic():
            run = TaskRun.objects.select_for_update().get(pk=run.pk)
            run.status = prior_status
            run.environment = prior_environment
            run.completed_at = prior_completed_at
            run.state = prior_state
            run.error_message = "Failed to start cloud workflow"
            run.save(update_fields=["status", "environment", "completed_at", "state", "error_message", "updated_at"])
        run.publish_stream_state_event()
        return "workflow_failed", None, None

    return "resumed", _task_run_detail_to_dto(run), None


# --- Task presentation CRUD + actions ---
# These back the thin ``TaskViewSet``. They mirror the original viewset's querysets
# (team scoping, ``task_visibility_q`` visibility, ordering, filters, annotations) and
# orchestration (title generation, signal-report linkage, workflow triggers, S3 artifact
# staging, presence beacons) byte-for-byte.


def signal_report_queryset():
    """The ``SignalReport`` manager queryset, for the task write serializer's report FK field.

    Kept here so presentation never imports the ``signals`` product's models directly; team
    scoping on the selected report is enforced by the serializer's ``validate_signal_report``.
    """
    from products.signals.backend.models import (  # noqa: PLC0415 — cross-product import kept off the api import path
        SignalReport,
    )

    return SignalReport.objects.all()


def channel_queryset():
    """Live ``Channel`` queryset for the task write serializer's channel FK field.

    Kept here so presentation never imports tasks models directly. Deliberately
    ``unscoped()``: the serializer is also instantiated without team context (e.g.
    drf-spectacular schema generation), where the fail-closed manager would raise.
    Team scoping comes from the serializer's team-scoped field, ownership of
    personal channels from ``validate_channel``.
    """
    return Channel.objects.unscoped().filter(deleted=False)


def is_internal_debug_team(team_id: int | None) -> bool:
    """Whether the team is the Insights-internal debugging team. Mirrors the original view helper."""
    from django.conf import settings  # noqa: PLC0415

    if settings.DEBUG and not settings.TEST:
        return team_id == 1
    return team_id == 2 and settings.CLOUD_DEPLOYMENT == "US"


def _task_detail_queryset():
    return Task.objects.select_related(
        "created_by", "team", "github_integration", "github_user_integration"
    ).prefetch_related("runs")


def _visible_task_qs(team_id: int, user_id: int | None, *, bypass_visibility: bool = False, for_control: bool = False):
    """Team-scoped live tasks, gated by read visibility — or by the narrower
    control predicate when ``for_control`` (mutations, runs, agent commands)."""
    qs = Task.objects.filter(team_id=team_id, deleted=False)
    if not bypass_visibility:
        qs = qs.filter(task_control_q(user_id) if for_control else task_visibility_q(user_id))
    return qs


def get_task_detail(
    task_id: str | UUID, team_id: int, user_id: int | None, *, bypass_visibility: bool = False
) -> contracts.TaskDetailDTO | None:
    """A single task as a detail DTO, team-scoped and visibility-gated.

    ``bypass_visibility`` mirrors the ``?ph_debug=true`` retrieve path for internal-debug teams.
    """
    task = (
        _visible_task_qs(team_id, user_id, bypass_visibility=bypass_visibility)
        .select_related("created_by", "team", "github_integration", "github_user_integration")
        .prefetch_related("runs")
        .filter(id=task_id)
        .first()
    )
    return _task_detail_to_dto(task) if task is not None else None


def get_conversation_task_dtos(task_ids: Sequence[str | UUID], team_id: int) -> dict[UUID, contracts.TaskDetailDTO]:
    """Task payloads for the Max conversation API, keyed by task id.

    Intentionally team-scoped, with no ``task_visibility_q(user_id)`` gate: the conversation is the
    read-share unit, so anyone who can ``retrieve`` a conversation may read its backing task. Broad
    read here does not grant action — write/send to a task stays creator-gated in the conversation
    viewset (``create``/``open``/``queue``) and at bind time (``validate_task_id``). Direct task reads
    (``get_task_detail``) gate by creator because that is the task-enumeration path; this one is not.

    ``latest_run`` (the nested run payload) stays excluded so conversation lists never presign per-row
    log URLs; instead a single ``latest_run_id`` subquery carries the latest run id, which the frontend
    needs to reconnect to sandbox logs.
    """
    if not task_ids:
        return {}

    latest_run_id_sq = (
        TaskRun.objects.filter(task=OuterRef("pk"), team_id=team_id).order_by("-created_at", "-id").values("id")[:1]
    )
    tasks = (
        Task.objects.filter(team_id=team_id, id__in=task_ids)
        .select_related("created_by", "team")
        .annotate(_latest_run_id=Subquery(latest_run_id_sq))
    )
    return {task.id: _task_detail_to_dto(task, include_latest_run=False) for task in tasks}


def pi_cloud_runtime_enabled(team: Team, user: User) -> bool:
    distinct_id = user.distinct_id or f"user_{user.id}"
    organization_id = str(team.organization_id)
    try:
        return bool(
            hanzo_insights.feature_enabled(
                PI_CLOUD_RUNTIME_FEATURE_FLAG,
                distinct_id,
                groups={"organization": organization_id},
                group_properties={"organization": {"id": organization_id}},
                only_evaluate_locally=False,
                send_feature_flag_events=False,
            )
        )
    except Exception:
        logger.exception("pi-harness flag check failed; treating as disabled")
        return False


def task_runtime(task_id: str | UUID, team_id: int, user_id: int | None, *, for_control: bool = False) -> str | None:
    return (
        _visible_task_qs(team_id, user_id, for_control=for_control)
        .filter(id=task_id)
        .values_list("runtime", flat=True)
        .first()
    )


def task_visible(task_id: str | UUID, team_id: int, user_id: int | None, *, for_control: bool = False) -> bool:
    """Whether a non-deleted task exists for the team and is visible to the user.

    Mirrors the existence gate ``TaskViewSet.get_object()`` applied (team + ``deleted=False`` +
    ``task_visibility_q``). Used by the ``run`` action to 404 before the usage gate, preserving
    the original ordering; the ``run`` action passes ``for_control`` since starting a run drives
    the task.
    """
    return _visible_task_qs(team_id, user_id, for_control=for_control).filter(id=task_id).exists()


def list_pinned_task_ids(team_id: int, user_id: int) -> list[UUID]:
    visible_tasks = _visible_task_qs(team_id, user_id).values("id")
    return list(
        TaskPin.objects.filter(user_id=user_id, task_id__in=Subquery(visible_tasks))
        .order_by("-pinned_at")
        .values_list("task_id", flat=True)
    )


def set_task_pinned(task_id: str | UUID, team_id: int, user_id: int, *, pinned: bool) -> bool | None:
    if not task_visible(task_id, team_id, user_id):
        return None
    if pinned:
        TaskPin.objects.get_or_create(user_id=user_id, task_id=task_id)
    else:
        TaskPin.objects.filter(user_id=user_id, task_id=task_id).delete()
    return pinned


async def select_repository_for_message(team_id: int, user_id: int, message: str, *, origin_product: str) -> str | None:
    """Pick the repository a free-form chat message is most likely about.

    Kept as a lazy facade wrapper so API importers do not load the repo-selection agent or
    sandbox/Temporal dependencies on their request import path.
    """
    from products.tasks.backend.logic.repo_selection.cascade import (  # noqa: PLC0415 — keeps repo-selection agent imports lazy
        select_repository_for_message as select_repository_for_message_impl,
    )

    return await select_repository_for_message_impl(
        team_id, user_id, message, origin_product=Task.OriginProduct(origin_product)
    )


def _list_tasks_queryset(
    team_id: int, user_id: int | None, *, filters: dict, bypass_visibility: bool = False
) -> QuerySet[Task]:
    latest_run = TaskRun.objects.filter(task=OuterRef("pk"), team_id=team_id).order_by("-created_at", "-id")
    qs = _visible_task_qs(team_id, user_id, bypass_visibility=bypass_visibility).order_by("-created_at", "-id")

    origin_product = filters.get("origin_product")
    if origin_product:
        qs = qs.filter(origin_product=origin_product)

    stage = filters.get("stage")
    if stage:
        stage_run = TaskRun.objects.filter(task=OuterRef("pk"), team_id=team_id, stage=stage)
        qs = qs.filter(Exists(stage_run))

    organization = filters.get("organization")
    repository = filters.get("repository")
    created_by = filters.get("created_by")
    search = filters.get("search")
    status_filter = filters.get("status")

    if repository:
        repo_str = repository.strip().lower()
        if "/" in repo_str:
            qs = qs.filter(repository__iexact=repo_str)
        else:
            qs = qs.filter(repository__iendswith=f"/{repo_str}")

    if organization:
        org_str = organization.strip().lower()
        qs = qs.filter(repository__istartswith=f"{org_str}/")

    if created_by:
        qs = qs.filter(created_by_id=created_by)

    channel = filters.get("channel")
    if channel:
        qs = qs.filter(channel_id=channel)

    if search:
        search_term = search.strip()
        if search_term:
            search_q = Q(title__icontains=search_term) | Q(description__icontains=search_term)
            number_part = search_term.split("-")[-1].strip()
            if number_part.isdigit():
                search_q |= Q(task_number=int(number_part))
            qs = qs.filter(search_q)

    if status_filter:
        latest_run_status = latest_run.values("status")[:1]
        qs = qs.annotate(_latest_run_status=Subquery(latest_run_status)).filter(_latest_run_status=status_filter)

    # `internal` controls default visibility, not access — task visibility (applied above) is the real
    # authorization boundary, open to any team member. `all` returns both, `true` returns only-internal,
    # and the default excludes internal tasks so the main task list stays clean.
    internal_param = filters.get("internal")
    if internal_param == "all":
        pass
    elif internal_param == "true":
        qs = qs.filter(internal=True)
    else:
        qs = qs.filter(internal=False)

    archived_param = filters.get("archived")
    if archived_param == "true":
        qs = qs.filter(archived=True)
    elif archived_param == "all":
        pass
    else:
        qs = qs.filter(archived=False)

    qs = qs.select_related("created_by", "team", "github_integration", "github_user_integration").annotate(
        _latest_run_id=Subquery(latest_run.values("id")[:1])
    )

    return qs


def _latest_runs_by_id(run_ids: Iterable[UUID], team_id: int) -> dict[UUID, TaskRun]:
    unique_run_ids = list(dict.fromkeys(run_ids))
    if not unique_run_ids:
        return {}

    return {run.id: run for run in TaskRun.objects.filter(id__in=unique_run_ids, team_id=team_id)}


def _tasks_to_dtos(tasks: Iterable[Task], team_id: int) -> list[contracts.TaskDetailDTO]:
    task_list = list(tasks)
    latest_run_ids_by_task_id = {
        task.id: latest_run_id
        for task in task_list
        if (latest_run_id := getattr(task, "_latest_run_id", None)) is not None
    }
    latest_runs_by_id = _latest_runs_by_id(latest_run_ids_by_task_id.values(), team_id)

    dtos = []
    for task in task_list:
        latest_run_id = latest_run_ids_by_task_id.get(task.id)
        latest_run = latest_runs_by_id.get(latest_run_id) if latest_run_id is not None else None
        dtos.append(_task_detail_to_dto(task, latest_run=latest_run))
    return dtos


def list_tasks(team_id: int, user_id: int | None, *, filters: dict) -> list[contracts.TaskDetailDTO]:
    """All visible tasks for the team as DTOs, mirroring the task list view filters."""
    return _tasks_to_dtos(_list_tasks_queryset(team_id, user_id, filters=filters), team_id)


def inaccessible_repositories_via_integration(team_id: int, integration_id: int, repositories: list[str]) -> list[str]:
    return _inaccessible_repositories_via_integration(team_id, integration_id, repositories)


def list_task_repositories(team_id: int, user_id: int | None) -> list[str]:
    """Distinct repositories used by non-deleted, non-internal visible tasks for the team."""
    tasks = Task.objects.filter(team_id=team_id, deleted=False, internal=False).filter(task_visibility_q(user_id))
    plural = (
        tasks.exclude(repositories=[])
        .annotate(repository_name=Func(F("repositories"), function="unnest", output_field=CharField()))
        .values_list("repository_name", flat=True)
        .distinct()
    )
    legacy = (
        tasks.filter(repositories=[])
        .exclude(repository__isnull=True)
        .exclude(repository="")
        .values_list("repository", flat=True)
        .distinct()
    )
    return sorted(set(plural) | {repository for repository in legacy if repository})


def get_task_summaries(team_id: int, user_id: int | None, *, ids: list) -> list[contracts.TaskSummaryDTO]:
    """Summary fields for the requested tasks, mirroring ``TaskViewSet.summaries``."""
    from django.db.models.functions import JSONObject  # noqa: PLC0415

    latest_run = (
        TaskRun.objects.filter(task=OuterRef("pk"), team_id=team_id)
        .order_by("-created_at", "-id")
        .annotate(_data=JSONObject(status="status", environment="environment"))
    )
    tasks = (
        Task.objects.filter(team_id=team_id, deleted=False, id__in=ids)
        .filter(task_visibility_q(user_id))
        .annotate(_latest_run=Subquery(latest_run.values("_data")[:1]))
        .order_by("-created_at", "id")
    )
    summaries: list[contracts.TaskSummaryDTO] = []
    for task in tasks:
        raw = getattr(task, "_latest_run", None)
        latest = (
            contracts.TaskLatestRunSummaryDTO(status=raw.get("status"), environment=raw.get("environment"))
            if isinstance(raw, dict)
            else None
        )
        summaries.append(
            contracts.TaskSummaryDTO(
                id=task.id,
                title=task.title,
                repository=task.repository,
                created_at=task.created_at,
                updated_at=task.updated_at,
                origin_product=task.origin_product,
                latest_run=latest,
            )
        )
    return summaries


def compute_repository_readiness(team_id: int, *, repository: str, window_days: int, refresh: bool) -> dict:
    """Autonomy-readiness details for a repository. Thin wrapper over the internal computation."""
    from insights.models import Team  # noqa: PLC0415

    from products.tasks.backend.repository_readiness import (  # noqa: PLC0415 — keep readiness deps off the api import path
        compute_repository_readiness as _compute,
    )

    team = Team.objects.get(id=team_id)
    return _compute(team=team, repository=repository, window_days=window_days, refresh=refresh)


def create_task(
    team_id: int,
    user_id: int | None,
    *,
    validated_data: dict,
    client_provenance: TaskClientProvenance | None = None,
) -> contracts.TaskDetailDTO:
    """Create a task, mirroring ``TaskSerializer.create`` byte-for-byte.

    Absorbs the cross-product ``SignalReportTask`` linkage, ``generate_task_title``, and
    ``resolve_user_github_integration_for_task`` so no internal/other-product import leaks into
    presentation. ``validated_data`` carries the validated write fields (integrations already
    resolved to instances by the write serializer's PK fields).
    """
    from insights.models import Team  # noqa: PLC0415

    from products.signals.backend.task_run_artefacts import (  # noqa: PLC0415 — cross-product write kept off the api import path
        record_report_task,
    )
    from products.tasks.backend.logic.services.title_generator import generate_task_title  # noqa: PLC0415
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        resolve_user_github_integration_for_task,
    )

    team = Team.objects.get(id=team_id)
    validated_data = dict(validated_data)
    validated_data["team"] = team
    validated_data.setdefault("origin_product", Task.OriginProduct.USER_CREATED)
    validated_data["client_provenance"] = client_provenance
    warm_branch_provided = "branch" in validated_data
    warm_branch = validated_data.pop("branch", None)
    warm_runtime_adapter = validated_data.pop("runtime_adapter", None)
    warm_model = validated_data.pop("model", None)
    warm_reasoning_effort = validated_data.pop("reasoning_effort", None)
    warm_sandbox_environment_id = validated_data.pop("sandbox_environment_id", None)
    warm_custom_image_id = validated_data.pop("custom_image_id", None)
    pending_user_message = (validated_data.pop("pending_user_message", None) or "").strip() or None
    pending_user_artifact_ids = validated_data.pop("pending_user_artifact_ids", None) or []
    warm_auto_publish = validated_data.pop("auto_publish", None)
    channel = validated_data.get("channel")
    if channel is not None and "repositories" not in validated_data and "repository" not in validated_data:
        validated_data["repositories"] = channel.repositories
        validated_data["github_integration"] = channel.github_integration
    if "repositories" in validated_data:
        repositories = validated_data["repositories"]
        validated_data["repository"] = repositories[0] if repositories else None
    elif "repository" in validated_data:
        validated_data["repositories"] = [validated_data["repository"]] if validated_data["repository"] else []

    if user_id is not None:
        validated_data["created_by"] = User.objects.get(id=user_id)

    if (
        warm_branch_provided
        and validated_data["origin_product"] == Task.OriginProduct.USER_CREATED
        and validated_data.get("repository")
        and len(validated_data.get("repositories", [])) <= 1
        and user_id is not None
    ):
        warm_run = _find_idling_warm_run(
            team_id,
            user_id,
            repository=validated_data["repository"],
            branch=warm_branch,
            runtime_adapter=warm_runtime_adapter,
            model=warm_model,
            reasoning_effort=warm_reasoning_effort,
            sandbox_environment_id=warm_sandbox_environment_id,
            custom_image_id=warm_custom_image_id,
        )
        if warm_run is not None and not _warm_sandbox_selection_is_accessible(
            team_id=team_id,
            task_created_by_id=user_id,
            sandbox_environment_id=warm_sandbox_environment_id,
            custom_image_id=warm_custom_image_id,
        ):
            warm_run = None
        if warm_run is not None and pending_user_artifact_ids:
            from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
                get_task_run_artifacts_by_id,
            )

            _, missing_artifact_ids = get_task_run_artifacts_by_id(warm_run, pending_user_artifact_ids)
            if missing_artifact_ids:
                logger.info(
                    "Skipping warm run reuse: %d pending artifact id(s) missing from warm run %s manifest",
                    len(missing_artifact_ids),
                    warm_run.id,
                )
                warm_run = None
        if warm_run is not None:
            warm_task = warm_run.task
            description = (validated_data.get("description") or "").strip()
            update_fields: list[str] = []
            if description and not (warm_task.title or "").strip():
                warm_task.title = generate_task_title(description)
                warm_task.title_manually_set = False
                update_fields += ["title", "title_manually_set"]
            if description and not (warm_task.description or "").strip():
                warm_task.description = description
                update_fields.append("description")
            channel = validated_data.get("channel")
            if channel is not None and warm_task.channel_id != channel.id:
                warm_task.channel = channel
                update_fields.append("channel")
            if update_fields:
                warm_task.save(update_fields=[*update_fields, "updated_at"])
            if warm_task.client_provenance is None and client_provenance is not None:
                Task.objects.filter(id=warm_task.id, client_provenance__isnull=True).update(
                    client_provenance=client_provenance
                )
                warm_task.client_provenance = client_provenance
            _activate_warm_run(
                warm_run,
                warm_task,
                team_id,
                message=pending_user_message or description or None,
                description=description or None,
                artifact_ids=pending_user_artifact_ids,
                auto_publish=warm_auto_publish,
            )
            return _task_detail_to_dto(_task_detail_queryset().get(pk=warm_task.pk))

    # The relationship the client asserted (validated by the serializer, which rejects `research`).
    # Popped so it isn't forwarded to the model; the link itself is recorded by record_report_task below.
    signal_report_task_relationship = validated_data.pop("signal_report_task_relationship", None)

    if not validated_data.get("github_integration"):
        default_integration = Integration.objects.filter(team=team, kind="github").first()
        if default_integration:
            validated_data["github_integration"] = default_integration

    # Inbox "Create PR" / "Discuss" don't pre-select a repo, so resolve one here rather than
    # creating a report-linked task that can never open a PR.
    signal_report = validated_data.get("signal_report")
    if (
        signal_report is not None
        and not validated_data.get("repository")
        and validated_data.get("origin_product") == Task.OriginProduct.SIGNAL_REPORT
    ):
        from products.signals.backend.facade.api import (  # noqa: PLC0415 — cross-product read kept off the api import path
            persisted_repo_selection,
        )
        from products.tasks.backend.logic.repo_selection.cascade import (  # noqa: PLC0415 — keeps repo-selection agent imports lazy
            cascade_select_repository,
        )

        # The report's own selection is authoritative — including a scout's deliberate no-repo
        # (`repository=None`), which must not fall through to the cascade.
        selection = persisted_repo_selection(str(signal_report.id))
        resolved_repository = (
            selection.repository
            if selection is not None
            else cascade_select_repository(
                team_id,
                user_id,
                validated_data.get("description") or "",
                team=team,
                single_repo_wins=True,
                allow_refresh=False,
            )
        )
        if resolved_repository:
            validated_data["repository"] = resolved_repository

    if (
        validated_data.get("repository")
        and validated_data.get("origin_product", Task.OriginProduct.USER_CREATED) == Task.OriginProduct.USER_CREATED
        and not validated_data.get("github_user_integration")
    ):
        task_stub = Task(
            team=team,
            created_by=validated_data.get("created_by"),
            origin_product=Task.OriginProduct.USER_CREATED,
            repository=validated_data["repository"],
            github_integration=validated_data.get("github_integration"),
        )
        github_user_integration = resolve_user_github_integration_for_task(task_stub, allow_refresh=False)
        if github_user_integration is not None:
            validated_data["github_user_integration"] = github_user_integration.integration

    title = (validated_data.get("title") or "").strip()
    if not title and validated_data.get("description"):
        validated_data["title"] = generate_task_title(validated_data["description"])
        validated_data.setdefault("title_manually_set", False)
    elif title:
        validated_data.setdefault("title_manually_set", True)

    # The write serializer binds the report as `signal_report` (a PK field); direct callers may
    # pass `signal_report_id`. Either way this is the manual "start work from a report" path.
    # Gated regardless of the relationship label: the label is client-selected and manually
    # created tasks run PR-capable by default, so a "discussion" label must not dodge the limit.
    report_ref = validated_data.get("signal_report") or validated_data.get("signal_report_id")
    if report_ref and validated_data.get("origin_product") == Task.OriginProduct.SIGNAL_REPORT:
        enforce_self_driving_pr_quota(team, report_id=str(getattr(report_ref, "id", report_ref)))

    logger.info("Creating task with data: %s", validated_data)
    with transaction.atomic():
        task = Task.objects.create(**validated_data)
        if task.signal_report_id and task.origin_product == Task.OriginProduct.SIGNAL_REPORT:
            # Record the task↔report association + work-log artefact for the asserted relationship
            # (defaults to implementation, which also writes the auto-start spend gate row) so a
            # manually-started task matches autostarted ones.
            record_report_task(
                team_id=task.team_id,
                report_id=str(task.signal_report_id),
                task_id=str(task.id),
                relationship=signal_report_task_relationship,
            )

    return _task_detail_to_dto(_task_detail_queryset().get(pk=task.pk))


def set_task_title(task_id: str | UUID, team_id: int, title: str) -> bool:
    """Set a task's title, team-scoped. For automated relabels — e.g. backfilling a Signals research
    task with ``"Research: <report title>"`` once research produces the title. Leaves
    ``title_manually_set`` untouched (this isn't a user edit) and clamps to the column length. Returns
    whether a row was updated.
    """
    return bool(Task.objects.filter(id=task_id, team_id=team_id).update(title=title[:255]))


def update_task(
    task_id: str | UUID, team_id: int, user_id: int | None, *, validated_data: dict
) -> contracts.TaskDetailDTO | None:
    """Update a task, mirroring ``TaskSerializer.update``. ``None`` if not found/controllable."""
    task = _visible_task_qs(team_id, user_id, for_control=True).filter(id=task_id).first()
    if task is None:
        return None

    validated_data = dict(validated_data)
    # Immutable after creation; origin_product controls visibility, signal_report is set-once.
    validated_data.pop("signal_report", None)
    validated_data.pop("signal_report_task_relationship", None)
    validated_data.pop("origin_product", None)
    validated_data.pop("branch", None)
    if "repositories" in validated_data:
        repositories = validated_data["repositories"]
        validated_data["repository"] = repositories[0] if repositories else None
    elif "repository" in validated_data:
        if len(task.repositories) <= 1:
            validated_data["repositories"] = [validated_data["repository"]] if validated_data["repository"] else []
        else:
            validated_data.pop("repository")
    if "title" in validated_data and "title_manually_set" not in validated_data:
        validated_data["title_manually_set"] = True
    if "archived" in validated_data and validated_data["archived"] != task.archived:
        validated_data["archived_at"] = django_timezone.now() if validated_data["archived"] else None

    logger.info("perform_update called for task %s with validated_data: %s", task.id, validated_data)
    for key, value in validated_data.items():
        setattr(task, key, value)
    task.save()
    logger.info("Task %s updated successfully", task.id)

    return _task_detail_to_dto(_task_detail_queryset().get(pk=task.pk))


def soft_delete_task(task_id: str | UUID, team_id: int, user_id: int | None) -> bool:
    """Soft-delete a task. Returns whether a task was found/controllable and deleted."""
    task = _visible_task_qs(team_id, user_id, for_control=True).filter(id=task_id).first()
    if task is None:
        return False
    logger.info("Soft deleting task %s", task.id)
    task.soft_delete()
    return True


# --- Task staged artifacts (S3 + cache, attached to the next run) ---


_TASK_STAGED_ARTIFACT_UPLOAD_FORM_OVERHEAD_BYTES = 64 * 1024


def prepare_task_staged_artifacts(
    task_id: str | UUID,
    team_id: int,
    user_id: int | None,
    *,
    artifacts: list[dict],
    upload_expiration_seconds: int,
) -> contracts.StagedArtifactPrepareResult | None:
    """Reserve S3 keys + presigned POST forms for task attachments. ``None`` if task not found."""
    from insights.storage import object_storage  # noqa: PLC0415

    from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415
        build_task_staged_artifact_storage_path,
        get_safe_artifact_name,
    )

    # Staged artifacts feed the task's next run, so this is control, not viewing.
    task = _visible_task_qs(team_id, user_id, for_control=True).filter(id=task_id).first()
    if task is None:
        return None

    prepared: list[contracts.StagedArtifactPreparedDTO] = []
    for artifact in artifacts:
        artifact_id = uuid4().hex
        safe_name = get_safe_artifact_name(artifact["name"])
        storage_path = build_task_staged_artifact_storage_path(task, artifact_id, safe_name)
        presigned_post = object_storage.get_presigned_post(
            storage_path,
            conditions=[
                ["content-length-range", 0, artifact["size"] + _TASK_STAGED_ARTIFACT_UPLOAD_FORM_OVERHEAD_BYTES]
            ],
            expiration=upload_expiration_seconds,
        )
        if not presigned_post:
            return contracts.StagedArtifactPrepareResult(error="Unable to generate upload URL")

        prepared.append(
            contracts.StagedArtifactPreparedDTO(
                id=artifact_id,
                name=safe_name,
                type=artifact["type"],
                source=artifact.get("source") or "",
                size=artifact["size"],
                content_type=artifact.get("content_type") or "",
                storage_path=storage_path,
                expires_in=upload_expiration_seconds,
                presigned_post=presigned_post,
                metadata=artifact.get("metadata"),
            )
        )

    return contracts.StagedArtifactPrepareResult(artifacts=prepared)


def finalize_task_staged_artifacts(
    task_id: str | UUID,
    team_id: int,
    user_id: int | None,
    *,
    artifacts: list[dict],
) -> contracts.StagedArtifactFinalizeResult | None:
    """Verify staged S3 uploads and cache their metadata. ``None`` if task not found."""
    from django.conf import settings  # noqa: PLC0415

    from insights.storage import object_storage  # noqa: PLC0415

    from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415
        STAGED_ARTIFACT_TTL_DAYS,
        build_task_artifact_entry,
        cache_task_staged_artifact,
        get_safe_artifact_name,
        tag_task_artifact,
    )
    from products.tasks.backend.presentation.serializers import (  # noqa: PLC0415
        build_task_run_artifact_size_error,
        get_task_run_artifact_max_size_bytes,
    )

    task = _visible_task_qs(team_id, user_id, for_control=True).filter(id=task_id).first()
    if task is None:
        return None

    artifact_prefix = f"{settings.OBJECT_STORAGE_TASKS_FOLDER}/artifacts/team_{task.team_id}/task_{task.id}/staged/"
    finalized: list[dict] = []
    for artifact in artifacts:
        artifact_id = artifact["id"]
        storage_path = artifact["storage_path"]
        if not storage_path.startswith(artifact_prefix) or f"/{artifact_id}/" not in storage_path:
            return contracts.StagedArtifactFinalizeResult(error="Artifact storage path is invalid for this task")

        s3_object = object_storage.head_object(storage_path)
        if not s3_object:
            return contracts.StagedArtifactFinalizeResult(error="Artifact upload not found in object storage")

        content_length = s3_object.get("ContentLength")
        if not isinstance(content_length, int):
            return contracts.StagedArtifactFinalizeResult(error="Artifact upload metadata is unavailable")

        safe_name = get_safe_artifact_name(artifact["name"])
        content_type = artifact.get("content_type") or s3_object.get("ContentType") or ""
        max_size_bytes = get_task_run_artifact_max_size_bytes(safe_name, content_type, artifact.get("type"))
        if content_length > max_size_bytes:
            return contracts.StagedArtifactFinalizeResult(
                error=build_task_run_artifact_size_error(safe_name, max_size_bytes)
            )

        finalized.append(
            build_task_artifact_entry(
                artifact_id=artifact_id,
                name=safe_name,
                artifact_type=artifact["type"],
                source=artifact.get("source") or "",
                size=content_length,
                content_type=content_type,
                storage_path=storage_path,
                metadata=artifact.get("metadata"),
            )
        )

    for entry in finalized:
        cache_task_staged_artifact(task, entry)
        tag_task_artifact(entry["storage_path"], ttl_days=STAGED_ARTIFACT_TTL_DAYS, team_id=task.team_id)

    return contracts.StagedArtifactFinalizeResult(artifacts=finalized)


def resolve_team_github_integration_id(team_id: int, github_integration_id: int) -> int | None:
    """Return the integration id only if it is a GitHub integration owned by this team.

    Re-scoping guard for the collection-level warm endpoint, which accepts a bare PK with
    no serializer team context. Returns ``None`` for any id that doesn't belong to the team —
    the caller treats that as "skip warming" (the submit later falls through to a cold create+run).
    """
    exists = Integration.objects.filter(id=github_integration_id, team_id=team_id, kind="github").exists()
    return github_integration_id if exists else None


def can_mint_readonly_github_token(team_id: int) -> bool:
    """Whether a repo-less sandbox requesting `github_read_access` would actually get a token.

    Preflight for callers that condition user-visible behavior (e.g. prompt guidance naming `gh`)
    on the read-only token being obtainable — true only when the team has a usable team-level
    GitHub installation. Never raises.
    """
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keeps the temporal stack off the facade import path
        can_mint_readonly_github_token as _can_mint,
    )

    return _can_mint(team_id)


def _find_idling_warm_run(
    team_id: int,
    user_id: int | None,
    *,
    repository: str | None,
    branch: str | None,
    runtime_adapter: str | None = None,
    model: str | None = None,
    reasoning_effort: str | None = None,
    sandbox_environment_id: str | UUID | None = None,
    custom_image_id: str | UUID | None = None,
) -> TaskRun | None:
    """Most-recent idling pre-warmed Run matching this user's cloud composing selection, or ``None``.

    A warm Run is a non-terminal ``USER_CREATED`` Run for the same repo+branch still awaiting its
    first user message (the ``await_user_message`` state marker). This is the backend's single source
    of truth for the warm pool: it dedupes warm provisioning (so a repeated ``warm`` call reuses the
    live Run instead of spawning a second) and lets the normal create+run path transparently reuse a
    warm Run on submit. Team + user scoped; branch compared as ``None``-normalized exact match.

    Reuse also requires the warm Run's runtime, sandbox environment, and custom image selections to
    match the request. A mismatch returns ``None`` so the caller cold-creates on the correct sandbox.
    The repo/branch/``await_user_message`` predicates stay in the query; the remaining selection is
    matched in Python over the small candidate set.
    """
    if user_id is None or not repository:
        return None
    candidates = (
        TaskRun.objects.filter(  # nosemgrep: idor-lookup-without-team — team_id filter applied via the task FK below
            task__team_id=team_id,
            task__created_by_id=user_id,
            task__origin_product=Task.OriginProduct.USER_CREATED,
            task__repository__iexact=repository,
            task__deleted=False,
            state__await_user_message=True,
            branch=branch or None,
        )
        .exclude(status__in=_TERMINAL_TASK_RUN_STATUSES)
        .select_related("task")
        .order_by("-created_at")[:20]
    )
    wanted = (
        runtime_adapter or None,
        model or None,
        reasoning_effort or None,
        str(sandbox_environment_id) if sandbox_environment_id else None,
        str(custom_image_id) if custom_image_id else None,
    )
    for run in candidates:
        state = run.state or {}
        have = (
            state.get("runtime_adapter") or None,
            state.get("model") or None,
            state.get("reasoning_effort") or None,
            state.get("sandbox_environment_id") or None,
            state.get("custom_image_id") or None,
        )
        if have == wanted:
            return run
    return None


def _warm_sandbox_selection_is_accessible(
    *,
    team_id: int,
    task_created_by_id: int | None,
    sandbox_environment_id: str | UUID | None,
    custom_image_id: str | UUID | None,
) -> bool:
    if sandbox_environment_id is not None:
        sandbox_environment = SandboxEnvironment.get_accessible_for_task(
            environment_id=sandbox_environment_id,
            team_id=team_id,
            task_created_by_id=task_created_by_id,
        )
        if sandbox_environment is None:
            return False

    if custom_image_id is not None:
        custom_image = SandboxCustomImage.get_accessible_for_task(
            image_id=custom_image_id,
            team_id=team_id,
            task_created_by_id=task_created_by_id,
        )
        if custom_image is None or not custom_image.is_ready:
            return False

    return True


def _idling_warm_run_for_task(task: Task) -> TaskRun | None:
    """The task's latest run iff it is an idling pre-warmed Run (non-terminal, awaiting first message)."""
    run = task.latest_run
    if run is None or run.is_terminal:
        return None
    if not (run.state or {}).get("await_user_message"):
        return None
    return run


def _attach_staged_artifacts_to_run(
    run: TaskRun, task: Task, *, staged_artifacts: list[dict], artifact_ids: list[str]
) -> None:
    from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415 — keep storage deps off the api import path
        RUN_ARTIFACT_TTL_DAYS,
        build_task_staged_artifact_cache_key,
        tag_task_artifact,
    )
    from products.tasks.backend.redis import get_tasks_cache  # noqa: PLC0415

    manifest = list(run.artifacts or [])
    for staged_artifact in staged_artifacts:
        storage_path = str(staged_artifact["storage_path"])
        if _find_artifact_manifest_entry(manifest, str(staged_artifact.get("id")), storage_path):
            continue
        tag_task_artifact(storage_path, ttl_days=RUN_ARTIFACT_TTL_DAYS, team_id=task.team_id)
        manifest.append(dict(staged_artifact))
    _save_artifact_manifest(run, manifest)
    get_tasks_cache().delete_many(
        [build_task_staged_artifact_cache_key(str(task.id), artifact_id) for artifact_id in artifact_ids]
    )


def _activate_warm_run(
    run: TaskRun,
    task: Task,
    team_id: int,
    *,
    message: str | None,
    artifact_ids: list[str],
    description: str | None = None,
    auto_publish: bool | None = None,
) -> None:
    """Activate an idling warm Run: set the draft Task's visible description from raw task text,
    forward the first message to the already-running agent, and drop the ``await_user_message`` marker
    so the Run leaves the warm pool. Mirrors ``message_routing._handle_first_message``; no fresh agent
    start.

    ``auto_publish`` is persisted into the Run's state before the message signal: the already-running
    agent-server can't take it as a launch flag, so it re-reads run state when the forwarded first
    message arrives (and resumes read it from carried state)."""
    from products.tasks.backend.metrics import (  # noqa: PLC0415 — keep prometheus deps off the api import path
        observe_prewarmed_activated,
    )

    if description and not (task.description or "").strip():
        task.description = description
        task.save(update_fields=["description", "updated_at"])
    if auto_publish is not None:
        # Before the signal: the agent-server re-reads run state when the forwarded
        # first message arrives, so the choice must already be persisted by then.
        TaskRun.update_state_atomic(run.id, updates={"auto_publish": auto_publish})
    signal_task_run_user_message(run.id, task.id, team_id, content=message, artifact_ids=artifact_ids)
    TaskRun.update_state_atomic(run.id, remove_keys=["await_user_message"])
    # Only count activations of Runs that actually carry the prewarmed marker, so the activation
    # numerator stays consistent with the workflow_start{prewarmed="true"} denominator — otherwise
    # warm Runs provisioned before this ships (await_user_message set, prewarmed absent) would push
    # the hit rate above 1 during the deploy transition.
    if (run.state or {}).get("prewarmed"):
        observe_prewarmed_activated(run)


def warm_task_sandbox(
    team_id: int,
    user_id: int,
    *,
    repository: str,
    github_integration_id: int,
    branch: str | None,
    runtime_adapter: str | None = None,
    model: str | None = None,
    reasoning_effort: str | None = None,
    sandbox_environment_id: str | UUID | None = None,
    custom_image_id: str | UUID | None = None,
    client_provenance: TaskClientProvenance | None = None,
) -> contracts.WarmTaskDTO | None:
    """Warm a full idling Run for a Code-app cloud task while the user composes.

    Births a draft Task (``USER_CREATED``), then ``SandboxWarmer.warm()`` provisions an interactive
    Run that boots + clones + checks out ``branch`` + starts the agent on the selected
    ``runtime_adapter``/``model``/``reasoning_effort`` (carried on the Run state and read by the
    agent-server at launch, so the sandbox boots on the right runtime), then idles awaiting the first
    ``user_message``. The Run is dispatched with ``create_pr=True`` so that, once activated on submit,
    it completes autonomously and opens a PR like a normal Code-app cloud task.

    Best-effort: returns ``None`` (not an HTTP error) when warming is gated — over quota
    (``QuotaLimitExceeded``), product not enabled (``PermissionDenied``), or the warm pool is full
    (``Throttled``). The caller treats ``None`` as "no warm run; fall through to a cold create+run".

    ``github_integration_id`` must already be re-scoped to ``team_id`` by the caller
    (see :func:`resolve_team_github_integration_id`).
    """
    from rest_framework.exceptions import (  # noqa: PLC0415 — keep DRF exception types off the api import path
        PermissionDenied,
        Throttled,
    )

    from insights.exceptions import QuotaLimitExceeded  # noqa: PLC0415 — keep billing deps off the api import path
    from insights.models import Team  # noqa: PLC0415

    from products.tasks.backend.logic.services.warm import (
        SandboxWarmer,  # noqa: PLC0415 — keep warming deps off the api import path
    )
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        RuntimeAdapter,
        get_provider_for_runtime_adapter,
    )

    team = Team.objects.get(id=team_id)
    github_integration = Integration.objects.filter(id=github_integration_id, team_id=team_id, kind="github").first()
    if github_integration is None:
        return None

    sandbox_environment = None
    if sandbox_environment_id is not None:
        sandbox_environment = SandboxEnvironment.get_accessible_for_task(
            environment_id=sandbox_environment_id,
            team_id=team_id,
            task_created_by_id=user_id,
        )
        if sandbox_environment is None:
            return None

    custom_image = None
    if custom_image_id is not None:
        custom_image = SandboxCustomImage.get_accessible_for_task(
            image_id=custom_image_id,
            team_id=team_id,
            task_created_by_id=user_id,
        )
        if custom_image is None or not custom_image.is_ready:
            return None

    existing = _find_idling_warm_run(
        team_id,
        user_id,
        repository=repository,
        branch=branch,
        runtime_adapter=runtime_adapter,
        model=model,
        reasoning_effort=reasoning_effort,
        sandbox_environment_id=sandbox_environment_id,
        custom_image_id=custom_image_id,
    )
    if existing is not None:
        return contracts.WarmTaskDTO(task_id=existing.task_id, run_id=existing.id)

    task = Task.create_without_run(
        team=team,
        title="",
        description="",
        origin_product=Task.OriginProduct.USER_CREATED,
        user_id=user_id,
        repository=repository,
        client_provenance=client_provenance,
    )
    assert task.created_by is not None  # create_without_run always sets created_by from user_id

    provider = get_provider_for_runtime_adapter(runtime_adapter)
    initial_permission_mode = "auto" if runtime_adapter == RuntimeAdapter.CODEX.value else "default"
    extra_state: dict = {
        "branch": branch,
        "initial_permission_mode": initial_permission_mode,
        "use_modal_network_allowlist": False,
    }
    if sandbox_environment is not None:
        extra_state["sandbox_environment_id"] = str(sandbox_environment.id)
    if custom_image is not None:
        extra_state["custom_image_id"] = str(custom_image.id)
    for key, value in {
        "runtime_adapter": runtime_adapter,
        "provider": provider.value if provider is not None else None,
        "model": model,
        "reasoning_effort": reasoning_effort,
    }.items():
        if value is not None:
            extra_state[key] = value

    try:
        result = SandboxWarmer(task, user=task.created_by).warm(
            mode="interactive",
            extra_state=extra_state,
            create_pr=True,
        )
    except (Throttled, PermissionDenied, QuotaLimitExceeded):
        task.soft_delete()
        return None

    return contracts.WarmTaskDTO(task_id=task.id, run_id=result.run.id)


# --- Task run (the ``run`` action) ---


def run_task(
    task_id: str | UUID, team_id: int, user_id: int | None, *, validated_data: dict
) -> contracts.TaskRunResult | None:
    """Create a run for a task and kick off its workflow, mirroring ``TaskViewSet.run``.

    Returns ``None`` if the task isn't found/visible (the view raises 404). Otherwise a
    ``TaskRunResult`` carrying the refreshed task detail DTO or a structured error. The usage
    gate (429) is applied by the view before calling this.
    """
    from products.tasks.backend.logic.services.staged_artifacts import get_task_staged_artifacts  # noqa: PLC0415
    from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
        PrAuthorshipMode,
        RunSource,
        cache_github_user_token,
        get_provider_for_runtime_adapter,
        get_reasoning_effort_error,
        parse_run_state,
    )

    task = _visible_task_qs(team_id, user_id, for_control=True).filter(id=task_id).first()
    if task is None:
        return None
    mode = validated_data.get("mode", "background")
    branch = validated_data.get("branch")
    resume_from_run_id = validated_data.get("resume_from_run_id")
    pending_user_message = validated_data.get("pending_user_message")
    pending_user_artifact_ids = validated_data.get("pending_user_artifact_ids") or []
    is_pi_task = task.runtime == Task.Runtime.PI

    if not resume_from_run_id:
        warm_run = _idling_warm_run_for_task(task)
        if warm_run is not None and (branch or None) == (warm_run.branch or None):
            warm_state = warm_run.state or {}
            warm_runtime_matches = (
                warm_state.get("runtime_adapter") or None,
                warm_state.get("model") or None,
                warm_state.get("reasoning_effort") or None,
                warm_state.get("context_window") or None,
                warm_state.get("fast_mode") or None,
                warm_state.get("sandbox_environment_id") or None,
                warm_state.get("custom_image_id") or None,
            ) == (
                validated_data.get("runtime_adapter") or None,
                validated_data.get("model") or None,
                validated_data.get("reasoning_effort") or None,
                validated_data.get("context_window") or None,
                validated_data.get("fast_mode") or None,
                str(validated_data["sandbox_environment_id"]) if validated_data.get("sandbox_environment_id") else None,
                str(validated_data["custom_image_id"]) if validated_data.get("custom_image_id") else None,
            )
            if warm_runtime_matches and _warm_sandbox_selection_is_accessible(
                team_id=team_id,
                task_created_by_id=task.created_by_id,
                sandbox_environment_id=warm_state.get("sandbox_environment_id"),
                custom_image_id=warm_state.get("custom_image_id"),
            ):
                warm_staged_artifacts, warm_missing_artifact_ids = (
                    get_task_staged_artifacts(task, pending_user_artifact_ids)
                    if pending_user_artifact_ids
                    else ([], [])
                )
                if not warm_missing_artifact_ids:
                    if warm_staged_artifacts:
                        _attach_staged_artifacts_to_run(
                            warm_run,
                            task,
                            staged_artifacts=warm_staged_artifacts,
                            artifact_ids=pending_user_artifact_ids,
                        )
                    _activate_warm_run(
                        warm_run,
                        task,
                        team_id,
                        message=pending_user_message or (task.description or None),
                        description=task.description or None,
                        artifact_ids=pending_user_artifact_ids,
                        auto_publish=validated_data.get("auto_publish"),
                    )
                    return contracts.TaskRunResult(task=get_task_detail(task.id, team_id, user_id))
    sandbox_environment_id = validated_data.get("sandbox_environment_id")
    sandbox_environment_id_supplied_by_user = sandbox_environment_id is not None
    custom_image_id = validated_data.get("custom_image_id")
    custom_image_id_supplied_by_user = custom_image_id is not None
    pr_authorship_mode = validated_data.get("pr_authorship_mode")
    auto_publish = validated_data.get("auto_publish")
    run_source = validated_data.get("run_source")
    signal_report_id = validated_data.get("signal_report_id")
    runtime_adapter = validated_data.get("runtime_adapter")
    model = validated_data.get("model")
    reasoning_effort = validated_data.get("reasoning_effort")
    context_window = validated_data.get("context_window")
    fast_mode = validated_data.get("fast_mode")
    github_user_token = validated_data.get("github_user_token")
    initial_permission_mode = validated_data.get("initial_permission_mode")
    imported_mcp_servers = validated_data.get("imported_mcp_servers")
    relayed_mcp_servers = validated_data.get("relayed_mcp_servers")
    if run_source == RunSource.SIGNAL_REPORT:
        pr_authorship_mode = PrAuthorshipMode.BOT

    runtime_state_fields = {
        "pr_authorship_mode": pr_authorship_mode,
        "auto_publish": auto_publish,
        "run_source": run_source,
        "signal_report_id": signal_report_id,
        "runtime_adapter": runtime_adapter,
        "model": model,
        "reasoning_effort": reasoning_effort,
        "context_window": context_window,
        "fast_mode": fast_mode,
    }

    extra_state: dict | None = None
    if pending_user_message is not None and not is_pi_task:
        extra_state = {"pending_user_message": pending_user_message}
    if pending_user_artifact_ids and not is_pi_task:
        extra_state = extra_state or {}
        extra_state["pending_user_artifact_ids"] = pending_user_artifact_ids
    if initial_permission_mode is not None:
        extra_state = extra_state or {}
        extra_state["initial_permission_mode"] = initial_permission_mode
    rtk_enabled = validated_data.get("rtk_enabled")
    if rtk_enabled is not None:
        extra_state = extra_state or {}
        extra_state["rtk_enabled"] = rtk_enabled

    if resume_from_run_id:
        previous_run = task.runs.filter(id=resume_from_run_id).first()
        if not previous_run:
            return contracts.TaskRunResult(
                error=contracts.TaskValidationError(kind="detail", detail="Invalid resume_from_run_id")
            )

        prev_state = parse_run_state(previous_run.state)
        extra_state = extra_state or {}
        if not is_pi_task:
            extra_state["resume_from_run_id"] = str(resume_from_run_id)
            extra_state.update(prev_state.resume_snapshot_carry_state())

        # The resumed agent still pushes the head branch baked into the original prompt, so the
        # PR webhook must be able to match this run, not the terminal predecessor.
        prev_wizard_head_branch = (previous_run.state or {}).get("wizard_head_branch")
        if prev_wizard_head_branch:
            extra_state["wizard_head_branch"] = prev_wizard_head_branch

        # Same reasoning for a signals self-driving run: the head branch is baked into the original
        # prompt, so the resumed run pushes it too, and the review carve-out's branch linkage
        # (find_signal_implementation_run) only matches the successor if its stamp is carried
        # forward — otherwise a resume (the usual recovery for a cancelled run) silently ends
        # re-reviews. The key is PATCH-protected, so this server-side copy is the only way it
        # reaches the successor run.
        prev_self_driving_head_branch = (previous_run.state or {}).get("self_driving_head_branch")
        if prev_self_driving_head_branch:
            extra_state["self_driving_head_branch"] = prev_self_driving_head_branch

        # A read-only GitHub grant describes how the task was created, not one run — without the
        # carry-forward, a resumed successor of a repo-less read-only run falls through to the
        # full credential path and regains the write-capable token. (The key is PATCH-protected,
        # so this server-side copy is the only way it reaches a successor run.)
        if (previous_run.state or {}).get("github_read_access") is True:
            extra_state["github_read_access"] = True

        if prev_state.sandbox_environment_id and sandbox_environment_id is None:
            sandbox_environment_id = prev_state.sandbox_environment_id

        if custom_image_id is None:
            custom_image_id = (previous_run.state or {}).get("custom_image_id")

        for field_name in runtime_state_fields:
            if runtime_state_fields[field_name] is None:
                runtime_state_fields[field_name] = getattr(prev_state, field_name)

        pr_authorship_mode = runtime_state_fields["pr_authorship_mode"]
        auto_publish = runtime_state_fields["auto_publish"]
        run_source = runtime_state_fields["run_source"]
        signal_report_id = runtime_state_fields["signal_report_id"]
        runtime_adapter = runtime_state_fields["runtime_adapter"]
        model = runtime_state_fields["model"]
        reasoning_effort = runtime_state_fields["reasoning_effort"]
        context_window = runtime_state_fields["context_window"]
        fast_mode = runtime_state_fields["fast_mode"]
        if branch is None and prev_state.pr_base_branch is not None:
            branch = prev_state.pr_base_branch

    provider = get_provider_for_runtime_adapter(runtime_adapter)

    run_state_values = {
        "pr_base_branch": branch,
        "pr_authorship_mode": pr_authorship_mode,
        "auto_publish": auto_publish,
        "run_source": run_source,
        "signal_report_id": signal_report_id,
        "runtime_adapter": runtime_adapter,
        "provider": provider,
        "model": model,
        "reasoning_effort": reasoning_effort,
        "context_window": context_window,
        "fast_mode": fast_mode,
    }
    if is_pi_task:
        for key in ("runtime_adapter", "provider", "model", "reasoning_effort"):
            run_state_values.pop(key)
    for key, value in run_state_values.items():
        if value is not None:
            extra_state = extra_state or {}
            extra_state[key] = value.value if hasattr(value, "value") else value

    reasoning_effort_error = get_reasoning_effort_error(
        runtime_adapter=runtime_adapter, model=model, reasoning_effort=reasoning_effort
    )
    if reasoning_effort_error is not None:
        return contracts.TaskRunResult(
            error=contracts.TaskValidationError(
                kind="validation_error", code="invalid_input", detail=reasoning_effort_error, attr="reasoning_effort"
            )
        )

    pr_authorship_mode, validation_error = _resolve_cloud_pr_authorship_mode(
        task,
        pr_authorship_mode=pr_authorship_mode,
        request_user_id=user_id,
        github_user_token=github_user_token,
    )
    if validation_error is not None:
        return contracts.TaskRunResult(
            error=contracts.TaskValidationError(
                kind=validation_error.kind,
                detail=validation_error.detail,
                code=validation_error.code,
                attr=validation_error.attr,
            )
        )
    if pr_authorship_mode is not None:
        extra_state = extra_state or {}
        extra_state["pr_authorship_mode"] = (
            pr_authorship_mode.value if hasattr(pr_authorship_mode, "value") else pr_authorship_mode
        )

    if credential_source := _github_credential_source_extra_state(pr_authorship_mode, github_user_token):
        extra_state = extra_state or {}
        extra_state.update(credential_source)

    if custom_image_id is not None:
        custom_image = SandboxCustomImage.get_accessible_for_task(
            image_id=custom_image_id, team_id=task.team_id, task_created_by_id=task.created_by_id
        )
        if custom_image is None:
            if custom_image_id_supplied_by_user:
                return contracts.TaskRunResult(
                    error=contracts.TaskValidationError(kind="detail", detail="Invalid custom_image_id")
                )
        elif not custom_image.is_ready:
            if custom_image_id_supplied_by_user:
                return contracts.TaskRunResult(
                    error=contracts.TaskValidationError(
                        kind="detail", detail=f"Custom image is not ready (status: {custom_image.status})"
                    )
                )
        else:
            extra_state = extra_state or {}
            extra_state["custom_image_id"] = str(custom_image.id)

    if sandbox_environment_id is not None:
        sandbox_environment = SandboxEnvironment.get_accessible_for_task(
            environment_id=sandbox_environment_id,
            team_id=task.team_id,
            task_created_by_id=task.created_by_id,
        )
        if sandbox_environment is None:
            if sandbox_environment_id_supplied_by_user:
                return contracts.TaskRunResult(
                    error=contracts.TaskValidationError(kind="detail", detail="Invalid sandbox_environment_id")
                )
        else:
            extra_state = extra_state or {}
            extra_state["sandbox_environment_id"] = str(sandbox_environment.id)
            logger.info(
                "Applying sandbox environment to task run",
                extra={
                    "task_id": str(task.id),
                    "sandbox_environment_id": str(sandbox_environment.id),
                    "sandbox_environment_name": sandbox_environment.name,
                    "network_access_level": sandbox_environment.network_access_level,
                },
            )

    staged_artifacts: list[dict] = []
    if pending_user_artifact_ids:
        staged_artifacts, missing_artifact_ids = get_task_staged_artifacts(task, pending_user_artifact_ids)
        if missing_artifact_ids:
            return contracts.TaskRunResult(
                error=contracts.TaskValidationError(
                    kind="detail",
                    detail="Some pending_user_artifact_ids are invalid or expired",
                    missing_artifact_ids=missing_artifact_ids,
                )
            )

    logger.info("Creating task run for task %s with mode=%s, branch=%s", task.id, mode, branch)
    task_run = task.create_run(mode=mode, branch=branch, extra_state=extra_state)
    if is_pi_task and resume_from_run_id:
        task_run.active_task_session = previous_run.active_task_session
        task_run.save(update_fields=["active_task_session", "updated_at"])

    if imported_mcp_servers or relayed_mcp_servers:
        update_fields = ["updated_at"]
        if imported_mcp_servers:
            # Kept out of `state` (a plain JSONField) because header values carry credentials.
            task_run.imported_mcp_servers = imported_mcp_servers
            update_fields.append("imported_mcp_servers")
        if relayed_mcp_servers:
            task_run.relayed_mcp_servers = relayed_mcp_servers
            update_fields.append("relayed_mcp_servers")
        task_run.save(update_fields=update_fields)

    if pending_user_artifact_ids:
        _attach_staged_artifacts_to_run(
            task_run, task, staged_artifacts=staged_artifacts, artifact_ids=pending_user_artifact_ids
        )

    if github_user_token and pr_authorship_mode == PrAuthorshipMode.USER:
        cache_github_user_token(str(task_run.id), github_user_token)

    logger.info("Triggering workflow for task %s, run %s", task.id, task_run.id)
    if is_pi_task:
        initial_message = (
            pending_user_message if resume_from_run_id else pending_user_message or task.description or None
        )
        _trigger_task_processing_workflow(
            task,
            task_run,
            user_id,
            initial_message=initial_message,
            initial_artifact_ids=pending_user_artifact_ids,
            raise_on_error=False,
        )
    else:
        _trigger_task_processing_workflow(task, task_run, user_id, raise_on_error=False)

    return contracts.TaskRunResult(task=get_task_detail(task.id, team_id, user_id))


# --- Task presence beacons ---


def beacon_task_presence(task_id: str | UUID, team_id: int, user_id: int | None, *, device_id) -> str:
    """Idempotent upsert of a presence row for a device watching a task.

    Returns ``"not_found"`` (task not visible or device_id doesn't match the caller's push
    token), or ``"ok"``. Mirrors ``TaskViewSet._presence_beacon``.
    """
    from insights.models.user_push_token import UserPushToken  # noqa: PLC0415

    from products.tasks.backend.models import TASK_PRESENCE_TTL_SECONDS, TaskPresence  # noqa: PLC0415

    if user_id is None:
        return "not_found"
    task = _visible_task_qs(team_id, user_id).filter(id=task_id).first()
    if task is None:
        return "not_found"
    push_token = UserPushToken.objects.filter(user_id=user_id, id=device_id).first()
    if push_token is None:
        return "not_found"

    now = django_timezone.now()
    # nosemgrep: idor-lookup-without-team — team scope is enforced by TaskScopedManager
    # and via the `task` FK whose row is fetched (visibility-gated) above.
    TaskPresence.objects.update_or_create(
        task=task,
        push_token=push_token,
        defaults={
            "team": task.team,
            "user_id": user_id,
            "expires_at": now + timedelta(seconds=TASK_PRESENCE_TTL_SECONDS),
        },
    )
    return "ok"


def leave_task_presence(task_id: str | UUID, team_id: int, user_id: int | None, *, device_id) -> str:
    """Best-effort delete of a presence row. ``"not_found"`` if the task isn't visible, else ``"ok"``.

    No 404 on missing presence rows — leave runs from blur/background handlers. Mirrors
    ``TaskViewSet._presence_leave``.
    """
    from products.tasks.backend.models import TaskPresence  # noqa: PLC0415

    if user_id is None:
        return "not_found"
    task = _visible_task_qs(team_id, user_id).filter(id=task_id).first()
    if task is None:
        return "not_found"
    TaskPresence.objects.filter(task=task, push_token_id=device_id, user_id=user_id).delete()
    return "ok"


# --- Slack thread context (internal debug) ---


def _temporal_workflow_url(workflow_id: str | None) -> str | None:
    from django.conf import settings  # noqa: PLC0415

    if not workflow_id:
        return None
    base = getattr(settings, "TEMPORAL_UI_HOST", None)
    namespace = getattr(settings, "TEMPORAL_NAMESPACE", None)
    if not base or not namespace:
        return None
    return f"{base.rstrip('/')}/namespaces/{namespace}/workflows/{workflow_id}"


def _slack_repo_research_dto(
    team_id: int, state: dict, repo_research_runs_by_id: dict, *, build_task_view_url
) -> contracts.SlackThreadContextRepoResearchDTO | None:
    from insights.storage import object_storage  # noqa: PLC0415

    research_task_id = state.get("repo_research_task_id")
    research_run_id = state.get("repo_research_run_id")
    if not research_task_id or not research_run_id:
        return None
    research_run = repo_research_runs_by_id.get(research_run_id)
    sandbox_url = None
    log_url = None
    run_status = None
    if research_run is not None:
        sandbox_url = (research_run.state if isinstance(research_run.state, dict) else {}).get("sandbox_url")
        run_status = research_run.status
        try:
            log_url = object_storage.get_presigned_url(research_run.log_url, expiration=3600)
        except Exception:
            logger.exception("slack_thread_context_research_log_presign_failed", extra={"run_id": research_run_id})
            log_url = None
    # Prefer the run's actual id (prefixed dispatches persist it); fall back to derived when the row is gone.
    workflow_id = (
        research_run.workflow_id
        if research_run is not None
        else TaskRun.get_workflow_id(research_task_id, research_run_id)
    )
    return contracts.SlackThreadContextRepoResearchDTO(
        task_id=research_task_id,
        run_id=research_run_id,
        status=run_status,
        task_processing_workflow_id=workflow_id,
        task_processing_workflow_url=_temporal_workflow_url(workflow_id),
        sandbox_url=sandbox_url,
        task_view_url=build_task_view_url(
            f"/project/{team_id}/tasks/{research_task_id}?runId={research_run_id}&ph_debug=true"
        ),
        log_url=log_url,
    )


def resolve_slack_thread_context(
    team_id: int, *, channel: str, thread_ts: str, url: str, build_url
) -> contracts.SlackThreadContextResult:
    """Resolve a parsed Slack permalink to its task, runs, and Temporal workflow handles.

    Caller passes the already-parsed ``(channel, thread_ts)`` and a ``build_url`` callable
    (``request.build_absolute_uri``) so the facade stays request-agnostic. Caller also enforces
    the internal-debug gate before calling. Mirrors ``TaskViewSet.slack_thread_context``.
    """
    from insights.storage import object_storage  # noqa: PLC0415

    from products.slack_app.backend.models import (  # noqa: PLC0415 — cross-product import kept off the api import path
        SlackThreadTaskMapping,
    )

    mapping = (
        SlackThreadTaskMapping.objects.select_related("task", "task__created_by")
        .filter(channel=channel, thread_ts=thread_ts)
        .first()
    )
    if mapping is None:
        return contracts.SlackThreadContextResult(
            outcome="no_mapping",
            no_mapping_thread=contracts.SlackThreadContextThreadDTO(
                url=url,
                channel=channel,
                thread_ts=thread_ts,
                slack_workspace_id=None,
                mentioning_slack_user_id=None,
            ),
        )

    task = mapping.task
    runs = list(TaskRun.objects.filter(task=task).order_by("created_at", "id"))
    repo_research_run_ids = [
        rid for run in runs if (rid := (run.state if isinstance(run.state, dict) else {}).get("repo_research_run_id"))
    ]
    repo_research_runs_by_id = (
        {str(r.id): r for r in TaskRun.objects.filter(team=task.team, id__in=repo_research_run_ids)}
        if repo_research_run_ids
        else {}
    )
    task_url = build_url(f"/project/{task.team_id}/tasks/{task.id}?ph_debug=true")

    run_dtos: list[contracts.SlackThreadContextRunDTO] = []
    for run in runs:
        state = run.state if isinstance(run.state, dict) else {}
        output = run.output if isinstance(run.output, dict) else {}
        task_processing_workflow_id = run.workflow_id
        mention_workflow_id = state.get("slack_mention_workflow_id")
        try:
            presigned_log_url = object_storage.get_presigned_url(run.log_url, expiration=3600)
        except Exception:
            logger.exception("slack_thread_context_log_presign_failed", extra={"run_id": str(run.id)})
            presigned_log_url = None
        run_dtos.append(
            contracts.SlackThreadContextRunDTO(
                id=str(run.id),
                status=run.status,
                created_at=run.created_at,
                completed_at=run.completed_at,
                sandbox_url=state.get("sandbox_url"),
                pr_url=output.get("pr_url"),
                error_message=run.error_message,
                task_processing_workflow_id=task_processing_workflow_id,
                task_processing_workflow_url=_temporal_workflow_url(task_processing_workflow_id),
                mention_workflow_id=mention_workflow_id,
                mention_workflow_url=_temporal_workflow_url(mention_workflow_id),
                task_view_url=build_url(f"/project/{task.team_id}/tasks/{task.id}?runId={run.id}&ph_debug=true"),
                log_url=presigned_log_url,
                repo_research=_slack_repo_research_dto(
                    task.team_id, state, repo_research_runs_by_id, build_task_view_url=build_url
                ),
            )
        )

    context = contracts.SlackThreadContextDTO(
        thread=contracts.SlackThreadContextThreadDTO(
            url=url,
            channel=channel,
            thread_ts=thread_ts,
            slack_workspace_id=mapping.slack_workspace_id,
            mentioning_slack_user_id=mapping.mentioning_slack_user_id,
        ),
        task=contracts.SlackThreadContextTaskDTO(
            id=str(task.id),
            team_id=task.team_id,
            title=task.title,
            repository=task.repository,
            origin_product=task.origin_product,
            created_at=task.created_at,
            url=task_url,
        ),
        runs=run_dtos,
    )
    return contracts.SlackThreadContextResult(outcome="ok", context=context)


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


def create_sandbox_connection_token(run_id: str | UUID, user_id: int, distinct_id: str) -> str:
    """Mint a short-lived connection token for talking to a run's live sandbox."""
    from products.tasks.backend.logic.services.connection_token import (  # noqa: PLC0415 — keep sandbox deps off the api import path
        create_sandbox_connection_token as _create,
    )

    run = TaskRun.objects.select_related("task").get(id=run_id)
    return _create(run, user_id, distinct_id)


def send_cancel(run_id: str | UUID, *, auth_token: str | None = None):
    """Cancel the agent running in a run's live sandbox."""
    from products.tasks.backend.logic.services.agent_command import (  # noqa: PLC0415 — keep sandbox deps off the api import path
        send_cancel as _send_cancel,
    )

    run = TaskRun.objects.select_related("task").get(id=run_id)
    return _send_cancel(run, auth_token=auth_token)


# --- Channels & task threads ---


def normalize_channel_name(name: str) -> str:
    """Slack-style channel key: lowercase, whitespace collapsed to dashes.

    Channels are resolved by name from client-side surfaces (folder names), so the
    stored key must be canonical for the (team, name) uniqueness to mean anything.
    """
    return re.sub(r"\s+", "-", name.strip().lower())[:128]


def _channel_to_dto(channel: Channel, *, starred: bool = False) -> contracts.ChannelDTO:
    return contracts.ChannelDTO(
        id=channel.id,
        name=channel.name,
        channel_type=channel.channel_type,
        github_integration=channel.github_integration_id,
        repositories=channel.repositories,
        created_at=channel.created_at,
        created_by=_user_basic_info(channel.created_by if channel.created_by_id else None),
        starred=starred,
    )


def _team_channels(team_id: int) -> QuerySet[Channel]:
    # for_team rather than a bare team_id filter so these reads also resolve outside request
    # scope (Temporal activities), where the fail-closed manager raises on an unscoped read.
    return Channel.objects.for_team(team_id)


def _ensure_personal_channel(team_id: int, user_id: int) -> Channel:
    # select_related so _channel_to_dto doesn't lazy-load created_by per call.
    channels = _team_channels(team_id).select_related("created_by")
    lookup = {
        "team_id": team_id,
        "created_by_id": user_id,
        "channel_type": Channel.ChannelType.PERSONAL,
        "deleted": False,
    }
    try:
        channel, _ = channels.get_or_create(**lookup, defaults={"name": Channel.PERSONAL_CHANNEL_NAME})
    except IntegrityError:
        channel = channels.get(**lookup)
    return channel


def ensure_personal_channel_id(team_id: int, user_id: int) -> UUID:
    """Get-or-create the user's personal "#me" channel and return its id.

    For callers outside a request (Temporal activities) that need somewhere to file a task.
    """
    return _ensure_personal_channel(team_id, user_id).id


def list_channels(team_id: int, user_id: int | None) -> list[contracts.ChannelDTO]:
    """All live public channels plus the requester's personal channel (provisioned lazily),
    personal first, then by name. ``starred`` reflects the requester's stars."""
    channels: list[Channel] = []
    if user_id is not None:
        channels.append(_ensure_personal_channel(team_id, user_id))
    channels.extend(
        Channel.objects.filter(team_id=team_id, channel_type=Channel.ChannelType.PUBLIC, deleted=False)
        .select_related("created_by")
        .order_by("name")
    )
    starred_ids: set = (
        set(ChannelStar.objects.filter(team_id=team_id, user_id=user_id).values_list("channel_id", flat=True))
        if user_id is not None
        else set()
    )
    return [_channel_to_dto(channel, starred=channel.id in starred_ids) for channel in channels]


def _emit_channel_created(channel: Channel, user_id: int | None) -> None:
    """Announce a newly-created public channel in its own feed as a system row
    ("Ann created this context"). Server-emitted so the announcement appears no
    matter which client (or integration) created the channel. Best-effort — a
    feed-write failure must never break channel creation. The fail-closed
    ``TeamScopedManager`` raises without team context, so callers outside a
    request (temporal, MCP) must wrap in ``team_scope()`` or the announcement
    is swallowed here and only logged."""
    try:
        ChannelFeedMessage.objects.create(
            team_id=channel.team_id,
            channel_id=channel.id,
            author_id=user_id,
            author_kind=ChannelFeedMessage.AuthorKind.SYSTEM,
            event="channel_created",
            payload={"channel_name": channel.name},
        )
    except Exception:
        logger.exception("Failed to emit channel_created feed message", extra={"channel_id": str(channel.id)})


def resolve_channel(team_id: int, user_id: int | None, *, name: str) -> contracts.ChannelDTO | None:
    """Resolve-or-create a public channel by (normalized) name. ``None`` for empty names.
    Emits a ``channel_created`` feed message the first time a channel is created."""
    normalized = normalize_channel_name(name)
    if not normalized:
        return None
    created = False
    try:
        channel, created = Channel.objects.select_related("created_by").get_or_create(
            team_id=team_id,
            name=normalized,
            channel_type=Channel.ChannelType.PUBLIC,
            deleted=False,
            defaults={"created_by_id": user_id},
        )
    except IntegrityError:
        channel = Channel.objects.select_related("created_by").get(
            team_id=team_id, name=normalized, channel_type=Channel.ChannelType.PUBLIC, deleted=False
        )
    if created:
        _emit_channel_created(channel, user_id)
    return _channel_to_dto(channel)


def update_channel(
    channel_id: str | UUID,
    team_id: int,
    user_id: int | None,
    *,
    name: str | None = None,
    github_integration: Integration | None = None,
    repositories: list[str] | None = None,
) -> contracts.ChannelDTO | str:
    """Update a channel, keeping repository configuration creator-owned."""
    channel = Channel.objects.filter(id=channel_id, team_id=team_id, deleted=False).first()
    if channel is None:
        return "not_found"
    if channel.channel_type == Channel.ChannelType.PERSONAL:
        if channel.created_by_id != user_id:
            return "not_found"
        if name is not None:
            return "personal"
    elif repositories is not None and channel.created_by_id != user_id:
        return "not_found"
    update_fields: list[str] = []
    if name is not None:
        normalized = normalize_channel_name(name)
        if not normalized:
            return "invalid_name"
        channel.name = normalized
        update_fields.append("name")
    if repositories is not None:
        channel.repositories = repositories
        channel.github_integration = github_integration if repositories else None
        update_fields.extend(["repositories", "github_integration"])
    if not update_fields:
        return _channel_to_dto(channel)
    try:
        channel.save(update_fields=[*update_fields, "updated_at"])
    except IntegrityError:
        return "name_taken"
    return _channel_to_dto(channel)


def delete_channel(channel_id: str | UUID, team_id: int) -> str:
    """Soft-delete a public channel. Returns ``ok`` / ``not_found`` / ``personal``."""
    channel = Channel.objects.filter(id=channel_id, team_id=team_id, deleted=False).first()
    if channel is None:
        return "not_found"
    if channel.channel_type == Channel.ChannelType.PERSONAL:
        return "personal"
    channel.deleted = True
    channel.save(update_fields=["deleted", "updated_at"])
    return "ok"


# Per-channel ceiling on feed rows — the feed holds rare lifecycle announcements, so the
# cap exists to stop one member making the feed unboundedly expensive to store and read.
CHANNEL_FEED_MAX_MESSAGES = 500


def _channel_feed_message_to_dto(message: ChannelFeedMessage) -> contracts.ChannelFeedMessageDTO:
    return contracts.ChannelFeedMessageDTO(
        id=message.id,
        channel=message.channel_id,
        author_kind=message.author_kind,
        event=message.event,
        payload=message.payload or {},
        content=message.content,
        created_at=message.created_at,
        author=_user_basic_info(message.author if message.author_id else None),
    )


def visible_channels_q(user_id: int | None, *, relation: Literal["", "channel"] = "") -> Q:
    """The channel-visibility rule as a queryset filter; see ``Channel.visible_to_q``
    for the semantics. Exported for cross-product callers filtering channel-joined
    querysets. Single-object callers use ``get_channel``."""
    return Channel.visible_to_q(user_id, relation=relation)


def channel_exists(team_id: int, channel_id: str | UUID, user_id: int | None) -> bool:
    """Whether ``channel_id`` is a live channel in this team that the user may see."""
    return Channel.objects.filter(Channel.visible_to_q(user_id), id=channel_id, team_id=team_id, deleted=False).exists()


def _visible_channel(channel_id: str | UUID, team_id: int, user_id: int | None) -> Channel | None:
    """A channel the requester may read: any live public channel on the team, or their
    own personal channel. ``None`` when it's missing or someone else's personal channel."""
    return (
        _team_channels(team_id)
        .select_related("created_by")
        .filter(visible_channels_q(user_id), id=channel_id, deleted=False)
        .first()
    )


def list_channel_feed_messages(
    channel_id: str | UUID, team_id: int, user_id: int | None
) -> list[contracts.ChannelFeedMessageDTO] | None:
    """A channel's system-announcement feed, ascending. ``None`` when the channel isn't visible.
    Bounded to the newest ``CHANNEL_FEED_MAX_MESSAGES``: the feed holds rare lifecycle
    events, and the write path caps a channel at the same count. Add real pagination
    before any per-task or per-thread event lands here."""
    if _visible_channel(channel_id, team_id, user_id) is None:
        return None
    messages = (
        ChannelFeedMessage.objects.filter(channel_id=channel_id, team_id=team_id, deleted=False)
        .select_related("author")
        .order_by("-created_at", "-id")[:CHANNEL_FEED_MAX_MESSAGES]
    )
    return [_channel_feed_message_to_dto(message) for message in reversed(messages)]


def create_channel_feed_message(
    channel_id: str | UUID,
    team_id: int,
    user_id: int | None,
    *,
    event: str,
    payload: dict,
    created_at: datetime | None = None,
) -> contracts.ChannelFeedMessageDTO | None | str:
    """Post an announcement into a channel's feed as the requester. ``None`` when the
    channel isn't visible; ``"full"`` when the channel's feed is at capacity. The row is
    marked human-authored — ``system``/``agent`` kinds are reserved for server-side
    writers, so a client can't forge rows other clients render as trusted. ``author``
    records the acting user so the client can render "Adam …". ``created_at`` lets a
    client order a burst of announcements deterministically (else the server stamps
    ``now``)."""
    if _visible_channel(channel_id, team_id, user_id) is None:
        return None
    if (
        ChannelFeedMessage.objects.filter(channel_id=channel_id, team_id=team_id, deleted=False).count()
        >= CHANNEL_FEED_MAX_MESSAGES
    ):
        return "full"
    fields: dict = {
        "team_id": team_id,
        "channel_id": channel_id,
        "author_id": user_id,
        "author_kind": ChannelFeedMessage.AuthorKind.HUMAN,
        "event": event,
        "payload": payload or {},
    }
    if created_at is not None:
        fields["created_at"] = created_at
    message = ChannelFeedMessage.objects.create(**fields)
    # Fresh row: author lazy-loads once for the DTO.
    return _channel_feed_message_to_dto(message)


def get_channel(channel_id: str | UUID, team_id: int, user_id: int | None) -> contracts.ChannelDTO | None:
    """One channel the requester may read, or ``None``."""
    channel = _visible_channel(channel_id, team_id, user_id)
    if channel is None:
        return None
    starred = user_id is not None and ChannelStar.objects.filter(channel_id=channel.id, user_id=user_id).exists()
    return _channel_to_dto(channel, starred=starred)


# --- Channel instructions (CONTEXT.md) ---

# Generous cap on the markdown blob; channel instructions are descriptions, not documents.
CHANNEL_INSTRUCTIONS_MAX_BYTES = 100_000
MAX_CHANNEL_INSTRUCTIONS_VERSION = 2000


@dataclass
class ChannelInstructionsTooLargeError(Exception):
    max_bytes: int


@dataclass
class ChannelInstructionsVersionConflictError(Exception):
    current_version: int


@dataclass
class ChannelInstructionsVersionLimitError(Exception):
    max_version: int


def _instructions_to_dto(row: ChannelInstructions) -> contracts.ChannelInstructionsDTO:
    return contracts.ChannelInstructionsDTO(
        channel=row.channel_id,
        content=row.content,
        version=row.version,
        created_at=row.created_at,
        created_by=_user_basic_info(row.created_by if row.created_by_id else None),
    )


def _blank_instructions_dto(channel: Channel) -> contracts.ChannelInstructionsDTO:
    """A channel with no published instructions reads as blank version 0, so
    readers never 404 and a first publish guards on ``base_version: 0``."""
    return contracts.ChannelInstructionsDTO(channel=channel.id, content="", version=0)


def get_channel_instructions(
    channel_id: str | UUID, team_id: int, user_id: int | None
) -> contracts.ChannelInstructionsDTO | None:
    """The channel's latest instructions (blank version 0 when none exist).
    ``None`` when the channel isn't visible."""
    channel = _visible_channel(channel_id, team_id, user_id)
    if channel is None:
        return None
    latest = (
        ChannelInstructions.objects.filter(channel_id=channel.id, deleted=False, is_latest=True)
        .select_related("created_by")
        .first()
    )
    return _instructions_to_dto(latest) if latest is not None else _blank_instructions_dto(channel)


def list_channel_instruction_versions(
    channel_id: str | UUID, team_id: int, user_id: int | None
) -> list[contracts.ChannelInstructionsDTO] | None:
    """The channel's instruction history, newest first. ``None`` when the channel isn't visible."""
    if _visible_channel(channel_id, team_id, user_id) is None:
        return None
    versions = (
        ChannelInstructions.objects.filter(channel_id=channel_id, team_id=team_id, deleted=False)
        .select_related("created_by")
        .order_by("-version", "-created_at", "-id")[:200]
    )
    return [_instructions_to_dto(row) for row in versions]


def publish_channel_instructions(
    channel_id: str | UUID,
    team_id: int,
    user_id: int | None,
    *,
    content: str,
    base_version: int | None = None,
) -> contracts.ChannelInstructionsDTO | None:
    """Publish a new instructions version, superseding the current latest.

    ``base_version`` guards against lost updates: when the current latest no
    longer matches, ``ChannelInstructionsVersionConflictError`` is raised.
    ``None`` when the channel isn't visible. Publishing clears the channel's
    in-progress context-generation marker.
    """
    channel = _visible_channel(channel_id, team_id, user_id)
    if channel is None:
        return None
    if len(content.encode("utf-8")) > CHANNEL_INSTRUCTIONS_MAX_BYTES:
        raise ChannelInstructionsTooLargeError(max_bytes=CHANNEL_INSTRUCTIONS_MAX_BYTES)
    with transaction.atomic():
        current_latest = (
            ChannelInstructions.objects.select_for_update()
            .filter(channel_id=channel.id, deleted=False, is_latest=True)
            .order_by("-version", "-created_at", "-id")
            .first()
        )
        current_version = current_latest.version if current_latest is not None else 0
        if base_version is not None and base_version != current_version:
            raise ChannelInstructionsVersionConflictError(current_version=current_version)
        if current_version >= MAX_CHANNEL_INSTRUCTIONS_VERSION:
            raise ChannelInstructionsVersionLimitError(max_version=MAX_CHANNEL_INSTRUCTIONS_VERSION)
        if current_latest is not None:
            ChannelInstructions.objects.filter(pk=current_latest.pk).update(is_latest=False)
        try:
            # Nested savepoint: a lost-update race (the select_for_update above
            # locks no row when none exists yet, and a concurrent delete clears
            # is_latest without adding a lockable row) makes the insert collide
            # with the (channel, version) uniqueness. Rolling back to the
            # savepoint keeps this transaction usable so we can read the winner.
            with transaction.atomic():
                published = ChannelInstructions.objects.create(
                    team_id=team_id,
                    channel_id=channel.id,
                    content=content,
                    version=current_version + 1,
                    is_latest=True,
                    created_by_id=user_id,
                )
        except IntegrityError:
            # Surface the race as the conflict the view maps to 409, not a 500.
            latest = (
                ChannelInstructions.objects.filter(channel_id=channel.id, deleted=False)
                .order_by("-version", "-created_at", "-id")
                .first()
            )
            raise ChannelInstructionsVersionConflictError(current_version=latest.version if latest is not None else 0)
        # Publishing produced a result, so drop the in-progress generation marker.
        ChannelContextGeneration.objects.filter(channel_id=channel.id).update(task_id=None)
    return _instructions_to_dto(published)


def delete_channel_instructions(channel_id: str | UUID, team_id: int, user_id: int | None) -> int | None:
    """Soft-delete every instructions version. Returns the count, or ``None``
    when the channel isn't visible."""
    channel = _visible_channel(channel_id, team_id, user_id)
    if channel is None:
        return None
    with transaction.atomic():
        count = (
            ChannelInstructions.objects.select_for_update()
            .filter(channel_id=channel.id, deleted=False)
            .update(deleted=True, is_latest=False)
        )
        ChannelContextGeneration.objects.filter(channel_id=channel.id).update(task_id=None)
    return count


def get_channel_context_generation(
    channel_id: str | UUID, team_id: int, user_id: int | None
) -> str | None | Literal["not_found"]:
    """The id of the task currently generating the channel's CONTEXT.md, or ``None``."""
    channel = _visible_channel(channel_id, team_id, user_id)
    if channel is None:
        return "not_found"
    marker = ChannelContextGeneration.objects.filter(channel_id=channel.id).first()
    return str(marker.task_id) if marker is not None and marker.task_id else None


def set_channel_context_generation(
    channel_id: str | UUID, team_id: int, user_id: int | None, *, task_id: str | UUID | None
) -> str | None | Literal["not_found", "invalid_task"]:
    """Set or clear the task associated with the channel's CONTEXT.md generation."""
    channel = _visible_channel(channel_id, team_id, user_id)
    if channel is None:
        return "not_found"
    if task_id is not None and not task_exists(task_id, team_id):
        return "invalid_task"
    ChannelContextGeneration.objects.update_or_create(
        channel_id=channel.id, defaults={"team_id": team_id, "task_id": task_id}
    )
    return str(task_id) if task_id else None


def star_channel(channel_id: str | UUID, team_id: int, user_id: int, *, starred: bool) -> bool:
    """Star or unstar a channel for the requesting user. False when the channel isn't visible."""
    channel = _visible_channel(channel_id, team_id, user_id)
    if channel is None:
        return False
    if starred:
        ChannelStar.objects.get_or_create(channel_id=channel.id, user_id=user_id, defaults={"team_id": team_id})
    else:
        ChannelStar.objects.filter(channel_id=channel.id, user_id=user_id).delete()
    return True


def _thread_message_to_dto(message: TaskThreadMessage) -> contracts.TaskThreadMessageDTO:
    return contracts.TaskThreadMessageDTO(
        id=message.id,
        task=message.task_id,
        author_kind=message.author_kind,
        event=message.event,
        payload=message.payload or {},
        content=message.content,
        created_at=message.created_at,
        author=_user_basic_info(message.author if message.author_id else None),
        forwarded_to_agent_at=message.forwarded_to_agent_at,
        forwarded_by=_user_basic_info(message.forwarded_by if message.forwarded_by_id else None),
    )


def _visible_task(task_id: str | UUID, team_id: int, user_id: int | None) -> Task | None:
    return _visible_task_qs(team_id, user_id).filter(id=task_id).first()


def list_thread_messages(
    task_id: str | UUID, team_id: int, user_id: int | None
) -> list[contracts.TaskThreadMessageDTO] | None:
    """A task's thread, ascending. ``None`` when the task isn't visible to the user."""
    if _visible_task(task_id, team_id, user_id) is None:
        return None
    messages = (
        TaskThreadMessage.objects.filter(task_id=task_id, team_id=team_id)
        # The thread is human-to-human plus artifact announcements; rows written
        # back when the agent finished a turn (a since-removed behavior) stay out.
        .exclude(event="turn_complete")
        .select_related("author", "forwarded_by")
        .order_by("created_at", "id")
    )
    return [_thread_message_to_dto(message) for message in messages]


def create_thread_message(
    task_id: str | UUID, team_id: int, user_id: int | None, *, content: str
) -> contracts.TaskThreadMessageDTO | None:
    """Add a thread message as the requester. ``None`` when the task isn't visible."""
    if _visible_task(task_id, team_id, user_id) is None:
        return None
    message = TaskThreadMessage.objects.create(team_id=team_id, task_id=task_id, author_id=user_id, content=content)
    try:
        project_thread_message_activity(message)
    except Exception:
        logger.exception("Failed to project thread message activity", extra={"message_id": str(message.id)})
    try:
        mentioned_user_ids = resolve_mentioned_user_ids(
            User, message.content, team_id=message.team_id, author_id=message.author_id
        )
    except Exception:
        mentioned_user_ids = []
        logger.exception("Failed to resolve thread message mentions", extra={"message_id": str(message.id)})
    try:
        _index_thread_message_mentions(message, mentioned_user_ids)
    except Exception:
        # Mention indexing is best-effort: a failure must never fail message creation or discard resolved recipients.
        logger.exception("Failed to index thread message mentions", extra={"message_id": str(message.id)})
    from products.tasks.backend.push_dispatcher import notify_task_thread_message  # noqa: PLC0415

    notify_task_thread_message(message, mentioned_user_ids)
    # Fresh message: forwarded_by is None (no query) and author lazy-loads once.
    return _thread_message_to_dto(message)


def _index_thread_message_mentions(message: TaskThreadMessage, mentioned_user_ids: Collection[int]) -> None:
    """Create mention index rows for @[Name](email) tokens in the message content.

    Emails resolve case-insensitively, only to members of the team's organization;
    self-mentions are skipped (they are never notifications).
    """
    mentions = [
        TaskThreadMessageMention(
            team_id=message.team_id,
            message_id=message.id,
            task_id=message.task_id,
            mentioned_user_id=mentioned_user_id,
            created_at=message.created_at,
        )
        for mentioned_user_id in mentioned_user_ids
    ]
    TaskThreadMessageMention.objects.for_team(message.team_id).bulk_create(
        mentions,
        ignore_conflicts=True,
    )
    for mention in mentions:
        TaskActivity.record(
            team_id=message.team_id,
            user_id=mention.mentioned_user_id,
            task_id=message.task_id,
            kind=TaskActivity.Kind.MENTION,
            activity_at=message.created_at,
            message_id=message.id,
        )


def list_mentions(
    team_id: int, user_id: int | None, *, since: datetime | None = None, limit: int = 100
) -> list[contracts.TaskMentionDTO]:
    """Thread-message mentions of the requester across tasks they can see, newest first."""
    if user_id is None:
        return []
    qs = TaskThreadMessageMention.objects.filter(
        team_id=team_id,
        mentioned_user_id=user_id,
        # task__in keeps the visibility rules single-sourced in _visible_task_qs.
        task__in=_visible_task_qs(team_id, user_id),
        # Legacy turn_complete rows are hidden from threads (see list_thread_messages),
        # so their indexed mentions must not surface notifications pointing at them.
    ).exclude(message__event="turn_complete")
    if since is not None:
        qs = qs.filter(created_at__gt=since)
    mentions = qs.select_related("message__author", "task__channel").order_by("-created_at")[:limit]
    return [
        contracts.TaskMentionDTO(
            id=mention.id,
            message_id=mention.message_id,
            task_id=mention.task_id,
            task_title=mention.task.title,
            channel_id=mention.task.channel_id,
            channel_name=mention.task.channel.name if mention.task.channel else None,
            content=mention.message.content,
            created_at=mention.created_at,
            author=_user_basic_info(mention.message.author if mention.message.author_id else None),
        )
        for mention in mentions
    ]


def project_thread_message_activity(message: TaskThreadMessage) -> None:
    """Project a new thread message onto the feed of everyone it concerns."""
    recipient_ids = {recipient_id for recipient_id in (message.author_id, message.task.created_by_id) if recipient_id}
    for recipient_id in recipient_ids:
        TaskActivity.record(
            team_id=message.team_id,
            user_id=recipient_id,
            task_id=message.task_id,
            kind=TaskActivity.Kind.MESSAGE,
            activity_at=message.created_at,
            message_id=message.id,
            actor_id=message.author_id,
        )


def project_awaiting_input_activity(task_run: "TaskRun") -> None:
    """Flag the task creator's feed row when a run stops and needs them.

    Called from ``push_dispatcher.notify_task_run_awaiting_input`` so every path that
    decides a run is waiting (stream ingest, agent proxy callback, sandbox relay) projects
    the same row. Deliberately outside the push feature flag and its Redis cooldown — the
    in-app feed should update even where the mobile push is off.
    """
    creator_id = task_run.task.created_by_id
    if creator_id is None:
        return
    TaskActivity.record(
        team_id=task_run.task.team_id,
        user_id=creator_id,
        task_id=task_run.task_id,
        kind=TaskActivity.Kind.AWAITING_INPUT,
        activity_at=django_timezone.now(),
    )


def project_completed_activity(task_run: "TaskRun") -> None:
    creator_id = task_run.task.created_by_id
    if creator_id is None:
        return
    TaskActivity.record(
        team_id=task_run.task.team_id,
        user_id=creator_id,
        task_id=task_run.task_id,
        kind=TaskActivity.Kind.COMPLETED,
        activity_at=task_run.completed_at or django_timezone.now(),
    )


def _task_activity_qs(team_id: int, user_id: int) -> QuerySet[TaskActivity]:
    """The requester's feed rows, gated to tasks they can still see.

    Rows outlive visibility changes (a task moving to a private channel, say), so the
    visibility gate belongs on read rather than being enforced when projecting.
    """
    visible_tasks = _visible_task_qs(team_id, user_id).filter(internal=False, archived=False)
    return TaskActivity.objects.filter(team_id=team_id, user_id=user_id, task__in=visible_tasks)


def count_unread_task_activity(team_id: int, user_id: int | None) -> int:
    """Unread tasks across the requester's whole feed. Backs the sidebar badge."""
    if user_id is None:
        return 0
    return _task_activity_qs(team_id, user_id).filter(read_at__isnull=True).count()


def list_task_activity(
    team_id: int,
    user_id: int | None,
    *,
    limit: int = 100,
    before: datetime | None = None,
    before_id: UUID | None = None,
) -> contracts.TaskActivityPageDTO:
    """The requester's feed: one row per task they are involved in, newest activity first.

    ``unread_count`` counts every unread row the requester can see, not just the ones in
    this page, so the sidebar badge stays honest past ``limit``.
    """
    if user_id is None:
        return contracts.TaskActivityPageDTO(results=[], unread_count=0)
    qs = _task_activity_qs(team_id, user_id)
    if before is not None and before_id is not None:
        qs = qs.filter(Q(activity_at__lt=before) | Q(activity_at=before, id__lt=before_id))
    rows = list(qs.select_related("task__channel", "message__author").order_by("-activity_at", "-id")[: limit + 1])
    has_more = len(rows) > limit
    rows = rows[:limit]
    next_row = rows[-1] if has_more else None
    return contracts.TaskActivityPageDTO(
        results=[
            contracts.TaskActivityDTO(
                id=row.id,
                task_id=row.task_id,
                task_title=row.task.title,
                channel_id=row.task.channel_id,
                channel_name=row.task.channel.name if row.task.channel else None,
                activity_at=row.activity_at,
                activity_kind=row.kind,
                snippet=row.message.content if row.message else "",
                latest_author=_user_basic_info(row.message.author if row.message and row.message.author_id else None),
                latest_message_id=row.message_id,
                is_unread=row.read_at is None,
            )
            for row in rows
        ],
        unread_count=_task_activity_qs(team_id, user_id).filter(read_at__isnull=True).count(),
        next_before=next_row.activity_at if next_row else None,
        next_before_id=next_row.id if next_row else None,
    )


def mark_task_activity_read(team_id: int, user_id: int | None, activities: Sequence[tuple[UUID, datetime]]) -> int:
    """Mark feed rows read only when their latest activity was visible to the requester."""
    if user_id is None or not activities:
        return 0
    activity_versions = Q()
    for task_id, seen_before in activities:
        activity_versions |= Q(task_id=task_id, activity_at__lte=seen_before)
    return (
        TaskActivity.objects.filter(team_id=team_id, user_id=user_id, read_at__isnull=True)
        .filter(activity_versions)
        .update(read_at=django_timezone.now())
    )


def delete_thread_message(message_id: str | UUID, task_id: str | UUID, team_id: int, user_id: int | None) -> str:
    """Delete own thread message. Returns ``ok`` / ``not_found`` / ``forbidden``."""
    message = TaskThreadMessage.objects.filter(id=message_id, task_id=task_id, team_id=team_id).first()
    if message is None or _visible_task(task_id, team_id, user_id) is None:
        return "not_found"
    if message.author_id != user_id:
        return "forbidden"
    message.delete()
    return "ok"


def forward_thread_message(
    message_id: str | UUID, task_id: str | UUID, team_id: int, user_id: int | None
) -> tuple[str, contracts.TaskThreadMessageDTO | None]:
    """Send a thread message to the task's agent. Task-author only.

    Returns ``(kind, dto)`` where kind is ``ok`` / ``not_found`` / ``forbidden`` /
    ``already_forwarded`` / ``no_run`` / ``signal_failed``.
    """
    task = _visible_task(task_id, team_id, user_id)
    if task is None:
        return "not_found", None
    if task.created_by_id != user_id:
        return "forbidden", None

    # Lock the message row so concurrent forwards of the same message can't
    # both pass the forwarded_to_agent_at check and double-signal the agent.
    with transaction.atomic():
        # of=("self",) locks only the message row: FOR UPDATE cannot span the nullable
        # outer joins that select_related on author/forwarded_by introduces.
        message = (
            TaskThreadMessage.objects.select_for_update(of=("self",))
            .select_related("author", "forwarded_by")
            .filter(id=message_id, task_id=task_id, team_id=team_id)
            .first()
        )
        if message is None:
            return "not_found", None
        if message.forwarded_to_agent_at is not None:
            return "already_forwarded", _thread_message_to_dto(message)
        run = task.latest_run
        if run is None or run.status in (TaskRun.Status.COMPLETED, TaskRun.Status.FAILED, TaskRun.Status.CANCELLED):
            return "no_run", None

        author = message.author
        author_name = (author.get_full_name() or author.email) if author else "A teammate"
        content = f"[Thread comment from {author_name}] {message.content}"
        signal_result = signal_task_run_user_message(run.id, task.id, team_id, content=content, artifact_ids=[])
        if not signal_result:
            return "signal_failed", None

        message.forwarded_to_agent_at = django_timezone.now()
        message.forwarded_by_id = user_id
        message.forwarded_run = run
        message.save(update_fields=["forwarded_to_agent_at", "forwarded_by", "forwarded_run"])
    return "ok", _thread_message_to_dto(message)


# Threads are a Channels (project-bluebird) surface, so agent-authored thread
# updates are gated on the same flag — evaluated for the task creator.
AGENT_THREAD_UPDATES_FLAG = "project-bluebird"


def _create_agent_thread_message(task: Task, content: str, *, event: str, payload: dict | None = None) -> None:
    """Write an agent-authored thread message and index its mentions.

    ``content`` is the rendered text (older clients show it as-is); ``event`` +
    ``payload`` are the structured record, mirroring ChannelFeedMessage, that
    lets clients render agent rows natively and dedupe them against live views.
    """
    # for_team: callers include non-request contexts (temporal relay) where the
    # fail-closed manager has no team scope.
    message = TaskThreadMessage.objects.for_team(task.team_id).create(
        team_id=task.team_id,
        task_id=task.id,
        author_id=None,
        author_kind=TaskThreadMessage.AuthorKind.AGENT,
        event=event,
        payload=payload or {},
        content=content,
    )
    project_thread_message_activity(message)
    try:
        mentioned_user_ids = resolve_mentioned_user_ids(
            User, message.content, team_id=message.team_id, author_id=message.author_id
        )
        _index_thread_message_mentions(message, mentioned_user_ids)
    except Exception:
        logger.exception("Failed to index thread message mentions", extra={"message_id": str(message.id)})


def _agent_thread_updates_enabled(creator: User | None) -> bool:
    """Fail closed: no creator to key the flag on, or a flag-service error, means no post."""
    if creator is None:
        return False
    # Local dev rarely has the server-side flag client wired up, and failing
    # closed there silently drops every agent thread update (PR and canvas
    # announcements vanish from task threads with nothing in the logs).
    if settings.DEBUG:
        return True
    distinct_id = creator.distinct_id or f"user_{creator.id}"
    try:
        return bool(
            hanzo_insights.feature_enabled(AGENT_THREAD_UPDATES_FLAG, distinct_id, send_feature_flag_events=False)
        )
    except Exception:
        logger.warning("Agent thread update flag check failed", extra={"user_id": creator.id}, exc_info=True)
        return False


def post_canvas_created_thread_update(
    task_id: str | UUID, team_id: int, *, acting_user_id: int | None, canvas_name: str, canvas_url: str | None
) -> None:
    """Announce a freshly created canvas in the generating task's thread.

    Posts "[name](url) has been created" as an agent message. Called on a canvas's
    first publish only — the caller owns that once-guard. ``acting_user_id`` must be
    the task's creator: the sandbox publishes with the creator's credentials, so this
    binds the attributed task to the caller's identity — a same-team caller can't
    plant agent messages in someone else's task thread by naming its id. Best-effort
    and never raises: the publish must not fail because its announcement couldn't
    be written.
    """
    try:
        task = Task.objects.select_related("created_by").filter(id=task_id, team_id=team_id).first()
        if task is None or task.created_by_id is None or task.created_by_id != acting_user_id:
            return
        if not _agent_thread_updates_enabled(task.created_by):
            return
        # Brackets and newlines in the name would break the [label](url) token.
        name = re.sub(r"[\[\]\n]", " ", canvas_name).strip() or "Canvas"
        content = f"[{name}]({canvas_url}) has been created" if canvas_url else f"{name} has been created"
        _create_agent_thread_message(
            task,
            content,
            event="canvas_created",
            payload={"canvas_name": name, "canvas_url": canvas_url},
        )
    except Exception:
        logger.exception("Failed to post canvas-created thread update", extra={"task_id": str(task_id)})


_GITHUB_PR_PATH_PATTERN = re.compile(r"/([^/]+)/([^/]+)/pull/(\d+)/?", re.IGNORECASE)

# Characters that could break out of a markdown [label](url) token or smuggle
# extra markdown into the rendered thread message.
_PR_URL_UNSAFE_CHARS = set(" \t\n\r()[]<>\"'`\\")

_PR_URL_MAX_LENGTH = 2048


def _is_safe_pr_url(pr_url: str) -> bool:
    """Whether ``pr_url`` is a plain http(s) URL safe to embed in a markdown link.

    ``pr_url`` originates from task-run output APIs, so it is caller-controlled.
    Real PR URLs never contain whitespace, quotes, brackets, or parentheses;
    anything that does is rejected rather than escaped.
    """
    if not pr_url or len(pr_url) > _PR_URL_MAX_LENGTH or any(char in _PR_URL_UNSAFE_CHARS for char in pr_url):
        return False
    parsed = urlparse(pr_url)
    return parsed.scheme in ("http", "https") and bool(parsed.netloc)


def _pr_display_label(pr_url: str) -> str:
    parsed = urlparse(pr_url)
    if parsed.hostname is None or parsed.hostname.lower() != "github.com":
        return pr_url
    match = _GITHUB_PR_PATH_PATTERN.fullmatch(parsed.path)
    if match:
        owner, repo, number = match.groups()
        return f"{owner}/{repo}#{number}"
    return pr_url


def post_pr_created_thread_update(run: TaskRun, pr_url: str) -> None:
    """Announce a run's freshly opened pull request in its task's thread.

    Posts "[owner/repo#N](url) has been opened" as an agent artifact message
    (``event="pr_created"``). Both the agent-output path and the GitHub webhook
    backstop can observe the same PR, so the announcement dedupes on the task's
    existing ``pr_created`` rows for this URL. Best-effort and never raises —
    recording the PR must not fail because its announcement couldn't be written.
    """
    try:
        if not _is_safe_pr_url(pr_url):
            logger.info("pr_created thread update skipped", extra={"task_id": str(run.task_id), "reason": "unsafe_url"})
            return
        # Unlike turn_complete's old channel guard, artifact rows post for
        # channel-less tasks too: every task has a thread panel.
        task = Task.objects.select_related("created_by").filter(id=run.task_id, team_id=run.team_id).first()
        if task is None:
            return
        if task.created_by is None or not _agent_thread_updates_enabled(task.created_by):
            logger.info(
                "pr_created thread update skipped",
                extra={"task_id": str(task.id), "reason": "no_creator" if task.created_by is None else "flag_off"},
            )
            return
        # The agent-output path and the webhook backstop can race on the same PR;
        # locking the task row makes the dedupe check-and-create atomic across them.
        with transaction.atomic():
            Task.objects.select_for_update().filter(id=task.id).first()
            if (
                TaskThreadMessage.objects.for_team(task.team_id)
                .filter(task_id=task.id, event="pr_created", payload__pr_url=pr_url)
                .exists()
            ):
                return
            label = _pr_display_label(pr_url)
            _create_agent_thread_message(
                task,
                f"[{label}]({pr_url}) has been opened",
                event="pr_created",
                payload={"pr_url": pr_url},
            )
    except Exception:
        logger.exception("Failed to post pr-created thread update", extra={"task_id": str(run.task_id)})