From bf1f52cb380962ffab397ad916538eed7fcba505 Mon Sep 17 00:00:00 2001 From: airmang <38392618+airmang@users.noreply.github.com> Date: Thu, 16 Jul 2026 11:40:33 +0900 Subject: [PATCH 1/2] feat: add canonical mixed-form transaction --- CHANGELOG.md | 23 + README.md | 20 +- pyproject.toml | 2 +- src/hwpx/agent/__init__.py | 26 + src/hwpx/agent/catalog.py | 85 ++ src/hwpx/agent/form_plan.py | 1685 ++++++++++++++++++++++++++++++++ src/hwpx/agent/model.py | 26 +- src/hwpx/table_patch.py | 133 ++- tests/test_agent_mixed_form.py | 726 ++++++++++++++ tests/test_mixed_form_plan.py | 426 ++++++++ 10 files changed, 3137 insertions(+), 15 deletions(-) create mode 100644 src/hwpx/agent/form_plan.py create mode 100644 tests/test_agent_mixed_form.py create mode 100644 tests/test_mixed_form_plan.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4469f58..9028054 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,29 @@ ## [Unreleased] +## [3.1.0] - 2026-07-16 + +### 추가 +- **타입드 혼합 양식 계획기**: `hwpx.mixed-form-plan/v1`이 네이티브 필드, 고유 라벨 셀, + revision-bound canonical path, 본문 직접 단일-run anchor를 하나의 엄격한 계획으로 받습니다. + 모든 대상을 먼저 canonical path로 해석한 뒤 기존 `hwpx.agent-batch/v1` 한 건으로 컴파일하며, + `apply_document_commands`만 실행기로 사용해 dry-run·rollback·idempotency·단일 저장 게이트를 + 그대로 보장합니다. +- 병합 셀의 논리 좌표를 실제 `cellAddr` 앵커로 정규화하는 비변형 `resolve_cell_target` 계획 + 프리미티브와 공개 plan/compiled-plan JSON Schema를 추가했습니다. + +### 변경 +- 에이전트 batch JSON Schema를 정본 카탈로그에 추가하고 quality 객체의 enum·boolean 필드를 + 엄격하게 검증해 알 수 없는 값이 실행 단계까지 전달되지 않도록 했습니다. +- 혼합 양식 출력이 입력과 동일한 resolved path·symlink·hardlink인 경우 계획과 적용 직전에 + 거부하고, 셀 라벨은 정규화된 정확 일치만 허용합니다. 컴파일 계획은 locator별 node/path/ + section/좌표 불변식을 검증하며, 멱등성 식별자에 공개 locator 요청 hash를 포함합니다. + +### 검증 +- 합성 1쪽 한국어 혼합 양식으로 네 locator 동시 적용, 드라이런 무출력, 주입 실패 rollback, + idempotent replay/conflict, 미변경 OPC member 바이트 보존, 재열기와 `openSafety.ok`를 고정했습니다. + `evalplan` 회귀는 유지하며 `exam` 모듈은 혼합 양식 경로에 가져오지 않습니다. + ## [3.0.0] - 2026-07-16 ### 변경 diff --git a/README.md b/README.md index 0c74b8f..e552708 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@

python-hwpx

- 한글 없이 HWPX 문서를 Python으로 읽고, 편집하고, 생성하고, 검증합니다. + 한글 없이 HWPX 문서를 Python으로 읽고, 편집하고, 생성하고, 구조를 검증합니다.

