feat(tasks): schedule one-off runs through API and MCP

products/tasks/backend/facade/api.py

@@ -601,6 +601,7 @@
         updated_at=run.updated_at,
         completed_at=run.completed_at,
         preview_available=task_run_preview_ready(run.state),
+        scheduled_at=run.scheduled_at,
     )
 
 
@@ -2803,7 +2804,7 @@
     )
 
     run = TaskRun.objects.filter(pk=run_id).first()
-    if run is None:
+    if run is None or (run.scheduled_at is not None and run.queued_at is None):
         return
     try:
         client = sync_connect()
@@ -3015,13 +3016,14 @@
         )
 
 
 def update_task_run(
     run_id: str | UUID,
     task_id: str | UUID,
     team_id: int,
     *,
     validated_data: dict,
     only_if_non_terminal: bool = False,
+    only_if_not_started: bool = False,
     caller_is_agent: bool = False,
 ) -> contracts.TaskRunDetailDTO | None:
     """Apply a PATCH to a run: merge output/state, set completion, then dispatch side effects.
@@ -3083,8 +3085,16 @@
     update_fields: set[str] = set()
 
     with transaction.atomic():
-        if has_output_merge or has_state_mutation or only_if_non_terminal or "status" in validated_data:
+        if (
+            has_output_merge
+            or has_state_mutation
+            or only_if_non_terminal
+            or only_if_not_started
+            or "status" in validated_data
+        ):
             run = TaskRun.objects.select_for_update().get(pk=run.pk)
+        if only_if_not_started and run.status != TaskRun.Status.NOT_STARTED:
+            return None
         if only_if_non_terminal and run.is_terminal:
             if validated_data.get("status") == run.status:
                 transaction.on_commit(lambda: resume_workflow_step_for_run(run))
@@ -3140,6 +3150,10 @@
             update_fields.add("state")
 
         new_status = validated_data.get("status")
+        if only_if_not_started and new_status == TaskRun.Status.CANCELLED:
+            # A dormant run has no workflow to complete its stream after cancellation.
+            run.state = {**(run.state or {}), "cancel_fallback_cleanup_complete": True}
+            update_fields.add("state")
         if (
             caller_is_agent
             and new_status == TaskRun.Status.COMPLETED
@@ -3947,7 +3961,7 @@
     return prepared, True
 
 
 def finalize_task_run_artifact_uploads(
     run_id: str | UUID,
     task_id: str | UUID,
     team_id: int,
@@ -4507,7 +4521,7 @@
     return missing_artifact_ids, True
 
 
 def signal_task_run_user_message(
     run_id: str | UUID,
     task_id: str | UUID,
     team_id: int,
@@ -4915,7 +4929,7 @@
     return bool(sandbox_id) and preview.get("sandbox_id") == sandbox_id
 
 
 def resolve_task_run_preview_redirect(
     run_id: str | UUID, task_id: str | UUID, team_id: int, *, user_id: int
 ) -> TaskRunPreviewRedirect | None:
     from products.tasks.backend.exceptions import (
@@ -5235,7 +5249,7 @@
     }
 
 
 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.
@@ -5417,16 +5431,15 @@
         enqueue_or_start_workflow,
     )
     from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
-        RunSource,
+        mcp_scopes_for_run_source,
         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
-    posthog_mcp_scopes: Literal["read_only", "full"] = "full" if run_source in full_mcp_run_sources else "read_only"
+    posthog_mcp_scopes = mcp_scopes_for_run_source(run_source)
     try:
         logger.info("Attempting to trigger task processing workflow for task %s, run %s", task.id, run.id)
         message = None
@@ -6518,6 +6531,7 @@
     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)
+    validated_data.pop("scheduled_at", None)
     # Names the task from the pasted content while `description` stays the bare prompt. Write-only,
     # never persisted, so it must be popped before `Task.objects.create(**validated_data)`.
     naming_source = (validated_data.pop("naming_source", None) or "").strip() or None
@@ -7394,12 +7408,16 @@
         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)
+        # Scheduled attachments are tagged before the run-creation transaction takes its locks.
+        if run.scheduled_at is None:
+            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]
-    )
+    cache_keys = [build_task_staged_artifact_cache_key(str(task.id), artifact_id) for artifact_id in artifact_ids]
+    if run.scheduled_at is not None:
+        transaction.on_commit(lambda: get_tasks_cache().delete_many(cache_keys), robust=True)
+    else:
+        get_tasks_cache().delete_many(cache_keys)
 
 
 REPORT_WARM_RUN_NOT_ACTIVATED = "This sandbox is waiting for the report's Ask AI question. Send it from the report."
@@ -7998,15 +8016,19 @@
         is_report_implementation_task,
     )
     from products.tasks.backend.logic.services.staged_artifacts import (  # noqa: PLC0415
+        RUN_ARTIFACT_TTL_DAYS,
         get_task_run_artifacts_by_id,
         get_task_staged_artifacts,
+        staged_artifacts_expire_by,
+        tag_task_artifact,
     )
     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,
+        mcp_scopes_for_run_source,
         parse_run_state,
     )
 
@@ -8049,6 +8071,7 @@
             )
     mode = validated_data.get("mode", "background")
     run_source = validated_data.get("run_source")
+    scheduled_at = validated_data.get("scheduled_at")
     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")
@@ -8113,7 +8136,16 @@
     if claude_model_access is None and previous_state is not None:
         claude_model_access = previous_state.claude_model_access
 
-    warm_run = None if run_source == RunSource.AGENT else _idling_warm_run_for_task(task)
+    if scheduled_at is not None and claude_model_access == "own-subscription":
+        return contracts.TaskRunResult(
+            error=contracts.TaskValidationError(
+                kind="validation_error",
+                code="invalid_input",
+                detail="Scheduled runs must use the PostHog gateway.",
+                attr="claude_model_access",
+            )
+        )
+    warm_run = None if scheduled_at is not None or run_source == RunSource.AGENT else _idling_warm_run_for_task(task)
     if warm_run is not None and claude_model_access == "own-subscription":
         warm_run = None
     if warm_run is not None:
@@ -8433,14 +8465,43 @@
                 )
             )
 
+    if scheduled_at is not None and staged_artifacts_expire_by(staged_artifacts, scheduled_at):
+        return contracts.TaskRunResult(
+            error=contracts.TaskValidationError(
+                kind="validation_error",
+                code="invalid_input",
+                detail="The attached files expire before this run can start. Choose an earlier time or upload new files.",
+                attr="scheduled_at",
+            )
+        )
+
     logger.info("Creating task run for task %s with mode=%s, branch=%s", task.id, mode, branch)
+    if scheduled_at is not None:
+        for staged_artifact in staged_artifacts:
+            tag_task_artifact(
+                str(staged_artifact["storage_path"]),
+                ttl_days=RUN_ARTIFACT_TTL_DAYS,
+                team_id=task.team_id,
+                raise_on_error=True,
+            )
+        extra_state["pending_dispatch"] = {
+            "user_id": user_id,
+            "create_pr": True,
+            "posthog_mcp_scopes": mcp_scopes_for_run_source(run_source),
+        }
     try:
         with transaction.atomic():
-            task_run = task.create_run(mode=mode, branch=branch, extra_state=extra_state, acting_user_id=user_id)
+            task_run = task.create_run(
+                mode=mode, branch=branch, extra_state=extra_state, acting_user_id=user_id, scheduled_at=scheduled_at
+            )
             if report_id_for_slot_check is not None:
                 enforce_report_implementation_rerun_cap(
                     team_id=team_id, report_id=report_id_for_slot_check, task_id=str(task.id)
                 )
+            if scheduled_at is not None and pending_user_artifact_ids:
+                _attach_staged_artifacts_to_run(
+                    task_run, task, staged_artifacts=staged_artifacts, artifact_ids=pending_user_artifact_ids
+                )
     except InvalidTaskOriginError as error:
         return contracts.TaskRunResult(
             error=contracts.TaskValidationError(
@@ -8465,14 +8526,19 @@
             update_fields.append("relayed_mcp_servers")
         task_run.save(update_fields=update_fields)
 
-    if pending_user_artifact_ids:
+    if pending_user_artifact_ids and scheduled_at is None:
         _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)
 
+    if scheduled_at is not None:
+        return contracts.TaskRunResult(
+            task=_task_detail_to_dto(task, latest_run=task_run, include_latest_run_log_url=False)
+        )
+
     logger.info("Triggering workflow for task %s, run %s", task.id, task_run.id)
     if is_pi_task:
         initial_message = (

products/tasks/backend/facade/cancellation.py

@@ -88,7 +88,7 @@ def _publish_cancel_fallback_completion(run: TaskRun) -> bool:
         publish_run_stream_completion(str(run.id))
         TaskRun.update_state_atomic(run.id, updates={"cancel_fallback_cleanup_complete": False})
     except Exception:
-        logger.warning("Failed to complete stream for workflow-gone run %s", run.id, exc_info=True)
+        logger.warning("Failed to complete stream for cancelled run %s", run.id, exc_info=True)
         return False
     return True
 
@@ -127,6 +127,18 @@ def cancel_task_run(
     status_at_request = run.status
     error_message = (reason or "").strip()[:500] or "Stopped by user"
 
+    def capture_cancel_request(task_run: TaskRun, signal_outcome: str) -> None:
+        task_run.capture_event(
+            "task_run_cancel_requested",
+            {
+                "cancel_source": source,
+                "cancel_reason": error_message,
+                "requested_by_user_id": requested_by_user_id,
+                "workflow_signal_outcome": signal_outcome,
+                "status_at_request": status_at_request,
+            },
+        )
+
     marker: dict[str, Any] = {
         "cancel_requested_at": django_timezone.now().isoformat(),
         "cancel_source": source,
@@ -154,6 +166,28 @@ def cancel_task_run(
         if only_if_awaiting_first_message:
             return "unavailable", tasks_api._task_run_detail_to_dto(run)
 
+    if run.scheduled_at is not None and run.status == TaskRun.Status.NOT_STARTED:
+        dto = tasks_api.update_task_run(
+            run.id,
+            task_id,
+            team_id,
+            validated_data={"status": TaskRun.Status.CANCELLED, "error_message": error_message},
+            only_if_not_started=True,
+        )
+        if dto is not None:
+            capture_cancel_request(run, "not_needed")
+            run.refresh_from_db(fields=["state"])
+            if not _publish_cancel_fallback_completion(run):
+                return "unavailable", dto
+            return "accepted", dto
+        run = tasks_api._get_visible_run(run_id, task_id, team_id)
+        if run is None:
+            return "not_found", None
+        if run.is_terminal:
+            if not _publish_cancel_fallback_completion(run):
+                return "unavailable", tasks_api._task_run_detail_to_dto(run)
+            return "already_terminal", tasks_api._task_run_detail_to_dto(run)
+
     _interrupt_agent_turn(run, requested_by_user_id, requested_by_distinct_id)
 
     signal_outcome = _signal_complete_task(run, TaskRun.Status.CANCELLED, error_message)
@@ -185,14 +219,5 @@ def cancel_task_run(
         push_dispatcher.notify_task_run_cancelled(run)
         dto = tasks_api._task_run_detail_to_dto(run)
 
-    run.capture_event(
-        "task_run_cancel_requested",
-        {
-            "cancel_source": source,
-            "cancel_reason": error_message,
-            "requested_by_user_id": requested_by_user_id,
-            "workflow_signal_outcome": signal_outcome,
-            "status_at_request": status_at_request,
-        },
-    )
+    capture_cancel_request(run, signal_outcome)
     return "accepted", dto

products/tasks/backend/facade/contracts.py

@@ -598,6 +598,7 @@ class TaskRunDetailDTO:
     updated_at: datetime | None = None
     completed_at: datetime | None = None
     preview_available: bool = False
+    scheduled_at: datetime | None = None
 
 
 @dataclass(frozen=True)

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

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+from datetime import datetime, timedelta
 from typing import Any
 
 from django.conf import settings
@@ -122,7 +123,20 @@ def get_task_run_artifacts_by_id(task_run: TaskRun, artifact_ids: list[str]) ->
     return resolved_artifacts, missing_ids
 
 
-def tag_task_artifact(storage_path: str, *, ttl_days: str, team_id: int) -> None:
+def staged_artifacts_expire_by(artifacts: list[dict], scheduled_at: datetime) -> bool:
+    for artifact in artifacts:
+        stored_object = object_storage.head_object_strict(str(artifact["storage_path"]))
+        uploaded_at = stored_object.get("LastModified") if stored_object else None
+        if (
+            not isinstance(uploaded_at, datetime)
+            or timezone.is_naive(uploaded_at)
+            or scheduled_at >= uploaded_at + timedelta(days=int(RUN_ARTIFACT_TTL_DAYS))
+        ):
+            return True
+    return False
+
+
+def tag_task_artifact(storage_path: str, *, ttl_days: str, team_id: int, raise_on_error: bool = False) -> None:
     try:
         object_storage.tag(
             storage_path,
@@ -139,3 +153,5 @@ def tag_task_artifact(storage_path: str, *, ttl_days: str, team_id: int) -> None
             team_id=team_id,
             error=str(exc),
         )
+        if raise_on_error:
+            raise