feat: integrate HITL functionality and enhance cloud command capabilities

- Added support for human-in-the-loop (HITL) interactions in cloud runs, allowing for real-time user input during task execution.
- Introduced new cloud command `respond` to submit HITL answers for ongoing runs, improving user engagement and control.
- Enhanced the `run` command to support live event streaming and clarification requests, providing a more interactive experience.
- Updated CLI components to reflect new features and improve usability, including better handling of clarification events and task statuses.
- Added new dependencies and updated configuration to support the integration of the DuckDuckGo search API.

Author: rithulkamesh

devsper/agents/agent.py

@@ -468,34 +468,124 @@ def cmd_cloud_run(args: Any) -> int:
             )
         else:
             console.print(f"[green]Queued[/green] run_id={run_id}")
+            console.print(f"Check status: [bold]devsper cloud status {run_id}[/bold]")
+            console.print(f"View logs:    [bold]devsper cloud logs {run_id}[/bold]")
         return 0
 
-    try:
-        final = api.poll_run(
-            run_id,
-            interval_seconds=float(getattr(args, "interval", 2.0)),
-            timeout_seconds=timeout_poll,
-            terminal_statuses=("completed", "failed", "cancelled", "timeout"),
-        )
-    except TimeoutError as e:
-        console.print(f"[yellow]{e}[/yellow]")
-        return 1
-    except PlatformAPIError as e:
-        console.print(f"[red]poll failed:[/red] {e}")
-        return 1
+    use_live_view = not getattr(args, "quiet", False) and not json_out and sys.stdout.isatty()
 
-    status = str(final.get("status") or "")
-    if json_out:
-        print(
-            json.dumps({"run_id": run_id, "status": status, "raw": final}, default=str)
-        )
+    if use_live_view:
+        import threading
+        import tempfile
+        import time
+
+        with tempfile.NamedTemporaryFile("w+", suffix=".jsonl", delete=False) as f:
+            log_path = f.name
+
+        stop_event = threading.Event()
+        run_status_final = {"status": "pending"}
+
+        def stream_events():
+            import httpx
+            import json
+
+            url = f"{api.base_url}/orgs/{api.org_slug}/runs/{run_id}/stream"
+            headers = api.build_headers()
+            try:
+                with httpx.stream("GET", url, headers=headers, timeout=None) as response:
+                    if response.status_code != 200:
+                        return
+                    for line in response.iter_lines():
+                        if stop_event.is_set():
+                            break
+                        if line.startswith("data: "):
+                            data = line[6:].strip()
+                            if data:
+                                try:
+                                    json.loads(data)
+                                    with open(log_path, "a", encoding="utf-8") as out_f:
+                                        out_f.write(data + "\n")
+                                except Exception:
+                                    pass
+            except Exception:
+                pass
+
+        def check_status():
+            while not stop_event.is_set():
+                try:
+                    payload = api.get_run(run_id)
+                    status = str(payload.get("status") or "")
+                    if status in ("completed", "failed", "cancelled", "timeout"):
+                        run_status_final["status"] = status
+                        return True
+                except Exception:
+                    pass
+                time.sleep(2.0)
+            return False
+
+        t_stream = threading.Thread(target=stream_events, daemon=True)
+        t_stream.start()
+
+        from devsper.cli.ui.run_view import run_live_view, print_run_summary
+
+        def submit_cloud_hitl(request_id: str, answers: dict[str, Any], skipped: bool) -> None:
+            api.submit_run_input(
+                run_id,
+                request_id=request_id,
+                answers=answers,
+                skipped=skipped,
+            )
+
+        try:
+            state = run_live_view(
+                log_path=log_path,
+                run_id=run_id,
+                worker_count=1,
+                poll_interval=0.1,
+                stop_check=check_status,
+                clarification_queue=None,
+                swarm=None,
+                cloud_hitl_submit=submit_cloud_hitl,
+            )
+        finally:
+            stop_event.set()
+            t_stream.join(1.0)
+            try:
+                os.remove(log_path)
+            except Exception:
+                pass
+
+        final = api.get_run(run_id)
+        status = str(final.get("status") or "")
+        print_run_summary(state, final.get("result") or {})
+        return 0 if status == "completed" else 1
     else:
-        console.print(f"run_id={run_id} status={status}")
-        result = final.get("result")
-        if result is not None:
-            console.print(json.dumps(result, indent=2, default=str))
+        try:
+            final = api.poll_run(
+                run_id,
+                interval_seconds=float(getattr(args, "interval", 2.0)),
+                timeout_seconds=timeout_poll,
+                terminal_statuses=("completed", "failed", "cancelled", "timeout"),
+            )
+        except TimeoutError as e:
+            console.print(f"[yellow]{e}[/yellow]")
+            return 1
+        except PlatformAPIError as e:
+            console.print(f"[red]poll failed:[/red] {e}")
+            return 1
+
+        status = str(final.get("status") or "")
+        if json_out:
+            print(
+                json.dumps({"run_id": run_id, "status": status, "raw": final}, default=str)
+            )
+        else:
+            console.print(f"run_id={run_id} status={status}")
+            result = final.get("result")
+            if result is not None:
+                console.print(json.dumps(result, indent=2, default=str))
 
