Enhancements to forms shown by agent for asking Qs from User

coworker/engine.py

@@ -971,6 +971,13 @@ async def _handle_ask_user(self, tool_call: ToolCall) -> AsyncIterator[Event]:
         from the Inbox when unattended), and return it as the tool result."""
         args = tool_call.arguments or {}
         question = str(args.get("question", "")).strip()
+        # Grouped form (OPE-51): `questions` alone is a valid call — the singular field may be
+        # empty. The asker normalizes/validates the entries; here only "is anything asked?".
+        if not question:
+            for entry in args.get("questions") or []:
+                if isinstance(entry, dict) and str(entry.get("question", "")).strip():
+                    question = str(entry["question"]).strip()
+                    break
         if self.question_asker is None or not question:
             result: dict[str, Any] = {
                 "answer": "",
@@ -992,7 +999,7 @@ async def _handle_ask_user(self, tool_call: ToolCall) -> AsyncIterator[Event]:
                 "error": "no response",
             }
 
-        status = "ok" if result.get("answer") else "denied"
+        status = "ok" if (result.get("answer") or result.get("answers")) else "denied"
         self.messages.append(_tool_result_message(tool_call, result))
         self._audit(
             tool_call,

coworker/inbox.py

@@ -78,11 +78,19 @@ class InboxItem:
     tool_call_id: Optional[str] = None
     # Question metadata (ask_user): optional quick-reply choices + a free-text escape, mirroring
     # the structured-but-always-answerable shape of Claude Code's AskUserQuestion.
-    options: list[str] = field(default_factory=list)
+    # An option is a plain string OR a rich {label, description, recommended, preview} object
+    # (OPE-51); old persisted items hold strings and stay valid.
+    options: list = field(default_factory=list)
     allow_text: bool = (
         True  # accept a typed answer even when options exist (the "Other" escape)
     )
     multi: bool = False  # allow choosing more than one option
+    header: str = ""  # short chip label for the card ("Region")
+    # Grouped form (OPE-51): up to 4 {question, header, options, allow_text, multi} entries
+    # rendered as a stepper. When non-empty the singular title/options fields above still hold
+    # the FIRST question (so old surfaces and channel mirrors degrade to something sensible),
+    # and the resolution is a JSON object string keyed by header-or-question.
+    questions: list[dict] = field(default_factory=list)
     # Kind-specific payload (directory: suggested path/writable; plan: the plan text; …).
     data: dict[str, Any] = field(default_factory=dict)
 
@@ -126,6 +134,8 @@ def add(
         options=None,
         allow_text: bool = True,
         multi: bool = False,
+        header: str = "",
+        questions=None,
         tool_call_id: Optional[str] = None,
     ) -> InboxItem:
         # Idempotent by (session_id, tool_call_id): a durable resume re-raises the same prompt, and
@@ -146,6 +156,8 @@ def add(
             options=list(options or []),
             allow_text=bool(allow_text),
             multi=bool(multi),
+            header=str(header or ""),
+            questions=list(questions or []),
             tool_call_id=tool_call_id,
         )
         with self._lock:
@@ -194,6 +206,8 @@ def add_question(
         options=None,
         allow_text=True,
         multi=False,
+        header="",
+        questions=None,
         tool_call_id=None,
     ) -> InboxItem:
         return self.add(
@@ -206,6 +220,8 @@ def add_question(
             options=options,
             allow_text=allow_text,
             multi=multi,
+            header=header,
+            questions=questions,
             tool_call_id=tool_call_id,
         )
 

coworker/interactions.py

@@ -17,6 +17,7 @@
 from typing import Optional
 
 from .inbox import KIND_APPROVAL, KIND_QUESTION
+from .tools.ask import option_label
 
 
 @dataclass
@@ -48,7 +49,15 @@ def buttons_for(item) -> list[Button]:
             Button("Approve", encode(item.id, "allow")),
             Button("Deny", encode(item.id, "deny")),
         ]
+    if item.kind == KIND_QUESTION and getattr(item, "questions", None):
+        # Grouped questions (OPE-51): one button row can't answer 2+ questions — send plain text
+        # with the open-the-app hint instead.
+        return []
     if item.kind == KIND_QUESTION and getattr(item, "options", None):
-        # One button per option; the resolution IS the chosen option text (what the agent gets).
-        return [Button(opt, encode(item.id, opt)) for opt in item.options]
+        # One button per option; the resolution IS the chosen option's label (what the agent
+        # gets). Rich {label, description, …} options button as their label.
+        return [
+            Button(option_label(opt), encode(item.id, option_label(opt)))
+            for opt in item.options
+        ]
     return []

coworker/server/app.py

@@ -1603,15 +1603,17 @@ async def approver(_request) -> ApprovalOutcome:
 
         async def question_asker(args: dict, tool_call_id=None) -> dict:
             # ask_user (engine does NOT emit the event — we do, only when attended).
+            from ..tools.ask import answer_result, question_item_fields
+
+            fields = question_item_fields(args)
+            if fields is None:  # engine guards too; belt-and-braces
+                return {"answer": "", "error": "no question"}
             item = manager.inbox.add_question(
                 session_id,
-                str(args.get("question", "")),
                 inbox=_route(),
                 visibility=_visibility(),
-                options=list(args.get("options") or []),
-                allow_text=bool(args.get("allow_text", True)),
-                multi=bool(args.get("multi", False)),
                 tool_call_id=tool_call_id,
+                **fields,
             )
             if item.state == "pending":
                 manager.persist_session(session_id)
@@ -1626,11 +1628,12 @@ async def question_asker(args: dict, tool_call_id=None) -> dict:
                                 "options": item.options,
                                 "allow_text": item.allow_text,
                                 "multi": item.multi,
-                                "header": str(args.get("header", "")),
+                                "header": item.header,
+                                "questions": item.questions,
                             },
                         }
                     )
-            return {"answer": await manager.inbox.wait(item.id)}
+            return answer_result(item.questions, await manager.inbox.wait(item.id))
 
         async def directory_requester(args: dict, tool_call_id=None) -> dict:
             # The engine has already emitted DIRECTORY_REQUESTED. Park, await, then apply the grant.

coworker/server/manager.py

@@ -751,27 +751,26 @@ def inbox_question_asker(self, session_id: str, agent: str):
         async def ask(
             args: dict[str, Any], tool_call_id: Optional[str] = None
         ) -> dict[str, Any]:
-            question = str(args.get("question", "")).strip()
-            if not question:
+            from ..tools.ask import answer_result, question_item_fields
+
+            fields = question_item_fields(args)
+            if fields is None:
                 return {"answer": "", "error": "no question"}
             inbox_name = self.inbox_routing.route_for(session_id, agent)
             item = self.inbox.add_question(
                 session_id,
-                title=question,
                 inbox=inbox_name,
-                options=list(args.get("options") or []),
-                allow_text=bool(args.get("allow_text", True)),
-                multi=bool(args.get("multi", False)),
                 tool_call_id=tool_call_id,
+                **fields,
             )
             if (
                 item.state != "pending"
             ):  # durable resume re-raised an already-answered prompt
-                return {"answer": item.resolution or ""}
+                return answer_result(item.questions, item.resolution)
             self.persist_session(session_id)  # the pending tool call is now on disk
             await self.mirror_inbox_item(item)
             answer = await self.inbox.wait(item.id)
-            return {"answer": answer}
+            return answer_result(item.questions, answer)
 
         return ask