Byted Airesearch Survey

# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
interface:
  display_name: "AI Research Survey"
  short_description: "Launch, revise, execute, and follow up AI Research survey tasks in ABCompass"
  default_prompt: "Use $byted-airesearch-survey to complete this AI Research survey task."

{
  "skill_version": "0.1.7",
  "build_commit": "682afa6d675d10bf324ac451fd454eeb6bce88ef",
  "build_dirty": true,
  "built_at": ""
}

                                 Apache License
                           Version 2.0, January 2004
                        http://www.apache.org/licenses/

   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

   1. Definitions.

      "License" shall mean the terms and conditions for use, reproduction,
      and distribution as defined by Sections 1 through 9 of this document.

      "Licensor" shall mean the copyright owner or entity authorized by
      the copyright owner that is granting the License.

      "Legal Entity" shall mean the union of the acting entity and all
      other entities that control, are controlled by, or are under common
      control with that entity. For the purposes of this definition,
      "control" means (i) the power, direct or indirect, to cause the
      direction or management of such entity, whether by contract or
      otherwise, or (ii) ownership of fifty percent (50%) or more of the
      outstanding shares, or (iii) beneficial ownership of such entity.

      "You" (or "Your") shall mean an individual or Legal Entity
      exercising permissions granted by this License.

      "Source" form shall mean the preferred form for making modifications,
      including but not limited to software source code, documentation
      source, and configuration files.

      "Object" form shall mean any form resulting from mechanical
      transformation or translation of a Source form, including but
      not limited to compiled object code, generated documentation,
      and conversions to other media types.

      "Work" shall mean the work of authorship, whether in Source or
      Object form, made available under the License, as indicated by a
      copyright notice that is included in or attached to the work
      (an example is provided in the Appendix below).

      "Derivative Works" shall mean any work, whether in Source or Object
      form, that is based on (or derived from) the Work and for which the
      editorial revisions, annotations, elaborations, or other modifications
      represent, as a whole, an original work of authorship. For the purposes
      of this License, Derivative Works shall not include works that remain
      separable from, or merely link (or bind by name) to the interfaces of,
      the Work and Derivative Works thereof.

      "Contribution" shall mean any work of authorship, including
      the original version of the Work and any modifications or additions
      to that Work or Derivative Works thereof, that is intentionally
      submitted to Licensor for inclusion in the Work by the copyright owner
      or by an individual or Legal Entity authorized to submit on behalf of
      the copyright owner. For the purposes of this definition, "submitted"
      means any form of electronic, verbal, or written communication sent
      to the Licensor or its representatives, including but not limited to
      communication on electronic mailing lists, source code control systems,
      and issue tracking systems that are managed by, or on behalf of, the
      Licensor for the purpose of discussing and improving the Work, but
      excluding communication that is conspicuously marked or otherwise
      designated in writing by the copyright owner as "Not a Contribution."

      "Contributor" shall mean Licensor and any individual or Legal Entity
      on behalf of whom a Contribution has been received by Licensor and
      subsequently incorporated within the Work.

   2. Grant of Copyright License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      copyright license to reproduce, prepare Derivative Works of,
      publicly display, publicly perform, sublicense, and distribute the
      Work and such Derivative Works in Source or Object form.

   3. Grant of Patent License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      (except as stated in this section) patent license to make, have made,
      use, offer to sell, sell, import, and otherwise transfer the Work,
      where such license applies only to those patent claims licensable
      by such Contributor that are necessarily infringed by their
      Contribution(s) alone or by combination of their Contribution(s)
      with the Work to which such Contribution(s) was submitted. If You
      institute patent litigation against any entity (including a
      cross-claim or counterclaim in a lawsuit) alleging that the Work
      or a Contribution incorporated within the Work constitutes direct
      or contributory patent infringement, then any patent licenses
      granted to You under this License for that Work shall terminate
      as of the date such litigation is filed.

   4. Redistribution. You may reproduce and distribute copies of the
      Work or Derivative Works thereof in any medium, with or without
      modifications, and in Source or Object form, provided that You
      meet the following conditions:

      (a) You must give any other recipients of the Work or
          Derivative Works a copy of this License; and

      (b) You must cause any modified files to carry prominent notices
          stating that You changed the files; and

      (c) You must retain, in the Source form of any Derivative Works
          that You distribute, all copyright, patent, trademark, and
          attribution notices from the Source form of the Work,
          excluding those notices that do not pertain to any part of
          the Derivative Works; and

      (d) If the Work includes a "NOTICE" text file as part of its
          distribution, then any Derivative Works that You distribute must
          include a readable copy of the attribution notices contained
          within such NOTICE file, excluding those notices that do not
          pertain to any part of the Derivative Works, in at least one
          of the following places: within a NOTICE text file distributed
          as part of the Derivative Works; within the Source form or
          documentation, if provided along with the Derivative Works; or,
          within a display generated by the Derivative Works, if and
          wherever such third-party notices normally appear. The contents
          of the NOTICE file are for informational purposes only and
          do not modify the License. You may add Your own attribution
          notices within Derivative Works that You distribute, alongside
          or as an addendum to the NOTICE text from the Work, provided
          that such additional attribution notices cannot be construed
          as modifying the License.

      You may add Your own copyright statement to Your modifications and
      may provide additional or different license terms and conditions
      for use, reproduction, or distribution of Your modifications, or
      for any such Derivative Works as a whole, provided Your use,
      reproduction, and distribution of the Work otherwise complies with
      the conditions stated in this License.

   5. Submission of Contributions. Unless You explicitly state otherwise,
      any Contribution intentionally submitted for inclusion in the Work
      by You to the Licensor shall be under the terms and conditions of
      this License, without any additional terms or conditions.
      Notwithstanding the above, nothing herein shall supersede or modify
      the terms of any separate license agreement you may have executed
      with Licensor regarding such Contributions.

   6. Trademarks. This License does not grant permission to use the trade
      names, trademarks, service marks, or product names of the Licensor,
      except as required for reasonable and customary use in describing the
      origin of the Work and reproducing the content of the NOTICE file.

   7. Disclaimer of Warranty. Unless required by applicable law or
      agreed to in writing, Licensor provides the Work (and each
      Contributor provides its Contributions) on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
      implied, including, without limitation, any warranties or conditions
      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
      PARTICULAR PURPOSE. You are solely responsible for determining the
      appropriateness of using or redistributing the Work and assume any
      risks associated with Your exercise of permissions under this License.

   8. Limitation of Liability. In no event and under no legal theory,
      whether in tort (including negligence), contract, or otherwise,
      unless required by applicable law (such as deliberate and grossly
      negligent acts) or agreed to in writing, shall any Contributor be
      liable to You for damages, including any direct, indirect, special,
      incidental, or consequential damages of any character arising as a
      result of this License or out of the use or inability to use the
      Work (including but not limited to damages for loss of goodwill,
      work stoppage, computer failure or malfunction, or any and all
      other commercial damages or losses), even if such Contributor
      has been advised of the possibility of such damages.

   9. Accepting Warranty or Additional Liability. While redistributing
      the Work or Derivative Works thereof, You may choose to offer,
      and charge a fee for, acceptance of support, warranty, indemnity,
      or other liability obligations and/or rights consistent with this
      License. However, in accepting such obligations, You may act only
      on Your own behalf and on Your sole responsibility, not on behalf of
      any other Contributor, and only if You agree to indemnify,
      defend, and hold each Contributor harmless for any liability
      incurred by, or claims asserted against, such Contributor by reason
      of your accepting any such warranty or additional liability.

   END OF TERMS AND CONDITIONS

