#!/usr/bin/env python3 """One-command text, audio, or explicit MQ AI probe. Mock is isolated by default. The Bailian adapter uses only server-side BAILIAN_* environment variables; real call mode still fails closed until the Cell/media authorization and real-call evidence gate is explicitly arranged. """ from __future__ import annotations import argparse import hashlib import importlib import json import sys import tempfile import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) _ai_runtime = importlib.import_module("agent_call.ai_runtime") _core = importlib.import_module("agent_call.core") _bailian = importlib.import_module("agent_call.bailian") AIConfigError = _ai_runtime.AIConfigError AIProviderError = _ai_runtime.AIProviderError ConversationEngine = _ai_runtime.ConversationEngine build_mock_config = _ai_runtime.build_mock_config config_digest = _ai_runtime.config_digest load_prompt = _ai_runtime.load_prompt pcm_to_wav = _ai_runtime.pcm_to_wav validate_wav = _ai_runtime.validate_wav BailianASR = _bailian.BailianASR BailianLLM = _bailian.BailianLLM BailianTTS = _bailian.BailianTTS build_bailian_config = _bailian.build_bailian_config AgentCallService = _core.AgentCallService ServiceError = _core.ServiceError DEFAULT_PROFILE = ROOT / "configs" / "ai-test.example.yaml" def _load_data(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError: try: import yaml # type: ignore[import-not-found] except ImportError as exc: raise AIConfigError( "PROFILE_FORMAT_UNSUPPORTED", "profile must be JSON-compatible YAML when PyYAML is unavailable", ) from exc value = yaml.safe_load(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise AIConfigError("PROFILE_INVALID", "profile must be an object") return value def _path(value: str) -> Path: candidate = Path(value) return candidate if candidate.is_absolute() else ROOT / candidate def _iso_now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _safe_result(result: dict[str, Any]) -> dict[str, Any]: clean = dict(result) if "audio" in clean: audio = clean.pop("audio") if isinstance(audio, bytes): clean["audio_bytes"] = len(audio) clean["audio_sha256"] = hashlib.sha256(audio).hexdigest() if isinstance(clean.get("segments"), list): clean["segments"] = [ {key: value for key, value in segment.items() if key != "audio"} for segment in clean["segments"] if isinstance(segment, dict) ] return clean def _write_report(report: dict[str, Any], output: str | None) -> None: payload = json.dumps(report, ensure_ascii=False, indent=2) + "\n" if output: destination = _path(output) destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text(payload, encoding="utf-8") print(payload, end="") def _args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("kind", choices=("text", "audio", "call")) parser.add_argument("--profile", default=str(DEFAULT_PROFILE)) parser.add_argument("--mode", choices=("mock", "mixed", "real")) parser.add_argument("--prompt-file", default="prompts/test-call.txt") parser.add_argument( "--input", help="UTF-8 text for text mode or mono PCM16 WAV for audio mode" ) parser.add_argument("--model", default=None) parser.add_argument("--tts-model", default=None) parser.add_argument("--voice", default=None) parser.add_argument("--language", default=None) parser.add_argument( "--max-duration", type=int, default=None, help="call duration limit in seconds" ) parser.add_argument("--tenant-id", default=None) parser.add_argument("--task-id", default=None) parser.add_argument("--callee", default=None) parser.add_argument("--agent-version-id", default=None) parser.add_argument( "--db", default=None, help="SQLite path for call mode or durable config evidence", ) parser.add_argument("--output", default=None, help="JSON report path") parser.add_argument( "--output-audio", default=None, help="WAV response path for text/audio mode" ) parser.add_argument( "--allow-real-call", action="store_true", help="explicitly acknowledge that call mode could place a call; mock never dials", ) parser.add_argument( "--human-listening", choices=("pass", "fail", "not_run"), default="not_run", help="manual listening result; it is evidence only, not a provider claim", ) return parser.parse_args() def _config( settings: dict[str, Any], args: argparse.Namespace, mode: str ) -> dict[str, Any]: prompt = load_prompt( _path(args.prompt_file), int(settings.get("max_prompt_bytes", 32768)) ) version_id = args.agent_version_id or settings.get( "agent_version_id", "agent_test_v1" ) if mode == "mock": config = build_mock_config( str(version_id), prompt, str(args.model or settings.get("model", "mock-chat-v1")), str(args.tts_model or settings.get("tts_model", "mock-tts-v1")), str(args.voice or settings.get("voice", "mock-neutral")), ) else: config = build_bailian_config( str(version_id), prompt, str(args.model or settings.get("model", "qwen-plus")), args.tts_model or settings.get("tts_model"), args.voice or settings.get("voice"), str(args.language or settings.get("language", "zh-CN")), settings.get("asr_model"), str(settings.get("opening", "")), int(settings.get("max_turns", 20)), int(settings.get("max_duration_ms", 120000)), ) return config def _base_report( settings: dict[str, Any], args: argparse.Namespace, config: dict[str, Any] ) -> dict[str, Any]: provider_modes = settings.get("provider_modes", {}) return { "schema_version": "1.0", "created_at": _iso_now(), "mode": settings.get("mode", "mock"), "requested_mode": args.mode or settings.get("mode", "mock"), "provider_modes": provider_modes, "agent_version_id": config["agent_version_id"], "agent_config_sha256": config_digest(config), "prompt_sha256": hashlib.sha256( config["prompt"]["text"].encode("utf-8") ).hexdigest(), "prompt_bytes": len(config["prompt"]["text"].encode("utf-8")), "llm": { "provider_ref": config["llm"]["provider_ref"], "requested_model": config["llm"]["model"], "provider_returned_model": None, "provider_returned_model_verified": False, }, "tts": { "provider_ref": config["tts"]["provider_ref"], "requested_model": config["tts"]["model"], "provider_returned_model": None, "provider_returned_model_verified": False, "voice": config["tts"]["voice"], "voice_listening_verified": False, "format": config["tts"]["format"], }, "asr": { "provider_ref": config["asr"]["provider_ref"], "language": config["asr"]["language"], }, "human_listening": args.human_listening, } def _run_real_cell( args: argparse.Namespace, settings: dict[str, Any], service_settings: dict[str, Any] ) -> tuple[int, dict[str, Any]]: tenant_id = str(args.tenant_id or settings.get("tenant_id", "tenant-demo")) tenant = next( ( item for item in service_settings.get("tenants", []) if item.get("tenant_id") == tenant_id ), None, ) if not isinstance(tenant, dict) or not isinstance(tenant.get("tenant_key"), str): raise AIConfigError( "TENANT_NOT_CONFIGURED", "real Cell call tenant is not configured" ) runner = importlib.import_module("scripts.run_real_cell_call") real_args = argparse.Namespace( profile=args.profile, prompt_file=args.prompt_file, callee=args.callee or settings.get("callee", ""), tenant_id=tenant_id, tenant_key=tenant["tenant_key"], broker_url=None, ledger=args.db or str(ROOT / ".local" / "agent-call-ai" / "real-cell.sqlite3"), wait_seconds=240, allow_real_call=args.allow_real_call, output=None, ) report = runner.run(real_args) return (0 if report.get("status") == "PASS" else 1), report def _run(args: argparse.Namespace) -> tuple[int, dict[str, Any]]: settings = _load_data(_path(args.profile)) service_profile = _path( str(settings.get("service_profile", "docs/contracts/mock-profile.json")) ) service_settings = _load_data(service_profile) requested_mode = args.mode or str( settings.get("mode", service_settings.get("mode", "mock")) ) if requested_mode not in {"mock", "mixed", "real"}: raise AIConfigError("MODE_INVALID", "mode must be mock, mixed, or real") if args.kind == "call" and requested_mode == "real": if not args.allow_real_call: return 2, { "schema_version": "1.0", "status": "BLOCKED", "requested_mode": requested_mode, "reason_code": "CALL_AUTHORIZATION_REQUIRED", "message": "call mode requires --allow-real-call before the RabbitMQ Cell worker is started.", } return _run_real_cell(args, settings, service_settings) if requested_mode != "mock" and args.kind == "call": return 2, { "schema_version": "1.0", "status": "BLOCKED", "requested_mode": requested_mode, "reason_code": "REAL_CALL_MODE_REQUIRED", "message": "Only the explicit real profile may start the Cell/media executor; no call was placed.", } config = _config(settings, args, requested_mode) report_settings = dict(settings) report_settings["mode"] = requested_mode report_settings["provider_modes"] = settings.get( "provider_modes", service_settings.get("provider_modes", {}) ) report = _base_report(report_settings, args, config) db_path: str temporary_db: tempfile.TemporaryDirectory[str] | None = None if args.db: db_path = str(_path(args.db)) Path(db_path).parent.mkdir(parents=True, exist_ok=True) else: temporary_db = tempfile.TemporaryDirectory(prefix="agent-call-ai-") db_path = str(Path(temporary_db.name) / "state.sqlite3") service = AgentCallService( db_path=db_path, profile_path=service_profile, start_background=False ) try: tenant_id = str(args.tenant_id or settings.get("tenant_id", "tenant-demo")) published = service.publish_agent_version( tenant_id, config["agent_version_id"], config, actor_id="ai-test-cli", ) trusted = service.get_agent_version(tenant_id, config["agent_version_id"]) report["publication"] = published report["trusted_snapshot"] = { "agent_version_id": trusted["agent_version_id"], "content_sha256": trusted["content_sha256"], "immutable": trusted["immutable"], } engine_kwargs: dict[str, Any] = {"journal": service.journal} if requested_mode != "mock": engine_kwargs["llm"] = BailianLLM.from_env() engine_kwargs["tts"] = BailianTTS.from_env() if trusted["config"]["asr"]["provider_ref"] != "mock": engine_kwargs["asr"] = BailianASR.from_env() engine = ConversationEngine(trusted["config"], **engine_kwargs) if args.kind == "text": if args.input is None: raise AIConfigError( "TEXT_INPUT_REQUIRED", "--input is required for text mode" ) result = engine.run_text(args.input) elif args.kind == "audio": if args.input is None: raise AIConfigError( "AUDIO_INPUT_REQUIRED", "--input is required for audio mode" ) audio_path = _path(args.input) audio = audio_path.read_bytes() result = engine.run_audio(audio) else: if not args.allow_real_call: report.update( { "status": "BLOCKED", "reason_code": "CALL_AUTHORIZATION_REQUIRED", "message": "call mode requires --allow-real-call even though the selected profile is Mock.", } ) return 2, report callee = str(args.callee or settings.get("callee", "")) authorized = { str(value) for value in settings.get("authorized_callees", [callee]) } if callee not in authorized: raise AIConfigError( "CALLEE_NOT_AUTHORIZED", "callee is not in the explicit test allow-list", ) fixture = _load_data(ROOT / "docs/contracts/examples/call.execute.json") now = datetime.now(timezone.utc) fixture["command_id"] = f"cmd_ai_{uuid.uuid4().hex}" fixture["trace_id"] = f"trace_ai_{uuid.uuid4().hex}" fixture["issued_at"] = now.isoformat().replace("+00:00", "Z") fixture["not_after"] = ( now.replace(year=now.year + 1).isoformat().replace("+00:00", "Z") ) fixture["tenant_id"] = tenant_id fixture["tenant_key"] = next( item["tenant_key"] for item in service.profile["tenants"] if item["tenant_id"] == tenant_id ) fixture["payload"]["execution_id"] = f"exec_ai_{uuid.uuid4().hex}" fixture["payload"]["task_id"] = str( args.task_id or settings.get("task_id", "task-demo") ) fixture["payload"]["callee"] = callee fixture["payload"]["agent_version_id"] = config["agent_version_id"] if args.max_duration is not None: if args.max_duration < 1: raise AIConfigError( "MAX_DURATION_INVALID", "--max-duration must be positive" ) fixture["payload"]["max_call_duration_ms"] = args.max_duration * 1000 published_command = service.publish_execute(fixture) service.wait_for_idle(10) call_row = service.store.one( "SELECT call_id FROM calls WHERE execution_id=?", (fixture["payload"]["execution_id"],), ) if call_row is None: raise ServiceError( "CALL_NOT_CREATED", "mock executor did not reserve a call" ) result = service.get_call(tenant_id, call_row["call_id"]) report["mq_publication"] = published_command report["call_id"] = call_row["call_id"] clean = _safe_result(result) if result.get("provider_returned_model"): report["llm"]["provider_returned_model"] = result["provider_returned_model"] report["llm"]["provider_returned_model_verified"] = True if result.get("tts_provider_returned_model"): report["tts"]["provider_returned_model"] = result[ "tts_provider_returned_model" ] report["tts"]["provider_returned_model_verified"] = True if isinstance(result.get("tts_model_evidence"), dict): report["tts"]["model_evidence"] = result["tts_model_evidence"] if args.kind in {"text", "audio"}: raw_audio = result.get("audio", b"") if isinstance(raw_audio, bytes) and raw_audio: wav = pcm_to_wav( raw_audio, int(config["tts"]["format"]["sample_rate_hz"]) ) validate_wav(wav) if args.output_audio: destination = _path(args.output_audio) else: destination = ( ROOT / ".local" / "agent-call-ai" / f"response-{uuid.uuid4().hex}.wav" ) destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(wav) report["audio_output"] = { "path": str(destination), "bytes": len(wav), "sha256": hashlib.sha256(wav).hexdigest(), } report["result"] = clean result_status = ( clean.get("status") or result.get("outcome") or result.get("call_state") ) report["status"] = ( "PASS" if result_status in {"completed", "succeeded", "ended"} else "FAIL" ) return (0 if report["status"] == "PASS" else 1), report finally: service.store.close() if temporary_db is not None: temporary_db.cleanup() def main() -> int: args = _args() try: code, report = _run(args) except (AIConfigError, AIProviderError, ServiceError, OSError, ValueError) as exc: code = 1 report = { "schema_version": "1.0", "status": "FAIL", "reason_code": getattr(exc, "code", type(exc).__name__), "message": str(exc), } _write_report(report, args.output) return code if __name__ == "__main__": raise SystemExit(main())