feat(tasks): accept side_question commands on task runs

products/tasks/backend/presentation/serializers.py

@@ -3288,13 +3288,18 @@ class TaskRunCommandRequestSerializer(serializers.Serializer):
         "pi/rpc",
         "queue_get",
         "queue_clear",
+        "side_question",
     ]
 
     # Cap on the serialized mcp_response params (docs/CLOUD-MCP-RELAY.md): the relayed JSON-RPC
     # response payload plus envelope must fit in 300 KB. Params are forwarded to the sandbox
     # verbatim and never persisted or captured — they carry data from the user's private systems.
     MAX_MCP_RESPONSE_PARAMS_BYTES = 300_000
 
+    # A side question is forwarded into an agent turn, so its length is the direct lever on
+    # model spend. Generous for a one-shot question, far under the request body cap.
+    MAX_SIDE_QUESTION_CHARS = 10_000
+
     jsonrpc = serializers.ChoiceField(
         choices=["2.0"],
         help_text="JSON-RPC version, must be '2.0'",
@@ -3360,6 +3365,14 @@ def validate(self, attrs):
                 raise serializers.ValidationError(
                     {"params": "user_message requires a non-empty content string, artifact_ids, or both"}
                 )
+        elif method == "side_question":
+            self._require_nonempty_string(params, "question")
+            question = params["question"].strip()
+            if len(question) > self.MAX_SIDE_QUESTION_CHARS:
+                raise serializers.ValidationError(
+                    {"params": f"question must be at most {self.MAX_SIDE_QUESTION_CHARS} characters"}
+                )
+            params["question"] = question
         elif method == "pi/rpc":
             command = params.get("command")
             if not isinstance(command, dict):

products/tasks/backend/presentation/views/api.py

@@ -2339,7 +2339,7 @@ def stream_token(self, request, pk=None, **kwargs):
         summary="Send command to task run",
         description="Queue user_message JSON-RPC commands through the task workflow and forward sandbox control "
         "commands to the agent server. Supports user_message, cancel, close, permission_response, "
-        "set_config_option, mcp_response, native Pi RPC commands, and Pi queue operations.",
+        "set_config_option, mcp_response, side_question, native Pi RPC commands, and Pi queue operations.",
         strict_request_validation=True,
     )
     @action(
@@ -2373,6 +2373,14 @@ def command(self, request, pk=None, **kwargs):
         request_id = request.validated_data.get("id")
         params = request.validated_data.get("params")
 
+        # A side question drives the agent and spends model tokens on the caller's behalf. Unlike
+        # user_message below, it has no Inbox surface to exempt, so every caller takes both gates.
+        if method == "side_question":
+            if access_response := code_access_required_response(request.user):
+                return access_response
+            if limit_response := usage_limit_response(request.user, self.team_id):
+                return limit_response
+
         if method == "user_message":
             # The Inbox starts interactive runs and drops the user straight into this composer,
             # so "Discuss" and scout chat replies must not require Desktop access. Their tasks

products/tasks/backend/tests/test_api.py

@@ -10445,6 +10445,68 @@ def test_command_proxies_cancel(self, mock_post):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertTrue(response.json()["result"]["cancelled"])
 
+    @override_settings(SANDBOX_JWT_PRIVATE_KEY=TEST_RSA_PRIVATE_KEY)
+    @patch("products.tasks.backend.presentation.views.api.http_requests.post")
+    def test_command_proxies_side_question(self, mock_post):
+        reset_sandbox_jwt_key_cache()
+        self._mock_agent_response(
+            mock_post,
+            {"jsonrpc": "2.0", "id": "req-btw", "result": {"answer": "It parses JSONL."}},
+        )
+
+        task = self.create_task()
+        run = self._create_run_with_sandbox(task)
+
+        response = self.client.post(
+            self._command_url(task, run),
+            {
+                "jsonrpc": "2.0",
+                "method": "side_question",
+                "params": {"question": "  what does it do?  "},
+                "id": "req-btw",
+            },
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.json()["result"]["answer"], "It parses JSONL.")
+        forwarded = mock_post.call_args[1]["json"]
+        self.assertEqual(forwarded["method"], "side_question")
+        self.assertEqual(forwarded["params"]["question"], "what does it do?")
+
+    @patch("products.tasks.backend.presentation.views.api.http_requests.post")
+    def test_command_side_question_requires_code_access(self, mock_post):
+        self.set_tasks_feature_flag(False)
+        task = self.create_task()
+        run = self._create_run_with_sandbox(task)
+
+        response = self.client.post(
+            self._command_url(task, run),
+            {"jsonrpc": "2.0", "method": "side_question", "params": {"question": "why?"}, "id": "req-btw"},
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(response.json()["code"], "code_access_required")
+        mock_post.assert_not_called()
+
+    @patch("products.tasks.backend.presentation.views.api.http_requests.post")
+    @patch("products.tasks.backend.logic.services.code_usage_gate.get_posthog_code_usage")
+    def test_command_side_question_respects_usage_limit(self, mock_gate, mock_post):
+        mock_gate.return_value = OVER_LIMIT
+        task = self.create_task()
+        run = self._create_run_with_sandbox(task)
+
+        response = self.client.post(
+            self._command_url(task, run),
+            {"jsonrpc": "2.0", "method": "side_question", "params": {"question": "why?"}, "id": "req-btw"},
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
+        self.assertEqual(response.json()["code"], "usage_limit_exceeded")
+        mock_post.assert_not_called()
+
     def test_empty_task_session_returns_read_only_storage_access(self):
         task = self.create_task(runtime=Task.Runtime.PI)
         run = self._create_run_with_sandbox(task)
@@ -11136,6 +11198,22 @@ def test_command_fails_without_sandbox_url(self):
                     },
                 },
             ),
+            (
+                "side_question_blank_question",
+                {"jsonrpc": "2.0", "method": "side_question", "params": {"question": "   "}},
+            ),
+            (
+                "side_question_empty_question",
+                {"jsonrpc": "2.0", "method": "side_question", "params": {"question": ""}},
+            ),
+            (
+                "side_question_missing_question",
+                {"jsonrpc": "2.0", "method": "side_question", "params": {}},
+            ),
+            (
+                "side_question_oversized_question",
+                {"jsonrpc": "2.0", "method": "side_question", "params": {"question": "x" * 10_001}},
+            ),
         ]
     )
     def test_command_rejects_invalid_payloads(self, _name, payload):