PyPI @@ -19,15 +19,27 @@ |---|---|---| | 📦 라이브러리 | **[`python-hwpx`](https://github.com/airmang/python-hwpx)** | 순수 파이썬 HWPX 파싱·편집·생성 코어 | | 🔌 MCP 서버 | [`hwpx-mcp-server`](https://github.com/airmang/hwpx-mcp-server) | MCP 클라이언트(Claude Desktop, VS Code 등)에서 HWPX 조작 | -| 🎯 에이전트 스킬 | [`hwpx-skill`](https://github.com/airmang/hwpx-plugins) | 에이전트가 HWPX를 바로 쓰게 해주는 공식 온보딩 스킬 | +| 🎯 에이전트 스킬 | [`hwpx-plugin`](https://github.com/airmang/hwpx-plugins) | 에이전트가 HWPX를 바로 쓰게 해주는 first-party 플러그인·스킬 번들 | + +`python-hwpx`는 HWPX 파싱·편집·생성을 제공하는 코어 라이브러리이며, +`hwpx-mcp-server`와 `hwpx-plugin`은 같은 프로젝트가 직접 유지보수하는 first-party 연동 구성요소입니다. +“first-party”는 프로젝트 유지보수 관계를 뜻하며, 한컴 또는 제3자의 공식 인증을 뜻하지 않습니다. + +현재 검증 중인 릴리스 후보와 최소 호환 코어 버전은 모두 `python-hwpx 3.1.0`입니다. +이 버전 좌표는 별도 소유자 승인 전까지 공개 또는 배포를 뜻하지 않습니다. +현재 패키지 분류는 `Development Status :: 3 - Alpha`입니다. 이 분류는 API와 제품의 +성숙도를 나타내며, 릴리스 후보 버전이나 플러그인의 최소 호환 버전을 대신하지 않습니다. --- ## 왜 python-hwpx인가 -- **한컴오피스 설치 불필요** — HWPX는 ZIP+XML(OWPML/OPC) 구조라, 순수 파이썬으로 Windows·macOS·Linux·CI 어디서나 읽고 씁니다. +- **코어 편집에 한컴오피스 설치 불필요** — HWPX는 ZIP+XML(OWPML/OPC) 구조라, 순수 파이썬으로 Windows·macOS·Linux·CI 어디서나 읽고 씁니다. - **읽기부터 생성까지 한 코어** — 텍스트/서식 추출, 문단·표·양식 편집, 새 문서 생성, XSD 스키마 검증을 하나의 API로 처리합니다. -- **에이전트·자동화 친화** — 같은 스택 위에서 `hwpx-mcp-server`와 공식 스킬이 직결됩니다. +- **에이전트·자동화 친화** — 같은 프로젝트가 유지보수하는 `hwpx-mcp-server`와 `hwpx-plugin`이 코어에 연결됩니다. + +문서 파싱·편집·생성은 순수 Python으로 수행할 수 있습니다. 다만 페이지 나눔, 표 넘침, +글꼴 대체 등 최종 시각 품질을 확언하려면 필요에 따라 실제 한컴 렌더 오라클을 별도로 사용합니다. ## 빠른 시작 diff --git a/pyproject.toml b/pyproject.toml index 88a46c0..3bece3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-hwpx" -version = "3.0.0" +version = "3.1.0" description = "한글 없이 HWPX 문서를 열고, 편집하고, 생성하고, 검증하는 Python 자동화 라이브러리" readme = { file = "README.md", content-type = "text/markdown" } license = "Apache-2.0" diff --git a/src/hwpx/agent/__init__.py b/src/hwpx/agent/__init__.py index c68638d..49d3998 100644 --- a/src/hwpx/agent/__init__.py +++ b/src/hwpx/agent/__init__.py @@ -18,6 +18,20 @@ from .catalog import agent_catalog, agent_json_schemas, catalog_hash, human_help, node_help from .commands import apply_document_commands from .document import HwpxAgentDocument, NodeRecord +from .form_plan import ( + MIXED_FORM_COMPILED_PLAN_SCHEMA, + MIXED_FORM_LOCATOR_KINDS, + MIXED_FORM_PLAN_SCHEMA, + MIXED_FORM_REQUEST_SCHEMA, + MixedFormPlan, + MixedFormResolution, + apply_mixed_form_fill, + apply_mixed_form_plan, + mixed_form_json_schemas, + plan_mixed_form_fill, + validate_mixed_form_plan, + validate_mixed_form_request, +) from .path import PathSegment, SemanticPath, canonicalize_path, parse_path from .query import QueryResult, SemanticSelector, parse_selector from .blueprint import ( @@ -52,6 +66,10 @@ "AGENT_COMMAND_SCHEMA", "AGENT_ERROR_SCHEMA", "AGENT_NODE_SCHEMA", + "MIXED_FORM_COMPILED_PLAN_SCHEMA", + "MIXED_FORM_LOCATOR_KINDS", + "MIXED_FORM_PLAN_SCHEMA", + "MIXED_FORM_REQUEST_SCHEMA", "BLUEPRINT_CATALOG_SCHEMA", "BLUEPRINT_REPLAY_RESULT_SCHEMA", "BLUEPRINT_REPLAY_SCHEMA", @@ -65,6 +83,8 @@ "BlueprintBundle", "BlueprintDumpResult", "HwpxAgentDocument", + "MixedFormPlan", + "MixedFormResolution", "NodeRecord", "PathSegment", "QueryResult", @@ -74,6 +94,8 @@ "agent_json_schemas", "agent_contract_manifest", "apply_document_commands", + "apply_mixed_form_fill", + "apply_mixed_form_plan", "blueprint_catalog", "blueprint_catalog_hash", "blueprint_hash", @@ -84,15 +106,19 @@ "canonicalize_path", "catalog_hash", "human_help", + "mixed_form_json_schemas", "node_help", "parse_path", "parse_selector", + "plan_mixed_form_fill", "dump_document_blueprint", "read_blueprint_bundle", "repack_blueprint_bundle", "replay_document_blueprint", "validate_agent_batch", "validate_agent_command", + "validate_mixed_form_plan", + "validate_mixed_form_request", "validate_blueprint_manifest", "validate_replay_request", "write_blueprint_bundle", diff --git a/src/hwpx/agent/catalog.py b/src/hwpx/agent/catalog.py index cfd1ee3..dce633d 100644 --- a/src/hwpx/agent/catalog.py +++ b/src/hwpx/agent/catalog.py @@ -9,10 +9,16 @@ from typing import Any from .model import ( + AGENT_BATCH_SCHEMA, AGENT_CATALOG_SCHEMA, + MAX_COMMANDS, AgentContractError, NODE_KINDS, NODE_PROPERTY_CATALOG_V1, + QUALITY_KEYS, + QUALITY_MODES, + REVISION_PATTERN, + VERIFICATION_REQUIREMENTS, agent_contract_manifest, ) @@ -78,6 +84,22 @@ def agent_json_schemas() -> dict[str, Any]: "properties": {key: value for key, value in properties.items() if key in allowed}, } ) + quality_object = { + "type": "object", + "additionalProperties": False, + "properties": { + "mode": {"enum": list(QUALITY_MODES)}, + "renderCheck": {"enum": ["off", "auto", "required"]}, + "xsdMode": {"enum": ["off", "lint"]}, + "overflowPolicy": {"enum": ["fail", "warn", "truncate"]}, + "layoutLint": {"enum": ["off", "warn", "strict"]}, + "preserveUnmodifiedParts": {"type": "boolean"}, + "requireReferenceIntegrity": {"type": "boolean"}, + }, + } + if set(quality_object["properties"]) != set(QUALITY_KEYS): # pragma: no cover + raise AssertionError("quality JSON Schema drifted from QUALITY_KEYS") + return { "node": { "$schema": "https://json-schema.org/draft/2020-12/schema", @@ -113,6 +135,69 @@ def agent_json_schemas() -> dict[str, Any]: "title": "HwpxAgentCommand v1", "oneOf": command_variants, }, + "batch": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "HwpxAgentBatch v1", + "type": "object", + "required": [ + "schemaVersion", + "input", + "output", + "commands", + "expectedRevision", + "idempotencyKey", + "dryRun", + "quality", + "verificationRequirements", + ], + "additionalProperties": False, + "properties": { + "schemaVersion": {"const": AGENT_BATCH_SCHEMA}, + "input": { + "type": "object", + "required": ["filename"], + "additionalProperties": False, + "properties": {"filename": {"type": "string", "minLength": 1}}, + }, + "output": { + "type": "object", + "required": ["filename", "overwrite"], + "additionalProperties": False, + "properties": { + "filename": {"type": "string", "minLength": 1}, + "overwrite": {"type": "boolean"}, + }, + }, + "commands": { + "type": "array", + "minItems": 1, + "maxItems": MAX_COMMANDS, + "items": {"oneOf": deepcopy(command_variants)}, + }, + "expectedRevision": { + "type": ["string", "null"], + "pattern": REVISION_PATTERN.pattern, + }, + "idempotencyKey": { + "type": ["string", "null"], + "minLength": 1, + "maxLength": 128, + }, + "dryRun": {"type": "boolean"}, + "quality": { + "oneOf": [ + {"enum": list(QUALITY_MODES)}, + deepcopy(quality_object), + {"type": "null"}, + ] + }, + "verificationRequirements": { + "type": "array", + "items": {"enum": list(VERIFICATION_REQUIREMENTS)}, + "uniqueItems": True, + }, + }, + }, "queryInput": { "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", diff --git a/src/hwpx/agent/form_plan.py b/src/hwpx/agent/form_plan.py new file mode 100644 index 0000000..80ffab6 --- /dev/null +++ b/src/hwpx/agent/form_plan.py @@ -0,0 +1,1685 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Strict mixed-anchor form-fill planning over the semantic agent facade. + +The module deliberately owns *planning only*. It resolves four public locator +types against one immutable document revision, compiles them to ordinary +``hwpx.agent-batch/v1`` ``set`` commands, and delegates execution to +``apply_document_commands``. There is no second mutation or save engine here. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterator, Mapping, MutableMapping, Sequence +from copy import deepcopy +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from hwpx.quality import SavePipeline +from hwpx.table_patch import ResolvedCellTarget, resolve_cell_target + +from .catalog import agent_json_schemas +from .commands import ( + DomainVerifier, + FaultInjector, + IdempotencyStore, + apply_document_commands, +) +from .document import HwpxAgentDocument, NodeRecord +from .model import ( + AGENT_BATCH_SCHEMA, + COMMAND_ID_PATTERN, + MAX_COMMANDS, + MAX_TEXT_CHARS, + REVISION_PATTERN, + STABILITY_LEVELS, + VERIFICATION_REQUIREMENTS, + AgentBatchResult, + AgentContractError, + _validate_quality, + validate_agent_batch, +) +from .path import parse_path + +# Public input authored by a client/agent (``operations[].target``). +MIXED_FORM_PLAN_SCHEMA = "hwpx.mixed-form-plan/v1" +# Internal normalized form (``operations[].locator``); not an MCP wire promise. +MIXED_FORM_REQUEST_SCHEMA = "hwpx.mixed-form-request/v1" +# Revision-bound output containing the single ``hwpx.agent-batch/v1`` request. +MIXED_FORM_COMPILED_PLAN_SCHEMA = "hwpx.mixed-form-compiled-plan/v1" +MIXED_FORM_LOCATOR_KINDS = ( + "nativeField", + "labelCell", + "canonicalPath", + "bodyAnchor", +) + +_FILLABLE_NODE_KINDS = frozenset({"paragraph", "run", "cell", "form-field"}) +_DIRECTIONS = frozenset({"right", "left", "below", "above"}) +_MIXED_FORM_IDENTITY_SCOPE = "hwpx.mixed-form-idempotency/v1" +_COMPILED_PLAN_KEYS = frozenset( + { + "schemaVersion", + "inputRevision", + "requestHash", + "resolutions", + "batch", + "planHash", + } +) +_RESOLUTION_KEYS = frozenset( + { + "operationId", + "locatorKind", + "path", + "nodeKind", + "stability", + "section", + "tableIndex", + "logicalRow", + "logicalColumn", + "physicalRow", + "physicalColumn", + } +) + + +def _object(value: object, name: str) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise AgentContractError("invalid_syntax", f"{name} must be an object", target=name) + return dict(value) + + +def _array(value: object, name: str) -> list[Any]: + if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)): + raise AgentContractError("invalid_syntax", f"{name} must be an array", target=name) + return list(value) + + +def _exact_keys( + value: Mapping[str, Any], + *, + required: set[str] | frozenset[str], + optional: set[str] | frozenset[str] = frozenset(), + name: str, +) -> None: + missing = required - set(value) + extra = set(value) - required - optional + if missing or extra: + raise AgentContractError( + "invalid_syntax", + f"{name} fields mismatch (missing={sorted(missing)}, extra={sorted(extra)})", + target=name, + ) + + +def _nonempty_string(value: object, name: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise AgentContractError("invalid_syntax", f"{name} must be a non-empty string", target=name) + if len(value) > MAX_TEXT_CHARS: + raise AgentContractError("resource_limit", f"{name} is too long", target=name) + return value + + +def _value_string(value: object, name: str) -> str: + if not isinstance(value, str): + raise AgentContractError("invalid_syntax", f"{name} must be a string", target=name) + if len(value) > MAX_TEXT_CHARS: + raise AgentContractError("resource_limit", f"{name} is too long", target=name) + return value + + +def _positive_integer(value: object, name: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 1: + raise AgentContractError( + "invalid_syntax", f"{name} must be a one-based integer", target=name + ) + return value + + +def _nonnegative_integer_or_none(value: object, name: str) -> int | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise AgentContractError( + "invalid_syntax", f"{name} must be a non-negative integer or null", target=name + ) + return value + + +def _section_index_from_path(value: object, name: str) -> int: + section_path = _nonempty_string(value, name) + parsed = parse_path(section_path) + if ( + len(parsed.segments) != 1 + or parsed.segments[0].kind != "section" + or parsed.segments[0].index is None + ): + raise AgentContractError( + "invalid_syntax", + f"{name} must be a positional section path such as /section[1]", + target=name, + ) + return parsed.segments[0].index + + +def _canonical_hash(value: Mapping[str, Any]) -> str: + try: + payload = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise AgentContractError( + "invalid_syntax", "mixed-form contract is not canonical JSON", target="mixedForm" + ) from exc + return "sha256:" + hashlib.sha256(payload).hexdigest() + + +def _assert_distinct_source_output(source: str, output: str) -> None: + """Fail closed when output can name the input document itself. + + The lexical/real-path comparison catches ``.``/``..`` and symlink aliases, + while ``samefile`` additionally catches distinct hard-link names. This is + intentionally run both while planning and immediately before execution so + an output created or replaced between those phases cannot overwrite input. + """ + + source_path = Path(source) + output_path = Path(output) + try: + source_resolved = source_path.resolve(strict=False) + output_resolved = output_path.resolve(strict=False) + except (OSError, RuntimeError) as exc: + raise AgentContractError( + "verification_failed", + f"source/output identity could not be resolved safely: {exc}", + target="output.filename", + ) from exc + if source_resolved == output_resolved: + raise AgentContractError( + "invariant_violation", + "mixed-form output must not resolve to the input document", + target="output.filename", + ) + try: + same_file = source_path.samefile(output_path) + except FileNotFoundError: + same_file = False + except OSError as exc: + raise AgentContractError( + "verification_failed", + f"source/output identity could not be checked safely: {exc}", + target="output.filename", + ) from exc + if same_file: + raise AgentContractError( + "invariant_violation", + "mixed-form output must not alias the input document", + target="output.filename", + ) + + +class _MixedFormIdempotencyStore(MutableMapping[str, Any]): + """Translate one mixed-form identity to the batch executor's store shape. + + ``apply_document_commands`` remains the sole executor and continues to see + its own batch hash. The caller-owned backing store, however, records a + scoped composite hash, so two distinct public locator requests cannot + replay merely because they compile to the same canonical batch. + """ + + def __init__( + self, + backing: IdempotencyStore, + *, + active_key: str, + mixed_form_request_hash: str, + batch_request_hash: str, + ) -> None: + self._backing = backing + self._active_key = active_key + self._mixed_form_request_hash = mixed_form_request_hash + self._batch_request_hash = batch_request_hash + self._composite_hash = _canonical_hash( + { + "scope": _MIXED_FORM_IDENTITY_SCOPE, + "mixedFormRequestHash": mixed_form_request_hash, + "batchRequestHash": batch_request_hash, + } + ) + + def __getitem__(self, key: str) -> Any: + cached = self._backing[key] + if key != self._active_key or not isinstance(cached, Mapping): + return cached + if ( + cached.get("requestHash") == self._composite_hash + and cached.get("identityScope") == _MIXED_FORM_IDENTITY_SCOPE + and cached.get("mixedFormRequestHash") == self._mixed_form_request_hash + and cached.get("batchRequestHash") == self._batch_request_hash + ): + executor_view = dict(cached) + executor_view["requestHash"] = self._batch_request_hash + return executor_view + # A standard batch entry, another mixed-form request, or a malformed + # entry must conflict inside the existing typed executor path. + conflict_view = dict(cached) + conflict_view["requestHash"] = None + return conflict_view + + def __setitem__(self, key: str, value: Any) -> None: + if key != self._active_key or not isinstance(value, Mapping): + self._backing[key] = value + return + stored = dict(value) + stored.update( + { + "requestHash": self._composite_hash, + "identityScope": _MIXED_FORM_IDENTITY_SCOPE, + "mixedFormRequestHash": self._mixed_form_request_hash, + "batchRequestHash": self._batch_request_hash, + } + ) + self._backing[key] = stored + + def __delitem__(self, key: str) -> None: + del self._backing[key] + + def __iter__(self) -> Iterator[str]: + return iter(self._backing) + + def __len__(self) -> int: + return len(self._backing) + + +def _batch_request_shape(normalized: Mapping[str, Any]) -> dict[str, Any]: + """Detach validator output into the request shape the executor accepts. + + ``validate_agent_command`` annotates normalized commands with their schema + version, while an ``hwpx.agent-batch/v1`` request intentionally omits that + derived field. The executor validates requests again, so compiled plans + must retain the request shape rather than feed normalized commands back + through a non-idempotent validator. + """ + + batch = deepcopy(dict(normalized)) + for command in batch["commands"]: + command.pop("schemaVersion", None) + return batch + + +def _validate_locator(value: object, *, name: str) -> dict[str, Any]: + locator = _object(value, name) + kind = str(locator.get("kind", "")) + if kind not in MIXED_FORM_LOCATOR_KINDS: + raise AgentContractError( + "unknown_kind", f"unsupported mixed-form locator: {kind!r}", target=f"{name}.kind" + ) + + if kind == "nativeField": + _exact_keys( + locator, + required={"kind"}, + optional={"fieldId", "name"}, + name=name, + ) + selectors = [key for key in ("fieldId", "name") if key in locator] + if len(selectors) != 1: + raise AgentContractError( + "invalid_syntax", + f"{name} requires exactly one of fieldId or name", + target=name, + ) + selector = selectors[0] + return {"kind": kind, selector: _nonempty_string(locator[selector], f"{name}.{selector}")} + + if kind == "labelCell": + _exact_keys( + locator, + required={"kind", "section", "cellAnchor"}, + optional={"tableAnchor", "tableIndex"}, + name=name, + ) + table_selectors = [key for key in ("tableAnchor", "tableIndex") if key in locator] + if len(table_selectors) != 1: + raise AgentContractError( + "invalid_syntax", + f"{name} requires exactly one of tableAnchor or tableIndex", + target=name, + ) + anchor = _object(locator["cellAnchor"], f"{name}.cellAnchor") + _exact_keys( + anchor, + required={"label"}, + optional={"direction"}, + name=f"{name}.cellAnchor", + ) + direction = str(anchor.get("direction", "right")).lower() + if direction not in _DIRECTIONS: + raise AgentContractError( + "invalid_syntax", + f"{name}.cellAnchor.direction is unsupported", + target=f"{name}.cellAnchor.direction", + ) + normalized: dict[str, Any] = { + "kind": kind, + "section": _positive_integer(locator["section"], f"{name}.section"), + "cellAnchor": { + "label": _nonempty_string(anchor["label"], f"{name}.cellAnchor.label"), + "direction": direction, + }, + } + table_selector = table_selectors[0] + if table_selector == "tableAnchor": + normalized[table_selector] = _nonempty_string( + locator[table_selector], f"{name}.tableAnchor" + ) + else: + index = locator[table_selector] + if isinstance(index, bool) or not isinstance(index, int) or index < 0: + raise AgentContractError( + "invalid_syntax", + f"{name}.tableIndex must be a zero-based integer", + target=f"{name}.tableIndex", + ) + normalized[table_selector] = index + return normalized + + if kind == "canonicalPath": + _exact_keys(locator, required={"kind", "path"}, name=name) + return {"kind": kind, "path": _nonempty_string(locator["path"], f"{name}.path")} + + _exact_keys(locator, required={"kind", "section", "anchor"}, name=name) + return { + "kind": kind, + "section": _positive_integer(locator["section"], f"{name}.section"), + "anchor": _nonempty_string(locator["anchor"], f"{name}.anchor"), + } + + +def _validate_plan_target(value: object, *, name: str) -> dict[str, Any]: + """Validate the public ``hwpx.mixed-form-plan/v1`` target vocabulary.""" + + target = _object(value, name) + kind = str(target.get("kind", "")) + if kind not in MIXED_FORM_LOCATOR_KINDS: + raise AgentContractError( + "unknown_kind", f"unsupported mixed-form target: {kind!r}", target=f"{name}.kind" + ) + + if kind == "nativeField": + _exact_keys(target, required={"kind"}, optional={"fieldId", "name"}, name=name) + selectors = [key for key in ("fieldId", "name") if key in target] + if len(selectors) != 1: + raise AgentContractError( + "invalid_syntax", + f"{name} requires exactly one of fieldId or name", + target=name, + ) + selector = selectors[0] + return { + "kind": kind, + selector: _nonempty_string(target[selector], f"{name}.{selector}"), + } + + if kind == "labelCell": + _exact_keys( + target, + required={"kind", "sectionPath", "cellAnchor"}, + optional={"tableAnchor", "tableIndex"}, + name=name, + ) + table_selectors = [key for key in ("tableAnchor", "tableIndex") if key in target] + if len(table_selectors) != 1: + raise AgentContractError( + "invalid_syntax", + f"{name} requires exactly one of tableAnchor or tableIndex", + target=name, + ) + anchor = _object(target["cellAnchor"], f"{name}.cellAnchor") + _exact_keys( + anchor, + required={"label", "direction"}, + name=f"{name}.cellAnchor", + ) + direction = _nonempty_string( + anchor["direction"], f"{name}.cellAnchor.direction" + ).lower() + if direction not in _DIRECTIONS: + raise AgentContractError( + "invalid_syntax", + f"{name}.cellAnchor.direction is unsupported", + target=f"{name}.cellAnchor.direction", + ) + normalized: dict[str, Any] = { + "kind": kind, + "section": _section_index_from_path(target["sectionPath"], f"{name}.sectionPath"), + "cellAnchor": { + "label": _nonempty_string(anchor["label"], f"{name}.cellAnchor.label"), + "direction": direction, + }, + } + table_selector = table_selectors[0] + if table_selector == "tableAnchor": + normalized[table_selector] = _nonempty_string( + target[table_selector], f"{name}.tableAnchor" + ) + else: + table_index = target[table_selector] + if isinstance(table_index, bool) or not isinstance(table_index, int) or table_index < 0: + raise AgentContractError( + "invalid_syntax", + f"{name}.tableIndex must be a zero-based integer", + target=f"{name}.tableIndex", + ) + normalized[table_selector] = table_index + return normalized + + if kind == "canonicalPath": + _exact_keys(target, required={"kind", "path"}, name=name) + return {"kind": kind, "path": _nonempty_string(target["path"], f"{name}.path")} + + _exact_keys( + target, + required={"kind", "sectionPath", "anchor", "expectedCount"}, + name=name, + ) + expected_count = target["expectedCount"] + if isinstance(expected_count, bool) or expected_count != 1: + raise AgentContractError( + "invalid_syntax", + f"{name}.expectedCount must be exactly 1", + target=f"{name}.expectedCount", + ) + return { + "kind": kind, + "section": _section_index_from_path(target["sectionPath"], f"{name}.sectionPath"), + "anchor": _nonempty_string(target["anchor"], f"{name}.anchor"), + "expectedCount": 1, + } + + +def _validate_public_plan_request(request: Mapping[str, Any]) -> dict[str, Any]: + """Normalize the P1-frozen public plan into the internal request shape.""" + + required = { + "schemaVersion", + "source", + "output", + "expectedRevision", + "idempotencyKey", + "dryRun", + "overwrite", + "quality", + "verificationRequirements", + "operations", + } + _exact_keys(request, required=required, name="mixedFormPlan") + source = _nonempty_string(request["source"], "mixedFormPlan.source") + output = _nonempty_string(request["output"], "mixedFormPlan.output") + if not isinstance(request["dryRun"], bool): + raise AgentContractError( + "invalid_syntax", "mixedFormPlan.dryRun must be boolean", target="mixedFormPlan.dryRun" + ) + if not isinstance(request["overwrite"], bool): + raise AgentContractError( + "invalid_syntax", + "mixedFormPlan.overwrite must be boolean", + target="mixedFormPlan.overwrite", + ) + + expected_revision = request["expectedRevision"] + if expected_revision is not None and ( + not isinstance(expected_revision, str) + or REVISION_PATTERN.fullmatch(expected_revision) is None + ): + raise AgentContractError( + "invalid_syntax", + "mixedFormPlan.expectedRevision must be sha256 or null", + target="mixedFormPlan.expectedRevision", + ) + idempotency_key = request["idempotencyKey"] + if idempotency_key is not None and ( + not isinstance(idempotency_key, str) or not 1 <= len(idempotency_key) <= 128 + ): + raise AgentContractError( + "resource_limit", + "mixedFormPlan.idempotencyKey length is invalid", + target="mixedFormPlan.idempotencyKey", + ) + + raw_operations = _array(request["operations"], "mixedFormPlan.operations") + if not raw_operations or len(raw_operations) > MAX_COMMANDS: + raise AgentContractError( + "resource_limit", + "mixedFormPlan.operations count is out of bounds", + target="mixedFormPlan.operations", + ) + operations: list[dict[str, Any]] = [] + operation_ids: list[str] = [] + for index, raw_operation in enumerate(raw_operations): + name = f"mixedFormPlan.operations[{index}]" + operation = _object(raw_operation, name) + _exact_keys(operation, required={"operationId", "target", "value"}, name=name) + operation_id = operation.get("operationId") + if not isinstance(operation_id, str) or not COMMAND_ID_PATTERN.fullmatch(operation_id): + raise AgentContractError( + "invalid_syntax", f"{name}.operationId is invalid", target=f"{name}.operationId" + ) + operation_ids.append(operation_id) + operations.append( + { + "operationId": operation_id, + "locator": _validate_plan_target(operation["target"], name=f"{name}.target"), + "value": _value_string(operation["value"], f"{name}.value"), + } + ) + if len(operation_ids) != len(set(operation_ids)): + raise AgentContractError( + "invariant_violation", + "mixed-form operationId values must be unique", + target="mixedFormPlan.operations", + ) + + requirements = _array( + request["verificationRequirements"], "mixedFormPlan.verificationRequirements" + ) + if any(not isinstance(item, str) for item in requirements): + raise AgentContractError( + "invalid_syntax", + "mixedFormPlan.verificationRequirements items must be strings", + target="mixedFormPlan.verificationRequirements", + ) + unknown_requirements = sorted(set(requirements) - set(VERIFICATION_REQUIREMENTS)) + if unknown_requirements: + raise AgentContractError( + "invalid_syntax", + f"unknown verification requirements: {unknown_requirements}", + target="mixedFormPlan.verificationRequirements", + ) + return { + "schemaVersion": MIXED_FORM_REQUEST_SCHEMA, + "input": {"filename": source}, + "output": {"filename": output, "overwrite": request["overwrite"]}, + "operations": operations, + "expectedRevision": expected_revision, + "idempotencyKey": idempotency_key, + "dryRun": request["dryRun"], + "quality": _validate_quality(request["quality"]), + "verificationRequirements": list(dict.fromkeys(requirements)), + } + + +def validate_mixed_form_request(value: Mapping[str, Any]) -> dict[str, Any]: + """Validate a public plan or internal envelope into the internal shape.""" + + request = _object(value, "mixedFormRequest") + schema_version = request.get("schemaVersion") + if schema_version == MIXED_FORM_PLAN_SCHEMA: + return _validate_public_plan_request(request) + if schema_version != MIXED_FORM_REQUEST_SCHEMA: + raise AgentContractError( + "invalid_syntax", "unsupported mixed-form request schema", target="schemaVersion" + ) + required = { + "schemaVersion", + "input", + "output", + "operations", + "expectedRevision", + "idempotencyKey", + "dryRun", + "quality", + "verificationRequirements", + } + _exact_keys(request, required=required, name="mixedFormRequest") + input_ref = _object(request["input"], "mixedFormRequest.input") + _exact_keys(input_ref, required={"filename"}, name="mixedFormRequest.input") + input_filename = _nonempty_string( + input_ref["filename"], "mixedFormRequest.input.filename" + ) + + output_ref = _object(request["output"], "mixedFormRequest.output") + _exact_keys( + output_ref, + required={"filename", "overwrite"}, + name="mixedFormRequest.output", + ) + output_filename = _nonempty_string( + output_ref["filename"], "mixedFormRequest.output.filename" + ) + if not isinstance(output_ref["overwrite"], bool): + raise AgentContractError( + "invalid_syntax", + "mixedFormRequest.output.overwrite must be boolean", + target="mixedFormRequest.output.overwrite", + ) + + raw_operations = _array(request["operations"], "mixedFormRequest.operations") + if not raw_operations or len(raw_operations) > MAX_COMMANDS: + raise AgentContractError( + "resource_limit", + "mixedFormRequest.operations count is out of bounds", + target="mixedFormRequest.operations", + ) + operations: list[dict[str, Any]] = [] + operation_ids: list[str] = [] + for index, raw_operation in enumerate(raw_operations): + name = f"mixedFormRequest.operations[{index}]" + operation = _object(raw_operation, name) + _exact_keys(operation, required={"operationId", "locator", "value"}, name=name) + operation_id = str(operation.get("operationId", "")) + if not COMMAND_ID_PATTERN.fullmatch(operation_id): + raise AgentContractError( + "invalid_syntax", f"{name}.operationId is invalid", target=f"{name}.operationId" + ) + operation_ids.append(operation_id) + operations.append( + { + "operationId": operation_id, + "locator": _validate_locator(operation["locator"], name=f"{name}.locator"), + "value": _value_string(operation["value"], f"{name}.value"), + } + ) + if len(operation_ids) != len(set(operation_ids)): + raise AgentContractError( + "invariant_violation", + "mixed-form operationId values must be unique", + target="mixedFormRequest.operations", + ) + + expected_revision = request["expectedRevision"] + if expected_revision is not None and not REVISION_PATTERN.fullmatch(str(expected_revision)): + raise AgentContractError( + "invalid_syntax", + "mixedFormRequest.expectedRevision must be sha256 or null", + target="mixedFormRequest.expectedRevision", + ) + idempotency_key = request["idempotencyKey"] + if idempotency_key is not None and ( + not isinstance(idempotency_key, str) or not 1 <= len(idempotency_key) <= 128 + ): + raise AgentContractError( + "resource_limit", + "mixedFormRequest.idempotencyKey length is invalid", + target="mixedFormRequest.idempotencyKey", + ) + if not isinstance(request["dryRun"], bool): + raise AgentContractError( + "invalid_syntax", + "mixedFormRequest.dryRun must be boolean", + target="mixedFormRequest.dryRun", + ) + + requirements = _array( + request["verificationRequirements"], + "mixedFormRequest.verificationRequirements", + ) + if any(not isinstance(item, str) for item in requirements): + raise AgentContractError( + "invalid_syntax", + "mixedFormRequest.verificationRequirements items must be strings", + target="mixedFormRequest.verificationRequirements", + ) + unknown_requirements = sorted(set(requirements) - set(VERIFICATION_REQUIREMENTS)) + if unknown_requirements: + raise AgentContractError( + "invalid_syntax", + f"unknown verification requirements: {unknown_requirements}", + target="mixedFormRequest.verificationRequirements", + ) + + return { + "schemaVersion": MIXED_FORM_REQUEST_SCHEMA, + "input": {"filename": input_filename}, + "output": { + "filename": output_filename, + "overwrite": output_ref["overwrite"], + }, + "operations": operations, + "expectedRevision": None if expected_revision is None else str(expected_revision), + "idempotencyKey": idempotency_key, + "dryRun": request["dryRun"], + "quality": _validate_quality(request["quality"]), + "verificationRequirements": list(dict.fromkeys(str(item) for item in requirements)), + } + + +@dataclass(frozen=True, slots=True) +class MixedFormResolution: + operation_id: str + locator_kind: str + path: str + node_kind: str + stability: str + section: int | None = None + table_index: int | None = None + logical_row: int | None = None + logical_column: int | None = None + physical_row: int | None = None + physical_column: int | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "operationId": self.operation_id, + "locatorKind": self.locator_kind, + "path": self.path, + "nodeKind": self.node_kind, + "stability": self.stability, + "section": self.section, + "tableIndex": self.table_index, + "logicalRow": self.logical_row, + "logicalColumn": self.logical_column, + "physicalRow": self.physical_row, + "physicalColumn": self.physical_column, + } + + +@dataclass(frozen=True, slots=True) +class MixedFormPlan: + input_revision: str + request_hash: str + resolutions: tuple[MixedFormResolution, ...] + batch: Mapping[str, Any] + plan_hash: str + + def to_dict(self) -> dict[str, Any]: + return { + "schemaVersion": MIXED_FORM_COMPILED_PLAN_SCHEMA, + "inputRevision": self.input_revision, + "requestHash": self.request_hash, + "resolutions": [resolution.to_dict() for resolution in self.resolutions], + "batch": deepcopy(dict(self.batch)), + "planHash": self.plan_hash, + } + + +def _section_record(agent: HwpxAgentDocument, section_index: int) -> NodeRecord: + matches = [ + record + for record in agent.records + if record.kind == "section" and record.summary.get("index") == section_index + ] + if not matches: + raise AgentContractError( + "not_found", f"section {section_index} was not found", target=f"section[{section_index}]" + ) + if len(matches) != 1: # pragma: no cover - projection invariant + raise AgentContractError( + "invariant_violation", + f"section {section_index} projected more than once", + target=f"section[{section_index}]", + ) + return matches[0] + + +def _native_field_record(agent: HwpxAgentDocument, locator: Mapping[str, Any]) -> NodeRecord: + records = [record for record in agent.records if record.kind == "form-field"] + if "fieldId" in locator: + wanted = str(locator["fieldId"]).strip() + matches = [ + record + for record in records + if wanted + in { + str(record.native.get("field_id") or ""), + str(record.native.get("id") or ""), + str(record.native.get("fieldid") or ""), + } + ] + target = f"nativeField.fieldId={wanted!r}" + else: + wanted = str(locator["name"]).strip().casefold() + matches = [ + record + for record in records + if wanted + in { + str(record.native.get("name") or "").strip().casefold(), + str(record.native.get("prompt") or "").strip().casefold(), + str(record.native.get("instruction") or "").strip().casefold(), + } + ] + target = f"nativeField.name={locator['name']!r}" + if not matches: + raise AgentContractError("not_found", "native form field was not found", target=target) + if len(matches) > 1: + raise AgentContractError( + "ambiguous_target", + f"native form-field locator matched {len(matches)} fields", + target=target, + ) + return matches[0] + + +def _table_resolution_error(exc: ValueError, *, target: str) -> AgentContractError: + message = str(exc) + lowered = message.casefold() + if "ambiguous" in lowered: + code = "ambiguous_target" + elif "invalid" in lowered: + code = "unsupported_content" + else: + code = "not_found" + return AgentContractError(code, message, target=target) + + +def _table_cell_record( + agent: HwpxAgentDocument, + section: NodeRecord, + target: ResolvedCellTarget, +) -> NodeRecord: + part_name = str(section.native.part_name) + tables = [ + record + for record in agent.records + if record.kind == "table" + and str(record.native.paragraph.section.part_name) == part_name + ] + if not 0 <= target.table_index < len(tables): + raise AgentContractError( + "invariant_violation", + "byte table ordinal is absent from the semantic projection", + target=f"table[{target.table_index}]", + ) + table = tables[target.table_index] + matches = [ + record + for record in agent.records + if record.kind == "cell" + and record.native.table.element is table.native.element + and tuple(record.native.address) == (target.row, target.col) + ] + if not matches: + raise AgentContractError( + "not_found", + "resolved physical cell is absent from the semantic projection", + target=table.path, + ) + if len(matches) > 1: # pragma: no cover - invalid projection invariant + raise AgentContractError( + "ambiguous_target", + "resolved physical cell projected more than once", + target=table.path, + ) + return matches[0] + + +def _label_cell_record( + agent: HwpxAgentDocument, + source_bytes: bytes, + locator: Mapping[str, Any], +) -> tuple[NodeRecord, ResolvedCellTarget]: + section_index = int(locator["section"]) + section = _section_record(agent, section_index) + byte_locator: dict[str, Any] = { + "section_path": str(section.native.part_name), + "cell_anchor": { + "label": locator["cellAnchor"]["label"], + "direction": locator["cellAnchor"]["direction"], + }, + } + if "tableAnchor" in locator: + byte_locator["table_anchor"] = locator["tableAnchor"] + else: + byte_locator["table_index"] = locator["tableIndex"] + try: + resolved = resolve_cell_target(source_bytes, byte_locator) + except ValueError as exc: + raise _table_resolution_error(exc, target="labelCell") from exc + return _table_cell_record(agent, section, resolved), resolved + + +def _local_name(element: Any) -> str: + return str(getattr(element, "tag", "")).rsplit("}", 1)[-1] + + +def _body_anchor_record( + agent: HwpxAgentDocument, + locator: Mapping[str, Any], +) -> tuple[NodeRecord, str]: + section_index = int(locator["section"]) + section = _section_record(agent, section_index) + anchor = str(locator["anchor"]) + paragraphs = [ + record + for record in agent.records + if record.kind == "paragraph" and record.parent_path == section.path + ] + occurrences: list[tuple[NodeRecord, int]] = [] + for paragraph in paragraphs: + count = str(paragraph.native.text).count(anchor) + if count: + occurrences.append((paragraph, count)) + total = sum(count for _paragraph, count in occurrences) + if total == 0: + raise AgentContractError( + "not_found", "direct-body anchor was not found", target=f"bodyAnchor={anchor!r}" + ) + if total != 1: + raise AgentContractError( + "ambiguous_target", + f"direct-body anchor matched {total} times", + target=f"bodyAnchor={anchor!r}", + ) + paragraph = occurrences[0][0] + runs = [ + record + for record in agent.records + if record.kind == "run" and record.parent_path == paragraph.path + ] + run_hits = [(record, str(record.native.text).count(anchor)) for record in runs] + run_hits = [(record, count) for record, count in run_hits if count] + if not run_hits: + raise AgentContractError( + "unsupported_content", + "direct-body anchor crosses run boundaries", + target=paragraph.path, + ) + if sum(count for _record, count in run_hits) != 1: + raise AgentContractError( + "ambiguous_target", + "direct-body anchor is ambiguous within its paragraph", + target=paragraph.path, + ) + run = run_hits[0][0] + children = list(run.native.element) + if not children or any(_local_name(child) != "t" or len(list(child)) for child in children): + raise AgentContractError( + "unsupported_content", + "direct-body anchor run contains non-plain inline content", + target=run.path, + ) + return run, str(run.native.text).replace(anchor, str(locator.get("value", "")), 1) + + +def _compile_operation( + agent: HwpxAgentDocument, + source_bytes: bytes, + operation: Mapping[str, Any], +) -> tuple[dict[str, Any], MixedFormResolution]: + operation_id = str(operation["operationId"]) + locator = operation["locator"] + locator_kind = str(locator["kind"]) + value = str(operation["value"]) + resolved_table: ResolvedCellTarget | None = None + + if locator_kind == "nativeField": + record = _native_field_record(agent, locator) + property_name = "value" + compiled_value = value + elif locator_kind == "labelCell": + record, resolved_table = _label_cell_record(agent, source_bytes, locator) + property_name = "text" + compiled_value = value + elif locator_kind == "canonicalPath": + record = agent.resolve_record(str(locator["path"]), expected_revision=agent.revision) + if record.kind not in _FILLABLE_NODE_KINDS: + raise AgentContractError( + "unsupported_operation", + f"canonical path targets non-fillable kind {record.kind!r}", + target=record.path, + ) + property_name = "value" if record.kind == "form-field" else "text" + compiled_value = value + else: + body_locator = dict(locator) + body_locator["value"] = value + record, compiled_value = _body_anchor_record(agent, body_locator) + property_name = "text" + + command = { + "commandId": operation_id, + "op": "set", + "path": record.path, + "properties": {property_name: compiled_value}, + } + section = int(locator["section"]) if "section" in locator else None + resolution = MixedFormResolution( + operation_id=operation_id, + locator_kind=locator_kind, + path=record.path, + node_kind=record.kind, + stability=record.stability, + section=section, + table_index=(resolved_table.table_index if resolved_table is not None else None), + logical_row=(resolved_table.logical_row if resolved_table is not None else None), + logical_column=(resolved_table.logical_col if resolved_table is not None else None), + physical_row=(resolved_table.row if resolved_table is not None else None), + physical_column=(resolved_table.col if resolved_table is not None else None), + ) + return command, resolution + + +def plan_mixed_form_fill(request: Mapping[str, Any]) -> MixedFormPlan: + """Resolve four target types and compile one revision-bound agent batch.""" + + normalized = validate_mixed_form_request(request) + input_path = Path(normalized["input"]["filename"]) + _assert_distinct_source_output( + normalized["input"]["filename"], + normalized["output"]["filename"], + ) + try: + source_bytes = input_path.read_bytes() + except OSError as exc: + raise AgentContractError( + "not_found", f"mixed-form input could not be read: {exc}", target="input.filename" + ) from exc + + actual_revision = "sha256:" + hashlib.sha256(source_bytes).hexdigest() + expected_revision = normalized["expectedRevision"] + if expected_revision not in {None, actual_revision}: + raise AgentContractError( + "stale_revision", + "mixed-form expectedRevision does not match input bytes", + target="expectedRevision", + ) + + commands: list[dict[str, Any]] = [] + resolutions: list[MixedFormResolution] = [] + try: + with HwpxAgentDocument.open(source_bytes) as agent: + for operation in normalized["operations"]: + command, resolution = _compile_operation(agent, source_bytes, operation) + commands.append(command) + resolutions.append(resolution) + except AgentContractError: + raise + except Exception as exc: + raise AgentContractError( + "verification_failed", + f"mixed-form input could not be projected safely: {type(exc).__name__}: {exc}", + target="input.filename", + ) from exc + + resolved_paths = [resolution.path for resolution in resolutions] + if len(resolved_paths) != len(set(resolved_paths)): + raise AgentContractError( + "identity_collision", + "mixed-form operations resolve to the same canonical target", + target="operations", + ) + + normalized_batch = validate_agent_batch( + { + "schemaVersion": AGENT_BATCH_SCHEMA, + "input": deepcopy(normalized["input"]), + "output": deepcopy(normalized["output"]), + "commands": commands, + # A compiled plan is always revision-bound, including when the + # planning request deliberately supplied null. + "expectedRevision": actual_revision, + "idempotencyKey": normalized["idempotencyKey"], + "dryRun": normalized["dryRun"], + "quality": deepcopy(normalized["quality"]), + "verificationRequirements": list(normalized["verificationRequirements"]), + } + ) + batch = _batch_request_shape(normalized_batch) + request_hash = _canonical_hash(normalized) + plan_payload: dict[str, Any] = { + "schemaVersion": MIXED_FORM_COMPILED_PLAN_SCHEMA, + "inputRevision": actual_revision, + "requestHash": request_hash, + "resolutions": [resolution.to_dict() for resolution in resolutions], + "batch": deepcopy(batch), + "planHash": None, + } + plan_hash = _canonical_hash(plan_payload) + return MixedFormPlan( + input_revision=actual_revision, + request_hash=request_hash, + resolutions=tuple(resolutions), + batch=batch, + plan_hash=plan_hash, + ) + + +def validate_mixed_form_plan(value: MixedFormPlan | Mapping[str, Any]) -> MixedFormPlan: + """Validate a detached compiled plan, including its content hash.""" + + plan = value.to_dict() if isinstance(value, MixedFormPlan) else _object(value, "mixedFormPlan") + _exact_keys(plan, required=_COMPILED_PLAN_KEYS, name="mixedFormPlan") + if plan["schemaVersion"] != MIXED_FORM_COMPILED_PLAN_SCHEMA: + raise AgentContractError( + "invalid_syntax", "unsupported mixed-form plan schema", target="schemaVersion" + ) + input_revision = str(plan["inputRevision"]) + request_hash = str(plan["requestHash"]) + plan_hash = str(plan["planHash"]) + for name, digest in ( + ("inputRevision", input_revision), + ("requestHash", request_hash), + ("planHash", plan_hash), + ): + if not REVISION_PATTERN.fullmatch(digest): + raise AgentContractError( + "invalid_syntax", f"mixedFormPlan.{name} must be sha256", target=name + ) + + raw_resolutions = _array(plan["resolutions"], "mixedFormPlan.resolutions") + resolutions: list[MixedFormResolution] = [] + for index, raw_resolution in enumerate(raw_resolutions): + name = f"mixedFormPlan.resolutions[{index}]" + resolution = _object(raw_resolution, name) + _exact_keys(resolution, required=_RESOLUTION_KEYS, name=name) + locator_kind = str(resolution["locatorKind"]) + node_kind = str(resolution["nodeKind"]) + stability = str(resolution["stability"]) + if locator_kind not in MIXED_FORM_LOCATOR_KINDS: + raise AgentContractError("unknown_kind", "unknown plan locator kind", target=name) + if node_kind not in _FILLABLE_NODE_KINDS: + raise AgentContractError("unknown_kind", "plan targets a non-fillable kind", target=name) + if stability not in STABILITY_LEVELS: + raise AgentContractError("invalid_syntax", "unknown path stability", target=name) + section = resolution["section"] + if section is not None: + section = _positive_integer(section, f"{name}.section") + table_index = _nonnegative_integer_or_none(resolution["tableIndex"], f"{name}.tableIndex") + logical_row = _nonnegative_integer_or_none(resolution["logicalRow"], f"{name}.logicalRow") + logical_column = _nonnegative_integer_or_none( + resolution["logicalColumn"], f"{name}.logicalColumn" + ) + physical_row = _nonnegative_integer_or_none( + resolution["physicalRow"], f"{name}.physicalRow" + ) + physical_column = _nonnegative_integer_or_none( + resolution["physicalColumn"], f"{name}.physicalColumn" + ) + path = _nonempty_string(resolution["path"], f"{name}.path") + parsed_path = parse_path(path) + if parsed_path.canonical != path or not parsed_path.segments: + raise AgentContractError( + "invalid_syntax", "resolution path must be canonical and non-root", target=f"{name}.path" + ) + path_kind = parsed_path.segments[-1].kind + if node_kind != path_kind: + raise AgentContractError( + "invariant_violation", + "resolution nodeKind does not match its canonical path kind", + target=name, + ) + coordinates = (table_index, logical_row, logical_column, physical_row, physical_column) + if locator_kind == "labelCell": + if section is None or any(item is None for item in coordinates) or node_kind != "cell": + raise AgentContractError( + "invariant_violation", "labelCell plan lacks a physical resolution", target=name + ) + elif any(item is not None for item in coordinates): + raise AgentContractError( + "invariant_violation", "non-table plan carries table coordinates", target=name + ) + if locator_kind == "nativeField" and (node_kind != "form-field" or section is not None): + raise AgentContractError( + "invariant_violation", + "nativeField plan must resolve only to a form-field without section metadata", + target=name, + ) + if locator_kind == "canonicalPath" and section is not None: + raise AgentContractError( + "invariant_violation", + "canonicalPath plan must not carry locator section metadata", + target=name, + ) + if locator_kind == "bodyAnchor" and (node_kind != "run" or section is None): + raise AgentContractError( + "invariant_violation", + "bodyAnchor plan must resolve to a run in its declared section", + target=name, + ) + if locator_kind in {"labelCell", "bodyAnchor"}: + first = parsed_path.segments[0] + if first.kind != "section" or first.index != section: + raise AgentContractError( + "invariant_violation", + "resolution path does not belong to its declared section", + target=name, + ) + operation_id = str(resolution["operationId"]) + if not COMMAND_ID_PATTERN.fullmatch(operation_id): + raise AgentContractError("invalid_syntax", "invalid resolution operationId", target=name) + resolutions.append( + MixedFormResolution( + operation_id=operation_id, + locator_kind=locator_kind, + path=path, + node_kind=node_kind, + stability=stability, + section=section, + table_index=table_index, + logical_row=logical_row, + logical_column=logical_column, + physical_row=physical_row, + physical_column=physical_column, + ) + ) + + normalized_batch = validate_agent_batch(_object(plan["batch"], "mixedFormPlan.batch")) + batch = _batch_request_shape(normalized_batch) + if normalized_batch["expectedRevision"] != input_revision: + raise AgentContractError( + "stale_revision", "compiled batch is not bound to inputRevision", target="batch.expectedRevision" + ) + if len(resolutions) != len(normalized_batch["commands"]): + raise AgentContractError( + "invariant_violation", "resolution and command counts differ", target="resolutions" + ) + if len({resolution.path for resolution in resolutions}) != len(resolutions): + raise AgentContractError( + "identity_collision", + "compiled plan contains duplicate canonical targets", + target="resolutions", + ) + for resolution, command in zip(resolutions, normalized_batch["commands"]): + if ( + command["commandId"] != resolution.operation_id + or command["op"] != "set" + or command["path"] != resolution.path + ): + raise AgentContractError( + "invariant_violation", + "compiled command does not match its resolution", + target=resolution.operation_id, + ) + expected_property = "value" if resolution.node_kind == "form-field" else "text" + if set(command["properties"]) != {expected_property}: + raise AgentContractError( + "invariant_violation", + "compiled command has a non-fill property", + target=resolution.operation_id, + ) + + hash_payload = deepcopy(plan) + hash_payload["planHash"] = None + if _canonical_hash(hash_payload) != plan_hash: + raise AgentContractError( + "verification_failed", "mixed-form plan hash mismatch", target="planHash" + ) + return MixedFormPlan( + input_revision=input_revision, + request_hash=request_hash, + resolutions=tuple(resolutions), + batch=batch, + plan_hash=plan_hash, + ) + + +def apply_mixed_form_plan( + plan: MixedFormPlan | Mapping[str, Any], + *, + idempotency_store: IdempotencyStore | None = None, + fault_injector: FaultInjector | None = None, + domain_verifier: DomainVerifier | None = None, + save_pipeline: SavePipeline | None = None, +) -> AgentBatchResult: + """Validate a compiled plan and delegate to the sole agent batch executor.""" + + normalized = validate_mixed_form_plan(plan) + _assert_distinct_source_output( + str(normalized.batch["input"]["filename"]), + str(normalized.batch["output"]["filename"]), + ) + executor_store = idempotency_store + idempotency_key = normalized.batch["idempotencyKey"] + if idempotency_store is not None and idempotency_key is not None: + executor_store = _MixedFormIdempotencyStore( + idempotency_store, + active_key=str(idempotency_key), + mixed_form_request_hash=normalized.request_hash, + batch_request_hash=_canonical_hash(validate_agent_batch(normalized.batch)), + ) + return apply_document_commands( + normalized.batch, + idempotency_store=executor_store, + fault_injector=fault_injector, + domain_verifier=domain_verifier, + save_pipeline=save_pipeline, + ) + + +def apply_mixed_form_fill( + request: Mapping[str, Any], + *, + idempotency_store: IdempotencyStore | None = None, + fault_injector: FaultInjector | None = None, + domain_verifier: DomainVerifier | None = None, + save_pipeline: SavePipeline | None = None, +) -> AgentBatchResult: + """Resolve every locator, then execute the resulting single atomic batch.""" + + return apply_mixed_form_plan( + plan_mixed_form_fill(request), + idempotency_store=idempotency_store, + fault_injector=fault_injector, + domain_verifier=domain_verifier, + save_pipeline=save_pipeline, + ) + + +def mixed_form_json_schemas() -> dict[str, Any]: + """Return the three explicit mixed-form contract schemas. + + ``plan`` is the public, human/agent-authored input. ``internalRequest`` is + the normalized locator envelope used inside the core. ``compiledPlan`` is + the revision-bound, hash-protected output consumed by + :func:`apply_mixed_form_plan`. + """ + + agent_schemas = agent_json_schemas() + locator_variants = [ + { + "type": "object", + "required": ["kind", "fieldId"], + "additionalProperties": False, + "properties": { + "kind": {"const": "nativeField"}, + "fieldId": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARS}, + }, + }, + { + "type": "object", + "required": ["kind", "name"], + "additionalProperties": False, + "properties": { + "kind": {"const": "nativeField"}, + "name": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARS}, + }, + }, + { + "type": "object", + "required": ["kind", "section", "tableAnchor", "cellAnchor"], + "additionalProperties": False, + "properties": { + "kind": {"const": "labelCell"}, + "section": {"type": "integer", "minimum": 1}, + "tableAnchor": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARS}, + "cellAnchor": {"$ref": "#/$defs/cellAnchor"}, + }, + }, + { + "type": "object", + "required": ["kind", "section", "tableIndex", "cellAnchor"], + "additionalProperties": False, + "properties": { + "kind": {"const": "labelCell"}, + "section": {"type": "integer", "minimum": 1}, + "tableIndex": {"type": "integer", "minimum": 0}, + "cellAnchor": {"$ref": "#/$defs/cellAnchor"}, + }, + }, + { + "type": "object", + "required": ["kind", "path"], + "additionalProperties": False, + "properties": { + "kind": {"const": "canonicalPath"}, + "path": {"type": "string", "minLength": 1}, + }, + }, + { + "type": "object", + "required": ["kind", "section", "anchor"], + "additionalProperties": False, + "properties": { + "kind": {"const": "bodyAnchor"}, + "section": {"type": "integer", "minimum": 1}, + "anchor": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARS}, + }, + }, + ] + request_schema = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "HWPX mixed-form request v1", + "type": "object", + "required": [ + "schemaVersion", + "input", + "output", + "operations", + "expectedRevision", + "idempotencyKey", + "dryRun", + "quality", + "verificationRequirements", + ], + "additionalProperties": False, + "$defs": { + "cellAnchor": { + "type": "object", + "required": ["label"], + "additionalProperties": False, + "properties": { + "label": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARS}, + "direction": {"enum": sorted(_DIRECTIONS)}, + }, + } + }, + "properties": { + "schemaVersion": {"const": MIXED_FORM_REQUEST_SCHEMA}, + "input": { + "type": "object", + "required": ["filename"], + "additionalProperties": False, + "properties": {"filename": {"type": "string", "minLength": 1}}, + }, + "output": { + "type": "object", + "required": ["filename", "overwrite"], + "additionalProperties": False, + "properties": { + "filename": {"type": "string", "minLength": 1}, + "overwrite": {"type": "boolean"}, + }, + }, + "operations": { + "type": "array", + "minItems": 1, + "maxItems": MAX_COMMANDS, + "items": { + "type": "object", + "required": ["operationId", "locator", "value"], + "additionalProperties": False, + "properties": { + "operationId": { + "type": "string", + "pattern": COMMAND_ID_PATTERN.pattern, + }, + "locator": {"oneOf": locator_variants}, + "value": {"type": "string", "maxLength": MAX_TEXT_CHARS}, + }, + }, + }, + "expectedRevision": { + "type": ["string", "null"], + "pattern": REVISION_PATTERN.pattern, + }, + "idempotencyKey": {"type": ["string", "null"], "minLength": 1, "maxLength": 128}, + "dryRun": {"type": "boolean"}, + "quality": deepcopy(agent_schemas["batch"]["properties"]["quality"]), + "verificationRequirements": { + "type": "array", + "items": {"enum": list(VERIFICATION_REQUIREMENTS)}, + "uniqueItems": True, + }, + }, + } + nullable_coordinate = {"type": ["integer", "null"], "minimum": 0} + embedded_batch_schema = deepcopy(agent_schemas["batch"]) + embedded_batch_schema.pop("$schema", None) + embedded_batch_schema.pop("title", None) + compiled_plan_schema = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "HWPX mixed-form compiled plan v1", + "type": "object", + "required": sorted(_COMPILED_PLAN_KEYS), + "additionalProperties": False, + "properties": { + "schemaVersion": {"const": MIXED_FORM_COMPILED_PLAN_SCHEMA}, + "inputRevision": {"type": "string", "pattern": REVISION_PATTERN.pattern}, + "requestHash": {"type": "string", "pattern": REVISION_PATTERN.pattern}, + "planHash": {"type": "string", "pattern": REVISION_PATTERN.pattern}, + "resolutions": { + "type": "array", + "minItems": 1, + "maxItems": MAX_COMMANDS, + "items": { + "type": "object", + "required": sorted(_RESOLUTION_KEYS), + "additionalProperties": False, + "properties": { + "operationId": {"type": "string", "pattern": COMMAND_ID_PATTERN.pattern}, + "locatorKind": {"enum": list(MIXED_FORM_LOCATOR_KINDS)}, + "path": {"type": "string", "minLength": 1}, + "nodeKind": {"enum": sorted(_FILLABLE_NODE_KINDS)}, + "stability": {"enum": list(STABILITY_LEVELS)}, + "section": {"type": ["integer", "null"], "minimum": 1}, + "tableIndex": deepcopy(nullable_coordinate), + "logicalRow": deepcopy(nullable_coordinate), + "logicalColumn": deepcopy(nullable_coordinate), + "physicalRow": deepcopy(nullable_coordinate), + "physicalColumn": deepcopy(nullable_coordinate), + }, + }, + }, + "batch": embedded_batch_schema, + }, + } + public_target_variants = [ + deepcopy(locator_variants[0]), + deepcopy(locator_variants[1]), + { + "type": "object", + "required": ["kind", "sectionPath", "tableAnchor", "cellAnchor"], + "additionalProperties": False, + "properties": { + "kind": {"const": "labelCell"}, + "sectionPath": {"type": "string", "pattern": r"^/section\[[1-9][0-9]*\]$"}, + "tableAnchor": { + "type": "string", + "minLength": 1, + "maxLength": MAX_TEXT_CHARS, + }, + "cellAnchor": {"$ref": "#/$defs/requiredCellAnchor"}, + }, + }, + { + "type": "object", + "required": ["kind", "sectionPath", "tableIndex", "cellAnchor"], + "additionalProperties": False, + "properties": { + "kind": {"const": "labelCell"}, + "sectionPath": {"type": "string", "pattern": r"^/section\[[1-9][0-9]*\]$"}, + "tableIndex": {"type": "integer", "minimum": 0}, + "cellAnchor": {"$ref": "#/$defs/requiredCellAnchor"}, + }, + }, + deepcopy(locator_variants[4]), + { + "type": "object", + "required": ["kind", "sectionPath", "anchor", "expectedCount"], + "additionalProperties": False, + "properties": { + "kind": {"const": "bodyAnchor"}, + "sectionPath": {"type": "string", "pattern": r"^/section\[[1-9][0-9]*\]$"}, + "anchor": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARS}, + "expectedCount": {"const": 1}, + }, + }, + ] + public_plan_schema = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "HWPX mixed-form plan v1", + "type": "object", + "required": [ + "schemaVersion", + "source", + "output", + "expectedRevision", + "idempotencyKey", + "dryRun", + "overwrite", + "quality", + "verificationRequirements", + "operations", + ], + "additionalProperties": False, + "$defs": { + "requiredCellAnchor": { + "type": "object", + "required": ["label", "direction"], + "additionalProperties": False, + "properties": { + "label": {"type": "string", "minLength": 1, "maxLength": MAX_TEXT_CHARS}, + "direction": {"enum": sorted(_DIRECTIONS)}, + }, + } + }, + "properties": { + "schemaVersion": {"const": MIXED_FORM_PLAN_SCHEMA}, + "source": {"type": "string", "minLength": 1}, + "output": {"type": "string", "minLength": 1}, + "expectedRevision": { + "type": ["string", "null"], + "pattern": REVISION_PATTERN.pattern, + }, + "idempotencyKey": { + "type": ["string", "null"], + "minLength": 1, + "maxLength": 128, + }, + "dryRun": {"type": "boolean"}, + "overwrite": {"type": "boolean"}, + "quality": deepcopy(agent_schemas["batch"]["properties"]["quality"]), + "verificationRequirements": { + "type": "array", + "items": {"enum": list(VERIFICATION_REQUIREMENTS)}, + "uniqueItems": True, + }, + "operations": { + "type": "array", + "minItems": 1, + "maxItems": MAX_COMMANDS, + "items": { + "type": "object", + "required": ["operationId", "target", "value"], + "additionalProperties": False, + "properties": { + "operationId": { + "type": "string", + "pattern": COMMAND_ID_PATTERN.pattern, + }, + "target": {"oneOf": public_target_variants}, + "value": {"type": "string", "maxLength": MAX_TEXT_CHARS}, + }, + }, + }, + }, + } + return { + "plan": deepcopy(public_plan_schema), + "compiledPlan": deepcopy(compiled_plan_schema), + "internalRequest": deepcopy(request_schema), + } + + +__all__ = [ + "MIXED_FORM_COMPILED_PLAN_SCHEMA", + "MIXED_FORM_LOCATOR_KINDS", + "MIXED_FORM_PLAN_SCHEMA", + "MIXED_FORM_REQUEST_SCHEMA", + "MixedFormPlan", + "MixedFormResolution", + "apply_mixed_form_fill", + "apply_mixed_form_plan", + "mixed_form_json_schemas", + "plan_mixed_form_fill", + "validate_mixed_form_plan", + "validate_mixed_form_request", +] diff --git a/src/hwpx/agent/model.py b/src/hwpx/agent/model.py index 7a1d18e..7871cd0 100644 --- a/src/hwpx/agent/model.py +++ b/src/hwpx/agent/model.py @@ -106,6 +106,16 @@ "requireReferenceIntegrity", } ) +_QUALITY_ENUM_VALUES = { + "mode": QUALITY_MODES, + "renderCheck": ("off", "auto", "required"), + "xsdMode": ("off", "lint"), + "overflowPolicy": ("fail", "warn", "truncate"), + "layoutLint": ("off", "warn", "strict"), +} +_QUALITY_BOOLEAN_KEYS = frozenset( + {"preserveUnmodifiedParts", "requireReferenceIntegrity"} +) # The v1 public property vocabulary. P2's command catalog consumes this exact # manifest. A property absent here cannot be set through generic commands. @@ -329,8 +339,20 @@ def _validate_quality(value: object) -> str | dict[str, Any] | None: f"batch.quality contains unknown fields: {sorted(extra)}", target="batch.quality", ) - if quality.get("mode", "transparent") not in QUALITY_MODES: - raise AgentContractError("invalid_syntax", "batch.quality.mode is unsupported", target="batch.quality.mode") + for name, choices in _QUALITY_ENUM_VALUES.items(): + if name in quality and quality[name] not in choices: + raise AgentContractError( + "invalid_syntax", + f"batch.quality.{name} is unsupported", + target=f"batch.quality.{name}", + ) + for name in _QUALITY_BOOLEAN_KEYS: + if name in quality and not isinstance(quality[name], bool): + raise AgentContractError( + "invalid_syntax", + f"batch.quality.{name} must be boolean", + target=f"batch.quality.{name}", + ) _validate_json_value(quality, name="batch.quality") return quality diff --git a/src/hwpx/table_patch.py b/src/hwpx/table_patch.py index 5f1d0cf..6ff4fe1 100644 --- a/src/hwpx/table_patch.py +++ b/src/hwpx/table_patch.py @@ -323,6 +323,35 @@ def to_dict(self) -> dict[str, Any]: } +@dataclass(frozen=True) +class ResolvedCellTarget: + """One non-mutating, merge-aware table-cell locator resolution. + + ``logical_row``/``logical_col`` record the address selected by the label + semantics. ``row``/``col`` are the physical ``cellAddr`` coordinates of + the covering cell and are therefore the coordinates a semantic document + projection can bind to exactly, even when the logical address lands inside + a row/column-spanning cell. + """ + + section_path: str + table_index: int + logical_row: int + logical_col: int + row: int + col: int + + def to_dict(self) -> dict[str, Any]: + return { + "sectionPath": self.section_path, + "tableIndex": self.table_index, + "logicalRow": self.logical_row, + "logicalColumn": self.logical_col, + "physicalRow": self.row, + "physicalColumn": self.col, + } + + @dataclass(frozen=True) class CellFillResult: data: bytes @@ -400,9 +429,17 @@ def _find_tables_by_anchor(section: bytes, anchor: str, spans: list[tuple[int, i matches.append(ti) return matches -def _resolve_cell_anchor(table: bytes, label: str, direction: str) -> list[tuple[int, int]]: +def _resolve_cell_anchor( + table: bytes, + label: str, + direction: str, + *, + exact: bool = False, +) -> list[tuple[int, int]]: """Logical (row, col) of the cell *direction* of the unique cell containing - *label*. direction in {right,left,below,above}.""" + *label*. ``exact=True`` requires normalized equality for canonical planning; + the compatibility default retains normalized substring matching. direction + in {right,left,below,above}.""" grid, _rep = build_grid(table) nl = _norm_anchor(label) if not nl: @@ -414,7 +451,9 @@ def _resolve_cell_anchor(table: bytes, label: str, direction: str) -> list[tuple if span_id in seen: continue seen.add(span_id) - if nl in _norm_anchor(_text_of(table[cell.start:cell.end])): + normalized_cell_text = _norm_anchor(_text_of(table[cell.start:cell.end])) + label_matches = normalized_cell_text == nl if exact else nl in normalized_cell_text + if label_matches: if direction == "left": tr, tc = cell.row, cell.col - 1 elif direction == "below": @@ -427,7 +466,12 @@ def _resolve_cell_anchor(table: bytes, label: str, direction: str) -> list[tuple targets.append((tr, tc)) return targets -def _resolve_anchor_cells(parts: Mapping[str, bytes], cells: Sequence[Any]) -> tuple[list[Any], list[CellSkipped]]: +def _resolve_anchor_cells( + parts: Mapping[str, bytes], + cells: Sequence[Any], + *, + exact_cell_labels: bool = False, +) -> tuple[list[Any], list[CellSkipped]]: """Turn anchor-bearing cell specs into concrete (table_index,row,col) specs. Coordinate specs pass through untouched; unresolvable anchors -> CellSkipped.""" out: list[Any] = [] @@ -470,7 +514,12 @@ def _resolve_anchor_cells(parts: Mapping[str, bytes], cells: Sequence[Any]) -> t cget = canchor.get if isinstance(canchor, Mapping) else (lambda k, d=None: getattr(canchor, k, d)) label = cget("label") or "" direction = str(cget("dir") or cget("direction") or "right") - m2 = _resolve_cell_anchor(sec[s:e], str(label), direction) + m2 = _resolve_cell_anchor( + sec[s:e], + str(label), + direction, + exact=exact_cell_labels, + ) if len(m2) == 1: row, col = m2[0] elif not m2: @@ -500,6 +549,70 @@ def _normalize(cell: Mapping[str, Any] | Any) -> tuple[str, int, int, int, str, return section, int(table_index), int(row), int(col), str(text), (int(max_lines) if max_lines else None) +def resolve_cell_target( + source: str | Path | bytes, + locator: Mapping[str, Any], +) -> ResolvedCellTarget: + """Resolve one existing table/cell locator without mutating the package. + + This is the narrow public planning seam for the same heading/table semantics + used by :func:`fill_cells`, with stricter normalized-exact adjacent labels. + It deliberately raises instead of returning a guessed target: zero/multiple + anchor matches, absent sections, invalid table indices, and out-of-range + cells all fail closed. + """ + + source_bytes = _read_source_bytes(source) + import io + import zipfile + + with zipfile.ZipFile(io.BytesIO(source_bytes), "r") as archive: + parts = { + info.filename: archive.read(info.filename) + for info in archive.infolist() + if not info.is_dir() + } + + # _resolve_anchor_cells also accepts concrete coordinates. A placeholder + # text value is supplied only because its historical fill-cell record shape + # includes text; no edit or serialization occurs here. + spec = dict(locator) + spec.setdefault("text", "") + # Mixed-form planning is a canonical, fail-closed path: after whitespace + # and punctuation normalization, a cell label must be exactly equal. The + # older fill_cells compatibility path deliberately keeps substring matching. + resolved, skipped = _resolve_anchor_cells(parts, [spec], exact_cell_labels=True) + if skipped: + raise ValueError(skipped[0].reason) + if len(resolved) != 1: + raise ValueError("cell locator did not resolve exactly one target") + + section_path, table_index, logical_row, logical_col, _text, _max_lines = _normalize( + resolved[0] + ) + section = parts.get(section_path) + if section is None: + raise ValueError("section part not found") + spans = _iter_table_spans(section) + if not 0 <= table_index < len(spans): + raise ValueError("table_index out of range") + start, end = spans[table_index] + grid, report = build_grid(section[start:end]) + if not report.ok: + raise ValueError("table grid is invalid: " + "; ".join(report.issues[:5])) + cell = grid.get((logical_row, logical_col)) + if cell is None: + raise ValueError("cell address out of range") + return ResolvedCellTarget( + section_path=section_path, + table_index=table_index, + logical_row=logical_row, + logical_col=logical_col, + row=cell.row, + col=cell.col, + ) + + def fill_cells( source: str | Path | bytes, cells: Sequence[Mapping[str, Any] | Any], @@ -527,7 +640,8 @@ def fill_cells( open_safety, _ = _finalize(source_bytes, output_path, source=source) return CellFillResult(source_bytes, (), (), (), True, "none", open_safety) - import io, zipfile + import io + import zipfile with zipfile.ZipFile(io.BytesIO(source_bytes), "r") as zf: parts = {i.filename: zf.read(i.filename) for i in zf.infolist() if not i.is_dir()} @@ -1233,7 +1347,8 @@ def _apply_cell_line_spacing( cells:[[r,c],...] 또는 rows:[r,...], line_spacing:int(PERCENT)}. 대상 셀의 각 문단 paraPr을 lineSpacing 변형으로 재매핑하고 linesegarray를 제거한다 (stale 캐시 줄겹침 방지). 중첩 표를 품은 셀은 refuse.""" - import io, zipfile + import io + import zipfile with zipfile.ZipFile(io.BytesIO(source_bytes), "r") as zf: parts = {i.filename: zf.read(i.filename) for i in zf.infolist() if not i.is_dir()} header_name = _header_part_name(parts) @@ -1338,7 +1453,8 @@ def _p_wrapper_span(section: bytes, table_start: int) -> tuple[int, int]: def _sections(data: bytes) -> dict[str, bytes]: - import io, zipfile + import io + import zipfile with zipfile.ZipFile(io.BytesIO(data)) as z: return {n: z.read(n) for n in z.namelist() if re.search(r"section\d+\.xml$", n)} @@ -1615,5 +1731,6 @@ def table_summary(source: str | Path | bytes) -> list[dict[str, Any]]: __all__ = [ "fill_cells", "build_grid", "GridReport", "CellFillResult", "CellApplied", "CellSkipped", + "ResolvedCellTarget", "resolve_cell_target", "apply_table_ops", "TableStructureError", "verify_fill", "RenderCheckRequired", "table_summary", ] diff --git a/tests/test_agent_mixed_form.py b/tests/test_agent_mixed_form.py new file mode 100644 index 0000000..2e9a66c --- /dev/null +++ b/tests/test_agent_mixed_form.py @@ -0,0 +1,726 @@ +# SPDX-License-Identifier: Apache-2.0 +from __future__ import annotations + +import ast +import copy +import hashlib +import io +import json +import zipfile +from pathlib import Path +from typing import Any + +import pytest + +from hwpx import HwpxDocument, validate_editor_open_safety +from hwpx.agent import ( + AGENT_BATCH_SCHEMA, + MIXED_FORM_COMPILED_PLAN_SCHEMA, + MIXED_FORM_LOCATOR_KINDS, + MIXED_FORM_PLAN_SCHEMA, + MIXED_FORM_REQUEST_SCHEMA, + AgentContractError, + HwpxAgentDocument, + apply_mixed_form_fill, + apply_mixed_form_plan, + mixed_form_json_schemas, + plan_mixed_form_fill, + validate_mixed_form_plan, + validate_mixed_form_request, +) +from hwpx.agent import form_plan as mixed_form_module +from hwpx.agent.catalog import agent_json_schemas +from hwpx.oxml.namespaces import HP +from hwpx.quality import SavePipeline +from hwpx.visual import NullOracle + + +def _append_native_field( + paragraph: Any, + *, + field_id: str, + name: str, + value: str, +) -> None: + begin_run = paragraph.element.makeelement(f"{HP}run", {"charPrIDRef": "0"}) + begin_control = begin_run.makeelement(f"{HP}ctrl", {"type": "FORM"}) + begin_control.append( + begin_control.makeelement( + f"{HP}fieldBegin", + { + "id": field_id, + "fieldName": name, + "type": "FORM", + "editable": "true", + }, + ) + ) + begin_run.append(begin_control) + paragraph.element.append(begin_run) + + value_run = paragraph.element.makeelement(f"{HP}run", {"charPrIDRef": "0"}) + text = value_run.makeelement(f"{HP}t", {}) + text.text = value + value_run.append(text) + paragraph.element.append(value_run) + + end_run = paragraph.element.makeelement(f"{HP}run", {"charPrIDRef": "0"}) + end_control = end_run.makeelement(f"{HP}ctrl", {"type": "FORM"}) + end_control.append( + end_control.makeelement( + f"{HP}fieldEnd", + {"beginIDRef": field_id, "fieldid": field_id}, + ) + ) + end_run.append(end_control) + paragraph.element.append(end_run) + paragraph.section.mark_dirty() + + +def _write_mixed_form_fixture( + path: Path, + *, + duplicate_body_anchor: bool = False, + split_body_anchor: bool = False, + duplicate_field_name: bool = False, + duplicate_table_anchor: bool = False, +) -> None: + """Build a one-page Korean form exercising all four public locators.""" + + with HwpxDocument.new() as document: + title = document.sections[0].paragraphs[0] + title.element.set("id", "1001") + title.text = "2026학년도 동아리 활동 신청서" + + heading = document.add_paragraph("1. 신청 정보") + heading.element.set("id", "1002") + table = heading.add_table(3, 2) + table.element.set("id", "2001") + table.rows[0].cells[0].text = "구분" + table.rows[0].cells[1].text = "미입력" + table.rows[1].cells[0].text = "동아리명" + table.rows[2].cells[0].text = "활동 장소" + table.rows[2].cells[1].text = "미정" + # The label at logical (1, 0) points right into a cell physically + # anchored at (0, 1), proving merge-aware physical normalization. + table.merge_cells(0, 1, 1, 1) + + if duplicate_table_anchor: + duplicate_heading = document.add_paragraph("2. 신청 정보") + duplicate_heading.element.set("id", "1006") + duplicate = duplicate_heading.add_table(2, 2) + duplicate.element.set("id", "2002") + duplicate.rows[0].cells[0].text = "구분" + duplicate.rows[0].cells[1].text = "내용" + duplicate.rows[1].cells[0].text = "동아리명" + duplicate.rows[1].cells[1].text = "미입력" + + if split_body_anchor: + purpose = document.add_paragraph("신청 목적: {{신청") + purpose.add_run("목적}}") + else: + purpose = document.add_paragraph("신청 목적: {{신청목적}}") + purpose.element.set("id", "1003") + if duplicate_body_anchor: + duplicate_purpose = document.add_paragraph("추가 목적: {{신청목적}}") + duplicate_purpose.element.set("id", "1007") + + teacher = document.add_paragraph("담당 교사: 미정") + teacher.element.set("id", "1004") + + representative = document.add_paragraph("대표 학생: ") + representative.element.set("id", "1005") + _append_native_field( + representative, + field_id="3001", + name="대표학생", + value="미입력", + ) + if duplicate_field_name: + second_representative = document.add_paragraph("부대표 학생: ") + second_representative.element.set("id", "1008") + _append_native_field( + second_representative, + field_id="3002", + name="대표학생", + value="미입력", + ) + document.save_to_path(path) + + +def _canonical_teacher_path(source: Path) -> str: + with HwpxAgentDocument.open(source) as agent: + return next( + record.path + for record in agent.records + if record.kind == "paragraph" and record.attributes.get("id") == "1004" + ) + + +def _public_plan( + source: Path, + output: Path, + *, + operations: list[dict[str, Any]] | None = None, + dry_run: bool = False, + idempotency_key: str | None = "mixed-form-test", +) -> dict[str, Any]: + if operations is None: + operations = [ + { + "operationId": "native", + "target": {"kind": "nativeField", "name": "대표학생"}, + "value": "홍길동", + }, + { + "operationId": "label", + "target": { + "kind": "labelCell", + "sectionPath": "/section[1]", + "tableAnchor": "신청 정보", + "cellAnchor": {"label": "동아리명", "direction": "right"}, + }, + "value": "인공지능 연구회", + }, + { + "operationId": "path", + "target": { + "kind": "canonicalPath", + "path": _canonical_teacher_path(source), + }, + "value": "담당 교사: 김교사", + }, + { + "operationId": "body", + "target": { + "kind": "bodyAnchor", + "sectionPath": "/section[1]", + "anchor": "{{신청목적}}", + "expectedCount": 1, + }, + "value": "교내 문제 해결", + }, + ] + return { + "schemaVersion": MIXED_FORM_PLAN_SCHEMA, + "source": str(source), + "output": str(output), + "expectedRevision": None, + "idempotencyKey": idempotency_key, + "dryRun": dry_run, + "overwrite": True, + "quality": "transparent", + "verificationRequirements": [ + "package", + "reopen", + "openSafety", + "semanticDiff", + "bytePreservation", + ], + "operations": operations, + } + + +def _single_operation_plan( + source: Path, + output: Path, + target: dict[str, Any], + *, + value: str = "채움값", +) -> dict[str, Any]: + return _public_plan( + source, + output, + operations=[{"operationId": "only", "target": target, "value": value}], + idempotency_key=None, + ) + + +def _member_payloads(path: Path) -> dict[str, bytes]: + with zipfile.ZipFile(io.BytesIO(path.read_bytes())) as archive: + return { + info.filename: archive.read(info.filename) + for info in archive.infolist() + if not info.is_dir() + } + + +def _rehash_compiled_plan(payload: dict[str, Any]) -> None: + material = copy.deepcopy(payload) + material["planHash"] = None + encoded = json.dumps( + material, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + payload["planHash"] = "sha256:" + hashlib.sha256(encoded).hexdigest() + + +def test_three_schema_roles_are_strict_shared_and_exam_free() -> None: + schemas = mixed_form_json_schemas() + assert set(schemas) == {"plan", "internalRequest", "compiledPlan"} + assert schemas["plan"]["properties"]["schemaVersion"] == { + "const": MIXED_FORM_PLAN_SCHEMA + } + assert schemas["internalRequest"]["properties"]["schemaVersion"] == { + "const": MIXED_FORM_REQUEST_SCHEMA + } + assert schemas["compiledPlan"]["properties"]["schemaVersion"] == { + "const": MIXED_FORM_COMPILED_PLAN_SCHEMA + } + + shared_batch = agent_json_schemas()["batch"] + embedded_batch = schemas["compiledPlan"]["properties"]["batch"] + assert embedded_batch["additionalProperties"] is False + assert embedded_batch["properties"]["commands"]["items"] == shared_batch["properties"][ + "commands" + ]["items"] + assert schemas["plan"]["properties"]["quality"] == shared_batch["properties"][ + "quality" + ] + assert schemas["internalRequest"]["properties"]["quality"] == shared_batch[ + "properties" + ]["quality"] + assert set(MIXED_FORM_LOCATOR_KINDS) == { + "nativeField", + "labelCell", + "canonicalPath", + "bodyAnchor", + } + assert "exam" not in json.dumps(schemas, ensure_ascii=False).casefold() + + +def test_public_plan_rejects_missing_unknown_and_non_exact_targets(tmp_path: Path) -> None: + source = tmp_path / "input.hwpx" + _write_mixed_form_fixture(source) + request = _public_plan(source, tmp_path / "output.hwpx") + + unknown = copy.deepcopy(request) + unknown["unexpected"] = True + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill(unknown) + assert exc.value.code == "invalid_syntax" + + missing = copy.deepcopy(request) + del missing["operations"][0]["value"] + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill(missing) + assert exc.value.code == "invalid_syntax" + + extra_target = copy.deepcopy(request) + extra_target["operations"][0]["target"]["index"] = 0 + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill(extra_target) + assert exc.value.code == "invalid_syntax" + + wrong_expected_count = copy.deepcopy(request) + wrong_expected_count["operations"][3]["target"]["expectedCount"] = 2 + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill(wrong_expected_count) + assert exc.value.code == "invalid_syntax" + + bad_quality = copy.deepcopy(request) + bad_quality["quality"] = {"renderCheck": "sometimes"} + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill(bad_quality) + assert exc.value.code == "invalid_syntax" + + +def test_plan_resolves_all_four_targets_to_one_revision_bound_batch(tmp_path: Path) -> None: + source = tmp_path / "mixed-anchor-input.hwpx" + output = tmp_path / "mixed-anchor-output.hwpx" + _write_mixed_form_fixture(source) + public_plan = _public_plan(source, output) + + internal = validate_mixed_form_request(public_plan) + assert internal["schemaVersion"] == MIXED_FORM_REQUEST_SCHEMA + assert all("locator" in operation for operation in internal["operations"]) + compiled = plan_mixed_form_fill(public_plan) + payload = compiled.to_dict() + + assert payload["schemaVersion"] == MIXED_FORM_COMPILED_PLAN_SCHEMA + assert payload["inputRevision"] == "sha256:" + hashlib.sha256(source.read_bytes()).hexdigest() + assert payload["requestHash"].startswith("sha256:") + assert payload["planHash"].startswith("sha256:") + assert payload["batch"]["schemaVersion"] == AGENT_BATCH_SCHEMA + assert payload["batch"]["expectedRevision"] == payload["inputRevision"] + assert len(payload["batch"]["commands"]) == 4 + assert all("schemaVersion" not in command for command in payload["batch"]["commands"]) + assert [item["locatorKind"] for item in payload["resolutions"]] == [ + "nativeField", + "labelCell", + "canonicalPath", + "bodyAnchor", + ] + label = payload["resolutions"][1] + assert (label["logicalRow"], label["logicalColumn"]) == (1, 1) + assert (label["physicalRow"], label["physicalColumn"]) == (0, 1) + assert payload["resolutions"][3]["nodeKind"] == "run" + assert validate_mixed_form_plan(payload).to_dict() == payload + + +def test_apply_uses_one_executor_preserves_members_and_reopens(tmp_path: Path) -> None: + source = tmp_path / "mixed-anchor-input.hwpx" + output = tmp_path / "mixed-anchor-output.hwpx" + _write_mixed_form_fixture(source) + + class CountingPipeline(SavePipeline): + def __init__(self) -> None: + super().__init__(oracle=NullOracle()) + self.calls = 0 + + def run(self, *args: Any, **kwargs: Any): + self.calls += 1 + return super().run(*args, **kwargs) + + pipeline = CountingPipeline() + result = apply_mixed_form_fill(_public_plan(source, output), save_pipeline=pipeline) + assert result.ok, result.to_dict() + assert result.rolled_back is False + assert pipeline.calls == 1 + assert output.exists() + assert validate_editor_open_safety(output).ok + byte_report = result.verification_report["bytePreservation"] + assert byte_report["ok"] is True + assert byte_report["changedMembers"] == ["Contents/section0.xml"] + assert byte_report["unchangedMemberCount"] > 0 + + before = _member_payloads(source) + after = _member_payloads(output) + assert set(before) == set(after) + assert all( + before[name] == after[name] + for name in before + if name != "Contents/section0.xml" + ) + + with HwpxAgentDocument.open(output) as agent: + field = next(record for record in agent.records if record.kind == "form-field") + teacher = next( + record + for record in agent.records + if record.kind == "paragraph" and record.attributes.get("id") == "1004" + ) + purpose = next( + record + for record in agent.records + if record.kind == "paragraph" and record.attributes.get("id") == "1003" + ) + merged_cell = next( + record + for record in agent.records + if record.kind == "cell" + and record.summary.get("row") == 1 + and record.summary.get("column") == 2 + ) + assert field.summary["value"] == "홍길동" + assert merged_cell.summary["text"] == "인공지능 연구회" + assert merged_cell.summary["rowSpan"] == 2 + assert teacher.summary["text"] == "담당 교사: 김교사" + assert purpose.summary["text"] == "신청 목적: 교내 문제 해결" + + +def test_dry_run_rollback_and_idempotent_replay(tmp_path: Path) -> None: + source = tmp_path / "mixed-anchor-input.hwpx" + _write_mixed_form_fixture(source) + + dry_output = tmp_path / "dry-output.hwpx" + dry_plan = plan_mixed_form_fill(_public_plan(source, dry_output, dry_run=True)) + dry_result = apply_mixed_form_plan(dry_plan) + assert dry_result.ok and dry_result.dry_run + assert not dry_output.exists() + + rollback_output = tmp_path / "rollback-output.hwpx" + rollback_output.write_bytes(b"existing destination") + rollback_plan = plan_mixed_form_fill(_public_plan(source, rollback_output)) + + def fail_third(stage: str, index: int | None) -> None: + if stage == "before_command" and index == 2: + raise RuntimeError("injected mixed-form failure") + + failed = apply_mixed_form_plan(rollback_plan, fault_injector=fail_third) + assert not failed.ok and failed.rolled_back + assert rollback_output.read_bytes() == b"existing destination" + + replay_output = tmp_path / "replay-output.hwpx" + replay_plan = plan_mixed_form_fill( + _public_plan(source, replay_output, idempotency_key="mixed-replay") + ) + store: dict[str, Any] = {} + first = apply_mixed_form_plan(replay_plan, idempotency_store=store) + first_bytes = replay_output.read_bytes() + replay = apply_mixed_form_plan(replay_plan, idempotency_store=store) + assert first.ok and replay.ok + assert replay.verification_report["idempotency"]["replayed"] is True + assert replay_output.read_bytes() == first_bytes + + +@pytest.mark.parametrize( + "target", + [ + {"kind": "nativeField", "name": "없는필드"}, + { + "kind": "labelCell", + "sectionPath": "/section[1]", + "tableAnchor": "없는표제", + "cellAnchor": {"label": "동아리명", "direction": "right"}, + }, + { + "kind": "bodyAnchor", + "sectionPath": "/section[1]", + "anchor": "{{없는본문}}", + "expectedCount": 1, + }, + {"kind": "canonicalPath", "path": "/section[1]/paragraph[999]"}, + ], +) +def test_zero_candidate_locators_fail_closed(tmp_path: Path, target: dict[str, Any]) -> None: + source = tmp_path / "input.hwpx" + _write_mixed_form_fixture(source) + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill(_single_operation_plan(source, tmp_path / "out.hwpx", target)) + assert exc.value.code == "not_found" + + +def test_multiple_candidate_locators_fail_closed(tmp_path: Path) -> None: + duplicate_field = tmp_path / "duplicate-field.hwpx" + _write_mixed_form_fixture(duplicate_field, duplicate_field_name=True) + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill( + _single_operation_plan( + duplicate_field, + tmp_path / "field-out.hwpx", + {"kind": "nativeField", "name": "대표학생"}, + ) + ) + assert exc.value.code == "ambiguous_target" + + duplicate_body = tmp_path / "duplicate-body.hwpx" + _write_mixed_form_fixture(duplicate_body, duplicate_body_anchor=True) + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill( + _single_operation_plan( + duplicate_body, + tmp_path / "body-out.hwpx", + { + "kind": "bodyAnchor", + "sectionPath": "/section[1]", + "anchor": "{{신청목적}}", + "expectedCount": 1, + }, + ) + ) + assert exc.value.code == "ambiguous_target" + + duplicate_table = tmp_path / "duplicate-table.hwpx" + _write_mixed_form_fixture(duplicate_table, duplicate_table_anchor=True) + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill( + _single_operation_plan( + duplicate_table, + tmp_path / "table-out.hwpx", + { + "kind": "labelCell", + "sectionPath": "/section[1]", + "tableAnchor": "신청 정보", + "cellAnchor": {"label": "동아리명", "direction": "right"}, + }, + ) + ) + assert exc.value.code == "ambiguous_target" + + +def test_body_anchor_crossing_runs_is_refused(tmp_path: Path) -> None: + source = tmp_path / "split-body-anchor.hwpx" + _write_mixed_form_fixture(source, split_body_anchor=True) + request = _single_operation_plan( + source, + tmp_path / "out.hwpx", + { + "kind": "bodyAnchor", + "sectionPath": "/section[1]", + "anchor": "{{신청목적}}", + "expectedCount": 1, + }, + ) + with pytest.raises(AgentContractError) as exc: + plan_mixed_form_fill(request) + assert exc.value.code == "unsupported_content" + assert "run boundaries" in str(exc.value) + + +def test_compiled_plan_hash_rejects_tampering(tmp_path: Path) -> None: + source = tmp_path / "input.hwpx" + _write_mixed_form_fixture(source) + compiled = plan_mixed_form_fill(_public_plan(source, tmp_path / "out.hwpx")).to_dict() + compiled["batch"]["commands"][0]["properties"]["value"] = "변조" + with pytest.raises(AgentContractError) as exc: + validate_mixed_form_plan(compiled) + assert exc.value.code == "verification_failed" + + extra = plan_mixed_form_fill(_public_plan(source, tmp_path / "out2.hwpx")).to_dict() + extra["unexpected"] = True + with pytest.raises(AgentContractError) as exc: + validate_mixed_form_plan(extra) + assert exc.value.code == "invalid_syntax" + + +def test_evalplan_and_exam_remain_separate_from_mixed_contract() -> None: + from hwpx.evalplan_fill import fill_evalplan + + tree = ast.parse(Path(mixed_form_module.__file__).read_text(encoding="utf-8")) + imported_modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imported_modules.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + imported_modules.add(node.module) + assert not any(name == "hwpx.exam" or name.startswith("hwpx.exam.") for name in imported_modules) + assert "exam" not in MIXED_FORM_LOCATOR_KINDS + assert "evalplan" not in MIXED_FORM_LOCATOR_KINDS + assert fill_evalplan.__module__ == "hwpx.evalplan_fill" + + +@pytest.mark.parametrize("alias_kind", ["exact", "lexical", "symlink", "hardlink"]) +def test_planning_rejects_every_source_output_alias( + tmp_path: Path, + alias_kind: str, +) -> None: + source = tmp_path / "source.hwpx" + _write_mixed_form_fixture(source) + source_before = source.read_bytes() + + if alias_kind == "exact": + output = source + elif alias_kind == "lexical": + output = tmp_path / "not-created" / ".." / source.name + else: + output = tmp_path / f"{alias_kind}.hwpx" + if alias_kind == "symlink": + output.symlink_to(source) + else: + output.hardlink_to(source) + + with pytest.raises(AgentContractError) as caught: + plan_mixed_form_fill(_public_plan(source, output)) + assert caught.value.code == "invariant_violation" + assert source.read_bytes() == source_before + + +@pytest.mark.parametrize("alias_kind", ["symlink", "hardlink"]) +def test_apply_rechecks_source_output_identity_after_planning( + tmp_path: Path, + alias_kind: str, +) -> None: + source = tmp_path / "source.hwpx" + output = tmp_path / "output.hwpx" + _write_mixed_form_fixture(source) + plan = plan_mixed_form_fill(_public_plan(source, output)) + source_before = source.read_bytes() + + if alias_kind == "symlink": + output.symlink_to(source) + else: + output.hardlink_to(source) + + with pytest.raises(AgentContractError) as caught: + apply_mixed_form_plan(plan) + assert caught.value.code == "invariant_violation" + assert source.read_bytes() == source_before + + +def test_mixed_form_cell_label_requires_normalized_exact_equality(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + _write_mixed_form_fixture(source) + target = { + "kind": "labelCell", + "sectionPath": "/section[1]", + "tableAnchor": "신청 정보", + "cellAnchor": {"label": "동아리", "direction": "right"}, + } + + with pytest.raises(AgentContractError) as caught: + plan_mixed_form_fill(_single_operation_plan(source, tmp_path / "out.hwpx", target)) + assert caught.value.code == "not_found" + + +def test_public_request_hash_participates_in_idempotency_identity(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + output = tmp_path / "output.hwpx" + _write_mixed_form_fixture(source) + + by_name_request = _public_plan(source, output, idempotency_key="locator-identity") + by_id_request = copy.deepcopy(by_name_request) + by_id_request["operations"][0]["target"] = { + "kind": "nativeField", + "fieldId": "3001", + } + by_name = plan_mixed_form_fill(by_name_request) + by_id = plan_mixed_form_fill(by_id_request) + assert by_name.batch == by_id.batch + assert by_name.request_hash != by_id.request_hash + + store: dict[str, Any] = {} + first = apply_mixed_form_plan(by_name, idempotency_store=store) + first_output = output.read_bytes() + first_identity = store["locator-identity"]["requestHash"] + conflict = apply_mixed_form_plan(by_id, idempotency_store=store) + + assert first.ok + assert not conflict.ok and conflict.error is not None + assert conflict.error.code == "idempotency_conflict" + assert conflict.verification_report["idempotency"]["replayed"] is False + assert output.read_bytes() == first_output + stored = store["locator-identity"] + assert stored["identityScope"] == "hwpx.mixed-form-idempotency/v1" + assert stored["mixedFormRequestHash"] == by_name.request_hash + assert stored["batchRequestHash"] != stored["requestHash"] + assert stored["requestHash"] == first_identity + + +@pytest.mark.parametrize( + ("operation_id", "mutation"), + [ + ( + "path", + lambda resolution: resolution.update({"locatorKind": "nativeField"}), + ), + ( + "body", + lambda resolution: resolution.update({"section": None}), + ), + ( + "label", + lambda resolution: resolution.update({"physicalColumn": None}), + ), + ( + "path", + lambda resolution: resolution.update({"nodeKind": "run"}), + ), + ], + ids=[ + "native-field-must-be-form-field", + "body-anchor-must-have-section", + "label-cell-must-have-coordinates", + "canonical-path-kind-must-match-node", + ], +) +def test_compiled_plan_enforces_locator_node_path_invariants( + tmp_path: Path, + operation_id: str, + mutation: Any, +) -> None: + source = tmp_path / "source.hwpx" + _write_mixed_form_fixture(source) + payload = plan_mixed_form_fill(_public_plan(source, tmp_path / "out.hwpx")).to_dict() + resolution = next( + item for item in payload["resolutions"] if item["operationId"] == operation_id + ) + mutation(resolution) + _rehash_compiled_plan(payload) + + with pytest.raises(AgentContractError) as caught: + validate_mixed_form_plan(payload) + assert caught.value.code == "invariant_violation" diff --git a/tests/test_mixed_form_plan.py b/tests/test_mixed_form_plan.py new file mode 100644 index 0000000..8084015 --- /dev/null +++ b/tests/test_mixed_form_plan.py @@ -0,0 +1,426 @@ +from __future__ import annotations + +import ast +import hashlib +import zipfile +from copy import deepcopy +from pathlib import Path +from typing import Any + +import pytest + +from hwpx import HwpxDocument, validate_editor_open_safety +from hwpx.agent import ( + AGENT_BATCH_SCHEMA, + MIXED_FORM_COMPILED_PLAN_SCHEMA, + MIXED_FORM_PLAN_SCHEMA, + AgentContractError, + HwpxAgentDocument, + apply_mixed_form_fill, + apply_mixed_form_plan, + mixed_form_json_schemas, + plan_mixed_form_fill, + validate_mixed_form_plan, + validate_mixed_form_request, +) +from hwpx.oxml.namespaces import HP + + +def _append(parent: Any, tag: str, attrs: dict[str, str] | None = None) -> Any: + child = parent.makeelement(tag, attrs or {}) + parent.append(child) + return child + + +def _add_native_project_field(document: HwpxDocument) -> None: + paragraph = document.add_paragraph("사업명: ") + paragraph.element.set("id", "240020") + begin_run = _append(paragraph.element, f"{HP}run", {"charPrIDRef": "0"}) + control = _append(begin_run, f"{HP}ctrl", {"type": "FORM", "id": "ctrl-240021"}) + field_begin = _append( + control, + f"{HP}fieldBegin", + { + "id": "240021", + "fieldid": "240021", + "type": "ClickHere", + "name": "사업명", + "prompt": "사업명", + "editable": "true", + }, + ) + parameters = _append(field_begin, f"{HP}parameters", {"count": "2"}) + _append(parameters, f"{HP}stringParam", {"name": "FieldName"}).text = "사업명" + _append(parameters, f"{HP}stringParam", {"name": "Instruction"}).text = "사업명" + value_run = _append(paragraph.element, f"{HP}run", {"charPrIDRef": "0"}) + _append(value_run, f"{HP}t").text = "여기를 누르세요" + end_run = _append(paragraph.element, f"{HP}run", {"charPrIDRef": "0"}) + end_control = _append(end_run, f"{HP}ctrl") + _append( + end_control, + f"{HP}fieldEnd", + {"beginIDRef": "240021", "fieldid": "240021"}, + ) + paragraph.section.mark_dirty() + + +def build_s079_korean_mixed_form_fixture( + path: Path, + *, + split_body_anchor: bool = False, + duplicate_body_anchor: bool = False, + duplicate_native_field: bool = False, + duplicate_label: bool = False, + include_merged_probe: bool = False, +) -> None: + """Build the synthetic, one-page Korean form frozen by Feature 024 P1.""" + + with HwpxDocument.new() as document: + title = document.sections[0].paragraphs[0] + title.element.set("id", "240010") + title.text = "S-079 혼합 양식 기준 문서" + _add_native_project_field(document) + if duplicate_native_field: + _add_native_project_field(document) + + table_paragraph = document.add_paragraph("담당 부서") + table_paragraph.element.set("id", "240030") + table = table_paragraph.add_table(2, 2) + table.element.set("id", "240031") + table.rows[0].cells[0].text = "담당 부서" if duplicate_label else "사업명" + table.rows[0].cells[1].text = "" + table.rows[1].cells[0].text = "담당 부서" + table.rows[1].cells[1].text = "" + + body = document.add_paragraph("", include_run=False) + body.element.set("id", "240040") + if split_body_anchor: + body.add_run("담당자: {{담") + body.add_run("당자}}") + else: + body.add_run("담당자: {{담당자}}") + if duplicate_body_anchor: + document.add_paragraph("비상 담당자: {{담당자}}") + + purpose = document.add_paragraph("행사 목적: 여기를 입력하세요") + purpose.element.set("id", "240050") + + if include_merged_probe: + merged_paragraph = document.add_paragraph("병합 대상") + merged_table = merged_paragraph.add_table(2, 2) + merged_table.rows[0].cells[0].text = "구분" + merged_table.rows[0].cells[1].text = "병합값" + merged_table.rows[1].cells[0].text = "" + merged_table.rows[1].cells[1].text = "" + merged_table.merge_cells("A2:B2") + + document.save_to_path(path) + + +def _revision(path: Path) -> str: + return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() + + +def _plan_request( + source: Path, + output: Path, + *, + dry_run: bool = False, + idempotency_key: str | None = "s079-mixed-anchor-reference-v1", +) -> dict[str, Any]: + return { + "schemaVersion": MIXED_FORM_PLAN_SCHEMA, + "source": str(source), + "output": str(output), + "expectedRevision": _revision(source), + "idempotencyKey": idempotency_key, + "dryRun": dry_run, + "overwrite": True, + "quality": "transparent", + "verificationRequirements": [ + "package", + "reopen", + "bytePreservation", + "openSafety", + ], + "operations": [ + { + "operationId": "native-project-name", + "target": {"kind": "nativeField", "fieldId": "240021"}, + "value": "AI 수업 나눔의 날", + }, + { + "operationId": "label-department", + "target": { + "kind": "labelCell", + "sectionPath": "/section[1]", + "tableAnchor": "담당 부서", + "cellAnchor": {"label": "담당 부서", "direction": "right"}, + }, + "value": "교육연구부", + }, + { + "operationId": "canonical-purpose", + "target": { + "kind": "canonicalPath", + "path": '/section[1]/paragraph[@id="240050"]', + }, + "value": "행사 목적: 교내 AI 활용 사례 공유", + }, + { + "operationId": "body-owner", + "target": { + "kind": "bodyAnchor", + "sectionPath": "/section[1]", + "anchor": "{{담당자}}", + "expectedCount": 1, + }, + "value": "김서현", + }, + ], + } + + +def _operation_only(request: dict[str, Any], operation_id: str) -> dict[str, Any]: + selected = deepcopy(request) + selected["operations"] = [ + operation + for operation in selected["operations"] + if operation["operationId"] == operation_id + ] + return selected + + +def test_p1_frozen_plan_compiles_four_locators_before_mutation(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + output = tmp_path / "output.hwpx" + build_s079_korean_mixed_form_fixture(source) + source_before = source.read_bytes() + + plan = plan_mixed_form_fill(_plan_request(source, output, dry_run=True)) + + assert source.read_bytes() == source_before + assert not output.exists() + assert plan.to_dict()["schemaVersion"] == MIXED_FORM_COMPILED_PLAN_SCHEMA + assert plan.input_revision == _revision(source) + assert [item.locator_kind for item in plan.resolutions] == [ + "nativeField", + "labelCell", + "canonicalPath", + "bodyAnchor", + ] + assert plan.batch["schemaVersion"] == AGENT_BATCH_SCHEMA + assert plan.batch["expectedRevision"] == plan.input_revision + assert len({item.path for item in plan.resolutions}) == 4 + assert [command["op"] for command in plan.batch["commands"]] == ["set"] * 4 + assert validate_mixed_form_plan(plan).plan_hash == plan.plan_hash + + +def test_success_preserves_untouched_members_and_reopens(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + output = tmp_path / "output.hwpx" + build_s079_korean_mixed_form_fixture(source) + + result = apply_mixed_form_fill(_plan_request(source, output)) + + assert result.ok, result.to_dict() + assert result.rolled_back is False + assert result.verification_report["openSafety"]["ok"] is True + assert result.verification_report["bytePreservation"]["ok"] is True + assert validate_editor_open_safety(output).ok + with HwpxDocument.open(output) as document: + assert document.list_form_fields()[0]["current_value"] == "AI 수업 나눔의 날" + matches = document.find_cell_by_label("담당 부서", direction="right") + match = matches["matches"][0] + assert match["target_cell"]["text"] == "교육연구부" + with HwpxAgentDocument.open(output) as agent: + purpose = agent.resolve_record('/section[1]/paragraph[@id="240050"]') + owner = agent.resolve_record('/section[1]/paragraph[@id="240040"]') + assert purpose.summary["text"] == "행사 목적: 교내 AI 활용 사례 공유" + assert owner.summary["text"] == "담당자: 김서현" + + changed = set(result.verification_report["bytePreservation"]["changedMembers"]) + with zipfile.ZipFile(source) as before, zipfile.ZipFile(output) as after: + common = set(before.namelist()) & set(after.namelist()) + assert "Contents/section0.xml" in changed + assert all(before.read(name) == after.read(name) for name in common - changed) + + +def test_dry_run_and_injected_failure_publish_nothing(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + dry_output = tmp_path / "dry.hwpx" + failed_output = tmp_path / "failed.hwpx" + build_s079_korean_mixed_form_fixture(source) + + dry = apply_mixed_form_fill(_plan_request(source, dry_output, dry_run=True)) + assert dry.ok and dry.dry_run + assert not dry_output.exists() + + prior = b"prior destination must survive" + failed_output.write_bytes(prior) + plan = plan_mixed_form_fill(_plan_request(source, failed_output)) + + def fail_after_second(stage: str, index: int | None) -> None: + if stage == "after_command" and index == 1: + raise RuntimeError("injected mixed-form failure") + + failed = apply_mixed_form_plan(plan, fault_injector=fail_after_second) + assert not failed.ok + assert failed.rolled_back + assert failed_output.read_bytes() == prior + assert source.exists() and validate_editor_open_safety(source).ok + + +def test_idempotent_replay_and_same_key_conflict(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + output = tmp_path / "output.hwpx" + build_s079_korean_mixed_form_fixture(source) + store: dict[str, Any] = {} + request = _plan_request(source, output) + plan = plan_mixed_form_fill(request) + + first = apply_mixed_form_plan(plan, idempotency_store=store) + replay = apply_mixed_form_plan(plan, idempotency_store=store) + assert first.ok and replay.ok + assert replay.document_revision == first.document_revision + assert replay.verification_report["idempotency"]["replayed"] is True + + conflicting_request = deepcopy(request) + conflicting_request["operations"][0]["value"] = "다른 사업명" + conflict_plan = plan_mixed_form_fill(conflicting_request) + conflict = apply_mixed_form_plan(conflict_plan, idempotency_store=store) + assert not conflict.ok and conflict.error is not None + assert conflict.error.code == "idempotency_conflict" + + +@pytest.mark.parametrize( + "mutation", + [ + lambda request: request.update({"unknown": True}), + lambda request: request["operations"][0]["target"].update({"raw": "forbidden"}), + lambda request: request["operations"][3]["target"].pop("expectedCount"), + lambda request: request["operations"][1]["target"].update( + {"sectionPath": "/section[@id=\"1\"]"} + ), + ], +) +def test_public_plan_is_strict_and_fail_closed(tmp_path: Path, mutation: Any) -> None: + source = tmp_path / "source.hwpx" + build_s079_korean_mixed_form_fixture(source) + request = _plan_request(source, tmp_path / "output.hwpx") + mutation(request) + + with pytest.raises(AgentContractError): + validate_mixed_form_request(request) + + schema = mixed_form_json_schemas()["plan"] + assert schema["additionalProperties"] is False + assert schema["properties"]["operations"]["items"]["additionalProperties"] is False + + +@pytest.mark.parametrize( + ("fixture_options", "expected_code"), + [ + ({"split_body_anchor": True}, "unsupported_content"), + ({"duplicate_body_anchor": True}, "ambiguous_target"), + ], +) +def test_body_anchor_zero_multiple_and_run_spanning_fail_closed( + tmp_path: Path, + fixture_options: dict[str, bool], + expected_code: str, +) -> None: + source = tmp_path / "source.hwpx" + build_s079_korean_mixed_form_fixture(source, **fixture_options) + request = _operation_only( + _plan_request(source, tmp_path / "output.hwpx"), + "body-owner", + ) + + with pytest.raises(AgentContractError) as caught: + plan_mixed_form_fill(request) + assert caught.value.code == expected_code + + missing = deepcopy(request) + missing["operations"][0]["target"]["anchor"] = "{{없는값}}" + with pytest.raises(AgentContractError) as absent: + plan_mixed_form_fill(missing) + assert absent.value.code == "not_found" + + +@pytest.mark.parametrize( + ("fixture_options", "operation_id", "expected_code"), + [ + ({"duplicate_native_field": True}, "native-project-name", "ambiguous_target"), + ({"duplicate_label": True}, "label-department", "ambiguous_target"), + ], +) +def test_native_and_label_locators_reject_multiple_matches( + tmp_path: Path, + fixture_options: dict[str, bool], + operation_id: str, + expected_code: str, +) -> None: + source = tmp_path / "source.hwpx" + build_s079_korean_mixed_form_fixture(source, **fixture_options) + request = _operation_only( + _plan_request(source, tmp_path / "output.hwpx"), + operation_id, + ) + + with pytest.raises(AgentContractError) as caught: + plan_mixed_form_fill(request) + assert caught.value.code == expected_code + + +def test_revision_binding_rejects_stale_source(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + build_s079_korean_mixed_form_fixture(source) + request = _plan_request(source, tmp_path / "output.hwpx") + request["expectedRevision"] = "sha256:" + "0" * 64 + + with pytest.raises(AgentContractError) as caught: + plan_mixed_form_fill(request) + assert caught.value.code == "stale_revision" + + +def test_label_cell_normalizes_merged_logical_coordinate(tmp_path: Path) -> None: + source = tmp_path / "source.hwpx" + build_s079_korean_mixed_form_fixture(source, include_merged_probe=True) + request = _operation_only( + _plan_request(source, tmp_path / "output.hwpx", dry_run=True), + "label-department", + ) + request["operations"][0]["target"] = { + "kind": "labelCell", + "sectionPath": "/section[1]", + "tableAnchor": "병합 대상", + "cellAnchor": {"label": "병합값", "direction": "below"}, + } + + plan = plan_mixed_form_fill(request) + resolution = plan.resolutions[0] + assert (resolution.logical_row, resolution.logical_column) == (1, 1) + assert (resolution.physical_row, resolution.physical_column) == (1, 0) + assert resolution.path.endswith("/row[2]/cell[1]") + + +def test_evalplan_remains_available_and_exam_is_not_imported() -> None: + import hwpx.agent.form_plan as form_plan + import hwpx.evalplan_fill as evalplan_fill + + tree = ast.parse(Path(form_plan.__file__).read_text(encoding="utf-8")) + imported = { + alias.name + for node in ast.walk(tree) + if isinstance(node, ast.Import) + for alias in node.names + } + imported.update( + node.module or "" + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) + ) + assert not any(name == "hwpx.exam" or name.startswith("hwpx.exam.") for name in imported) + assert callable(evalplan_fill.parse_review_md) + assert callable(evalplan_fill.fill_evalplan) From cbed9a3c5a0e42bc415c742ffc063c0937cc2990 Mon Sep 17 00:00:00 2001 From: airmang <38392618+airmang@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:55:08 +0900 Subject: [PATCH 2/2] docs: mark python-hwpx 3.1.0 released --- README.md | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index e552708..2aef887 100644 --- a/README.md +++ b/README.md @@ -25,10 +25,9 @@ `hwpx-mcp-server`와 `hwpx-plugin`은 같은 프로젝트가 직접 유지보수하는 first-party 연동 구성요소입니다. “first-party”는 프로젝트 유지보수 관계를 뜻하며, 한컴 또는 제3자의 공식 인증을 뜻하지 않습니다. -현재 검증 중인 릴리스 후보와 최소 호환 코어 버전은 모두 `python-hwpx 3.1.0`입니다. -이 버전 좌표는 별도 소유자 승인 전까지 공개 또는 배포를 뜻하지 않습니다. +현재 공개 릴리스와 최소 호환 코어 버전은 모두 `python-hwpx 3.1.0`입니다. 현재 패키지 분류는 `Development Status :: 3 - Alpha`입니다. 이 분류는 API와 제품의 -성숙도를 나타내며, 릴리스 후보 버전이나 플러그인의 최소 호환 버전을 대신하지 않습니다. +성숙도를 나타내며, 공개 버전이나 플러그인의 최소 호환 버전을 대신하지 않습니다. ---