Add Deep QA benchmark mode and Telegram intent handling

gaia/cli.py

@@ -71,7 +71,24 @@
 CONTROL_CHOICES = ("local", "telegram")
 TELEGRAM_MODE_CHOICES = ("polling", "webhook")
 TELEGRAM_SETUP_CHOICES = ("reuse", "fresh")
-TERMINAL_PURPOSE_CHOICES = ("실제 사용 모드 실행", "벤치마크 용도 실행")
+TERMINAL_ACTUAL_PURPOSE_LABEL = "실제 사용 모드 실행"
+TERMINAL_BENCHMARK_PURPOSE_LABEL = "벤치마크 용도 실행"
+TERMINAL_DEEP_QA_BENCHMARK_PURPOSE_LABEL = "Deep QA 전용 벤치마크 실행"
+TERMINAL_PURPOSE_CHOICES = (
+    TERMINAL_ACTUAL_PURPOSE_LABEL,
+    TERMINAL_BENCHMARK_PURPOSE_LABEL,
+    TERMINAL_DEEP_QA_BENCHMARK_PURPOSE_LABEL,
+)
+TERMINAL_PURPOSE_BY_LABEL = {
+    TERMINAL_ACTUAL_PURPOSE_LABEL: "actual",
+    TERMINAL_BENCHMARK_PURPOSE_LABEL: "benchmark",
+    TERMINAL_DEEP_QA_BENCHMARK_PURPOSE_LABEL: "deep_qa_benchmark",
+}
+TERMINAL_PURPOSE_PROFILE_LABEL = {
+    "actual": TERMINAL_ACTUAL_PURPOSE_LABEL,
+    "benchmark": TERMINAL_BENCHMARK_PURPOSE_LABEL,
+    "deep_qa_benchmark": TERMINAL_DEEP_QA_BENCHMARK_PURPOSE_LABEL,
+}
 PROVIDER_CHOICES = ("openai", "gemini", "ollama")
 DEFAULT_TELEGRAM_TOKEN_FILE = str(Path.home() / ".gaia" / "telegram_bot_token")
 TELEGRAM_BRIDGE_PID_FILE = Path.home() / ".gaia" / "telegram_bridge.pid"