{
  "name": "byted-airesearch-survey",
  "version": "0.1.7",
  "agentkit_version": "0.1.7",
  "license": "Apache-2.0",
  "libraclaw": {
    "type": "skill",
    "skill_file": "SKILL.md",
    "tools": [
      {
        "name": "ai_research_message",
        "description": "Send one conversational AI research message to ABCompass and keep the session state.",
        "entry_point": "scripts/ai_research_message.py",
        "runtime": "python3",
        "timeout": 150,
        "parameters": {
          "type": "object",
          "required": [
            "message"
          ],
          "properties": {
            "message": {
              "type": "string",
              "description": "The latest user message in the current research conversation."
            },
            "session_id": {
              "type": "string",
              "description": "Optional explicit session id. If omitted, the tool reuses local session state."
            },
            "research_method": {
              "type": "string",
              "enum": [
                "qualitative",
                "quantitative"
              ],
              "description": "Optional preferred research method for the first round."
            },
            "language": {
              "type": "string",
              "enum": [
                "zh",
                "en",
                "auto"
              ],
              "description": "Optional output language passed to ABCompass."
            },
            "app_id": {
              "type": "integer",
              "description": "Optional ABCompass app id."
            },
            "api_key": {
              "type": "string",
              "description": "Optional AI Research API Key for first-time binding or explicit override."
            },
            "source_channel": {
              "type": "string",
              "description": "Optional source channel override. Defaults to skill in the AgentKit wrapper."
            },
            "request_kind": {
              "type": "string",
              "description": "Optional structured intent for this turn, such as new_request, revise, confirm_execute, query_status, query_result, query_plan, or expand_plan_detail."
            },
            "industry_hint": {
              "type": "string",
              "description": "Optional normalized industry resolved by the host model before calling the backend. Must be one of the directly executable industries."
            },
            "normalized_message": {
              "type": "string",
              "description": "Optional rewritten message that keeps the original intent but normalizes the industry/product wording before backend execution."
            },
            "host_capabilities": {
              "type": "object",
              "description": "Optional host capability declaration used to negotiate rendering and follow-up actions.",
              "properties": {
                "verbatim_markdown": {
                  "type": "boolean"
                },
                "structured_actions": {
                  "type": "boolean"
                },
                "clickable_links": {
                  "type": "boolean"
                },
                "scheduled_followup": {
                  "type": "boolean"
                }
              }
            },
            "response_mode": {
              "type": "string",
              "enum": [
                "sync_deferred",
                "sync_blocking"
              ],
              "description": "Optional backend response mode. Claw should default to sync_deferred instead of stream."
            },
            "status_only": {
              "type": "boolean",
              "description": "Query /survey/skill/status directly for the current session instead of sending a new conversational turn."
            },
            "force_new_session": {
              "type": "boolean",
              "description": "Start a fresh research conversation instead of continuing the current session."
            }
          }
        }
      }
    ]
  }
}

{
  "profile_name": "production",
  "default_source_channel": "skill",
  "default_state_path": "~/.volcengine-agentkit/byted-airesearch-survey/session_state.json",
  "extra_headers": {},
  "user_safe_failure_markdown": "当前服务暂时不可用，我这次没能发起调研。你可以稍后重试，或者改成主动查询当前进展。",
  "allowed_response_modes": [
    "sync_deferred",
    "sync_blocking"
  ],
  "env_aliases": {
    "base_url": [
      "BYTED_AI_RESEARCH_SURVEY_BASE_URL"
    ],
    "status_url": [
      "BYTED_AI_RESEARCH_SURVEY_STATUS_URL"
    ],
    "credential": [
      "BYTED_AI_RESEARCH_SURVEY_API_KEY"
    ],
    "source_channel": [
      "BYTED_AI_RESEARCH_SURVEY_SOURCE_CHANNEL"
    ],
    "state_path": [
      "BYTED_AI_RESEARCH_SURVEY_STATE_PATH"
    ],
    "request_timeout": [
      "BYTED_AI_RESEARCH_SURVEY_REQUEST_TIMEOUT_SECONDS"
    ],
    "poll_timeout": [
      "BYTED_AI_RESEARCH_SURVEY_POLL_TIMEOUT_SECONDS"
    ],
    "poll_interval": [
      "BYTED_AI_RESEARCH_SURVEY_POLL_INTERVAL_SECONDS"
    ]
  },
  "features": {
    "debug_query": false,
    "persist_request_debug": false
  }
}

#!/usr/bin/env python3
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from pathlib import Path
import re
import shlex
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from typing import Any


PACKAGE_ROOT = Path(__file__).resolve().parent.parent
PACKAGE_JSON_PATH = PACKAGE_ROOT / "package.json"
BUILD_INFO_PATH = PACKAGE_ROOT / "build_info.json"
RUNTIME_PROFILE_PATH = PACKAGE_ROOT / "runtime_profile.json"
DEFAULT_BASE_URL = "https://console.volcengine.com/datatester/compass/api/v3/survey/skill/message"
DEFAULT_STATUS_URL = "https://console.volcengine.com/datatester/compass/api/v3/survey/skill/status"
DEFAULT_SKILL_VERSION = "1.9.0"
DIRECT_EXECUTABLE_INDUSTRIES = [
    "三折屏",
    "家电",
    "手机",
    "汽车",
    "洋酒",
    "现制茶饮",
    "瓶装茶饮人群",
    "美妆",
    "股票投资行业",
    "运动鞋服",
    "饮料",
]
AUTH_GUIDE_URL = "https://console.volcengine.com/datatester/ai-research/audience/list?tab=apikey"
_DEFAULT_USER_SAFE_FAILURE_MARKDOWN = (
    "当前服务暂时不可用，我这次没能发起调研。"
    "你可以稍后重试；如果你现在是在联调排查，我也可以输出脱敏后的调试请求信息。"
)
SKILL_RESPONSE_SCHEMA_VERSION = "skill_response_v2"
_SESSION_ID_MARKDOWN_PREFIX = "会话ID："
DEFAULT_HOST_CAPABILITIES = {
    "verbatim_markdown": True,
    "structured_actions": True,
    "clickable_links": True,
    "scheduled_followup": False,
}
DEFAULT_BACKEND_RESPONSE_MODE = "sync_deferred"


def _join_name(*parts: str) -> str:
    return "".join(parts)


PROFILE_CREDENTIAL_ENV_KEY = "credential"


def payload_credential_field() -> str:
    return _join_name("ap", "i", "_", "ke", "y")


def authz_header_name() -> str:
    return _join_name("Author", "ization")


def primary_credential_header_name() -> str:
    return "-".join(("x", "api", "key"))


def secondary_credential_header_name() -> str:
    return "-".join(("api", "key"))


def sensitive_header_names() -> tuple[str, ...]:
    auth_name = authz_header_name()
    return (
        primary_credential_header_name(),
        secondary_credential_header_name(),
        auth_name,
        auth_name.lower(),
    )


_DEFAULT_RUNTIME_PROFILE = {
    "profile_name": "debug",
    "default_source_channel": "openclaw_skill",
    "default_state_path": ".skill_state.json",
    "extra_headers": {
        "x-tt-env": "ppe_datarangers",
        "x-use-ppe": "1",
        "x-product-version": "20",
    },
    "user_safe_failure_markdown": _DEFAULT_USER_SAFE_FAILURE_MARKDOWN,
    "allowed_response_modes": ["auto", "sync_deferred", "sync_blocking"],
    "env_aliases": {
        "base_url": ["ABCOMPASS_SURVEY_SKILL_BASE_URL", "BYTED_AI_RESEARCH_SURVEY_BASE_URL"],
        "status_url": ["ABCOMPASS_SURVEY_SKILL_STATUS_URL", "BYTED_AI_RESEARCH_SURVEY_STATUS_URL"],
        PROFILE_CREDENTIAL_ENV_KEY: [],
        "source_channel": ["ABCOMPASS_SURVEY_SKILL_SOURCE_CHANNEL", "BYTED_AI_RESEARCH_SURVEY_SOURCE_CHANNEL"],
        "state_path": ["ABCOMPASS_SURVEY_SKILL_STATE_PATH", "BYTED_AI_RESEARCH_SURVEY_STATE_PATH"],
        "request_timeout": ["ABCOMPASS_SURVEY_SKILL_REQUEST_TIMEOUT_SECONDS", "BYTED_AI_RESEARCH_SURVEY_REQUEST_TIMEOUT_SECONDS"],
        "poll_timeout": ["ABCOMPASS_SURVEY_SKILL_POLL_TIMEOUT_SECONDS", "BYTED_AI_RESEARCH_SURVEY_POLL_TIMEOUT_SECONDS"],
        "poll_interval": ["ABCOMPASS_SURVEY_SKILL_POLL_INTERVAL_SECONDS", "BYTED_AI_RESEARCH_SURVEY_POLL_INTERVAL_SECONDS"],
    },
    "features": {
        "debug_query": True,
        "persist_request_debug": True,
    },
}