-    return 0 if status == "completed" else 1
+        return 0 if status == "completed" else 1
 
 
 def cmd_cloud_status(args: Any) -> int:
@@ -530,6 +620,59 @@ def cmd_cloud_status(args: Any) -> int:
     return 0
 
 
+def cmd_cloud_respond(args: Any) -> int:
+    """POST a HITL answer for a cloud run (Redis devsper:inputs:{run_id} via platform API)."""
+    run_id = (getattr(args, "run_id", None) or "").strip()
+    if not run_id:
+        console.print("[red]run_id required.[/red]")
+        return 1
+
+    request_id = (getattr(args, "request_id", None) or "").strip()
+    if not request_id:
+        console.print("[red]--request-id is required.[/red]")
+        return 1
+
+    api = _builder_from_args(
+        getattr(args, "api_url", None),
+        getattr(args, "org", None),
+        getattr(args, "token", None),
+    )
+    if not api.enabled():
+        console.print("[red]Platform not configured.[/red]")
+        return 1
+
+    skipped = bool(getattr(args, "skipped", False))
+    answers: dict[str, Any] = {}
+    if not skipped:
+        raw = (getattr(args, "answers_json", None) or "").strip()
+        if not raw:
+            console.print("[red]Pass --answers '{\"...\": \"...\"}' or use --skipped.[/red]")
+            return 1
+        try:
+            parsed = json.loads(raw)
+            if not isinstance(parsed, dict):
+                console.print("[red]--answers must be a JSON object.[/red]")
+                return 1
+            answers = parsed
+        except json.JSONDecodeError as e:
+            console.print(f"[red]Invalid JSON in --answers:[/red] {e}")
+            return 1
+
+    try:
+        api.submit_run_input(
+            run_id,
+            request_id=request_id,
+            answers=answers,
+            skipped=skipped,
+        )
+    except PlatformAPIError as e:
+        console.print(f"[red]submit_run_input failed:[/red] {e}")
+        return 1
+
+    console.print("[green]Response sent.[/green] The run should resume if it is waiting for input.")
+    return 0
+
+
 def cmd_cloud_logs(args: Any) -> int:
     run_id = (getattr(args, "run_id", None) or "").strip()
     if not run_id:

devsper/cli/commands/cloud.py

@@ -445,10 +445,19 @@ def _run_swarm(args: object) -> int:
     summary_only = getattr(args, "summary", False)
     json_output = getattr(args, "json_output", False)
     plain = getattr(args, "plain", False) or not sys.stdout.isatty()
-    use_live_view = not quiet and not plain and sys.stdout.isatty()
+    reporter = getattr(args, "reporter", False)
+    use_live_view = not quiet and not plain and not reporter and sys.stdout.isatty()
 
     cfg = get_config()
-    event_log = EventLog(events_folder_path=cfg.events_dir)
+    from devsper.platform.redis_results_sink import build_reporter_sinks_chain
+
+    _prid = (os.environ.get("DEVSPER_PLATFORM_RUN_ID") or "").strip()
+    _chain = build_reporter_sinks_chain(cfg.events_dir)
+    event_log = EventLog(
+        events_folder_path=cfg.events_dir,
+        run_id=_prid if _prid else None,
+        platform_sink=_chain,
+    )
     log_path = getattr(event_log, "log_path", None)
     memory_router = MemoryRouter(
         store=get_default_store(),
@@ -460,7 +469,7 @@ def _run_swarm(args: object) -> int:
     )
     workers = getattr(cfg.swarm, "workers", 2)
     clarification_queue = None
-    if use_live_view:
+    if use_live_view or reporter:
         import queue
 
         clarification_queue = queue.Queue()
@@ -476,6 +485,35 @@ def _run_swarm(args: object) -> int:
     results_holder: list[dict] = []
     run_id = getattr(event_log, "run_id", "") or ""
 
