#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import re import socket import sys from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") ROOT = Path(__file__).resolve().parents[1] DEFAULT_REPORT = ROOT / "reports" / "1c-mcp-selector-chain-smoke.json" EXPECTED_CONTRACT_VERSION = "onec-selector-contract.v1" FORBIDDEN_CONCRETE_SELECTOR_VALUES = ( "ПечатьЭтикеток", "УОП_Печать", "ЦенаСоСкидкой", "fs_Отчеты", ) DEFAULT_DISCOVERY_KINDS = ("DataProcessor", "Report", "Document", "Catalog", "InformationRegister") LIVE_HELP_SELECTOR_METHODS = ( "metadata.definition.find", "metadata.object.modules", "modules.read", "code.search", "code.read", ) WORKING_STATE_METHODS = { "extension.objects.find", "metadata.resolve_overrides", "metadata.object.forms", "metadata.object.form.details", "metadata.form.decode", "modules.search", "code.search", } SEARCH_TOKEN_SKIP = { "procedure", "function", "endprocedure", "endfunction", "процедура", "функция", "конецпроцедуры", "конецфункции", } IDENTIFIER_LIKE_RE = re.compile(r"^[A-Za-zА-Яа-я_][A-Za-zА-Яа-я0-9_]*$") def is_identifier_like(value: Any) -> bool: return bool(IDENTIFIER_LIKE_RE.fullmatch(str(value or "").strip())) def text_from_module_read(result: dict[str, Any]) -> str: for key in ("text", "preview"): value = result.get(key) if isinstance(value, str) and value.strip(): return value selection = result.get("selection") if isinstance(selection, dict): for key in ("text", "preview"): value = selection.get(key) if isinstance(value, str) and value.strip(): return value return "" def search_token_from_text(text: str) -> str | None: for match in re.finditer(r"[A-Za-zА-Яа-я_][A-Za-zА-Яа-я0-9_]{3,}", text): token = match.group(0) if token.casefold() not in SEARCH_TOKEN_SKIP: return token return None def search_token_from_module_read(result: dict[str, Any]) -> tuple[str | None, str | None]: text_token = search_token_from_text(text_from_module_read(result)) if text_token: return text_token, "modules.read.text" for routine in result.get("routines") or []: if not isinstance(routine, dict): continue name = routine.get("name") if is_identifier_like(name): return str(name), "modules.read.routines" return "Процедура", "generic_bsl_keyword" def routine_name_from_module_read(result: dict[str, Any]) -> str | None: for routine in result.get("routines") or []: if not isinstance(routine, dict): continue name = routine.get("name") if is_identifier_like(name): return str(name) selection = result.get("selection") if isinstance(selection, dict): for key in ("routine_name", "name"): name = selection.get(key) if is_identifier_like(name): return str(name) return None def first_write_plan_target_from_saved_state(result: dict[str, Any]) -> dict[str, Any] | None: for module in result.get("modules") or []: if not isinstance(module, dict): continue for stream in module.get("streams") or []: if not isinstance(stream, dict): continue target = stream.get("write_plan_target") if not isinstance(target, dict): continue has_internal_handle = bool(target.get("module_ref") and target.get("expected_sha1")) has_public_selector = bool(target.get("ref") and target.get("module")) if has_internal_handle or has_public_selector: return target return None def first_public_form_write_plan_candidate(result: dict[str, Any]) -> dict[str, Any] | None: candidates: list[dict[str, Any]] = [] for form in result.get("forms") or []: if not isinstance(form, dict): continue for match in form.get("matches") or []: if not isinstance(match, dict): continue selector = match.get("selector") if isinstance(match.get("selector"), dict) else {} if not selector.get("ref") or not selector.get("form"): continue if not any(selector.get(key) for key in ("command", "element", "attribute")): continue if any(key in selector for key in ("table", "file_name", "guid", "form_guid")): continue for prop in match.get("writable_properties") or []: if not isinstance(prop, dict) or prop.get("value_type") != "string": continue property_name = str(prop.get("presentation") or prop.get("property") or "").strip() if not property_name: continue old_value = str(prop.get("value") or "") normalized_property = str(prop.get("property") or "").strip().casefold() priority = 0 if normalized_property == "title" else (2 if normalized_property in {"id", "name"} else 1) candidates.append( { "priority": priority, "target": { "kind": "form", **{ key: value for key, value in selector.items() if key in {"extension", "ref", "form", "command", "element", "attribute"} and value not in (None, "") }, }, "edit": { "property": property_name, "value": f"{old_value} [NAME-FIRST PLAN]", }, } ) if not candidates: return None candidate = sorted(candidates, key=lambda item: int(item.get("priority") or 0))[0] candidate.pop("priority", None) return candidate def live_coverage_from_steps(steps: list[dict[str, Any]]) -> dict[str, Any]: def last_step(name: str) -> dict[str, Any]: for step in reversed(steps): if isinstance(step, dict) and step.get("name") == name: return step return {} resolve = last_step("metadata.resolve_overrides") saved_state = last_step("metadata.saved_state.modules.search") write_plan = last_step("metadata.write.plan") form_search = last_step("metadata.saved_state.forms.search") form_write_plan = last_step("metadata.write.plan.form") skips = [ {"step": str(step.get("name")), "status": step.get("status"), **({"reason": step.get("reason")} if step.get("reason") else {})} for step in steps if isinstance(step.get("status"), str) and str(step.get("status")).startswith("skipped") ] return { "resolve_overrides": { "attempted": bool(resolve), "status": resolve.get("status"), "write_plan_evidence": bool(resolve.get("write_plan_evidence")), "next_method": resolve.get("next_method"), }, "saved_state_resolution": { "attempted": bool(saved_state), "status": saved_state.get("status"), "modules": int(saved_state.get("modules") or 0), "write_plan_target": bool(saved_state.get("write_plan_target")), }, "write_plan_composition": { "attempted": bool(write_plan), "status": write_plan.get("status"), "composed": write_plan.get("allowed") is True, "from_write_plan_target": bool(write_plan.get("from_write_plan_target")), }, "form_write_plan_composition": { "attempted": bool(form_write_plan), "search_attempted": bool(form_search), "search_status": form_search.get("status"), "candidate": bool(form_search.get("write_plan_candidate")), "status": form_write_plan.get("status"), "composed": form_write_plan.get("allowed") is True, "name_first": bool(form_write_plan.get("name_first")), }, "skips": skips, } def live_write_plan_composition_required_issue(coverage: dict[str, Any]) -> dict[str, Any] | None: composition = coverage.get("write_plan_composition") if isinstance(coverage.get("write_plan_composition"), dict) else {} if composition.get("composed") is True: return None return { "code": "live_write_plan_composition_required", "status": composition.get("status"), "saved_state_resolution": coverage.get("saved_state_resolution"), } def live_form_write_plan_composition_required_issue(coverage: dict[str, Any]) -> dict[str, Any] | None: composition = coverage.get("form_write_plan_composition") if isinstance(coverage.get("form_write_plan_composition"), dict) else {} if composition.get("composed") is True and composition.get("name_first") is True: return None return { "code": "live_form_write_plan_composition_required", "status": composition.get("status"), "search_status": composition.get("search_status"), "candidate": composition.get("candidate"), } def live_object_summary(*sources: dict[str, Any] | None) -> dict[str, Any]: summary: dict[str, Any] = {} for source in sources: if not isinstance(source, dict): continue for key in ("kind", "name", "guid", "ref"): value = source.get(key) if value is not None and value != "": summary[key] = value if "ref" not in summary and summary.get("kind") and summary.get("name"): summary["ref"] = f"{summary['kind']}.{summary['name']}" return summary def selector_from_ref(ref: str | None) -> dict[str, str]: text = str(ref or "").strip() if "." not in text: return {} kind, name = text.split(".", 1) if not kind.strip() or not name.strip(): return {} return {"kind": kind.strip(), "name": name.strip(), "ref": text} def post_json(url: str, payload: dict[str, Any], *, timeout: float, headers: dict[str, str] | None = None) -> tuple[dict[str, str], dict[str, Any]]: request = Request( url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), method="POST", headers={"Content-Type": "application/json", **(headers or {})}, ) try: with urlopen(request, timeout=timeout) as response: raw = response.read().decode("utf-8") data = json.loads(raw) if raw else {} return dict(response.headers), data if isinstance(data, dict) else {"status": "error", "error": "response is not an object", "response": data} except (TimeoutError, socket.timeout) as exc: return {}, {"status": "error", "error": "timeout", "message": str(exc)} def rest_rpc_call(adapter_url: str, method: str, payload: dict[str, Any], *, timeout: float) -> dict[str, Any]: body = json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8") service_token = os.environ.get("ONEC_ADAPTER_SERVICE_TOKEN", "").strip() headers = {"Content-Type": "application/json"} if service_token: headers["Authorization"] = f"Bearer {service_token}" request = Request( adapter_url.rstrip("/") + "/rpc", data=body, method="POST", headers=headers, ) try: with urlopen(request, timeout=timeout) as response: raw = response.read().decode("utf-8") data = json.loads(raw) if raw else {} return data if isinstance(data, dict) else {"status": "error", "error": "rpc response is not an object", "response": data} except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") return {"status": "error", "error": "http_error", "http_status": exc.code, "response": raw} except URLError as exc: return {"status": "error", "error": "url_error", "message": str(exc)} except (TimeoutError, socket.timeout) as exc: return {"status": "error", "error": "timeout", "message": str(exc)} def mcp_rpc_call(mcp_url: str, method: str, payload: dict[str, Any], *, timeout: float, session_id: str | None = None) -> dict[str, Any]: headers = {"Accept": "application/json, text/event-stream"} if session_id: headers["Mcp-Session-Id"] = session_id body = { "jsonrpc": "2.0", "id": f"smoke-{method}", "method": "tools/call", "params": {"name": "onec_request", "arguments": {"method": method, "payload": payload}}, } try: _, data = post_json(mcp_url.rstrip("/") + "/mcp", body, timeout=timeout, headers=headers) except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") return {"status": "error", "error": "mcp_http_error", "http_status": exc.code, "response": raw} except URLError as exc: return {"status": "error", "error": "mcp_url_error", "message": str(exc)} if "error" in data: return {"status": "error", "error": "mcp_jsonrpc_error", "details": data.get("error")} result = data.get("result") if isinstance(data.get("result"), dict) else {} content = result.get("content") if isinstance(result, dict) else None if not isinstance(content, list) or not content: return {"status": "error", "error": "mcp_tool_content_missing", "result": result} first = content[0] if isinstance(content[0], dict) else {} text = first.get("text") if not isinstance(text, str): return {"status": "error", "error": "mcp_tool_text_missing", "content": content} try: decoded = json.loads(text) except json.JSONDecodeError as exc: return {"status": "error", "error": "mcp_tool_text_not_json", "message": str(exc), "text": text[:500]} return decoded if isinstance(decoded, dict) else {"status": "error", "error": "mcp_tool_payload_not_object", "payload": decoded} def mcp_initialize(mcp_url: str, *, timeout: float) -> tuple[str | None, dict[str, Any]]: body = { "jsonrpc": "2.0", "id": "smoke-initialize", "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "smoke_1c_mcp_selector_chain", "version": "1"}, }, } try: headers, data = post_json( mcp_url.rstrip("/") + "/mcp", body, timeout=timeout, headers={"Accept": "application/json, text/event-stream"}, ) except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") return None, {"status": "error", "error": "mcp_initialize_http_error", "http_status": exc.code, "response": raw} except URLError as exc: return None, {"status": "error", "error": "mcp_initialize_url_error", "message": str(exc)} except (TimeoutError, socket.timeout) as exc: return None, {"status": "error", "error": "mcp_initialize_timeout", "message": str(exc)} if "error" in data: return None, {"status": "error", "error": "mcp_initialize_jsonrpc_error", "details": data.get("error")} return headers.get("Mcp-Session-Id"), data def mcp_onec_request_schema_check(mcp_url: str, *, timeout: float, session_id: str | None) -> tuple[dict[str, Any], list[dict[str, Any]]]: headers = {"Accept": "application/json, text/event-stream"} if session_id: headers["Mcp-Session-Id"] = session_id body = {"jsonrpc": "2.0", "id": "smoke-tools-list", "method": "tools/list", "params": {}} try: _, data = post_json(mcp_url.rstrip("/") + "/mcp", body, timeout=timeout, headers=headers) except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") return {"status": "error", "error": "mcp_tools_list_http_error", "http_status": exc.code}, [{"code": "live_mcp_tools_list_failed", "response": raw}] except URLError as exc: return {"status": "error", "error": "mcp_tools_list_url_error", "message": str(exc)}, [{"code": "live_mcp_tools_list_failed", "message": str(exc)}] if "error" in data: return {"status": "error", "error": "mcp_tools_list_jsonrpc_error", "details": data.get("error")}, [{"code": "live_mcp_tools_list_failed", "details": data.get("error")}] result = data.get("result") if isinstance(data.get("result"), dict) else {} tools = result.get("tools") or [] onec_request = next((tool for tool in tools if isinstance(tool, dict) and tool.get("name") == "onec_request"), None) contract_version = result.get("contract_version") summary = { "status": "ok", "contract_version": contract_version, "tools": len(tools), "onec_request": bool(onec_request), "selector_guidance": False, "placeholder_examples": False, } issues: list[dict[str, Any]] = [] if contract_version != EXPECTED_CONTRACT_VERSION: issues.append({"code": "live_mcp_contract_version_mismatch", "expected": EXPECTED_CONTRACT_VERSION, "actual": contract_version}) if not isinstance(onec_request, dict): issues.append({"code": "live_mcp_onec_request_missing"}) return summary, issues schema = onec_request.get("inputSchema") if isinstance(onec_request.get("inputSchema"), dict) else {} description = str(onec_request.get("description") or "") payload_schema = (schema.get("properties") or {}).get("payload") or {} payload_description = str(payload_schema.get("description") or "") selector_terms = ("ref", "kind/name/guid", "object_type/object_name/object_guid") summary["selector_guidance"] = all(term in description and term in payload_description for term in selector_terms) summary["placeholder_examples"] = "." in json.dumps(schema.get("examples") or [], ensure_ascii=False) if schema.get("required") != ["method"]: issues.append({"code": "live_mcp_onec_request_required_not_generic", "required": schema.get("required")}) if "oneOf" in schema: issues.append({"code": "live_mcp_onec_request_enumerates_methods"}) for term in selector_terms: if term not in description or term not in payload_description: issues.append({"code": "live_mcp_selector_guidance_missing", "term": term}) if not summary["placeholder_examples"]: issues.append({"code": "live_mcp_placeholder_examples_missing"}) return summary, issues def rpc_call(endpoint_url: str, method: str, payload: dict[str, Any], *, timeout: float, transport: str = "rest", session_id: str | None = None) -> dict[str, Any]: if transport == "mcp": return mcp_rpc_call(endpoint_url, method, payload, timeout=timeout, session_id=session_id) return rest_rpc_call(endpoint_url, method, payload, timeout=timeout) def with_working_state(payload: dict[str, Any], *, transport: str) -> dict[str, Any]: if payload.get("state") or payload.get("source_state"): return payload if transport == "mcp": return {**payload, "source_state": "working"} return {**payload, "state": "working"} def adapter_help_methods_contract(endpoint_url: str, *, timeout: float, transport: str, session_id: str | None = None) -> tuple[dict[str, Any], list[dict[str, Any]]]: result = rpc_call(endpoint_url, "help.methods", {}, timeout=timeout, transport=transport, session_id=session_id) methods = { str(item.get("name") or ""): item for item in result.get("methods") or [] if isinstance(item, dict) and item.get("name") } contract_version = result.get("contract_version") summary = {"status": result.get("status") or ("ok" if methods else "error"), "contract_version": contract_version, "methods": len(methods), "selector_capabilities": False} issues: list[dict[str, Any]] = [] if contract_version != EXPECTED_CONTRACT_VERSION: issues.append({"code": "live_adapter_contract_version_mismatch", "expected": EXPECTED_CONTRACT_VERSION, "actual": contract_version}) if not methods or result.get("status") == "error": return summary, [{"code": "live_adapter_help_methods_failed", "status": result.get("status"), "error": result.get("error"), "diagnostics": result.get("diagnostics")}] for method in LIVE_HELP_SELECTOR_METHODS: capabilities = (methods.get(method) or {}).get("selector_capabilities") if not isinstance(capabilities, dict): issues.append({"code": "live_adapter_selector_capabilities_missing", "method": method}) continue if not capabilities.get("accepts_ref") or not capabilities.get("accepts_object_aliases"): issues.append({"code": "live_adapter_selector_capabilities_incomplete", "method": method, "selector_capabilities": capabilities}) definition_capabilities = (methods.get("metadata.definition.find") or {}).get("selector_capabilities") or {} if not definition_capabilities.get("allows_global_areas") or not definition_capabilities.get("requires_query"): issues.append({"code": "live_adapter_definition_capabilities_incomplete", "selector_capabilities": definition_capabilities}) if not ((methods.get("modules.read") or {}).get("selector_capabilities") or {}).get("accepts_module_ref"): issues.append({"code": "live_adapter_modules_read_module_ref_missing"}) if not ((methods.get("code.read") or {}).get("selector_capabilities") or {}).get("accepts_module_ref"): issues.append({"code": "live_adapter_code_read_module_ref_missing"}) if not ((methods.get("code.search") or {}).get("selector_capabilities") or {}).get("requires_query"): issues.append({"code": "live_adapter_code_search_requires_query_missing"}) summary["selector_capabilities"] = not issues return summary, issues def first_object_with_modules(endpoint_url: str, base_id: str, *, timeout: float, kinds: tuple[str, ...], transport: str = "rest", session_id: str | None = None) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]: attempts: list[dict[str, Any]] = [] for kind in kinds: listed = rpc_call( endpoint_url, "metadata.objects.list", {"base_id": base_id, "kind": kind, "limit": 50, "include_storage": False}, timeout=timeout, transport=transport, session_id=session_id, ) objects = [item for item in listed.get("objects") or [] if isinstance(item, dict)] objects = sorted(objects, key=lambda item: (not is_identifier_like(item.get("name")), str(item.get("name") or ""))) attempts.append({"method": "metadata.objects.list", "kind": kind, "status": listed.get("status"), "objects": len(objects)}) for obj in objects: if not obj.get("name") and not obj.get("guid"): continue selector = { "base_id": base_id, "kind": obj.get("kind") or kind, **({"name": obj.get("name")} if obj.get("name") else {}), **({"guid": obj.get("guid")} if obj.get("guid") else {}), "include_storage": False, } modules = rpc_call(endpoint_url, "metadata.object.modules", selector, timeout=timeout, transport=transport, session_id=session_id) module_items = [item for item in modules.get("modules") or [] if isinstance(item, dict)] attempts.append( { "method": "metadata.object.modules", "kind": selector.get("kind"), "status": modules.get("status"), "modules": len(module_items), } ) if modules.get("status") == "ok" and module_items: module_owner = modules.get("object") if isinstance(modules.get("object"), dict) else {} object_card: dict[str, Any] = {} if obj.get("guid") or module_owner.get("guid"): card = rpc_call( endpoint_url, "metadata.object.get", { "base_id": base_id, "kind": module_owner.get("kind") or obj.get("kind") or kind, "guid": module_owner.get("guid") or obj.get("guid"), "include_semantic": False, }, timeout=timeout, transport=transport, session_id=session_id, ) object_card = card.get("object") if isinstance(card.get("object"), dict) else {} attempts.append({"method": "metadata.object.get", "kind": selector.get("kind"), "status": card.get("status"), "resolved": bool(object_card)}) merged_object = { **obj, **{key: value for key, value in module_owner.items() if value is not None and value != ""}, **{key: value for key, value in object_card.items() if value is not None and value != ""}, } return {"object": merged_object, "modules": modules, "module": module_items[0]}, attempts return None, attempts def selected_object_with_modules(endpoint_url: str, base_id: str, selector: dict[str, Any], *, timeout: float, transport: str = "rest", session_id: str | None = None) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]: attempts: list[dict[str, Any]] = [] payload = {"base_id": base_id, **{key: value for key, value in selector.items() if value is not None and value != ""}, "include_storage": False} modules = rpc_call(endpoint_url, "metadata.object.modules", payload, timeout=timeout, transport=transport, session_id=session_id) module_items = [item for item in modules.get("modules") or [] if isinstance(item, dict)] attempts.append({"method": "metadata.object.modules", "kind": payload.get("kind"), "status": modules.get("status"), "modules": len(module_items), "selected": True}) if modules.get("status") != "ok" or not module_items: return None, attempts module_owner = modules.get("object") if isinstance(modules.get("object"), dict) else {} object_card: dict[str, Any] = {} get_payload = { "base_id": base_id, **({"kind": module_owner.get("kind") or payload.get("kind")} if module_owner.get("kind") or payload.get("kind") else {}), **({"name": module_owner.get("name") or payload.get("name")} if module_owner.get("name") or payload.get("name") else {}), **({"guid": module_owner.get("guid") or payload.get("guid")} if module_owner.get("guid") or payload.get("guid") else {}), **({"ref": module_owner.get("ref") or payload.get("ref")} if module_owner.get("ref") or payload.get("ref") else {}), "include_semantic": False, } card = rpc_call(endpoint_url, "metadata.object.get", get_payload, timeout=timeout, transport=transport, session_id=session_id) object_card = card.get("object") if isinstance(card.get("object"), dict) else {} attempts.append({"method": "metadata.object.get", "kind": get_payload.get("kind"), "status": card.get("status"), "resolved": bool(object_card), "selected": True}) merged_object = { **selector, **{key: value for key, value in module_owner.items() if value is not None and value != ""}, **{key: value for key, value in object_card.items() if value is not None and value != ""}, } return {"object": merged_object, "modules": modules, "module": module_items[0]}, attempts def build_live_report( endpoint_url: str, base_id: str, *, timeout: float, kinds: tuple[str, ...] = DEFAULT_DISCOVERY_KINDS, transport: str = "rest", object_selector: dict[str, Any] | None = None, require_write_plan_composition: bool = False, require_form_write_plan_composition: bool = False, ) -> dict[str, Any]: steps: list[dict[str, Any]] = [] issues: list[dict[str, Any]] = [] session_id: str | None = None if transport == "mcp": session_id, initialize_result = mcp_initialize(endpoint_url, timeout=timeout) initialize_contract_version = (((initialize_result.get("result") or {}).get("serverInfo") or {}).get("contract_version") if isinstance(initialize_result.get("result"), dict) else None) steps.append( { "name": "mcp.initialize", "status": "ok" if session_id and initialize_contract_version == EXPECTED_CONTRACT_VERSION else "error", "session": bool(session_id), "contract_version": initialize_contract_version, } ) if not session_id: issues.append({"code": "live_mcp_initialize_failed", "result": initialize_result}) return {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "endpoint_url": endpoint_url, "transport": transport, "steps": steps, "issues": issues} if initialize_contract_version != EXPECTED_CONTRACT_VERSION: issues.append({"code": "live_mcp_initialize_contract_version_mismatch", "expected": EXPECTED_CONTRACT_VERSION, "actual": initialize_contract_version}) return {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "endpoint_url": endpoint_url, "transport": transport, "steps": steps, "issues": issues} schema_summary, schema_issues = mcp_onec_request_schema_check(endpoint_url, timeout=timeout, session_id=session_id) steps.append( { "name": "mcp.tools.list", "status": "ok" if not schema_issues else "error", "contract_version": schema_summary.get("contract_version"), "onec_request": bool(schema_summary.get("onec_request")), "selector_guidance": bool(schema_summary.get("selector_guidance")), "placeholder_examples": bool(schema_summary.get("placeholder_examples")), } ) if schema_issues: issues.extend(schema_issues) return {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "endpoint_url": endpoint_url, "transport": transport, "steps": steps, "issues": issues} help_summary, help_issues = adapter_help_methods_contract(endpoint_url, timeout=timeout, transport=transport, session_id=session_id) steps.append( { "name": "adapter.help.methods", "status": "ok" if not help_issues else "error", "contract_version": help_summary.get("contract_version"), "methods": int(help_summary.get("methods") or 0), "selector_capabilities": bool(help_summary.get("selector_capabilities")), } ) if help_issues: issues.extend(help_issues) return {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "endpoint_url": endpoint_url, "transport": transport, "steps": steps, "issues": issues} if object_selector: discovered, discovery_attempts = selected_object_with_modules(endpoint_url, base_id, object_selector, timeout=timeout, transport=transport, session_id=session_id) else: discovered, discovery_attempts = first_object_with_modules(endpoint_url, base_id, timeout=timeout, kinds=kinds, transport=transport, session_id=session_id) steps.extend({"name": "discover", **attempt} for attempt in discovery_attempts) if not discovered: issues.append({"code": "live_object_with_modules_not_found", **({"selector": object_selector} if object_selector else {"kinds": list(kinds)})}) return {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "steps": steps, "issues": issues} obj = discovered["object"] object_kind = obj.get("kind") object_name = obj.get("name") object_guid = obj.get("guid") object_ref = f"{object_kind}.{object_name}" if object_kind and object_name else None if not object_ref and not object_guid: issues.append({"code": "live_discovered_object_ref_missing", "object": obj}) return {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "steps": steps, "issues": issues} definition_query = str(object_guid or object_ref) definition_payload = { "base_id": base_id, "query": definition_query, "areas": ["metadata"], "max_matches": 5, "exact_only": True, "timeout_seconds": int(timeout), **({"kind": object_kind} if object_kind else {}), **({"guid": object_guid} if object_guid else {}), } if object_ref and is_identifier_like(object_name): definition_payload["ref"] = object_ref definition = rpc_call( endpoint_url, "metadata.definition.find", definition_payload, timeout=timeout, transport=transport, session_id=session_id, ) steps.append({"name": "metadata.definition.find", "status": definition.get("status"), "ref_present": bool(definition_payload.get("ref")), "guid_present": bool(definition_payload.get("guid"))}) related = definition.get("related_selectors") if isinstance(definition.get("related_selectors"), dict) else {} modules_selector = related.get("modules") if isinstance(related.get("modules"), dict) else None canonical_object = definition.get("object") if isinstance(definition.get("object"), dict) else {} if definition.get("status") != "ok" or not modules_selector: issues.append( { "code": "live_definition_related_modules_missing", "status": definition.get("status"), "error": definition.get("error"), "diagnostics": definition.get("diagnostics"), } ) return { "schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "endpoint_url": endpoint_url, "transport": transport, "base_id": base_id, "object": live_object_summary({"kind": object_kind, "name": object_name, "guid": object_guid, "ref": object_ref}, canonical_object, modules_selector), "steps": steps, "issues": issues, } canonical_summary = live_object_summary({"kind": object_kind, "name": object_name, "guid": object_guid, "ref": object_ref}, canonical_object, modules_selector) object_kind = canonical_summary.get("kind") or object_kind object_name = canonical_summary.get("name") or object_name object_guid = canonical_summary.get("guid") or object_guid object_ref = canonical_summary.get("ref") or object_ref if not modules_selector.get("ref"): issues.append({"code": "live_related_selector_ref_missing", "selector": modules_selector}) modules = rpc_call(endpoint_url, str(modules_selector.get("method") or "metadata.object.modules"), modules_selector, timeout=timeout, transport=transport, session_id=session_id) module_items = [item for item in modules.get("modules") or [] if isinstance(item, dict)] steps.append({"name": "related_selectors.modules", "status": modules.get("status"), "modules": len(module_items)}) if modules.get("status") != "ok" or not module_items: issues.append({"code": "live_related_modules_empty", "status": modules.get("status")}) return {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "steps": steps, "issues": issues} module_ordinal = module_items[0].get("module_ordinal") or 1 read_payload = { **{key: value for key, value in modules_selector.items() if key != "method" and value is not None and value != ""}, "base_id": modules_selector.get("base_id") or base_id, "module_ordinal": module_ordinal, "preview": True, "max_chars": 1000, } if not read_payload.get("ref") and object_ref: read_payload["ref"] = object_ref read_result = rpc_call(endpoint_url, "modules.read", read_payload, timeout=timeout, transport=transport, session_id=session_id) steps.append({"name": "modules.read", "status": read_result.get("status"), "module_ordinal": module_ordinal, "used_ref": bool(read_payload.get("ref"))}) if read_result.get("status") not in {"ok", "summary"}: issues.append({"code": "live_modules_read_failed", "status": read_result.get("status"), "diagnostics": read_result.get("diagnostics")}) return { "schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "endpoint_url": endpoint_url, "transport": transport, "base_id": base_id, "object": live_object_summary({"kind": object_kind, "name": object_name, "guid": object_guid, "ref": object_ref}, canonical_object, modules_selector), "steps": steps, "issues": issues, } routine_name = routine_name_from_module_read(read_result) if routine_name: resolve_payload = { "base_id": base_id, "method_name": routine_name, "max_items": 10, "timeout_seconds": int(timeout), **({"ref": object_ref} if object_ref else {}), **({"object_type": object_kind} if object_kind else {}), **({"object_name": object_name} if object_name else {}), **({"object_guid": object_guid} if object_guid else {}), } resolve_payload = with_working_state(resolve_payload, transport=transport) resolve_result = rpc_call( endpoint_url, "metadata.resolve_overrides", resolve_payload, timeout=timeout, transport=transport, session_id=session_id, ) evidence = resolve_result.get("write_plan_evidence") if isinstance(resolve_result.get("write_plan_evidence"), dict) else {} evidence_target = evidence.get("target") if isinstance(evidence.get("target"), dict) else {} next_resolution = evidence.get("next_resolution") if isinstance(evidence.get("next_resolution"), dict) else {} next_params = next_resolution.get("params") if isinstance(next_resolution.get("params"), dict) else {} steps.append( { "name": "metadata.resolve_overrides", "status": resolve_result.get("status"), "routine": routine_name, "chain": len([item for item in resolve_result.get("chain") or [] if isinstance(item, dict)]), "write_plan_evidence": bool(evidence), "next_method": next_resolution.get("method"), "working_state": resolve_payload.get("source_state") or resolve_payload.get("state"), } ) if not evidence: issues.append( { "code": "live_resolve_overrides_write_plan_evidence_missing", "status": resolve_result.get("status"), "error": resolve_result.get("error"), "diagnostics": resolve_result.get("diagnostics"), } ) if evidence and evidence.get("method") != "metadata.write.plan": issues.append({"code": "live_resolve_overrides_write_plan_method_mismatch", "actual": evidence.get("method")}) if evidence and evidence_target.get("routine_name") != routine_name: issues.append( { "code": "live_resolve_overrides_evidence_routine_mismatch", "expected": routine_name, "actual": evidence_target.get("routine_name"), } ) if evidence and next_resolution.get("method") != "metadata.saved_state.modules.search": issues.append({"code": "live_resolve_overrides_next_method_mismatch", "actual": next_resolution.get("method")}) if evidence and next_params.get("base_id") != base_id: issues.append({"code": "live_resolve_overrides_next_base_id_mismatch", "expected": base_id, "actual": next_params.get("base_id")}) if evidence and next_params.get("query") != routine_name: issues.append({"code": "live_resolve_overrides_next_query_mismatch", "expected": routine_name, "actual": next_params.get("query")}) if evidence and object_guid and evidence_target.get("object_guid") != object_guid: issues.append( { "code": "live_resolve_overrides_evidence_owner_guid_mismatch", "expected": object_guid, "actual": evidence_target.get("object_guid"), } ) if evidence and next_resolution.get("method") == "metadata.saved_state.modules.search" and next_params: saved_state = rpc_call( endpoint_url, "metadata.saved_state.modules.search", next_params, timeout=timeout, transport=transport, session_id=session_id, ) saved_modules = [item for item in saved_state.get("modules") or [] if isinstance(item, dict)] write_plan_target = first_write_plan_target_from_saved_state(saved_state) steps.append( { "name": "metadata.saved_state.modules.search", "status": saved_state.get("status"), "modules": len(saved_modules), "write_plan_target": bool(write_plan_target), "from_next_resolution": True, } ) if saved_state.get("status") != "ok": issues.append( { "code": "live_saved_state_modules_search_failed", "status": saved_state.get("status"), "error": saved_state.get("error"), "diagnostics": saved_state.get("diagnostics"), } ) elif write_plan_target: module_text = text_from_module_read(read_result) control_fragment = search_token_from_text(module_text) or routine_name plan_target = {**evidence_target, **write_plan_target} plan_intent = evidence.get("intent") if isinstance(evidence.get("intent"), dict) else {} plan_intent = { "operation": plan_intent.get("operation") or "replace_with_control", "control_fragment": plan_intent.get("control_fragment") or control_fragment, "expected_old_contains": plan_intent.get("expected_old_contains") or control_fragment, "new": plan_intent.get("new") or module_text or control_fragment, } write_plan = rpc_call( endpoint_url, "metadata.write.plan", { "base_id": base_id, "target": plan_target, "intent": plan_intent, "resolve_origin": False, }, timeout=timeout, transport=transport, session_id=session_id, ) route = write_plan.get("route") if isinstance(write_plan.get("route"), dict) else {} hint = route.get("apply_payload_hint") if isinstance(route.get("apply_payload_hint"), dict) else {} hint_payload = hint.get("payload") if isinstance(hint.get("payload"), dict) else {} steps.append( { "name": "metadata.write.plan", "status": write_plan.get("status"), "allowed": write_plan.get("allowed"), "apply_method": route.get("apply_method"), "ready_for_apply_method": hint.get("ready_for_apply_method"), "from_write_plan_target": True, } ) if write_plan.get("allowed") is not True: issues.append({"code": "live_composed_write_plan_not_allowed", "status": write_plan.get("status"), "problems": write_plan.get("problems")}) if route.get("apply_method") not in {"metadata.module.write_apply", "code.write"}: issues.append({"code": "live_composed_write_plan_apply_method_mismatch", "actual": route.get("apply_method")}) if hint.get("method") and hint.get("method") != route.get("apply_method"): issues.append( { "code": "live_composed_write_plan_hint_method_mismatch", "route_method": route.get("apply_method"), "hint_method": hint.get("method"), } ) if hint.get("ready_for_apply_method") is not True: issues.append({"code": "live_composed_write_plan_hint_not_ready", "hint": hint}) if write_plan_target.get("module_ref") and hint_payload.get("module_ref") != write_plan_target.get("module_ref"): issues.append({"code": "live_composed_write_plan_module_ref_mismatch", "expected": write_plan_target.get("module_ref"), "actual": hint_payload.get("module_ref")}) if write_plan_target.get("expected_sha1") and hint_payload.get("expected_sha1") != write_plan_target.get("expected_sha1"): issues.append({"code": "live_composed_write_plan_expected_sha1_mismatch"}) if write_plan_target.get("ref"): name_resolution = route.get("name_resolution") if isinstance(route.get("name_resolution"), dict) else {} resolution_selector = name_resolution.get("selector") if isinstance(name_resolution.get("selector"), dict) else {} if resolution_selector.get("ref") != write_plan_target.get("ref"): issues.append( { "code": "live_composed_write_plan_name_resolution_mismatch", "expected": write_plan_target.get("ref"), "actual": resolution_selector.get("ref"), } ) else: steps.append({"name": "metadata.write.plan", "status": "skipped_no_saved_state_target", "from_write_plan_target": False}) else: steps.append({"name": "metadata.resolve_overrides", "status": "skipped", "reason": "routine_name_not_found"}) search_token, search_token_source = search_token_from_module_read(read_result) if not search_token: issues.append({"code": "live_modules_read_search_token_missing", "module_ordinal": module_ordinal}) else: code_search_payload = { **{key: value for key, value in modules_selector.items() if key != "method" and value is not None and value != ""}, "base_id": modules_selector.get("base_id") or base_id, "query": search_token, "module_ordinal": module_ordinal, "limit": 5, } if not code_search_payload.get("ref") and object_ref: code_search_payload["ref"] = object_ref code_search_payload = with_working_state(code_search_payload, transport=transport) code_search = rpc_call(endpoint_url, "code.search", code_search_payload, timeout=timeout, transport=transport, session_id=session_id) items = [item for item in code_search.get("items") or [] if isinstance(item, dict)] steps.append( { "name": "code.search", "status": code_search.get("status"), "items": len(items), "query_source": search_token_source, "working_state": code_search_payload.get("source_state") or code_search_payload.get("state"), } ) if code_search.get("status") not in {"ok", "partial"} or not items: issues.append({"code": "live_code_search_empty", "status": code_search.get("status"), "query_source": search_token_source, "error": code_search.get("error"), "diagnostics": code_search.get("diagnostics")}) else: item_selector = items[0].get("read_selector") if isinstance(items[0].get("read_selector"), dict) else None if not item_selector or item_selector.get("method") != "code.read": issues.append({"code": "live_code_search_read_selector_missing", "selector": item_selector}) else: code_read_payload = {key: value for key, value in item_selector.items() if key != "method"} code_read_payload.setdefault("base_id", base_id) code_read_payload.setdefault("preview", True) code_read_payload.setdefault("max_chars", 1000) code_read = rpc_call(endpoint_url, "code.read", code_read_payload, timeout=timeout, transport=transport, session_id=session_id) steps.append({"name": "code.read", "status": code_read.get("status"), "used_read_selector": True}) if code_read.get("status") not in {"ok", "summary"}: issues.append({"code": "live_code_read_failed", "status": code_read.get("status"), "diagnostics": code_read.get("diagnostics")}) if require_form_write_plan_composition: form_search = rpc_call( endpoint_url, "metadata.saved_state.forms.search", { "base_id": base_id, "limit": 20, "scan_limit": 1000, "timeout_seconds": int(timeout), }, timeout=timeout, transport=transport, session_id=session_id, ) form_candidate = first_public_form_write_plan_candidate(form_search) steps.append( { "name": "metadata.saved_state.forms.search", "status": form_search.get("status"), "forms": int((form_search.get("counts") or {}).get("forms") or 0), "write_plan_candidate": bool(form_candidate), "include_storage": False, } ) if form_search.get("status") != "ok": issues.append({"code": "live_saved_state_forms_search_failed", "status": form_search.get("status")}) elif not form_candidate: steps.append( { "name": "metadata.write.plan.form", "status": "skipped_no_public_form_target", "allowed": False, "name_first": False, } ) else: form_plan_payload = { "base_id": base_id, "target": form_candidate["target"], "edits": [form_candidate["edit"]], "resolve_origin": False, } form_plan = rpc_call( endpoint_url, "metadata.write.plan", form_plan_payload, timeout=timeout, transport=transport, session_id=session_id, ) form_route = form_plan.get("route") if isinstance(form_plan.get("route"), dict) else {} form_hint = form_route.get("apply_payload_hint") if isinstance(form_route.get("apply_payload_hint"), dict) else {} form_name_resolution = form_route.get("name_resolution") if isinstance(form_route.get("name_resolution"), dict) else {} steps.append( { "name": "metadata.write.plan.form", "status": form_plan.get("status"), "allowed": form_plan.get("allowed"), "apply_method": form_route.get("apply_method"), "ready_for_apply_method": form_hint.get("ready_for_apply_method"), "name_resolution": form_name_resolution.get("status"), "name_first": not any( key in form_plan_payload["target"] for key in ("table", "file_name", "guid", "form_guid", "module_ref") ), } ) if form_plan.get("allowed") is not True: issues.append({"code": "live_form_write_plan_not_allowed", "status": form_plan.get("status"), "problems": form_plan.get("problems")}) if form_route.get("apply_method") != "metadata.form.element.write_apply": issues.append({"code": "live_form_write_plan_apply_method_mismatch", "actual": form_route.get("apply_method")}) if form_hint.get("ready_for_apply_method") is not True: issues.append({"code": "live_form_write_plan_hint_not_ready", "hint": form_hint}) if form_name_resolution.get("status") != "resolved": issues.append({"code": "live_form_write_plan_name_resolution_missing", "name_resolution": form_name_resolution}) coverage = live_coverage_from_steps(steps) if require_write_plan_composition: strict_issue = live_write_plan_composition_required_issue(coverage) if strict_issue: issues.append(strict_issue) if require_form_write_plan_composition: strict_form_issue = live_form_write_plan_composition_required_issue(coverage) if strict_form_issue: issues.append(strict_form_issue) return { "schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": not issues, "endpoint_url": endpoint_url, "transport": transport, "base_id": base_id, "object": live_object_summary({"kind": object_kind, "name": object_name, "guid": object_guid, "ref": object_ref}, canonical_object, modules_selector), "coverage": coverage, "steps": steps, "issues": issues, } def selector_chain_examples() -> list[dict[str, Any]]: return [ { "name": "definition_to_related_modules", "steps": [ { "tool": "onec_request", "method": "metadata.definition.find", "payload": { "base_id": "", "query": ".", "areas": ["metadata"], }, "next": "Use related_selectors.modules or related_selectors.modules_search from the result.", }, { "tool": "onec_request", "method": "", "payload": "", "next": "Use a module read_selector returned by modules or switch to related_selectors.code_search.", }, ], }, { "name": "definition_to_code_read", "steps": [ { "tool": "onec_request", "method": "metadata.definition.find", "payload": { "base_id": "", "ref": ".", "areas": ["metadata", "object", "module"], }, "next": "Use related_selectors.code_search and add query.", }, { "tool": "onec_request", "method": "code.search", "payload": { "base_id": "", "ref": ".", "query": "", "source_state": "working", "limit": 10, }, "next": "Pass item.read_selector directly to code.read.", }, { "tool": "onec_request", "method": "", "payload": "", "expected_method": "code.read", }, ], }, { "name": "global_code_search_to_direct_read", "steps": [ { "tool": "onec_request", "method": "code.search", "payload": { "base_id": "", "query": "", "scope": "all", "source_state": "working", "resolve_owners": True, "owner_scan_limit": 40, }, "next": "If owner is unresolved, still pass item.read_selector to code.read before broadening search.", }, { "tool": "onec_request", "method": "", "payload": "", "expected_method": "code.read", }, ], }, { "name": "report_to_separate_form_description", "steps": [ { "tool": "onec_request", "method": "metadata.object.forms", "payload": { "base_id": "", "ref": "Report.", "source_state": "working", }, "next": "Forms are references of the report. Select a returned public form name; do not treat the report card as the form description.", }, { "tool": "onec_request", "method": "metadata.object.form.details", "payload": { "base_id": "", "ref": "Report.", "form": "", "source_state": "working", "include_parameters": True, }, "next": "Read the separate form description: attributes, parameters, commands, items, and form module summary. A command has no module; its handler is a routine in this form module when decoded evidence provides the link.", }, { "tool": "onec_request", "method": "metadata.form.decode", "payload": { "base_id": "", "ref": "Report.", "form": "", "source_state": "working", "view": "structure", }, "next": "Use the static structure projection only. Parent/child edges remain unresolved unless the adapter reports proven codec evidence.", }, ], }, ] def collect_strings(value: Any) -> list[str]: if isinstance(value, str): return [value] if isinstance(value, dict): strings: list[str] = [] for key, item in value.items(): strings.extend(collect_strings(key)) strings.extend(collect_strings(item)) return strings if isinstance(value, list): strings = [] for item in value: strings.extend(collect_strings(item)) return strings return [] def validate_chain(chain: dict[str, Any]) -> list[dict[str, Any]]: issues: list[dict[str, Any]] = [] steps = chain.get("steps") if not isinstance(steps, list) or not steps: return [{"code": "chain_steps_missing", "chain": chain.get("name")}] for index, step in enumerate(steps): if not isinstance(step, dict): issues.append({"code": "chain_step_not_object", "chain": chain.get("name"), "index": index}) continue if step.get("tool") != "onec_request": issues.append({"code": "chain_step_tool_not_onec_request", "chain": chain.get("name"), "index": index}) method = step.get("method") if not isinstance(method, str) or not method: issues.append({"code": "chain_step_method_missing", "chain": chain.get("name"), "index": index}) payload = step.get("payload") if isinstance(payload, dict) and "base_id" not in payload: issues.append({"code": "chain_step_base_id_missing", "chain": chain.get("name"), "index": index}) if method in {"code.search", "modules.search", "metadata.definition.find"} and isinstance(payload, dict): selector_present = any(key in payload for key in ("ref", "query", "kind", "name", "guid", "object_type", "object_name", "object_guid")) if not selector_present: issues.append({"code": "chain_step_selector_or_query_missing", "chain": chain.get("name"), "index": index}) if method in WORKING_STATE_METHODS and isinstance(payload, dict) and not (payload.get("source_state") or payload.get("state")): issues.append({"code": "chain_step_working_state_missing", "chain": chain.get("name"), "index": index, "method": method}) if step.get("expected_method") and "" not in str(method): issues.append({"code": "chain_step_does_not_use_read_selector_method", "chain": chain.get("name"), "index": index}) for text in collect_strings(chain): for forbidden in FORBIDDEN_CONCRETE_SELECTOR_VALUES: if forbidden in text: issues.append({"code": "chain_contains_concrete_selector_value", "chain": chain.get("name"), "value": forbidden}) return issues def build_report() -> dict[str, Any]: chains = selector_chain_examples() issues: list[dict[str, Any]] = [] for chain in chains: issues.extend(validate_chain(chain)) return { "schema": "onec_mcp_selector_chain_smoke.v1", "passed": not issues, "chains": chains, "counts": {"chains": len(chains), "steps": sum(len(chain.get("steps") or []) for chain in chains)}, "issues": issues, } def main() -> int: parser = argparse.ArgumentParser(description="Smoke check for generic 1C MCP selector chains.") parser.add_argument("--json", action="store_true", help="Print JSON report.") parser.add_argument("--report", type=Path, default=DEFAULT_REPORT, help="Write JSON report.") parser.add_argument("--no-report", action="store_true", help="Do not write a report file.") parser.add_argument("--live", action="store_true", help="Run optional live adapter smoke through /rpc.") parser.add_argument("--transport", choices=("rest", "mcp"), default="rest", help="Live smoke transport.") parser.add_argument("--adapter-url", default="http://docker.cin.su:8011", help="1C REST adapter base URL for --live.") parser.add_argument("--mcp-url", default="http://docker.cin.su:8021", help="1C MCP proxy base URL for --live --transport mcp.") parser.add_argument("--base-id", help="Concrete 1C base id for --live.") parser.add_argument("--ref", help="Optional concrete public object ref for deterministic --live checks, for example Kind.ObjectName.") parser.add_argument("--kind", help="Optional concrete metadata kind for deterministic --live checks.") parser.add_argument("--name", help="Optional concrete metadata object name for deterministic --live checks.") parser.add_argument("--guid", help="Optional concrete metadata object GUID for deterministic --live checks.") parser.add_argument("--timeout", type=float, default=15.0, help="HTTP timeout for --live.") parser.add_argument("--require-write-plan-composition", action="store_true", help="Fail live smoke when saved-state lookup does not produce a concrete write_plan_target and composed metadata.write.plan.") parser.add_argument("--require-form-write-plan-composition", action="store_true", help="Fail live smoke unless a public saved-state form selector composes metadata.write.plan without storage identifiers.") args = parser.parse_args() if args.live and not args.base_id: report = {"schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": False, "issues": [{"code": "live_base_id_required"}]} elif args.live: endpoint_url = args.mcp_url if args.transport == "mcp" else args.adapter_url object_selector = {**selector_from_ref(args.ref), **{key: value for key, value in {"kind": args.kind, "name": args.name, "guid": args.guid}.items() if value}} report = build_live_report( endpoint_url, args.base_id, timeout=args.timeout, transport=args.transport, object_selector=object_selector or None, require_write_plan_composition=args.require_write_plan_composition, require_form_write_plan_composition=args.require_form_write_plan_composition, ) else: report = build_report() if not args.no_report: args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") if args.json: print(json.dumps(report, ensure_ascii=False, indent=2)) else: status = "passed" if report["passed"] else "failed" counts = report.get("counts") if isinstance(report.get("counts"), dict) else {} if counts: print(f"1C MCP selector chain smoke {status}: {counts.get('chains')} chains, {counts.get('steps')} steps.") else: print( json.dumps( { "status": status, "schema": report.get("schema"), "issues": report.get("issues") or [], "steps": report.get("steps") or [], }, ensure_ascii=False, indent=2, ) ) return 0 if report["passed"] else 1 if __name__ == "__main__": raise SystemExit(main())