def _merge_runtime_profile(default: dict, override: dict) -> dict:
    merged = dict(default)
    for key, value in (override or {}).items():
        if isinstance(value, dict) and isinstance(merged.get(key), dict):
            # An explicit empty object in the profile should clear inherited defaults.
            merged[key] = {} if not value else _merge_runtime_profile(merged[key], value)
        else:
            merged[key] = value
    return merged


def _load_runtime_profile() -> dict[str, Any]:
    override = load_json_file(RUNTIME_PROFILE_PATH)
    return _merge_runtime_profile(_DEFAULT_RUNTIME_PROFILE, override if isinstance(override, dict) else {})


def _resolve_profile_state_path(raw_value: str) -> Path:
    normalized = str(raw_value or "").strip() or ".skill_state.json"
    candidate = Path(normalized).expanduser()
    if not candidate.is_absolute():
        candidate = (PACKAGE_ROOT / candidate).resolve()
    return candidate


def _profile_env_aliases(key: str) -> tuple[str, ...]:
    env_aliases = _RUNTIME_PROFILE.get("env_aliases") if isinstance(_RUNTIME_PROFILE.get("env_aliases"), dict) else {}
    values = env_aliases.get(key) if isinstance(env_aliases, dict) else None
    if isinstance(values, list):
        return tuple(str(item).strip() for item in values if str(item).strip())
    return ()


def read_stdin(raw: str) -> dict:
    raw = (raw or "").strip()
    if not raw:
        return {}
    data = json.loads(raw)
    if not isinstance(data, dict):
        raise ValueError("tool input must be a JSON object")
    return data


def load_json_file(path: Path) -> dict:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


_RUNTIME_PROFILE = _load_runtime_profile()
DEFAULT_STATE_PATH = _resolve_profile_state_path(str(_RUNTIME_PROFILE.get("default_state_path") or ".skill_state.json"))
DEFAULT_SOURCE_CHANNEL = str(_RUNTIME_PROFILE.get("default_source_channel") or "openclaw_skill").strip() or "openclaw_skill"
PROFILE_EXTRA_HEADERS = dict(_RUNTIME_PROFILE.get("extra_headers") or {})
USER_SAFE_FAILURE_MARKDOWN = str(_RUNTIME_PROFILE.get("user_safe_failure_markdown") or _DEFAULT_USER_SAFE_FAILURE_MARKDOWN)
ALLOWED_RESPONSE_MODES = tuple(
    str(item).strip()
    for item in (_RUNTIME_PROFILE.get("allowed_response_modes") or [])
    if str(item).strip()
) or ("auto", "sync_deferred", "sync_blocking")
_FEATURES = _RUNTIME_PROFILE.get("features") if isinstance(_RUNTIME_PROFILE.get("features"), dict) else {}
DEBUG_QUERY_ENABLED = bool(_FEATURES.get("debug_query", True))
PERSIST_REQUEST_DEBUG = bool(_FEATURES.get("persist_request_debug", True))
_BASE_URL_ENV_NAMES = _profile_env_aliases("base_url")
_STATUS_URL_ENV_NAMES = _profile_env_aliases("status_url")
_CREDENTIAL_ENV_NAMES = _profile_env_aliases(PROFILE_CREDENTIAL_ENV_KEY)
_SOURCE_CHANNEL_ENV_NAMES = _profile_env_aliases("source_channel")
_STATE_PATH_ENV_NAMES = _profile_env_aliases("state_path")
REQUEST_TIMEOUT_ENV_NAMES = _profile_env_aliases("request_timeout")
POLL_TIMEOUT_ENV_NAMES = _profile_env_aliases("poll_timeout")
POLL_INTERVAL_ENV_NAMES = _profile_env_aliases("poll_interval")



def load_state(state_path: Path) -> dict:
    return load_json_file(state_path)



def save_state(state_path: Path, state: dict) -> None:
    state_path.parent.mkdir(parents=True, exist_ok=True)
    state_path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")


def env_first(*names: str, default: str = "") -> str:
    for name in names:
        value = str(os.environ.get(name) or "").strip()
        if value:
            return value
    return default



def resolve_session_id(payload: dict, state: dict) -> str:
    explicit = str(payload.get("session_id") or "").strip()
    if explicit:
        return explicit
    if payload.get("force_new_session"):
        return uuid.uuid4().hex
    cached = str(state.get("session_id") or "").strip()
    if cached:
        return cached
    return uuid.uuid4().hex



def build_request_body(payload: dict, message: str, session_id: str, source_channel: str) -> dict:
    normalized_message = str(payload.get("normalized_message") or "").strip()
    request_kind = str(payload.get("request_kind") or "").strip()
    request_body = {
        "message": message,
        "session_id": session_id,
        "source_channel": source_channel,
        "force_new_session": bool(payload.get("force_new_session")),
    }
    if normalized_message:
        request_body["normalized_message"] = normalized_message
    industry_hint = str(payload.get("industry_hint") or "").strip()
    if industry_hint:
        request_body["industry_hint"] = industry_hint
    if request_kind:
        request_body["request_kind"] = request_kind
    host_capabilities = payload.get("host_capabilities")
    if isinstance(host_capabilities, dict) and host_capabilities:
        request_body["host_capabilities"] = host_capabilities
    response_mode = str(payload.get("response_mode") or DEFAULT_BACKEND_RESPONSE_MODE).strip()
    if response_mode and response_mode not in ALLOWED_RESPONSE_MODES:
        response_mode = DEFAULT_BACKEND_RESPONSE_MODE
    if response_mode and not payload.get("status_only"):
        request_body["response_mode"] = response_mode
    for key in ("research_method", "language", "app_id"):
        value = payload.get(key)
        if value not in (None, ""):
            request_body[key] = value
    return request_body



def resolve_request_credential(payload: dict) -> str:
    explicit = str(payload.get(payload_credential_field()) or "").strip()
    if explicit:
        return explicit
    env_value = env_first(*_CREDENTIAL_ENV_NAMES)
    if env_value:
        return env_value
    return ""



def extract_request_credential_from_text(text: str) -> str:
    normalized = str(text or "").strip()
    if not normalized:
        return ""

    if re.fullmatch(r"[A-Fa-f0-9]{40}", normalized):
        return normalized

    match = re.search(
        r"(?i)(?:api\s*key|apikey)[^A-Za-z0-9_-]*([A-Za-z0-9_-]{20,128})",
        normalized,
    )
    if match:
        return str(match.group(1) or "").strip()
    return ""



def _coerce_bool(value: Any, default: bool) -> bool:
    if isinstance(value, bool):
        return value
    if value is None:
        return default
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in {"1", "true", "yes", "on"}:
            return True
        if normalized in {"0", "false", "no", "off"}:
            return False
    return bool(value)


def normalize_host_capabilities(value: Any) -> dict[str, bool]:
    raw = value if isinstance(value, dict) else {}
    return {
        key: _coerce_bool(raw.get(key), default)
        for key, default in DEFAULT_HOST_CAPABILITIES.items()
    }