+    # Headless HITL bridge: publish clarification_requested and wait for answers.
+    stop_event = None
+    bridge_thread = None
+    bridge_event_sink = None
+    if reporter and run_id:
+        try:
+            import redis
+
+            redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379")
+            r = redis.Redis.from_url(redis_url, decode_responses=True)
+
+            from devsper.platform.reporter import hitl_bridge_thread
+
+            stop_event = threading.Event()
+            # `event_log` uses `platform_sink=_chain` when reporter is enabled.
+            bridge_event_sink = _chain
+            bridge_thread = threading.Thread(
+                target=hitl_bridge_thread,
+                args=(r, run_id, clarification_queue, swarm, stop_event, bridge_event_sink),
+                daemon=True,
+            )
+            bridge_thread.start()
+        except Exception:
+            from devsper.cli.ui import console
+
+            console.print(
+                "[yellow]Reporter HITL bridge could not start; run may not pause for user input.[/yellow]"
+            )
+
     hitl_resolver = None
     if (
         getattr(getattr(cfg, "hitl", None), "enabled", False)
@@ -570,6 +608,10 @@ def _run() -> None:
             sys.stderr.write("\n")
             sys.stderr.flush()
         thread.join()
+        if stop_event is not None:
+            stop_event.set()
+        if bridge_thread is not None:
+            bridge_thread.join(timeout=1.0)
         results = results_holder[0] if results_holder else {}
         if json_output:
             import json
@@ -2725,6 +2767,7 @@ def _run_cloud_dispatch(args: object) -> int:
         cmd_cloud_run,
         cmd_cloud_logs,
         cmd_cloud_status,
+        cmd_cloud_respond,
         cmd_cloud_import_keys,
     )
 
@@ -2733,6 +2776,7 @@ def _run_cloud_dispatch(args: object) -> int:
         "logout": cmd_cloud_logout,
         "run": cmd_cloud_run,
         "status": cmd_cloud_status,
+        "respond": cmd_cloud_respond,
         "logs": cmd_cloud_logs,
         "import-keys": cmd_cloud_import_keys,
     }
@@ -2826,6 +2870,7 @@ def main() -> int:
 Examples:
   devsper run "Summarize swarm intelligence in one paragraph"
   devsper run "Analyze diffusion models" -q
+  devsper run --reporter "task"   # headless; set DEVSPER_PLATFORM_RUN_ID + REDIS_URL for live platform SSE
 """,
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
@@ -2872,6 +2917,12 @@ def main() -> int:
         default=180.0,
         help="Platform run polling timeout seconds.",
     )
+    run_parser.add_argument(
+        "--reporter",
+        action="store_true",
+        help="Headless integration: skip TUI; stream events via DEVSPER_PLATFORM_RUN_ID + REDIS_URL "
+        "and/or DEVSPER_PLATFORM_RUNTIME_EVENTS (see docs).",
+    )
     run_parser.set_defaults(func=lambda a: _run_swarm(a))
 
     meta_parser = subparsers.add_parser(
@@ -3679,6 +3730,7 @@ def main() -> int:
   devsper cloud login --api-url http://localhost:8080 --email you@example.com
   devsper cloud run "Summarize the platform README in three bullets."
   devsper cloud status <run_id>
+  devsper cloud respond <run_id> --request-id <uuid> --answers '{"Q":"A"}'
   devsper cloud logs <run_id>
 """,
         formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -3749,7 +3801,7 @@ def main() -> int:
         help="Optional x-devsper-run-manifest-version header",
     )
     cloud_run_p.add_argument(
-        "--no-wait", action="store_true", help="Print run_id only; do not poll"
+        "--no-wait", "--detach", action="store_true", help="Print run_id only; do not poll"
     )
     cloud_run_p.add_argument(
         "--timeout", type=float, default=300.0, help="Poll timeout seconds"
@@ -3777,6 +3829,34 @@ def main() -> int:
     )
     cloud_status_p.set_defaults(cloud_cmd="status")
 
+    cloud_respond_p = cloud_sub.add_parser(
+        "respond",
+        help="Send a human-in-the-loop answer for a waiting cloud run",
+    )
+    cloud_respond_p.add_argument("run_id", help="Run UUID")
+    cloud_respond_p.add_argument(
+        "--request-id",
+        required=True,
+        help="Clarification request_id from run stream or logs",
+    )
+    cloud_respond_p.add_argument(
+        "--answers",
+        dest="answers_json",
+        default="",
+        help='JSON object of answers, keys = field questions (e.g. \'{"Which API?":"REST"}\')',
+    )
+    cloud_respond_p.add_argument(
+        "--skipped",
+        action="store_true",
+        help="Tell the worker to proceed with defaults / without user answers",
+    )
+    cloud_respond_p.add_argument(
+        "--api-url", default=None, help="Override platform API URL"
+    )
+    cloud_respond_p.add_argument("--org", default=None, help="Override org slug")
+    cloud_respond_p.add_argument("--token", default=None, help="Override JWT")
+    cloud_respond_p.set_defaults(cloud_cmd="respond")
+
     cloud_logs_p = cloud_sub.add_parser("logs", help="List run events (history)")
     cloud_logs_p.add_argument("run_id", help="Run UUID")
     cloud_logs_p.add_argument(