@@ -750,17 +767,17 @@ def _resolve_terminal_launch_purpose(
         return "actual"
     if not sys.stdin.isatty():
         return "actual"
-    default = TERMINAL_PURPOSE_CHOICES[0]
-    if str(profile.get("last_terminal_purpose") or "").strip().lower() == "benchmark":
-        default = TERMINAL_PURPOSE_CHOICES[1]
+    last_purpose = str(profile.get("last_terminal_purpose") or "").strip().lower()
+    default = TERMINAL_PURPOSE_PROFILE_LABEL.get(last_purpose, TERMINAL_ACTUAL_PURPOSE_LABEL)
     selected = _prompt_select(
         "테스트 용도 인가요?",
         TERMINAL_PURPOSE_CHOICES,
         default=default,
     )
-    profile["last_terminal_purpose"] = "benchmark" if selected == TERMINAL_PURPOSE_CHOICES[1] else "actual"
+    purpose = TERMINAL_PURPOSE_BY_LABEL.get(selected, "actual")
+    profile["last_terminal_purpose"] = purpose
     _save_profile(profile)
-    return "benchmark" if selected == TERMINAL_PURPOSE_CHOICES[1] else "actual"
+    return purpose
 
 
 def _resolve_telegram_setup_strategy(parsed: argparse.Namespace, profile: dict[str, str]) -> str:
@@ -1234,7 +1251,12 @@ def _dispatch_plan(
     return run_gui(forwarded)
 
 
-def _run_terminal_benchmark_mode(*, workspace_root: Path, push_metrics: bool = False) -> int:
+def _run_terminal_benchmark_mode(
+    *,
+    workspace_root: Path,
+    push_metrics: bool = False,
+    qa_mode: str | None = None,
+) -> int:
     from gaia.src.terminal_benchmark_mode import run_terminal_benchmark_mode
 
     return run_terminal_benchmark_mode(
@@ -1244,6 +1266,7 @@ def _run_terminal_benchmark_mode(*, workspace_root: Path, push_metrics: bool = F
         prompt_non_empty=_prompt_non_empty,
         emit=print,
         push_metrics=push_metrics,
+        qa_mode=qa_mode,
     )
 
 
@@ -1593,11 +1616,14 @@ def run_launcher(argv: Sequence[str] | None = None) -> int:
     pending_user_input = dict(saved_state.pending_user_input) if saved_state else {}
     profile = _load_profile()
     terminal_purpose = _resolve_terminal_launch_purpose(args, profile, runtime=runtime)
-    if terminal_purpose == "benchmark":
-        return _run_terminal_benchmark_mode(
-            workspace_root=Path(__file__).resolve().parent.parent,
-            push_metrics=bool(getattr(args, "push_metrics", False)),
-        )
+    if terminal_purpose in {"benchmark", "deep_qa_benchmark"}:
+        benchmark_kwargs = {
+            "workspace_root": Path(__file__).resolve().parent.parent,
+            "push_metrics": bool(getattr(args, "push_metrics", False)),
+        }
+        if terminal_purpose == "deep_qa_benchmark":
+            benchmark_kwargs["qa_mode"] = DEEP_ADAPTIVE_QA_MODE
+        return _run_terminal_benchmark_mode(**benchmark_kwargs)
 
     url = _resolve_url(args, profile, required=True)
     if not url:

gaia/harness/cli_runtime.py

@@ -10,11 +10,13 @@
 from gaia.harness.registry import HarnessTask, TaskRegistry, load_builtin_registry, load_registry
 from gaia.harness.runner import (
     ARTIFACT_ROOT,
+    benchmark_mode_label,
     _grade_task_result,
     _latest_report_path,
     _summarize_grades,
     _summarize_results,
     _write_markdown,
+    normalize_harness_qa_mode,
     run_task,
 )
 
@@ -71,6 +73,12 @@ def build_harness_parser(prog: str = "gaia harness") -> argparse.ArgumentParser:
     run_parser.add_argument("--repeats", type=int, default=1, help="Run the selected tasks multiple times.")
     run_parser.add_argument("--timeout-sec", type=int, default=180)
     run_parser.add_argument("--session-prefix", default="harness")
+    run_parser.add_argument(
+        "--qa-mode",
+        choices=("off", "adaptive", "deep", "adaptive_qa", "deep_qa", "deep_adaptive_qa"),
+        default="off",
+        help="Run selected harness tasks with adaptive QA expansion.",
+    )
     run_parser.add_argument("--json", action="store_true", help="Emit JSON instead of a human-readable summary.")
 
     report_parser = subparsers.add_parser("report", help="Show a harness report.")
@@ -409,6 +417,7 @@ def _run_registry(
     timeout_sec: int = 1800,
     env: Mapping[str, str] | None = None,
     session_prefix: str = "harness",
+    qa_mode: str | None = None,
 ) -> dict[str, Any]:
     tasks = _select_tasks(
         registry,
@@ -421,6 +430,7 @@ def _run_registry(
     repeat_count = max(int(repeats), 1)
     results: list[dict[str, Any]] = []
     task_groups: dict[str, list[dict[str, Any]]] = {}
+    normalized_qa_mode = normalize_harness_qa_mode(qa_mode)
 
     ARTIFACT_ROOT.mkdir(parents=True, exist_ok=True)
     for repeat_index in range(1, repeat_count + 1):
@@ -432,6 +442,7 @@ def _run_registry(
                 timeout_sec=timeout_sec,
                 env=env,
                 session_id=session_id,
+                qa_mode=normalized_qa_mode,
             )
             grades = _grade_task_result(task, row)
             row["task_index"] = index
@@ -453,6 +464,8 @@ def _run_registry(
                 "suite_id": task.suite_id,
                 "goal": task.goal,
                 "url": task.url,
+                "qa_mode": normalized_qa_mode or "off",
+                "benchmark_mode": benchmark_mode_label(normalized_qa_mode),
                 "repeats": repeat_count,
                 "rows": task_rows,
                 "attempts": task_rows,
@@ -495,7 +508,10 @@ def _run_registry(
             "suite_ids": list(_normalize_values(suite_ids)),
             "tags": list(_normalize_values(tags)),
             "contains": list(_normalize_values(contains)),
+            "qa_mode": normalized_qa_mode or "off",
         },
+        "qa_mode": normalized_qa_mode or "off",
+        "benchmark_mode": benchmark_mode_label(normalized_qa_mode),
         "results": results,
         "tasks": task_reports,
         "summary": summary,
@@ -557,6 +573,7 @@ def run_harness_cli(argv: Sequence[str] | None = None) -> int:
             contains=getattr(parsed, "contains", None),
             timeout_sec=max(10, int(parsed.timeout_sec)),
             session_prefix=str(parsed.session_prefix or "harness"),
+            qa_mode=str(getattr(parsed, "qa_mode", "off") or "off"),
         )
         if parsed.json:
             print(json.dumps(payload, ensure_ascii=False, indent=2))

gaia/harness/runner.py

@@ -26,6 +26,28 @@
     sys.path.insert(0, str(WORKSPACE_ROOT))
 
 ARTIFACT_ROOT = WORKSPACE_ROOT / "artifacts" / "harness"
+ADAPTIVE_QA_MODE = "adaptive_qa"
+DEEP_ADAPTIVE_QA_MODE = "deep_adaptive_qa"
+
+
+def normalize_harness_qa_mode(value: str | None) -> str | None:
+    raw = str(value or "").strip().lower()
+    if raw in {"", "off", "none", "default", "false", "0"}:
+        return None
+    if raw in {"adaptive", ADAPTIVE_QA_MODE, "progressive_qa"}:
+        return ADAPTIVE_QA_MODE
+    if raw in {"deep", "deep_qa", "aggressive_qa", DEEP_ADAPTIVE_QA_MODE}:
+        return DEEP_ADAPTIVE_QA_MODE
+    return None
+
+
+def benchmark_mode_label(qa_mode: str | None) -> str:
+    normalized = normalize_harness_qa_mode(qa_mode)
+    if normalized == DEEP_ADAPTIVE_QA_MODE:
+        return "deep_qa"
+    if normalized == ADAPTIVE_QA_MODE:
+        return "adaptive_qa"
+    return "standard"
 
 
 def _normalize_status(summary: Mapping[str, Any], exit_code: int) -> str:
@@ -39,18 +61,34 @@ def _task_payload(task: HarnessTask) -> dict[str, Any]:
     return task.as_dict()
 
 
-def _build_child_code(task: Mapping[str, Any], session_id: str) -> str:
-    payload = json.dumps({"task": task, "session_id": session_id}, ensure_ascii=False)
+def _build_child_code(task: Mapping[str, Any], session_id: str, qa_mode: str | None = None) -> str:
+    payload = json.dumps(
+        {
+            "task": task,
+            "session_id": session_id,
+            "qa_mode": normalize_harness_qa_mode(qa_mode) or "",
+        },
+        ensure_ascii=False,
+    )
     return f"""
 import contextlib, io, json, os
 from gaia.terminal import _build_test_goal, run_chat_terminal_once
 payload = json.loads({payload!r})
 task = payload["task"]
 session_id = payload["session_id"]
+benchmark_qa_mode = str(payload.get("qa_mode") or "").strip()
 prepared_goal = _build_test_goal(url=task["url"], query=task["goal"])
 constraints = task.get("constraints") if isinstance(task.get("constraints"), dict) else {{}}
 expected_signals = task.get("expected_signals") if isinstance(task.get("expected_signals"), list) else []
 goal_test_data = dict(getattr(prepared_goal, "test_data", {{}}) or {{}})
+if benchmark_qa_mode:
+    goal_test_data["qa_mode"] = benchmark_qa_mode
+    if benchmark_qa_mode == "deep_adaptive_qa":
+        goal_test_data.pop("adaptive_qa", None)
+        goal_test_data["deep_adaptive_qa"] = {{"enabled": True}}
+    elif benchmark_qa_mode == "adaptive_qa":
+        goal_test_data.pop("deep_adaptive_qa", None)
+        goal_test_data["adaptive_qa"] = {{"enabled": True}}
 prepared_goal.expected_signals = [str(item) for item in expected_signals if str(item).strip()]
 if prepared_goal.expected_signals:
     goal_test_data["harness_expected_signals"] = list(prepared_goal.expected_signals)
@@ -90,13 +128,21 @@ def run_task(
     timeout_sec: int = 1800,
     env: Mapping[str, str] | None = None,
     session_id: str | None = None,
+    qa_mode: str | None = None,
 ) -> Dict[str, Any]:
     task_payload = _task_payload(task)
-    code = _build_child_code(task_payload, session_id or task.id)
+    normalized_qa_mode = normalize_harness_qa_mode(qa_mode)
+    code = _build_child_code(task_payload, session_id or task.id, qa_mode=normalized_qa_mode)
     started = time.monotonic()
     run_env = dict(os.environ)
     if env:
         run_env.update(env)
+    run_env.pop("GAIA_ADAPTIVE_QA", None)
+    run_env.pop("GAIA_DEEP_ADAPTIVE_QA", None)
+    if normalized_qa_mode == DEEP_ADAPTIVE_QA_MODE:
+        run_env["GAIA_DEEP_ADAPTIVE_QA"] = "1"
+    elif normalized_qa_mode == ADAPTIVE_QA_MODE:
+        run_env["GAIA_ADAPTIVE_QA"] = "1"
     run_env.setdefault("PYTHONUNBUFFERED", "1")
     try:
         proc = subprocess.run(
@@ -116,6 +162,8 @@ def run_task(
             "goal": task.goal,
             "url": task.url,
             "constraints": dict(task.constraints),
+            "qa_mode": normalized_qa_mode or "off",
+            "benchmark_mode": benchmark_mode_label(normalized_qa_mode),
             "status": "FAIL",
             "final_status": "FAIL",
             "reason": f"harness_timeout({timeout_sec}s)",
@@ -152,6 +200,8 @@ def run_task(
         "goal": task.goal,
         "url": task.url,
         "constraints": dict(task.constraints),
+        "qa_mode": normalized_qa_mode or "off",
+        "benchmark_mode": benchmark_mode_label(normalized_qa_mode),
         "status": status,
         "final_status": status,
         "reason": reason,
@@ -355,6 +405,8 @@ def _write_markdown(path: Path, payload: Mapping[str, Any]) -> None:
         f"- generated_at: {payload.get('generated_at')}",
         f"- task_count: {payload.get('task_count')}",
         f"- repeats: {payload.get('repeats')}",
+        f"- qa_mode: {payload.get('qa_mode', 'off')}",
+        f"- benchmark_mode: {payload.get('benchmark_mode', 'standard')}",
         "",
         "## Summary",
         "",
@@ -391,6 +443,7 @@ def run_registry(
     env: Mapping[str, str] | None = None,
     session_prefix: str = "harness",
     repeats: int = 1,
+    qa_mode: str | None = None,
 ) -> dict[str, Any]:
     tasks = _select_tasks(registry, task_id=task_id, limit=limit)
     if suite_id is not None:
@@ -400,6 +453,7 @@ def run_registry(
     ARTIFACT_ROOT.mkdir(parents=True, exist_ok=True)
     task_reports = []
     repeats = max(1, int(repeats))
+    normalized_qa_mode = normalize_harness_qa_mode(qa_mode)
     for index, task in enumerate(tasks, start=1):
         task_rows = []
         for attempt in range(1, repeats + 1):
@@ -410,6 +464,7 @@ def run_registry(
                 timeout_sec=timeout_sec,
                 env=env,
                 session_id=session_id,
+                qa_mode=normalized_qa_mode,
             )
             row["attempt"] = attempt
             row["attempt_index"] = attempt
@@ -432,6 +487,8 @@ def run_registry(
                 "suite_id": task.suite_id,
                 "goal": task.goal,
                 "url": task.url,
+                "qa_mode": normalized_qa_mode or "off",
+                "benchmark_mode": benchmark_mode_label(normalized_qa_mode),
                 "repeats": repeats,
                 "rows": task_rows,
                 "attempts": task_rows,
@@ -474,6 +531,8 @@ def run_registry(
         "registry": str(registry.source) if registry.source else None,
         "task_count": len(tasks),
         "repeats": repeats,
+        "qa_mode": normalized_qa_mode or "off",
+        "benchmark_mode": benchmark_mode_label(normalized_qa_mode),
         "results": results,
         "tasks": task_reports,
         "summary": summary,
@@ -492,6 +551,7 @@ def run_registry(
     "_latest_report_path",
     "load_builtin_registry",
     "load_registry",
+    "normalize_harness_qa_mode",
     "run_registry",
     "run_task",
 ]