def build_generating_reply_markdown(host_capabilities: Any = None) -> str:
    normalized_caps = normalize_host_capabilities(host_capabilities)
    lines = [
        "# AI调研方案生成中",
        "",
        "- 已收到你的调研需求，正在生成完整方案卡。",
        "- 通常还需要几十秒到几分钟，期间不需要重复发送相同请求。",
        "- 你可以稍后在当前会话直接发送“查看进度”或“查看方案”继续跟进。",
    ]
    if normalized_caps.get("scheduled_followup"):
        lines.append("- 当前环境支持定时跟进，你也可以直接使用定时任务或跟进动作。")
    return "\n".join(lines)


def append_session_id_markdown(reply_markdown: str, session_id: str) -> str:
    body = str(reply_markdown or "").rstrip()
    normalized_session_id = str(session_id or "").strip()
    if not normalized_session_id:
        return body
    if _SESSION_ID_MARKDOWN_PREFIX in body and normalized_session_id in body:
        return body
    session_line = f"{_SESSION_ID_MARKDOWN_PREFIX}`{normalized_session_id}`"
    if not body:
        return session_line
    return f"{body}\n\n{session_line}"


def build_response_envelope(
    payload: dict | None = None,
    *,
    status: str = "",
    reply_markdown: str = "",
    session_id: str = "",
    session_event: str = "",
    local_only: bool = False,
    presentation: dict | None = None,
    next_actions: list[dict] | None = None,
    artifacts: dict | None = None,
    capability_hints: dict | None = None,
    host_capabilities: dict | None = None,
) -> dict:
    current = dict(payload or {})
    if status and not current.get("status"):
        current["status"] = status
    if reply_markdown and not current.get("reply_markdown"):
        current["reply_markdown"] = reply_markdown
    if session_id and not current.get("session_id"):
        current["session_id"] = session_id
    current["reply_markdown"] = append_session_id_markdown(
        str(current.get("reply_markdown") or ""),
        str(current.get("session_id") or session_id or ""),
    )
    if local_only:
        current["local_only"] = True
    current["schema_version"] = str(current.get("schema_version") or SKILL_RESPONSE_SCHEMA_VERSION)
    current["session_event"] = str(session_event or current.get("session_event") or "")
    current.setdefault("supported_actions", [])

    normalized_caps = normalize_host_capabilities(host_capabilities)

    merged_artifacts = dict(current.get("artifacts") or {}) if isinstance(current.get("artifacts"), dict) else {}
    if isinstance(artifacts, dict):
        merged_artifacts.update(artifacts)
    for key in ("task_console_url", "result_url", "result_summary", "skill_version", "supported_industries", "auth_url"):
        value = current.get(key)
        if value not in (None, "", [], {}):
            merged_artifacts.setdefault(key, value)
    current["artifacts"] = merged_artifacts

    merged_presentation = dict(current.get("presentation") or {}) if isinstance(current.get("presentation"), dict) else {}
    if isinstance(presentation, dict):
        merged_presentation.update(presentation)
    merged_presentation.setdefault("render_mode", "verbatim_markdown")
    merged_presentation.setdefault("card_type", "info" if local_only else "default")
    merged_presentation.setdefault("detail_available", bool(current.get("detail_available")))
    merged_presentation.setdefault("detail_prompt", str(current.get("detail_prompt") or ""))
    merged_presentation.setdefault(
        "show_task_link",
        bool(normalized_caps.get("clickable_links") and merged_artifacts.get("task_console_url")),
    )
    merged_presentation.setdefault(
        "show_result_link",
        bool(normalized_caps.get("clickable_links") and (merged_artifacts.get("result_url") or merged_artifacts.get("task_console_url"))),
    )
    merged_presentation.setdefault("show_internal_ids", False)
    current["presentation"] = merged_presentation

    current["next_actions"] = (
        list(current.get("next_actions"))
        if isinstance(current.get("next_actions"), list)
        else list(next_actions or [])
    )
    current["capability_hints"] = (
        dict(current.get("capability_hints") or {})
        if isinstance(current.get("capability_hints"), dict)
        else {}
    )
    if isinstance(capability_hints, dict):
        current["capability_hints"].update(capability_hints)
    current["capability_hints"].setdefault("followup_supported", bool(normalized_caps.get("scheduled_followup")))
    current["capability_hints"].setdefault("followup_requires_host_scheduler", True)
    return current


def build_auth_required_response(
    reason: str = "missing",
    *,
    session_id: str = "",
    session_event: str = "pending_request_saved",
    host_capabilities: dict | None = None,
) -> dict:
    if reason == "invalid":
        reply = (
            "当前绑定的 API Key 无效或已失效。"
            f"请前往 {AUTH_GUIDE_URL} 重新获取 API Key，拿到后直接发给我，"
            "我会继续刚才这条调研需求，不需要重新描述。"
        )
    else:
        reply = (
            "首次使用 AI Research 需要先绑定 API Key。"
            f"请先前往 {AUTH_GUIDE_URL} 获取 API Key，拿到后直接发给我，"
            "我会继续刚才这条调研需求，不需要重新描述。"
        )
    return build_response_envelope(
        {
            "auth_required": True,
            "auth_url": AUTH_GUIDE_URL,
        },
        status="AUTH_REQUIRED",
        reply_markdown=reply,
        session_id=session_id,
        session_event="auth_invalid" if reason == "invalid" else session_event,
        local_only=True,
        presentation={"card_type": "auth_required"},
        host_capabilities=host_capabilities,
    )



def build_request_headers(credential_value: str) -> dict[str, str]:
    headers = {
        "Content-Type": "application/json",
        **PROFILE_EXTRA_HEADERS,
    }
    if credential_value:
        headers[primary_credential_header_name()] = credential_value
    return headers


def _normalize_message_url(raw: str) -> str:
    normalized = str(raw or "").strip().rstrip("/")
    if not normalized:
        return DEFAULT_BASE_URL
    if normalized.endswith("/message"):
        return normalized
    if normalized.endswith("/message/stream"):
        return normalized[: -len("/stream")]
    if normalized.endswith("/status"):
        return f"{normalized[: -len('/status')]}/message"
    if normalized.endswith("/datatester/compass/api/v3/survey/skill"):
        return f"{normalized}/message"
    if normalized.endswith("/datatester/compass/api/v3/survey/skill/"):
        return f"{normalized.rstrip('/')}/message"
    return f"{normalized}/datatester/compass/api/v3/survey/skill/message"


def resolve_state_path(value: str | Path | None = None) -> Path:
    if isinstance(value, Path):
        return value.expanduser().resolve()
    explicit = str(value or env_first(*_STATE_PATH_ENV_NAMES, default=str(DEFAULT_STATE_PATH))).strip()
    return Path(explicit).expanduser().resolve()


def resolve_base_url(base_url: str | None = None) -> str:
    raw = str(base_url or env_first(*_BASE_URL_ENV_NAMES, default=DEFAULT_BASE_URL)).strip()
    return _normalize_message_url(raw)


def resolve_status_url(base_url: str | None = None) -> str:
    explicit = env_first(*_STATUS_URL_ENV_NAMES)
    if explicit:
        return explicit.rstrip("/")
    message_url = resolve_base_url(base_url)
    if message_url.endswith("/message"):
        return f"{message_url[:-len('/message')]}/status"
    return DEFAULT_STATUS_URL


def resolve_source_channel(payload: dict | None = None) -> str:
    current = payload if isinstance(payload, dict) else {}
    return str(
        current.get("source_channel")
        or env_first(*_SOURCE_CHANNEL_ENV_NAMES, default=DEFAULT_SOURCE_CHANNEL)
        or DEFAULT_SOURCE_CHANNEL,
    ).strip() or DEFAULT_SOURCE_CHANNEL



