From 39c8c4ab180c5fb4d75bb578fae391c54799d169 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 26 Jul 2026 17:30:04 +0300 Subject: [PATCH] Compose name-first form write plans --- plugins/1c/connector/adapter_1c_server.py | 182 ++++++++++++++++- scripts/check_1c_verify_reports.py | 29 ++- scripts/check_powershell_scripts.py | 2 + scripts/smoke_1c_mcp_selector_chain.py | 154 ++++++++++++++ scripts/verify_1c_adapter_deployment.ps1 | 10 +- tests/1c/test_payload_codec.py | 236 ++++++++++++++++++++++ 6 files changed, 603 insertions(+), 10 deletions(-) diff --git a/plugins/1c/connector/adapter_1c_server.py b/plugins/1c/connector/adapter_1c_server.py index 61eecba..d2c26ab 100644 --- a/plugins/1c/connector/adapter_1c_server.py +++ b/plugins/1c/connector/adapter_1c_server.py @@ -41169,14 +41169,29 @@ def metadata_saved_state_forms_search(payload: dict[str, Any]) -> dict[str, Any] break if len(rows) >= int(limit or 50): break - public_rows = rows if include_storage else [ - public_saved_state_form_search_row( - row, - requested_owner_kind=owner_kind, - requested_extension=extension_filter or None, - ) - for row in rows - ] + extension_names_by_guid: dict[str, str] = {} + if not include_storage and not extension_filter and any(row.get("table") == "ConfigCASSave" for row in rows): + extension_names_by_guid = { + guid: str(item.get("name") or guid) + for guid, item in extension_map_by_guid(base_id).items() + } + public_rows = rows if include_storage else [] + if not include_storage: + for row in rows: + row_extension = extension_filter or None + if not row_extension and row.get("table") == "ConfigCASSave": + row_source = row.get("source") if isinstance(row.get("source"), dict) else {} + row_file_name = str(row.get("file_name") or row_source.get("file_name") or "") + extension_guid_from_file = row_file_name.split("__", 1)[0].strip().lower() if "__" in row_file_name else "" + if extension_guid_from_file: + row_extension = extension_names_by_guid.get(extension_guid_from_file) + public_rows.append( + public_saved_state_form_search_row( + row, + requested_owner_kind=owner_kind, + requested_extension=row_extension, + ) + ) return { "schema": "onec_saved_state_form_search.v1", "status": "ok", @@ -46265,6 +46280,26 @@ def metadata_write_plan_apply_payload_hint( hint["payload"]["attribute"] = path_resolution.get("attribute_name") elif form_member_path: hint["payload"]["element"] = form_member_path[-1] + for field in ( + "extension", + "ref", + "object_ref", + "form", + "form_name", + "command", + "element", + "element_name", + "attribute", + "element_path", + "element_id", + "id", + ): + value = target.get(field) if target.get(field) not in (None, "") else payload.get(field) + if value not in (None, ""): + hint["payload"][field] = value + edits = payload.get("edits") if isinstance(payload.get("edits"), list) else intent.get("edits") + if isinstance(edits, list) and edits: + hint["payload"]["edits"] = edits for field in ("table", "property", "value"): value = metadata_write_plan_first_value(payload, target, intent, field) if value is not None: @@ -46681,6 +46716,115 @@ def metadata_write_plan_resolve_public_module_target( } +def metadata_write_plan_resolve_public_form_target( + payload: dict[str, Any], + target: dict[str, Any], + intent: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any]]: + """Resolve a public 1C form selector to an internal saved-state handle.""" + selector_ref = str(target.get("ref") or target.get("object_ref") or payload.get("ref") or payload.get("object_ref") or "").strip() + selector_kind = str( + target.get("object_kind") + or target.get("object_type") + or payload.get("object_kind") + or payload.get("object_type") + or "" + ).strip() + selector_name = str(target.get("object_name") or payload.get("object_name") or "").strip() + form_name = str(target.get("form") or target.get("form_name") or payload.get("form") or payload.get("form_name") or "").strip() + parsed_ref = parse_1c_object_path(selector_ref) if selector_ref else {} + if not selector_kind and parsed_ref.get("kind"): + selector_kind = str(parsed_ref.get("kind") or "") + if not selector_name and parsed_ref.get("name"): + selector_name = str(parsed_ref.get("name") or "") + if not form_name and canonical_kind(selector_kind) == "CommonForm": + form_name = selector_name + extension = str(target.get("extension") or payload.get("extension") or "").strip() + member_selector: dict[str, Any] = {} + for key in ("command", "element", "element_name", "attribute", "element_path", "element_id", "id"): + value = target.get(key) if target.get(key) not in (None, "") else payload.get(key) + if value not in (None, ""): + member_selector[key] = value + + public_selector = { + **({"extension": extension} if extension else {}), + **({"ref": selector_ref} if selector_ref else {}), + **({"kind": canonical_kind(selector_kind)} if selector_kind and not selector_ref else {}), + **({"name": selector_name} if selector_name and not selector_ref else {}), + **({"form": form_name} if form_name else {}), + **member_selector, + } + if not (selector_ref or (selector_kind and selector_name)): + return target, { + "status": "not_attempted", + "error": "public_object_selector_missing", + "selector": public_selector, + } + if not form_name and canonical_kind(selector_kind) != "CommonForm": + return target, { + "status": "not_attempted", + "error": "public_form_selector_missing", + "selector": public_selector, + } + + first_edit = next( + (item for item in payload.get("edits") or [] if isinstance(item, dict)), + {}, + ) + property_name = metadata_write_plan_first_value(payload, target, intent, "property") + if property_name in (None, ""): + property_name = first_edit.get("property") + value_present = any( + isinstance(source, dict) and "value" in source + for source in (intent, target, payload, first_edit) + ) + property_value = metadata_write_plan_first_value(payload, target, intent, "value") + if "value" in first_edit and property_value is None: + property_value = first_edit.get("value") + + resolve_payload: dict[str, Any] = { + "base_id": payload.get("base_id"), + "table": str(target.get("table") or payload.get("table") or ("ConfigCASSave" if extension else "ConfigSave")), + "include_storage": True, + "search_limit": int(payload.get("search_limit") or 10), + "scan_limit": int(payload.get("scan_limit") or 1000), + "timeout_seconds": int(payload.get("timeout_seconds") or 60), + **public_selector, + **({"property": property_name} if property_name not in (None, "") else {}), + **({"value": property_value} if value_present else {}), + } + resolved = metadata_form_write_target_resolve(resolve_payload) + if resolved.get("status") != "ok": + return target, { + "status": str(resolved.get("status") or "error"), + "error": "public_form_target_not_resolved", + "selector": public_selector, + "resolution_status": resolved.get("status"), + **({"counts": resolved.get("counts")} if isinstance(resolved.get("counts"), dict) else {}), + } + + source = resolved.get("source") if isinstance(resolved.get("source"), dict) else {} + file_name = str(source.get("file_name") or "").strip() + table = str(source.get("table") or resolve_payload.get("table") or "").strip() + if not file_name or table not in FORM_ELEMENT_SAVED_STATE_TABLES: + return target, { + "status": "error", + "error": "public_form_target_storage_handle_missing", + "selector": public_selector, + } + resolved_target = { + **target, + "table": table, + "file_name": file_name, + } + return resolved_target, { + "status": "resolved", + "method": FORM_WRITE_TARGET_RESOLVE_METHOD, + "selector": public_selector, + "counts": {"matches": int((resolved.get("counts") or {}).get("matches") or 1)}, + } + + def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]: method = METADATA_WRITE_PLAN_METHOD target = dict(payload.get("target")) if isinstance(payload.get("target"), dict) else {} @@ -46695,6 +46839,7 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]: ) target_kind_hint = str(target_kind_raw or "").strip().casefold() module_name_resolution: dict[str, Any] | None = None + form_name_resolution: dict[str, Any] | None = None has_public_module_selector = bool( target.get("ref") or target.get("object_ref") @@ -46706,6 +46851,17 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]: and has_public_module_selector ): target, module_name_resolution = metadata_write_plan_resolve_public_module_target(payload, target) + has_public_form_selector = bool( + target.get("ref") + or target.get("object_ref") + or ((target.get("object_kind") or target.get("object_type")) and target.get("object_name")) + ) and bool(target.get("form") or target.get("form_name") or target.get("command") or target.get("element") or target.get("attribute")) + if ( + target_kind_hint in {"form", "форма"} + and not metadata_write_concrete_reference_info(payload, target) + and has_public_form_selector + ): + target, form_name_resolution = metadata_write_plan_resolve_public_form_target(payload, target, intent) canonical_path = str( target.get("canonical_path") @@ -46757,6 +46913,14 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]: "resolution_status": module_name_resolution.get("status"), } ) + if form_name_resolution and form_name_resolution.get("status") != "resolved": + problems.append( + { + "code": str(form_name_resolution.get("error") or "public_form_target_not_resolved"), + "message": "The public 1C form selector did not resolve to exactly one saved-state form target.", + "resolution_status": form_name_resolution.get("status"), + } + ) if not path_resolution.get("is_full_path") and not concrete_reference: problems.append( { @@ -46833,6 +46997,8 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]: } if module_name_resolution: route["name_resolution"] = module_name_resolution + elif form_name_resolution: + route["name_resolution"] = form_name_resolution if extension_action: route["extension_action"] = extension_action if operation_was_inferred_from_extension_action: diff --git a/scripts/check_1c_verify_reports.py b/scripts/check_1c_verify_reports.py index 9c4f81f..5f3abd5 100644 --- a/scripts/check_1c_verify_reports.py +++ b/scripts/check_1c_verify_reports.py @@ -146,6 +146,7 @@ def validate_selector_chain( resolve_overrides = coverage.get("resolve_overrides") if isinstance(coverage.get("resolve_overrides"), dict) else {} saved_state_resolution = coverage.get("saved_state_resolution") if isinstance(coverage.get("saved_state_resolution"), dict) else {} composition = coverage.get("write_plan_composition") if isinstance(coverage.get("write_plan_composition"), dict) else {} + form_composition = coverage.get("form_write_plan_composition") if isinstance(coverage.get("form_write_plan_composition"), dict) else {} skips = coverage.get("skips") if isinstance(coverage.get("skips"), list) else None summary = { "path": str(path), @@ -161,6 +162,9 @@ def validate_selector_chain( "write_plan_target": saved_state_resolution.get("write_plan_target"), "composition_status": composition.get("status"), "composed": composition.get("composed"), + "form_composition_status": form_composition.get("status"), + "form_composed": form_composition.get("composed"), + "form_name_first": form_composition.get("name_first"), } if not report: return summary @@ -177,7 +181,7 @@ def validate_selector_chain( ) if report.get("passed") is not True: failures.append({"code": "selector_chain_not_passed", "label": label, "path": str(path), "issues": report.get("issues")}) - for section in ("resolve_overrides", "saved_state_resolution", "write_plan_composition", "skips"): + for section in ("resolve_overrides", "saved_state_resolution", "write_plan_composition", "form_write_plan_composition", "skips"): if section not in coverage: failures.append({"code": "selector_chain_coverage_section_missing", "label": label, "section": section, "path": str(path)}) if resolve_overrides.get("attempted") is not True: @@ -207,10 +211,23 @@ def validate_selector_chain( }) if composition.get("composed") is True and composition.get("status") in {None, "skipped_no_saved_state_target"}: failures.append({"code": "selector_chain_composed_status_unexpected", "label": label, "path": str(path), "composition": composition}) + if not isinstance(form_composition.get("attempted"), bool): + failures.append({"code": "selector_chain_form_composition_attempted_not_boolean", "label": label, "path": str(path), "composition": form_composition}) + if not isinstance(form_composition.get("composed"), bool): + failures.append({"code": "selector_chain_form_composed_not_boolean", "label": label, "path": str(path), "composition": form_composition}) + if form_composition.get("candidate") is True and form_composition.get("composed") is not True: + failures.append({"code": "selector_chain_form_target_not_composed", "label": label, "path": str(path), "composition": form_composition}) + if form_composition.get("composed") is True and form_composition.get("name_first") is not True: + failures.append({"code": "selector_chain_form_composition_not_name_first", "label": label, "path": str(path), "composition": form_composition}) if skips is None: failures.append({"code": "selector_chain_skips_not_list", "label": label, "path": str(path), "skips": coverage.get("skips")}) if require_composition and composition.get("composed") is not True: failures.append({"code": "selector_chain_composition_required", "label": label, "path": str(path), "composition": composition}) + if require_composition and ( + form_composition.get("composed") is not True + or form_composition.get("name_first") is not True + ): + failures.append({"code": "selector_chain_form_composition_required", "label": label, "path": str(path), "composition": form_composition}) steps = report.get("steps") if isinstance(report.get("steps"), list) else [] for step in steps: if not isinstance(step, dict): @@ -1445,6 +1462,15 @@ def write_self_test_reports(report_dir: Path, *, base_id: str, composed: bool, s "composed": composed, "from_write_plan_target": composed, }, + "form_write_plan_composition": { + "attempted": composed, + "search_attempted": composed, + "search_status": "ok" if composed else None, + "candidate": composed, + "status": "planned" if composed else None, + "composed": composed, + "name_first": composed, + }, "skips": [] if composed else [{"step": "metadata.write.plan", "status": "skipped_no_saved_state_target"}], }, "steps": [ @@ -1970,6 +1996,7 @@ def run_self_test() -> dict[str, Any]: strict_codes = {str(failure.get("code")) for failure in strict_skip["failures"]} for expected in ( "selector_chain_composition_required", + "selector_chain_form_composition_required", "saved_state_form_write_required", "saved_state_module_write_required", ): diff --git a/scripts/check_powershell_scripts.py b/scripts/check_powershell_scripts.py index 64c6bb0..b68e8e6 100644 --- a/scripts/check_powershell_scripts.py +++ b/scripts/check_powershell_scripts.py @@ -169,6 +169,8 @@ def check_adapter_verify_wiring(scripts: list[Path], executable: str) -> list[st failures.append("verify_1c_adapter_deployment.ps1 must use -SavedStateTable for copy plan and saved-state smoke commands.") if verify_text.count("--require-write-plan-composition") < 2: failures.append("verify_1c_adapter_deployment.ps1 must pass --require-write-plan-composition to both REST and MCP selector-chain smoke commands.") + if verify_text.count("--require-form-write-plan-composition") < 2: + failures.append("verify_1c_adapter_deployment.ps1 must pass --require-form-write-plan-composition to both REST and MCP selector-chain smoke commands.") if "--allow-empty-saved-state" not in verify_text or "if (-not $RequireSavedStateWriteSmoke)" not in verify_text: failures.append("verify_1c_adapter_deployment.ps1 must allow empty saved-state only when -RequireSavedStateWriteSmoke is not set.") if '"skipped_no_saved_state_candidate"' not in verify_text: diff --git a/scripts/smoke_1c_mcp_selector_chain.py b/scripts/smoke_1c_mcp_selector_chain.py index c057340..3b511e6 100644 --- a/scripts/smoke_1c_mcp_selector_chain.py +++ b/scripts/smoke_1c_mcp_selector_chain.py @@ -126,6 +126,54 @@ def first_write_plan_target_from_saved_state(result: dict[str, Any]) -> dict[str 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): @@ -136,6 +184,8 @@ def live_coverage_from_steps(steps: list[dict[str, Any]]) -> dict[str, Any]: 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 @@ -160,6 +210,15 @@ def live_coverage_from_steps(steps: list[dict[str, Any]]) -> dict[str, Any]: "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, } @@ -175,6 +234,18 @@ def live_write_plan_composition_required_issue(coverage: dict[str, Any]) -> dict } +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: @@ -504,6 +575,7 @@ def build_live_report( 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]] = [] @@ -856,11 +928,91 @@ def build_live_report( 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", @@ -1037,6 +1189,7 @@ def main() -> int: 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: @@ -1051,6 +1204,7 @@ def main() -> int: 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() diff --git a/scripts/verify_1c_adapter_deployment.ps1 b/scripts/verify_1c_adapter_deployment.ps1 index edf3a34..e6525d9 100644 --- a/scripts/verify_1c_adapter_deployment.ps1 +++ b/scripts/verify_1c_adapter_deployment.ps1 @@ -147,7 +147,7 @@ function Assert-SelectorChainReport { if (-not $report.coverage) { throw "$Label report is missing coverage: $Path" } - foreach ($section in @("resolve_overrides", "saved_state_resolution", "write_plan_composition", "skips")) { + foreach ($section in @("resolve_overrides", "saved_state_resolution", "write_plan_composition", "form_write_plan_composition", "skips")) { if ($report.coverage.PSObject.Properties.Name -notcontains $section) { throw "$Label report coverage is missing '$section': $Path" } @@ -155,6 +155,12 @@ function Assert-SelectorChainReport { if ($RequireComposition -and $report.coverage.write_plan_composition.composed -ne $true) { throw "$Label report did not compose metadata.write.plan in strict mode: $Path" } + if ($RequireComposition -and ( + $report.coverage.form_write_plan_composition.composed -ne $true -or + $report.coverage.form_write_plan_composition.name_first -ne $true + )) { + throw "$Label report did not compose a name-first form metadata.write.plan in strict mode: $Path" + } foreach ($step in @($report.steps)) { if ($step.name -in @("metadata.resolve_overrides", "code.search") -and $step.status -notlike "skipped*") { if ($step.working_state -ne "working") { @@ -441,6 +447,7 @@ try { ) if ($RequireSelectorChainWritePlanComposition) { $restCommand += "--require-write-plan-composition" + $restCommand += "--require-form-write-plan-composition" } Invoke-CheckedCommand -Label "REST adapter selector-chain live smoke ($currentBaseId)" -Command (Add-ObjectSelectorArgs -Command $restCommand) Assert-SelectorChainReport -Label "REST adapter selector-chain live smoke ($currentBaseId)" -Path $selectorChainReport -RequireComposition:$RequireSelectorChainWritePlanComposition @@ -736,6 +743,7 @@ try { ) if ($RequireSelectorChainWritePlanComposition) { $mcpCommand += "--require-write-plan-composition" + $mcpCommand += "--require-form-write-plan-composition" } Invoke-CheckedCommand -Label "MCP proxy selector-chain live smoke ($currentBaseId)" -Command (Add-ObjectSelectorArgs -Command $mcpCommand) Assert-SelectorChainReport -Label "MCP proxy selector-chain live smoke ($currentBaseId)" -Path $selectorChainReport -RequireComposition:$RequireSelectorChainWritePlanComposition diff --git a/tests/1c/test_payload_codec.py b/tests/1c/test_payload_codec.py index b329276..31e02af 100644 --- a/tests/1c/test_payload_codec.py +++ b/tests/1c/test_payload_codec.py @@ -3129,6 +3129,67 @@ def test_saved_state_forms_search_default_is_name_first(monkeypatch: pytest.Monk assert storage_marker not in str(result) +def test_saved_state_forms_search_infers_public_extension_name_without_filter(monkeypatch: pytest.MonkeyPatch) -> None: + extension_guid = "fb26cf42-7609-11f1-828f-005056b0d483" + file_name = f"{extension_guid}__form-guid.0" + monkeypatch.setattr( + adapter_server, + "extension_map_by_guid", + lambda base_id: {extension_guid: {"guid": extension_guid, "name": "test2"}}, + ) + monkeypatch.setattr( + adapter_server, + "storage_files_list", + lambda payload: {"status": "ok", "files": [{"FileName": file_name}]}, + ) + monkeypatch.setattr( + adapter_server, + "saved_state_form_search_row", + lambda **kwargs: { + "table": "ConfigCASSave", + "file_name": file_name, + "name": "t_Форма", + "owner": {"status": "resolved", "kind": "Catalog", "name": "test2"}, + "form": {"name": "t_Форма"}, + "source": {"table": "ConfigCASSave", "file_name": file_name}, + "matches": [ + { + "name": "ЗаменаДомена", + "section": "commands", + "writable_properties": [ + { + "property": "title", + "presentation": "Заголовок", + "value": "Замена домена", + "value_type": "string", + } + ], + } + ], + }, + ) + + result = adapter_server.metadata_saved_state_forms_search( + { + "base_id": "upo_test", + "tables": ["ConfigCASSave"], + "limit": 1, + } + ) + + form = result["forms"][0] + assert form["selector"]["extension"] == "test2" + assert form["matches"][0]["selector"] == { + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + "qualified_name": "test2.t_Форма", + "command": "ЗаменаДомена", + } + assert extension_guid not in str(result) + assert "ConfigCASSave" not in str(result["forms"]) + + def test_saved_state_form_search_resolves_owner_for_internal_name_first_write(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( adapter_server, @@ -14015,6 +14076,87 @@ def test_mcp_selector_chain_smoke_uses_generic_placeholders() -> None: assert concrete_value not in examples_json +def test_mcp_selector_chain_selects_name_first_form_write_plan_candidate() -> None: + candidate = mcp_selector_chain_smoke.first_public_form_write_plan_candidate( + { + "forms": [ + { + "selector": { + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + }, + "matches": [ + { + "selector": { + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + "command": "ЗаменаДомена", + }, + "writable_properties": [ + { + "property": "name", + "presentation": "Имя", + "value": "ЗаменаДомена", + "value_type": "string", + }, + { + "property": "title", + "presentation": "Заголовок", + "value": "Замена домена", + "value_type": "string", + }, + ], + } + ], + } + ] + } + ) + + assert candidate == { + "target": { + "kind": "form", + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + "command": "ЗаменаДомена", + }, + "edit": { + "property": "Заголовок", + "value": "Замена домена [NAME-FIRST PLAN]", + }, + } + assert not any(key in candidate["target"] for key in ("table", "file_name", "guid", "form_guid")) + + +def test_mcp_selector_chain_form_composition_strict_check() -> None: + missing = mcp_selector_chain_smoke.live_form_write_plan_composition_required_issue( + { + "form_write_plan_composition": { + "attempted": True, + "status": "skipped_no_public_form_target", + "composed": False, + "name_first": False, + } + } + ) + ready = mcp_selector_chain_smoke.live_form_write_plan_composition_required_issue( + { + "form_write_plan_composition": { + "attempted": True, + "status": "planned", + "composed": True, + "name_first": True, + } + } + ) + + assert missing["code"] == "live_form_write_plan_composition_required" + assert ready is None + + def test_mcp_selector_chain_smoke_rejects_concrete_selector_values() -> None: issues = mcp_selector_chain_smoke.validate_chain( { @@ -21551,6 +21693,100 @@ def test_metadata_write_plan_resolves_public_module_selector_internally(monkeypa assert hint["payload"]["expected_sha1"] == "saved-payload-sha1" +def test_metadata_write_plan_resolves_public_form_selector_internally(monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, Any] = {} + + def fake_resolve(payload: dict[str, Any]) -> dict[str, Any]: + seen.update(payload) + return { + "schema": "onec_form_write_target_resolution.v1", + "status": "ok", + "source": { + "kind": "live_sql", + "table": "ConfigCASSave", + "file_name": "extension-guid__form-guid.0", + }, + "counts": {"matches": 1}, + } + + monkeypatch.setattr(adapter_server, "metadata_form_write_target_resolve", fake_resolve) + + result = adapter_server.metadata_write_plan( + { + "base_id": "upo_test", + "target": { + "kind": "form", + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + "command": "ЗаменаДомена", + }, + "edits": [{"property": "Заголовок", "value": "Новое имя"}], + "resolve_origin": False, + } + ) + + assert seen["table"] == "ConfigCASSave" + assert seen["ref"] == "Catalog.test2" + assert seen["form"] == "t_Форма" + assert seen["command"] == "ЗаменаДомена" + assert seen["property"] == "Заголовок" + assert seen["value"] == "Новое имя" + assert seen["include_storage"] is True + assert result["allowed"] is True + assert result["status"] == "planned" + assert result["route"]["name_resolution"] == { + "status": "resolved", + "method": "metadata.form.write_target.resolve", + "selector": { + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + "command": "ЗаменаДомена", + }, + "counts": {"matches": 1}, + } + hint = result["route"]["apply_payload_hint"] + assert hint["ready_for_apply_method"] is True + assert hint["payload"]["table"] == "ConfigCASSave" + assert hint["payload"]["file_name"] == "extension-guid__form-guid.0" + assert hint["payload"]["ref"] == "Catalog.test2" + assert hint["payload"]["form"] == "t_Форма" + assert hint["payload"]["command"] == "ЗаменаДомена" + assert hint["payload"]["edits"] == [{"property": "Заголовок", "value": "Новое имя"}] + + +def test_metadata_write_plan_reports_unresolved_public_form_selector(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + adapter_server, + "metadata_form_write_target_resolve", + lambda payload: { + "schema": "onec_form_write_target_resolution.v1", + "status": "not_found", + "counts": {"matches": 0}, + }, + ) + + result = adapter_server.metadata_write_plan( + { + "base_id": "upo_test", + "target": { + "kind": "form", + "ref": "Catalog.Контрагенты", + "form": "ФормаЭлемента", + "element": "КнопкаЗаписать", + }, + "edits": [{"property": "Заголовок", "value": "Записать"}], + "resolve_origin": False, + } + ) + + assert result["allowed"] is False + assert result["status"] == "needs_origin" + assert result["route"]["name_resolution"]["status"] == "not_found" + assert any(problem["code"] == "public_form_target_not_resolved" for problem in result["problems"]) + + def test_metadata_write_preflight_verifies_concrete_saved_state(monkeypatch: pytest.MonkeyPatch) -> None: def fake_diff(payload: dict[str, Any]) -> dict[str, Any]: assert payload["module_ref"] == "ConfigCASSave:object-guid__module-guid.0#stream:4"