def post_json(url: str, body: dict, credential_value: str, timeout_seconds: int = 30) -> dict:
    request = urllib.request.Request(
        url,
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers=build_request_headers(credential_value),
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
            payload = json.loads(response.read().decode("utf-8"))
            if isinstance(payload, dict):
                payload.setdefault("code", 0)
                payload["_http_status"] = getattr(response, "status", 200)
                return payload
            return {"code": 0, "data": payload, "_http_status": getattr(response, "status", 200)}
    except urllib.error.HTTPError as exc:
        try:
            payload = json.loads(exc.read().decode("utf-8"))
        except Exception:
            payload = {"error": f"http error {exc.code}"}
        if not isinstance(payload, dict):
            payload = {"error": str(payload)}
        payload.setdefault("code", exc.code)
        payload["_http_status"] = exc.code
        return payload
    except urllib.error.URLError as exc:
        reason = str(getattr(exc, "reason", "") or exc)
        return {
            "code": 599,
            "error": reason or "network error",
            "_http_status": 0,
            "_transport_error": reason or "network error",
        }



def get_json(url: str, query: dict[str, Any], credential_value: str, timeout_seconds: int = 15) -> dict:
    encoded_query = urllib.parse.urlencode(
        {key: value for key, value in (query or {}).items() if value not in (None, "")},
        doseq=True,
    )
    request_url = f"{url}?{encoded_query}" if encoded_query else url
    request = urllib.request.Request(
        request_url,
        headers=build_request_headers(credential_value),
        method="GET",
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
            payload = json.loads(response.read().decode("utf-8"))
            if isinstance(payload, dict):
                payload.setdefault("code", 0)
                payload["_http_status"] = getattr(response, "status", 200)
                return payload
            return {"code": 0, "data": payload, "_http_status": getattr(response, "status", 200)}
    except urllib.error.HTTPError as exc:
        try:
            payload = json.loads(exc.read().decode("utf-8"))
        except Exception:
            payload = {"error": f"http error {exc.code}"}
        if not isinstance(payload, dict):
            payload = {"error": str(payload)}
        payload.setdefault("code", exc.code)
        payload["_http_status"] = exc.code
        return payload
    except urllib.error.URLError as exc:
        reason = str(getattr(exc, "reason", "") or exc)
        return {
            "code": 599,
            "error": reason or "network error",
            "_http_status": 0,
            "_transport_error": reason or "network error",
        }



def load_build_metadata() -> dict[str, Any]:
    package_info = load_json_file(PACKAGE_JSON_PATH)
    build_info = load_json_file(BUILD_INFO_PATH)
    return {
        "skill_version": str(build_info.get("skill_version") or package_info.get("version") or DEFAULT_SKILL_VERSION),
        "build_commit": str(build_info.get("build_commit") or ""),
        "build_dirty": bool(build_info.get("build_dirty", False)),
        "built_at": str(build_info.get("built_at") or ""),
    }


def format_supported_industries_markdown() -> str:
    items = "\n".join(f"- {item}" for item in DIRECT_EXECUTABLE_INDUSTRIES)
    return f"# 当前支持直接执行的行业\n\n{items}"



def redact_credential_value(value: str) -> str:
    return "<redacted_credential>" if str(value or "").strip() else ""



def redact_headers(headers: dict[str, Any]) -> dict[str, Any]:
    redacted = dict(headers)
    for key in sensitive_header_names():
        if key in redacted:
            redacted[key] = (
                "bearer <redacted_credential>"
                if str(key).lower() == authz_header_name().lower() and str(redacted.get(key) or "").strip()
                else redact_credential_value(str(redacted.get(key) or ""))
            )
    return redacted



def summarize_response_payload(payload: Any) -> Any:
    if not isinstance(payload, dict):
        return payload
    summary: dict[str, Any] = {}
    for key in ("code", "message", "msg", "error"):
        value = payload.get(key)
        if value not in (None, ""):
            summary[key] = value
    data = payload.get("data")
    if isinstance(data, dict):
        data_summary = {}
        for key in ("session_id", "status", "plan_id", "result_url"):
            value = data.get(key)
            if value not in (None, ""):
                data_summary[key] = value
        if data_summary:
            summary["data"] = data_summary
    return summary or {key: value for key, value in payload.items() if not str(key).startswith("_")}



def build_redacted_curl(
    url: str,
    body: dict,
    headers: dict[str, Any],
    *,
    method: str = "POST",
    query_params: dict[str, Any] | None = None,
) -> str:
    encoded_query = urllib.parse.urlencode(
        {key: value for key, value in (query_params or {}).items() if value not in (None, "")},
        doseq=True,
    )
    request_url = f"{url}?{encoded_query}" if encoded_query else url
    normalized_method = str(method or "POST").upper()
    lines = [f"curl -X {normalized_method} {shlex.quote(request_url)} \\"]
    header_items = list(headers.items())
    for index, (key, value) in enumerate(header_items):
        suffix = " \\" if index < len(header_items) - 1 or (body and normalized_method != "GET") else ""
        lines.append(f"  -H {shlex.quote(f'{key}: {value}')}" + suffix)
    if body and normalized_method != "GET":
        body_str = json.dumps(body, ensure_ascii=False, indent=2)
        lines.append(f"  -d {shlex.quote(body_str)}")
    return "\n".join(lines)



def build_request_debug_snapshot(
    *,
    base_url: str,
    request_body: dict,
    headers: dict[str, Any],
    response_payload: Any = None,
    http_status: int | None = None,
    transport_error: str = "",
    method: str = "POST",
    query_params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    redacted_headers = redact_headers(headers)
    snapshot = {
        "captured_at": datetime.now(timezone.utc).isoformat(),
        "effective_base_url": base_url,
        "request_headers": redacted_headers,
        "request_body": request_body,
        "redacted_curl": build_redacted_curl(
            base_url,
            request_body,
            redacted_headers,
            method=method,
            query_params=query_params,
        ),
    }
    if query_params:
        snapshot["request_query"] = {
            key: value for key, value in (query_params or {}).items() if value not in (None, "")
        }
    if http_status not in (None, ""):
        snapshot["http_status"] = http_status  # type: ignore[assignment]
    if transport_error:
        snapshot["transport_error"] = transport_error
    if response_payload not in (None, ""):
        snapshot["response_summary"] = summarize_response_payload(response_payload)
    return snapshot



def save_last_request_debug(state_path: Path, state: dict, snapshot: dict[str, Any]) -> None:
    if not PERSIST_REQUEST_DEBUG:
        return
    state["last_request_debug"] = snapshot
    save_state(state_path, state)



def build_user_safe_failure_response(
    *,
    session_id: str = "",
    session_event: str = "request_failed",
    host_capabilities: dict | None = None,
) -> dict:
    return build_response_envelope(
        {
            "request_failed": True,
            "retryable": True,
        },
        status="REQUEST_FAILED",
        reply_markdown=USER_SAFE_FAILURE_MARKDOWN,
        session_id=session_id,
        session_event=session_event,
        local_only=True,
        presentation={"card_type": "request_failed"},
        host_capabilities=host_capabilities,
    )



def build_business_error_response(
    message: str,
    *,
    session_id: str = "",
    session_event: str = "business_error",
    host_capabilities: dict | None = None,
) -> dict:
    return build_response_envelope(
        {
            "business_error": True,
        },
        status="BUSINESS_ERROR",
        reply_markdown=(message or "请求未通过，请调整后重试。"),
        session_id=session_id,
        session_event=session_event,
        local_only=True,
        presentation={"card_type": "business_error"},
        host_capabilities=host_capabilities,
    )



def build_debug_report(state: dict, include_last_request: bool = True) -> dict:
    """Build a redacted debug snapshot for internal use by ai_research_message only."""
    metadata = load_build_metadata()
    current_credential = str(env_first(*_CREDENTIAL_ENV_NAMES) or "").strip()
    report = {
        "debug": True,
        **metadata,
        "effective_base_url": resolve_base_url(),
        "effective_source_channel": env_first(*_SOURCE_CHANNEL_ENV_NAMES, default=DEFAULT_SOURCE_CHANNEL),
        "has_credential": bool(current_credential),
        "credential_hint": redact_credential_value(current_credential),
        "session_id": str(state.get("session_id") or ""),
        "has_pending_request": isinstance(state.get("pending_request"), dict),
    }
    if include_last_request:
        report["last_request"] = state.get("last_request_debug") if isinstance(state.get("last_request_debug"), dict) else None
    return report

#!/usr/bin/env python3
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import json
from pathlib import Path
import re
import sys
import time

CURRENT_DIR = Path(__file__).resolve().parent
if str(CURRENT_DIR) not in sys.path:
    sys.path.insert(0, str(CURRENT_DIR))

from ai_research_common import (  # noqa: E402
    DEBUG_QUERY_ENABLED,
    DIRECT_EXECUTABLE_INDUSTRIES,
    DEFAULT_STATE_PATH,
    POLL_INTERVAL_ENV_NAMES,
    POLL_TIMEOUT_ENV_NAMES,
    REQUEST_TIMEOUT_ENV_NAMES,
    build_auth_required_response,
    build_business_error_response,
    build_debug_report,
    build_generating_reply_markdown,
    build_response_envelope,
    build_request_body,
    build_request_debug_snapshot,
    build_request_headers,
    build_user_safe_failure_response,
    extract_request_credential_from_text,
    format_supported_industries_markdown,
    get_json,
    load_build_metadata,
    load_state,
    payload_credential_field,
    post_json,
    read_stdin,
    env_first,
    resolve_request_credential,
    resolve_base_url,
    resolve_session_id,
    resolve_source_channel,
    resolve_state_path,
    resolve_status_url,
    save_last_request_debug,
    save_state,
    normalize_host_capabilities,
)


STATE_PATH = DEFAULT_STATE_PATH
DEFAULT_REQUEST_TIMEOUT_SECONDS = 120
DEFAULT_STATUS_POLL_TIMEOUT_SECONDS = 60
DEFAULT_STATUS_POLL_INTERVAL_SECONDS = 2.0
_POLLABLE_FAILURE_CODES = {502, 503, 504, 599}
_POLLABLE_STATUSES = {"GENERATING"}
_READY_STATUSES = {
    "WAITING_CONFIRM",
    "UNSUPPORTED_INDUSTRY",
    "CONFIRMED",
    "AUDIENCE_RUNNING",
    "TASK_RUNNING",
    "FINISHED",
    "FAILED",
}
_VERSION_QUERY_PATTERNS = (
    r"哪个版本",
    r"什么版本",
    r"当前版本",
    r"skill版本",
    r"技能版本",
    r"\bskill\s+version\b",
)
_REQUEST_DEBUG_PATTERNS = (
    r"\bcurl\b",
    r"请求头",
    r"调试信息",
    r"debug",
    r"最近请求",
)
_SUPPORTED_INDUSTRY_PATTERNS = (
    r"支持.*行业",
    r"有哪些行业",
    r"行业列表",
)
# NOTE: These keyword lists provide a client-side hint to the backend's
# request_kind parameter. The backend (skill_service.py) is the authority.
# Keep these lists in sync with skill_service.py or make them a strict
# subset. False positives are acceptable; false negatives cause extra
# backend re-inference but no correctness issues.
_RESULT_QUERY_KINDS = {"query_status", "query_result", "query_plan"}
_SHORT_FOLLOWUP_MAX_LENGTH = 64
_CONFIRM_EXECUTE_EXACT_MESSAGES = {
    "yes",
    "confirm",
    "start",
    "execute",
    "run",
    "是",
    "确认",
    "开始",
    "执行",
    "启动",
    "开跑",
}
_CONFIRM_EXECUTE_PHRASES = (
    "确认方案",
    "确认执行",
    "确认并执行",
    "开始执行",
    "开始投放",
    "直接投放",
    "按这个执行",
    "就按这个执行",
    "开始调研",
)
_QUERY_VERB_KEYWORDS = (
    "查看",
    "查询",
    "查下",
    "看看",
    "看下",
    "看一下",
    "当前",
    "现在",
    "最新",
)
_STATUS_TOPIC_KEYWORDS = ("状态", "进度", "进展")
_STATUS_QUESTION_KEYWORDS = ("如何", "怎么样", "到哪", "哪一步")
_RESULT_TOPIC_KEYWORDS = ("结果", "报告")
_PLAN_TOPIC_KEYWORDS = ("方案", "计划")
_DETAIL_SIGNAL_KEYWORDS = ("完整", "详细", "详情", "全部")
_DETAIL_TARGET_KEYWORDS = ("方案", "计划", "问卷", "题单", "题目", "题", "选项", "信息", "概览")
_DETAIL_GRANULAR_TARGET_KEYWORDS = ("问卷", "题单", "题目", "题", "选项")
_DETAIL_ACTION_KEYWORDS = ("给我看", "看一下", "看下", "具体")
_FAILURE_REASON_KEYWORDS = ("原因", "详情", "为什么", "为何", "为啥", "卡在哪")
_RESTART_SIGNAL_KEYWORDS = ("重发", "重来", "重新")
_RESTART_TARGET_KEYWORDS = ("发起", "生成", "调研", "来一轮", "来一次", "做一轮", "做一次")


def _load_state() -> dict:
    return load_state(STATE_PATH)



def _save_state(state: dict) -> None:
    save_state(STATE_PATH, state)


def _load_state_at(state_path: Path) -> dict:
    return load_state(state_path)


def _save_state_at(state_path: Path, state: dict) -> None:
    save_state(state_path, state)


def _resolve_runtime_state_path(payload: dict, state_path_override: Path | None = None) -> Path:
    if state_path_override is not None:
        return Path(state_path_override).expanduser().resolve()
    if payload.get("state_path") not in (None, ""):
        return resolve_state_path(payload.get("state_path"))
    resolved = resolve_state_path()
    if resolved != DEFAULT_STATE_PATH:
        return resolved
    return STATE_PATH



def _resolve_session_id(payload: dict, state: dict) -> str:
    return resolve_session_id(payload, state)



def _build_request_body(payload: dict, message: str, session_id: str, source_channel: str) -> dict:
    return build_request_body(payload, message, session_id, source_channel)



def _resolve_credential(payload: dict) -> str:
    return resolve_request_credential(payload)



def _build_auth_required_response(reason: str = "missing", **kwargs) -> dict:
    return build_auth_required_response(reason, **kwargs)



def _build_request_headers(credential_value: str) -> dict[str, str]:
    return build_request_headers(credential_value)



def _post_json(url: str, body: dict, credential_value: str, timeout_seconds: int = DEFAULT_REQUEST_TIMEOUT_SECONDS) -> dict:
    return post_json(url, body, credential_value, timeout_seconds=timeout_seconds)



def _get_json(url: str, query: dict, credential_value: str, timeout_seconds: int = 15) -> dict:
    return get_json(url, query, credential_value, timeout_seconds=timeout_seconds)



def _resolve_request_timeout_seconds(payload: dict | None = None) -> int:
    raw = str((payload or {}).get("timeout_seconds") or env_first(*REQUEST_TIMEOUT_ENV_NAMES) or "").strip()
    try:
        value = int(raw)
    except Exception:
        value = DEFAULT_REQUEST_TIMEOUT_SECONDS
    return max(5, value)



def _resolve_status_poll_timeout_seconds(payload: dict | None = None) -> int:
    raw = str((payload or {}).get("poll_timeout_seconds") or env_first(*POLL_TIMEOUT_ENV_NAMES) or "").strip()
    try:
        value = int(raw)
    except Exception:
        value = DEFAULT_STATUS_POLL_TIMEOUT_SECONDS
    return max(0, value)



def _resolve_status_poll_interval_seconds(payload: dict | None = None) -> float:
    raw = str((payload or {}).get("poll_interval_seconds") or env_first(*POLL_INTERVAL_ENV_NAMES) or "").strip()
    try:
        value = float(raw)
    except Exception:
        value = DEFAULT_STATUS_POLL_INTERVAL_SECONDS
    return max(0.5, value)



def _extract_data(payload: dict) -> dict:
    data = payload.get("data") or {}
    return data if isinstance(data, dict) else {}


def _ensure_generating_reply_markdown(
    data: dict,
    session_id: str,
    host_capabilities: dict | None = None,
) -> dict:
    current = dict(data or {})
    current.setdefault("session_id", session_id)
    current["status"] = _extract_status(current) or "GENERATING"
    reply = build_generating_reply_markdown(host_capabilities)
    created_at = str(current.get("created_at") or "").strip()
    if created_at:
        try:
            from datetime import datetime as _dt
            created = _dt.fromisoformat(created_at.replace("Z", "+00:00"))
            now = _dt.now(created.tzinfo) if created.tzinfo else _dt.now()
            elapsed_minutes = int((now - created).total_seconds() / 60)
            if elapsed_minutes >= 1:
                reply += f"\n- 当前已等待约 {elapsed_minutes} 分钟。"
        except Exception:
            pass
    current["reply_markdown"] = reply
    return current


def _extract_status(data: dict) -> str:
    return str((data or {}).get("status") or "").strip().upper()


def _normalize_message(message: str) -> str:
    return re.sub(r"\s+", " ", str(message or "").strip())


def _compact_message_text(message: str) -> str:
    normalized = _normalize_message(message).lower()
    return re.sub(r"[\s\.,!?，。！？:：;；'\"`~\\-_/\\\\()\\[\\]{}]+", "", normalized)


def _contains_any(text: str, keywords: tuple[str, ...]) -> bool:
    return any(keyword in text for keyword in keywords)


def _is_short_followup_message(message: str) -> bool:
    return len(_normalize_message(message)) <= _SHORT_FOLLOWUP_MAX_LENGTH


def _is_confirm_execute_message(message: str) -> bool:
    compact = _compact_message_text(message)
    if compact in {_compact_message_text(item) for item in _CONFIRM_EXECUTE_EXACT_MESSAGES}:
        return True
    lowered = _normalize_message(message).lower()
    return _contains_any(lowered, _CONFIRM_EXECUTE_PHRASES)


def _is_status_query_message(message: str) -> bool:
    lowered = _normalize_message(message).lower()
    if re.search(r"\b(status|progress)\b", lowered):
        return True
    if _contains_any(lowered, _STATUS_TOPIC_KEYWORDS) and _contains_any(lowered, _QUERY_VERB_KEYWORDS + _STATUS_QUESTION_KEYWORDS):
        return True
    return "到哪一步" in lowered


def _is_result_query_message(message: str) -> bool:
    lowered = _normalize_message(message).lower()
    if re.search(r"\b(result|report)\b", lowered):
        return True
    if _contains_any(lowered, _RESULT_TOPIC_KEYWORDS) and _contains_any(lowered, _QUERY_VERB_KEYWORDS + ("有", "出来", "呢")):
        return True
    return "调研结果" in lowered


def _is_plan_query_message(message: str) -> bool:
    lowered = _normalize_message(message).lower()
    if "show plan" in lowered or "showplan" in lowered:
        return True
    if _contains_any(lowered, _PLAN_TOPIC_KEYWORDS) and _contains_any(lowered, _QUERY_VERB_KEYWORDS + ("呢",)):
        return True
    return False


def _is_detail_query_message(message: str) -> bool:
    lowered = _normalize_message(message).lower()
    if _contains_any(lowered, _DETAIL_SIGNAL_KEYWORDS) and _contains_any(lowered, _DETAIL_TARGET_KEYWORDS):
        return True
    return _contains_any(lowered, _DETAIL_GRANULAR_TARGET_KEYWORDS) and _contains_any(lowered, _DETAIL_ACTION_KEYWORDS)


def _is_failure_reason_query_message(message: str) -> bool:
    lowered = _normalize_message(message).lower()
    return "失败" in lowered and _contains_any(lowered, _FAILURE_REASON_KEYWORDS)


def _is_restart_message(message: str) -> bool:
    lowered = _normalize_message(message).lower()
    if "重发" in lowered or "重来" in lowered:
        return True
    return "重新" in lowered and _contains_any(lowered, _RESTART_TARGET_KEYWORDS)


def _infer_request_kind_from_message(message: str) -> str:
    if not _is_short_followup_message(message):
        return ""
    if _is_confirm_execute_message(message):
        return "confirm_execute"
    if _is_detail_query_message(message):
        return "expand_plan_detail"
    if _is_failure_reason_query_message(message):
        return "query_failure_reason"
    if _is_result_query_message(message):
        return "query_result"
    if _is_plan_query_message(message):
        return "query_plan"
    if _is_status_query_message(message):
        return "query_status"
    if _is_restart_message(message):
        return "restart"
    return ""


def _matches_any_pattern(text: str, patterns: tuple[str, ...]) -> bool:
    normalized = str(text or "").strip().lower()
    return any(re.search(pattern, normalized, flags=re.IGNORECASE) for pattern in patterns)


def _is_version_query(message: str) -> bool:
    return _matches_any_pattern(message, _VERSION_QUERY_PATTERNS)


def _is_request_debug_query(message: str) -> bool:
    return _matches_any_pattern(message, _REQUEST_DEBUG_PATTERNS)


def _is_supported_industry_query(message: str) -> bool:
    return _matches_any_pattern(message, _SUPPORTED_INDUSTRY_PATTERNS)


def _resolve_host_capabilities(payload: dict) -> dict[str, bool]:
    return normalize_host_capabilities(payload.get("host_capabilities"))


def _ensure_response_envelope(
    payload: dict,
    *,
    status: str = "",
    reply_markdown: str = "",
    session_id: str = "",
    session_event: str = "",
    local_only: bool = False,
    presentation: dict | None = None,
    host_capabilities: dict | None = None,
) -> dict:
    return build_response_envelope(
        payload,
        status=status,
        reply_markdown=reply_markdown,
        session_id=session_id,
        session_event=session_event,
        local_only=local_only,
        presentation=presentation,
        host_capabilities=host_capabilities,
    )


def _build_version_response(*, host_capabilities: dict | None = None) -> dict:
    metadata = load_build_metadata()
    version = str(metadata.get("skill_version") or "unknown").strip() or "unknown"
    return _ensure_response_envelope(
        {
            "skill_version": version,
        },
        status="LOCAL_INFO",
        reply_markdown=f"当前 AI Research skill 版本为 v{version}。",
        local_only=True,
        session_event="local_query_only",
        presentation={"card_type": "info"},
        host_capabilities=host_capabilities,
    )


def _build_supported_industry_response(*, host_capabilities: dict | None = None) -> dict:
    return _ensure_response_envelope(
        {
            "supported_industries": list(DIRECT_EXECUTABLE_INDUSTRIES),
        },
        status="LOCAL_INFO",
        reply_markdown=format_supported_industries_markdown(),
        local_only=True,
        session_event="local_query_only",
        presentation={"card_type": "info"},
        host_capabilities=host_capabilities,
    )


def _build_request_debug_response(
    state: dict,
    *,
    include_last_request: bool,
    host_capabilities: dict | None = None,
) -> dict:
    report = build_debug_report(state, include_last_request=include_last_request)
    last_request = report.get("last_request") if isinstance(report.get("last_request"), dict) else None
    if not last_request:
        reply = (
            "当前还没有可用的最近请求快照。"
            "请先发起一次调研请求；如果需要调试信息，再明确说明即可。"
        )
    else:
        curl_text = str(last_request.get("redacted_curl") or "").strip()
        lines = [
            "下面是最近一次请求的脱敏调试信息：",
            "",
            "```bash",
            curl_text or "# 暂无可展示的 curl 快照",
            "```",
        ]
        reply = "\n".join(lines)
    return _ensure_response_envelope(
        {
            "debug": True,
            "skill_version": str(report.get("skill_version") or ""),
            "last_request": last_request,
        },
        status="DEBUG_INFO",
        reply_markdown=reply,
        local_only=True,
        session_event="debug_query_only",
        presentation={"card_type": "debug"},
        host_capabilities=host_capabilities,
    )



def _build_status_query(request_body: dict) -> dict:
    query = {
        "session_id": request_body.get("session_id"),
        "source_channel": request_body.get("source_channel"),
    }
    app_id = request_body.get("app_id")
    if app_id not in (None, ""):
        query["app_id"] = app_id
    host_capabilities = request_body.get("host_capabilities")
    if isinstance(host_capabilities, dict) and host_capabilities:
        query["host_capabilities"] = json.dumps(host_capabilities, ensure_ascii=False, separators=(",", ":"))
    return query
def _query_status_once(status_url: str, request_body: dict, credential_value: str) -> dict:
    return _get_json(status_url, _build_status_query(request_body), credential_value, timeout_seconds=15)


def _should_recover_via_status_query(response: dict, http_status: int, transport_error: str) -> bool:
    code = int(response.get("code", 0) or 0)
    if code in {400, 401}:
        return False
    if transport_error:
        return True
    return code in _POLLABLE_FAILURE_CODES or http_status in _POLLABLE_FAILURE_CODES


def run_payload(
    payload: dict,
    *,
    state_path_override: Path | None = None,
) -> tuple[dict, int]:
    try:
        payload = dict(payload or {})
        message = str(payload.get("message") or "").strip()
        if not message:
            return {"error": "message is required"}, 1

        state_path = _resolve_runtime_state_path(payload, state_path_override)
        state = _load_state_at(state_path)
        if state.pop(payload_credential_field(), None) is not None:
            _save_state_at(state_path, state)
        debug_mode = bool(payload.get("debug_mode")) and DEBUG_QUERY_ENABLED
        normalized_host_capabilities = _resolve_host_capabilities(payload)

        if _is_version_query(message):
            return _build_version_response(host_capabilities=normalized_host_capabilities), 0

        if _is_supported_industry_query(message):
            return _build_supported_industry_response(host_capabilities=normalized_host_capabilities), 0

        if debug_mode and _is_request_debug_query(message):
            return (
                _build_request_debug_response(
                    state,
                    include_last_request=True,
                    host_capabilities=normalized_host_capabilities,
                ),
                0,
            )

        explicit_request_kind = str(payload.get("request_kind") or "").strip()
        inferred_request_kind = _infer_request_kind_from_message(message) if not explicit_request_kind else ""
        request_kind = explicit_request_kind or inferred_request_kind
        if request_kind:
            payload["request_kind"] = request_kind
        if request_kind in _RESULT_QUERY_KINDS and "status_only" not in payload:
            payload["status_only"] = True

        session_id = _resolve_session_id(payload, state)
        base_url = resolve_base_url(payload.get("base_url"))
        status_url = resolve_status_url(payload.get("base_url"))
        source_channel = resolve_source_channel(payload)
        current_request = _build_request_body(payload, message, session_id, source_channel)
        status_only = bool(payload.get("status_only"))

        inline_credential = extract_request_credential_from_text(message)
        credential_value = _resolve_credential(payload) or inline_credential
        pending_request = state.get("pending_request") if isinstance(state.get("pending_request"), dict) else None
        if not credential_value:
            session_event = "pending_request_waiting_auth"
            if not pending_request:
                state["pending_request"] = current_request
                state["session_id"] = session_id
                _save_state_at(state_path, state)
                session_event = "pending_request_saved"
            return (
                _build_auth_required_response(
                    session_id=session_id,
                    session_event=session_event,
                    host_capabilities=normalized_host_capabilities,
                ),
                0,
            )

        explicit_credential = str(payload.get(payload_credential_field()) or inline_credential or "").strip()
        request_body = dict(pending_request) if explicit_credential and pending_request else current_request
        if "source_channel" not in request_body or not request_body.get("source_channel"):
            request_body["source_channel"] = source_channel
        if "session_id" not in request_body or not request_body.get("session_id"):
            request_body["session_id"] = session_id
        if "message" not in request_body or not str(request_body.get("message") or "").strip():
            request_body["message"] = message

        headers = _build_request_headers(credential_value)
        state["session_id"] = str(request_body.get("session_id") or session_id)

        request_query = _build_status_query(request_body) if status_only else None
        effective_request_url = status_url if status_only else base_url
        if status_only:
            response = _query_status_once(status_url, request_body, credential_value)
        else:
            response = _post_json(
                base_url,
                request_body,
                credential_value,
                timeout_seconds=_resolve_request_timeout_seconds(payload),
            )
        http_status = int(response.pop("_http_status", 0) or 0)
        transport_error = str(response.pop("_transport_error", "") or "").strip()
        if int(response.get("code", 0) or 0) == 0 and _extract_status(_extract_data(response)) in _POLLABLE_STATUSES:
            response["data"] = _ensure_generating_reply_markdown(
                _extract_data(response),
                str(request_body.get("session_id") or session_id),
                normalized_host_capabilities,
            )

        if not status_only and _should_recover_via_status_query(response, http_status, transport_error):
            recovered = _query_status_once(status_url, request_body, credential_value)
            recovered_http_status = int(recovered.pop("_http_status", 0) or 0)
            recovered_transport_error = str(recovered.pop("_transport_error", "") or "").strip()
            if int(recovered.get("code", 0) or 0) == 401:
                _save_state_at(state_path, state)
                return (
                    _build_auth_required_response(
                        reason="invalid",
                        session_id=str(request_body.get("session_id") or session_id),
                        host_capabilities=normalized_host_capabilities,
                    ),
                    0,
                )
            if int(recovered.get("code", 0) or 0) == 0:
                response = recovered
                http_status = recovered_http_status
                transport_error = recovered_transport_error
                request_query = _build_status_query(request_body)
                effective_request_url = status_url
                if _extract_status(_extract_data(response)) in _POLLABLE_STATUSES:
                    response["data"] = _ensure_generating_reply_markdown(
                        _extract_data(response),
                        str(request_body.get("session_id") or session_id),
                        normalized_host_capabilities,
                    )

        save_last_request_debug(
            state_path,
            state,
            build_request_debug_snapshot(
                base_url=effective_request_url,
                request_body=request_body,
                headers=headers,
                response_payload=response,
                http_status=http_status,
                transport_error=transport_error,
                method="GET" if request_query else "POST",
                query_params=request_query,
            ),
        )

        code = int(response.get("code", 0) or 0)
        if code != 0:
            if code == 401:
                _save_state_at(state_path, state)
                return (
                    _build_auth_required_response(
                        reason="invalid",
                        session_id=str(request_body.get("session_id") or session_id),
                        host_capabilities=normalized_host_capabilities,
                    ),
                    0,
                )

            message_text = str(response.get("msg") or response.get("message") or response.get("error") or "").strip()
            if code == 400 and message_text:
                _save_state_at(state_path, state)
                return (
                    build_business_error_response(
                        message_text,
                        session_id=str(request_body.get("session_id") or session_id),
                        host_capabilities=normalized_host_capabilities,
                    ),
                    0,
                )

            _save_state_at(state_path, state)
            return (
                build_user_safe_failure_response(
                    session_id=str(request_body.get("session_id") or session_id),
                    host_capabilities=normalized_host_capabilities,
                ),
                0,
            )

        data = _extract_data(response)
        session_event = "pending_request_replayed" if explicit_credential and pending_request else ""
        data = _ensure_response_envelope(
            data,
            session_id=str(data.get("session_id") or request_body.get("session_id") or session_id),
            session_event=session_event,
            host_capabilities=normalized_host_capabilities,
        )
        if explicit_credential and pending_request:
            state.pop("pending_request", None)
        state["session_id"] = str(data.get("session_id") or request_body.get("session_id") or session_id)
        _save_state_at(state_path, state)
        return data, 0
    except Exception as exc:
        return {"error": str(exc)}, 1


def main() -> int:
    payload = read_stdin(sys.stdin.read())
    response, code = run_payload(payload)
    print(json.dumps(response, ensure_ascii=False))
    return code


if __name__ == "__main__":
    raise SystemExit(main())