diff --git a/.github/workflows/pr-ci.yml b/.github/workflows/pr-ci.yml index e72af0d92..6495e36c6 100644 --- a/.github/workflows/pr-ci.yml +++ b/.github/workflows/pr-ci.yml @@ -49,6 +49,9 @@ jobs: - name: Run lint run: make lint + - name: Check generated parse-result contracts + run: make check-contracts + typecheck: name: Typecheck runs-on: ubuntu-latest @@ -109,6 +112,9 @@ jobs: - name: Install test system dependencies run: uv run python apps/api/scripts/ensure_test_environment.py --install + - name: Run parse-result contract tests + run: uv run pytest packages/shared-python/shared/tests/test_parse_result_contract.py packages/shared-python/shared/tests/test_zip_result_contract.py packages/shared-python/shared/tests/test_zip_package_writer_contract.py -q + # api/worker both expose top-level `app`; run separately to avoid module shadowing. - name: Run API tests run: uv run pytest apps/api/tests -q diff --git a/Makefile b/Makefile index 073def0e8..a5de7173e 100644 --- a/Makefile +++ b/Makefile @@ -32,3 +32,10 @@ check: lint typecheck test-doc-agent: cd apps/worker && $(UV_RUN_ENV) $(UV) run pytest tests/document_agent + +.PHONY: sync-contracts check-contracts +sync-contracts: + $(UV_RUN) python scripts/sync_parse_result_contracts.py + +check-contracts: + $(UV_RUN) python scripts/sync_parse_result_contracts.py --check diff --git a/apps/worker/tests/contract/test_parse_task_contract.py b/apps/worker/tests/contract/test_parse_task_contract.py index da630eaac..0a964befa 100644 --- a/apps/worker/tests/contract/test_parse_task_contract.py +++ b/apps/worker/tests/contract/test_parse_task_contract.py @@ -134,6 +134,9 @@ def test_parse_task_should_process_uploaded_file_through_real_contract_boundarie ) manifest_payload = result_zip["manifest"] + assert manifest_payload["schema_version"] == chunks_payload["schema_version"] == 1 + with zipfile.ZipFile(result_zip["path"]) as archive: + assert json.loads(archive.read("doc_nav.json"))["schema_version"] == 1 assert manifest_payload["source_file_name"] == job["source_file_name"] assert manifest_payload["statistics"]["total_chunks"] == len(job_chunks) @@ -1278,3 +1281,29 @@ def _identity_eval_md_headings( "oversized.pdf/Chapter 1", "oversized.pdf/Chapter 2", ] + + +def test_parse_task_rejects_invalid_result_contract_before_upload( + worker_contract_environment: None, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from shared.services.storage.zip_result_schema import ZipResultSchemaBuilder + + contract = WorkerParseContract.create() + contract.use_workspace_root(monkeypatch, tmp_path) + contract.use_billing(monkeypatch, is_enabled=False) + job = contract.create_file_job(source_file_name="malformed.xlsx", job_id_prefix="job_bad_contract") + contract.upload_source_file(local_file_path=_SAMPLE_XLSX_PATH, s3_key=job["s3_key"]) + monkeypatch.setattr( + ZipResultSchemaBuilder, "format_chunks", + lambda *args, **kwargs: [{"chunk_id": "bad", "type": "unsupported"}], + ) + result = contract.enqueue_parse_task(job_id=job["job_id"], user_id=job["user_id"]) + assert result.failed() + assert result.result.details["reason"] == "PARSE_RESULT_CONTRACT_VIOLATION" + assert contract.get_task_status(job["job_id"]) == "failed" + assert contract.observe_job_status(job["job_id"])["status"] == "failed" + assert contract.count_job_results(job["job_id"]) == 0 + result_key = contract.storage.build_result_zip_key(job_id=job["job_id"]) + assert contract.verify_result_zip_object(result_key)["exists"] is False diff --git a/apps/worker/tests/support/worker_parse_contract.py b/apps/worker/tests/support/worker_parse_contract.py index 108409a42..c82db6993 100644 --- a/apps/worker/tests/support/worker_parse_contract.py +++ b/apps/worker/tests/support/worker_parse_contract.py @@ -142,6 +142,8 @@ def create_file_job( "summary_image": False, "summary_table": False, "summary_txt": False, + "summary_use_llm": False, + "top_summary_use_llm": False, }, }, billing_status=billing_status, diff --git a/docs/design/parsed-result-contract.md b/docs/design/parsed-result-contract.md new file mode 100644 index 000000000..00c7265c5 --- /dev/null +++ b/docs/design/parsed-result-contract.md @@ -0,0 +1,128 @@ +# Parsed result ZIP contract + +**Status:** Draft implementation; policy choices below remain unaccepted upstream. +**Related issue:** [#45](https://github.com/Ontos-AI/knowhere/issues/45) + +The worker emits one versioned result ZIP. Its three JSON roots contain additive +`schema_version: 1`: `chunks.json`, `doc_nav.json`, and `manifest.json`. The existing +`doc_nav.version` (`"1.0"`) and `manifest.version` (`"2.0"`) remain unchanged. +Pydantic v2 models in `packages/shared-python/shared/contracts/parse_result/` are +its source of truth. Generated Draft 2020-12 JSON Schema lives in the +language-neutral `packages/contracts/parse_result/` directory for SDK vendoring. +Schema `$id` values identify the documents; they do not imply a hosted registry. + +## Producer boundary + +`ZipResultService.generate_zip_package` validates the prepared JSON before +calling the writer. `ZipPackageWriter` validates the closed ZIP before returning +its checksum. This precedes `finalize_parse_success` uploading results and +calling `finalize_job_success`. A contract violation raises the permanent domain +exception `ParseResultContractException`; it follows the existing task failure +handling and is not a transient storage retry. Invalid ZIPs are removed and are +never returned to the upload path. + +New successful packages require all three roots. Navigation generation failure, +which previously logged a warning and omitted `doc_nav.json`, now fails the +package. The already enriched navigation is retained, including extension fields +such as `top_summary`; no on-disk corpus files are rewritten. +Missing, unreadable, or unrecognized navigation keeps the existing rebuild +fallback. A readable object claiming an unsupported `schema_version` fails before +that fallback, even when its section layout is not recognized. + +Multiple chunks may reference the same image, table, or page asset. The writer +stores one ZIP member per asset path. Repeated paths with different contents +fail with `conflicting_asset_path` rather than creating ambiguous ZIP entries. + +## Version and compatibility policy + +- Version 1 captures current text, image, table, and page chunk shapes. Table + `content` may contain an asset path rather than inline HTML. Text `tokens` may + contain an integer count or a string list. `order` is not required. +- Required scalar fields are strictly typed. Shared metadata requires `length`, + `summary`, and `page_nums`. Type-specific metadata, including page citation + assets and connections, is validated when present. Nested sections may omit + `level` in existing enriched navigation; their nesting defines depth. +- Unknown fields are accepted and retained, including optional processing and + parser metadata. Additive optional fields do not bump the integer version. + Removing, renaming, or changing required fields/types requires a new version. +- One ZIP has one supported version. Unknown or noninteger versions fail. The + legacy string `version` fields remain independent of `schema_version`. +- `validate_parse_result` and `validate_parse_result_archive` default to a + compatibility read: missing root versions are interpreted as legacy v1 and + returned as warnings. Legacy unversioned manifests may omit navigation; a + package declaring version 1 must include it. Producer calls explicitly set + `allow_legacy=False`. The compatibility reader has no silent expiry; removing + legacy support requires a separately reviewed change. + +```python +from shared.contracts.parse_result import validate_parse_result_archive + +result = validate_parse_result_archive("result.zip") +print(result.manifest.schema_version) +print(result.warnings) # legacy-version/navigation compatibility notices +``` + +The helpers make no network calls. Consumers can vendor the JSON Schema and use +any Draft 2020-12 validator for individual JSON shapes. Cross-artifact checks and +ZIP membership checks require equivalent consumer logic or the Python helper. +This PR does not change the standalone Node or Python SDK repositories. + +## Hard failures and warnings + +Hard failures include malformed required JSON fields, unsupported versions, +missing required artifacts, disagreement between actual chunk types/counts and +manifest/navigation statistics, and invalid connection character spans. Manifest +heading hierarchy is recursive string-keyed dictionaries with dictionary leaves. + +Archive validation reads members without extraction. It rejects unsafe member +names, duplicate members, malformed JSON (including duplicate keys and nonfinite +numbers), missing referenced `metadata.file_path` / page `artifact_ref` assets, +and unreadable archives. Asset references must be relative paths under `images/`, +`tables/`, or `page_citation_assets/`. Binary bytes, image dimensions, and model +output quality are outside this contract. Consumer reads default to a 128 MiB +limit for each JSON member; callers can choose `max_json_bytes`. The local writer +validates its generated archive without that size limit. + +Only legacy root-version omission and legacy navigation omission are warnings. +Errors expose the existing canonical `INTERNAL_ERROR` and a stable details object: + +```json +{ + "reason": "PARSE_RESULT_CONTRACT_VIOLATION", + "schema_version": 1, + "violations": [{ + "artifact": "chunks.json", + "field": "chunks.0.metadata.length", + "reason": "int_type" + }] +} +``` + +Error details contain artifact and field locations and machine-readable reasons, +never document values, Pydantic input/context, archive paths, or hierarchy titles. +The canonical exception logger records these structured fields without the +validation traceback, so diagnostic log sinks cannot expand document locals. + +## Updating schemas + +```bash +make sync-contracts +make check-contracts +``` + +The check runs in PR CI and fails when checked-in schema differs from generated +models. Fixtures cover all four chunk types, enriched navigation, table asset +paths, and page citation assets. Unit tests validate fixtures with both Pydantic +and JSON Schema and exercise actual ZIP writing without external accounts. +Worker task contract tests cover successful versioned packages and a malformed +producer payload failing before upload. + +## Draft choices for review + +The schema location, hard failure policy, required navigation for newly emitted +packages, integer versioning, and continued legacy compatibility are local design +choices implementing the unanswered scope questions in #45. They have not been +confirmed by maintainers. The implementation is additive for successful current +JSON output, with the intentional change that malformed output fails the job +instead of reaching SDK consumers. There are no database migrations, new runtime +dependencies, environment variables, or ZIP member renames. diff --git a/packages/contracts/parse_result/chunks.schema.json b/packages/contracts/parse_result/chunks.schema.json new file mode 100644 index 000000000..79404e587 --- /dev/null +++ b/packages/contracts/parse_result/chunks.schema.json @@ -0,0 +1,352 @@ +{ + "$defs": { + "ChunkMetadata": { + "additionalProperties": true, + "properties": { + "connect_to": { + "anyOf": [ + { + "items": { + "anyOf": [ + { + "type": "string" + }, + { + "$ref": "#/$defs/Connection" + } + ] + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Connect To" + }, + "file_path": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "File Path" + }, + "keywords": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Keywords" + }, + "length": { + "minimum": 0, + "title": "Length", + "type": "integer" + }, + "page_assets": { + "anyOf": [ + { + "items": { + "$ref": "#/$defs/PageAsset" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Page Assets" + }, + "page_nums": { + "items": { + "type": "integer" + }, + "title": "Page Nums", + "type": "array" + }, + "summary": { + "title": "Summary", + "type": "string" + }, + "tokens": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Tokens" + } + }, + "required": [ + "length", + "summary", + "page_nums" + ], + "title": "ChunkMetadata", + "type": "object" + }, + "ChunkRecord": { + "additionalProperties": true, + "properties": { + "chunk_id": { + "minLength": 1, + "title": "Chunk Id", + "type": "string" + }, + "content": { + "title": "Content", + "type": "string" + }, + "metadata": { + "$ref": "#/$defs/ChunkMetadata" + }, + "path": { + "title": "Path", + "type": "string" + }, + "type": { + "enum": [ + "text", + "image", + "table", + "page" + ], + "title": "Type", + "type": "string" + } + }, + "required": [ + "chunk_id", + "type", + "content", + "path", + "metadata" + ], + "title": "ChunkRecord", + "type": "object" + }, + "Connection": { + "additionalProperties": true, + "properties": { + "keywords": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Keywords" + }, + "position": { + "anyOf": [ + { + "$ref": "#/$defs/Position" + }, + { + "type": "null" + } + ], + "default": null + }, + "ref": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Ref" + }, + "relation": { + "title": "Relation", + "type": "string" + }, + "same_as_owner": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Same As Owner" + }, + "score": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Score" + }, + "target": { + "minLength": 1, + "title": "Target", + "type": "string" + } + }, + "required": [ + "target", + "relation" + ], + "title": "Connection", + "type": "object" + }, + "PageAsset": { + "additionalProperties": true, + "properties": { + "artifact_ref": { + "minLength": 1, + "title": "Artifact Ref", + "type": "string" + }, + "asset_url": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Asset Url" + }, + "content_type": { + "minLength": 1, + "title": "Content Type", + "type": "string" + }, + "height": { + "anyOf": [ + { + "exclusiveMinimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Height" + }, + "page_num": { + "exclusiveMinimum": 0, + "title": "Page Num", + "type": "integer" + }, + "source": { + "minLength": 1, + "title": "Source", + "type": "string" + }, + "width": { + "anyOf": [ + { + "exclusiveMinimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Width" + } + }, + "required": [ + "page_num", + "artifact_ref", + "content_type", + "source" + ], + "title": "PageAsset", + "type": "object" + }, + "Position": { + "additionalProperties": true, + "properties": { + "end": { + "minimum": 0, + "title": "End", + "type": "integer" + }, + "start": { + "minimum": 0, + "title": "Start", + "type": "integer" + } + }, + "required": [ + "start", + "end" + ], + "title": "Position", + "type": "object" + } + }, + "$id": "https://knowhereto.ai/contracts/parse_result/v1/chunks.schema.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": true, + "properties": { + "chunks": { + "items": { + "$ref": "#/$defs/ChunkRecord" + }, + "title": "Chunks", + "type": "array" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + } + }, + "required": [ + "schema_version", + "chunks" + ], + "title": "Chunks", + "type": "object" +} diff --git a/packages/contracts/parse_result/doc_nav.schema.json b/packages/contracts/parse_result/doc_nav.schema.json new file mode 100644 index 000000000..aad69048c --- /dev/null +++ b/packages/contracts/parse_result/doc_nav.schema.json @@ -0,0 +1,185 @@ +{ + "$defs": { + "NavigationStatistics": { + "additionalProperties": true, + "properties": { + "image_chunks": { + "minimum": 0, + "title": "Image Chunks", + "type": "integer" + }, + "max_depth": { + "minimum": 0, + "title": "Max Depth", + "type": "integer" + }, + "page_chunks": { + "default": 0, + "minimum": 0, + "title": "Page Chunks", + "type": "integer" + }, + "table_chunks": { + "minimum": 0, + "title": "Table Chunks", + "type": "integer" + }, + "text_chunks": { + "minimum": 0, + "title": "Text Chunks", + "type": "integer" + }, + "total_chunks": { + "minimum": 0, + "title": "Total Chunks", + "type": "integer" + } + }, + "required": [ + "total_chunks", + "text_chunks", + "image_chunks", + "table_chunks", + "max_depth" + ], + "title": "NavigationStatistics", + "type": "object" + }, + "Resource": { + "additionalProperties": true, + "properties": { + "path": { + "title": "Path", + "type": "string" + }, + "summary": { + "title": "Summary", + "type": "string" + } + }, + "required": [ + "path", + "summary" + ], + "title": "Resource", + "type": "object" + }, + "Resources": { + "additionalProperties": true, + "properties": { + "images": { + "items": { + "$ref": "#/$defs/Resource" + }, + "title": "Images", + "type": "array" + }, + "tables": { + "items": { + "$ref": "#/$defs/Resource" + }, + "title": "Tables", + "type": "array" + } + }, + "required": [ + "images", + "tables" + ], + "title": "Resources", + "type": "object" + }, + "Section": { + "additionalProperties": true, + "properties": { + "children": { + "items": { + "$ref": "#/$defs/Section" + }, + "title": "Children", + "type": "array" + }, + "chunk_count": { + "minimum": 0, + "title": "Chunk Count", + "type": "integer" + }, + "level": { + "anyOf": [ + { + "exclusiveMinimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Level" + }, + "path": { + "title": "Path", + "type": "string" + }, + "summary": { + "title": "Summary", + "type": "string" + }, + "title": { + "title": "Title", + "type": "string" + } + }, + "required": [ + "title", + "path", + "summary", + "chunk_count", + "children" + ], + "title": "Section", + "type": "object" + } + }, + "$id": "https://knowhereto.ai/contracts/parse_result/v1/doc_nav.schema.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": true, + "properties": { + "file_name": { + "title": "File Name", + "type": "string" + }, + "resources": { + "$ref": "#/$defs/Resources" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + }, + "sections": { + "items": { + "$ref": "#/$defs/Section" + }, + "title": "Sections", + "type": "array" + }, + "stats": { + "$ref": "#/$defs/NavigationStatistics" + }, + "version": { + "title": "Version", + "type": "string" + } + }, + "required": [ + "schema_version", + "version", + "file_name", + "stats", + "sections", + "resources" + ], + "title": "DocNav", + "type": "object" +} diff --git a/packages/contracts/parse_result/manifest.schema.json b/packages/contracts/parse_result/manifest.schema.json new file mode 100644 index 000000000..31d4ff83f --- /dev/null +++ b/packages/contracts/parse_result/manifest.schema.json @@ -0,0 +1,249 @@ +{ + "$defs": { + "Hierarchy": { + "additionalProperties": { + "$ref": "#/$defs/Hierarchy" + }, + "title": "Hierarchy", + "type": "object" + }, + "ManifestStatistics": { + "additionalProperties": true, + "properties": { + "image_chunks": { + "minimum": 0, + "title": "Image Chunks", + "type": "integer" + }, + "page_chunks": { + "default": 0, + "minimum": 0, + "title": "Page Chunks", + "type": "integer" + }, + "table_chunks": { + "minimum": 0, + "title": "Table Chunks", + "type": "integer" + }, + "text_chunks": { + "minimum": 0, + "title": "Text Chunks", + "type": "integer" + }, + "total_chunks": { + "minimum": 0, + "title": "Total Chunks", + "type": "integer" + }, + "total_pages": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Total Pages" + } + }, + "required": [ + "total_chunks", + "text_chunks", + "image_chunks", + "table_chunks" + ], + "title": "ManifestStatistics", + "type": "object" + }, + "Processing": { + "additionalProperties": true, + "properties": { + "billing_status": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Billing Status" + }, + "cost": { + "$ref": "#/$defs/ProcessingCost" + }, + "page_count": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Page Count" + }, + "stages": { + "additionalProperties": true, + "title": "Stages", + "type": "object" + }, + "timing": { + "$ref": "#/$defs/ProcessingTiming" + } + }, + "required": [ + "page_count", + "billing_status", + "cost", + "timing", + "stages" + ], + "title": "Processing", + "type": "object" + }, + "ProcessingCost": { + "additionalProperties": true, + "properties": { + "credits": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Credits" + }, + "micro_dollars": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Micro Dollars" + } + }, + "required": [ + "micro_dollars", + "credits" + ], + "title": "ProcessingCost", + "type": "object" + }, + "ProcessingTiming": { + "additionalProperties": true, + "properties": { + "completed_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Completed At" + }, + "duration_ms": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Duration Ms" + }, + "started_at": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Started At" + } + }, + "required": [ + "started_at", + "completed_at", + "duration_ms" + ], + "title": "ProcessingTiming", + "type": "object" + } + }, + "$id": "https://knowhereto.ai/contracts/parse_result/v1/manifest.schema.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": true, + "properties": { + "HIERARCHY": { + "$ref": "#/$defs/Hierarchy" + }, + "data_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data Id" + }, + "job_id": { + "minLength": 1, + "title": "Job Id", + "type": "string" + }, + "processing": { + "$ref": "#/$defs/Processing" + }, + "processing_date": { + "pattern": "^\\d{4}-\\d{2}-\\d{2}T.+Z$", + "title": "Processing Date", + "type": "string" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + }, + "source_file_name": { + "title": "Source File Name", + "type": "string" + }, + "statistics": { + "$ref": "#/$defs/ManifestStatistics" + }, + "version": { + "title": "Version", + "type": "string" + } + }, + "required": [ + "schema_version", + "version", + "job_id", + "data_id", + "source_file_name", + "processing_date", + "processing", + "statistics", + "HIERARCHY" + ], + "title": "Manifest", + "type": "object" +} diff --git a/packages/shared-python/shared/contracts/__init__.py b/packages/shared-python/shared/contracts/__init__.py new file mode 100644 index 000000000..97a4fdfc1 --- /dev/null +++ b/packages/shared-python/shared/contracts/__init__.py @@ -0,0 +1 @@ +"""Versioned public artifact contracts.""" diff --git a/packages/shared-python/shared/contracts/parse_result/__init__.py b/packages/shared-python/shared/contracts/parse_result/__init__.py new file mode 100644 index 000000000..ce82c1954 --- /dev/null +++ b/packages/shared-python/shared/contracts/parse_result/__init__.py @@ -0,0 +1,8 @@ +"""Public parse-result ZIP contract and compatibility reader.""" + +from .models import Chunks, ChunkRecord, DocNav, Manifest, SCHEMA_VERSION +from .validation import ( + ValidatedParseResult, + validate_parse_result, + validate_parse_result_archive, +) diff --git a/packages/shared-python/shared/contracts/parse_result/models.py b/packages/shared-python/shared/contracts/parse_result/models.py new file mode 100644 index 000000000..69e27d9d7 --- /dev/null +++ b/packages/shared-python/shared/contracts/parse_result/models.py @@ -0,0 +1,155 @@ +"""Parse-result ZIP v1. Unknown fields are additive; scalar coercion is disabled.""" + +from __future__ import annotations + +from typing import Annotated, Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator + +SCHEMA_VERSION = 1 +NonNegativeInt = Annotated[int, Field(ge=0)] +PositiveInt = Annotated[int, Field(gt=0)] +NonEmptyString = Annotated[str, Field(min_length=1)] + + +class ContractModel(BaseModel): + model_config = ConfigDict(extra="allow", strict=True, allow_inf_nan=False) + + +class VersionedArtifact(ContractModel): + schema_version: Literal[1] + + @field_validator("schema_version", mode="before") + @classmethod + def require_integer_version(cls, value: Any) -> Any: + if type(value) is not int: + raise ValueError("schema_version must be an integer") + return value + + +class Position(ContractModel): + start: NonNegativeInt + end: NonNegativeInt + + +class Connection(ContractModel): + target: NonEmptyString + relation: str + ref: str | None = None + position: Position | None = None + score: float | None = None + keywords: list[str] | None = None + same_as_owner: str | None = None + + +class PageAsset(ContractModel): + page_num: PositiveInt + artifact_ref: NonEmptyString + content_type: NonEmptyString + source: NonEmptyString + asset_url: str | None = None + width: PositiveInt | None = None + height: PositiveInt | None = None + + +class ChunkMetadata(ContractModel): + length: NonNegativeInt + summary: str + page_nums: list[int] + # Existing text writers support both token count and pretokenized terms. + tokens: NonNegativeInt | list[str] | None = None + keywords: list[str] | None = None + connect_to: list[str | Connection] | None = None + file_path: str | None = None + page_assets: list[PageAsset] | None = None + + +class ChunkRecord(ContractModel): + chunk_id: NonEmptyString + type: Literal["text", "image", "table", "page"] + content: str + path: str + metadata: ChunkMetadata + + +class Chunks(VersionedArtifact): + chunks: list[ChunkRecord] + + +class ChunkStatistics(ContractModel): + total_chunks: NonNegativeInt + text_chunks: NonNegativeInt + image_chunks: NonNegativeInt + table_chunks: NonNegativeInt + # Older enriched navigation files predate page chunks. + page_chunks: NonNegativeInt = 0 + + +class ManifestStatistics(ChunkStatistics): + total_pages: NonNegativeInt | None = None + + +class NavigationStatistics(ChunkStatistics): + max_depth: NonNegativeInt + + +class Section(ContractModel): + title: str + path: str + # Existing enriched navigation may omit level; nesting defines depth. + level: PositiveInt | None = None + summary: str + chunk_count: NonNegativeInt + children: list[Section] + + +class Resource(ContractModel): + path: str + summary: str + + +class Resources(ContractModel): + images: list[Resource] + tables: list[Resource] + + +class DocNav(VersionedArtifact): + version: str + file_name: str + stats: NavigationStatistics + sections: list[Section] + resources: Resources + + +class ProcessingCost(ContractModel): + micro_dollars: int | None + credits: float | None + + +class ProcessingTiming(ContractModel): + started_at: str | None + completed_at: str | None + duration_ms: int | None + + +class Processing(ContractModel): + page_count: int | None + billing_status: str | None + cost: ProcessingCost + timing: ProcessingTiming + stages: dict[str, Any] + + +class Hierarchy(RootModel[dict[str, "Hierarchy"]]): + model_config = ConfigDict(strict=True) + + +class Manifest(VersionedArtifact): + version: str + job_id: NonEmptyString + data_id: str | None + source_file_name: str + processing_date: Annotated[str, Field(pattern=r"^\d{4}-\d{2}-\d{2}T.+Z$")] + processing: Processing + statistics: ManifestStatistics + HIERARCHY: Hierarchy diff --git a/packages/shared-python/shared/contracts/parse_result/validation.py b/packages/shared-python/shared/contracts/parse_result/validation.py new file mode 100644 index 000000000..b5ad507b3 --- /dev/null +++ b/packages/shared-python/shared/contracts/parse_result/validation.py @@ -0,0 +1,267 @@ +"""Shared producer/consumer validation without rewriting or extracting artifacts.""" + +from __future__ import annotations + +import json +import math +import zipfile +import zlib +from dataclasses import dataclass +from pathlib import PurePosixPath +from typing import Any, NoReturn + +from pydantic import BaseModel, ValidationError + +from shared.core.exceptions.domain_exceptions import ParseResultContractException +from .models import Chunks, DocNav, Manifest, SCHEMA_VERSION + + +@dataclass(frozen=True) +class ValidatedParseResult: + manifest: Manifest + chunks: Chunks + doc_nav: DocNav | None + warnings: tuple[str, ...] + + +def _fail(artifact: str, field: str, reason: str) -> NoReturn: + raise ParseResultContractException( + violations=[{"artifact": artifact, "field": field, "reason": reason}], + schema_version=SCHEMA_VERSION, + ) from None + + +def _model( + model: type[BaseModel], + payload: Any, + artifact: str, + allow_legacy: bool, + warnings: list[str], +) -> Any: + if not isinstance(payload, dict): + _fail(artifact, "$", "object_required") + value = dict(payload) + if "schema_version" not in value and allow_legacy: + value["schema_version"] = SCHEMA_VERSION + warnings.append(f"{artifact}: missing schema_version treated as legacy v1") + version = value.get("schema_version") + if type(version) is not int or version != SCHEMA_VERSION: + _fail(artifact, "schema_version", "unsupported_schema_version") + try: + return model.model_validate(value) + except ValidationError as exc: + # Suppress the raw validation traceback so logs cannot expose input values. + # Never copy Pydantic input/ctx/message, which may contain document text. + raise ParseResultContractException( + violations=[ + { + "artifact": artifact, + "field": ( + "HIERARCHY" + if item["loc"] and item["loc"][0] == "HIERARCHY" + else ".".join(map(str, item["loc"])) or "$" + ), + "reason": item["type"], + } + for item in exc.errors(include_input=False, include_context=False) + ], + schema_version=SCHEMA_VERSION, + ) from None + + +def validate_parse_result( + manifest: Any, + chunks: Any, + doc_nav: Any, + *, + allow_legacy: bool = True, +) -> ValidatedParseResult: + """Validate v1 JSON roots. Compatibility reads accept old unversioned ZIPs. + + Producers pass allow_legacy=False, requiring explicit versions and doc_nav. + Unknown fields remain permitted for additive backwards-compatible evolution. + """ + warnings: list[str] = [] + parsed_manifest: Manifest = _model( + Manifest, manifest, "manifest.json", allow_legacy, warnings + ) + parsed_chunks: Chunks = _model( + Chunks, chunks, "chunks.json", allow_legacy, warnings + ) + parsed_nav: DocNav | None = None + if doc_nav is None: + if ( + allow_legacy + and isinstance(manifest, dict) + and "schema_version" not in manifest + and isinstance(chunks, dict) + and "schema_version" not in chunks + ): + warnings.append("doc_nav.json: absent in legacy v1 package") + else: + _fail("doc_nav.json", "$", "required_artifact_missing") + else: + parsed_nav = _model(DocNav, doc_nav, "doc_nav.json", allow_legacy, warnings) + + counts = {kind: 0 for kind in ("text", "image", "table", "page")} + for index, chunk in enumerate(parsed_chunks.chunks): + counts[chunk.type] += 1 + metadata = chunk.metadata + if metadata.file_path: + _validate_asset_path( + metadata.file_path, "chunks.json", f"chunks.{index}.metadata.file_path" + ) + for asset_index, asset in enumerate(metadata.page_assets or []): + _validate_asset_path( + asset.artifact_ref, + "chunks.json", + f"chunks.{index}.metadata.page_assets.{asset_index}.artifact_ref", + ) + for connection_index, connection in enumerate(metadata.connect_to or []): + if not isinstance(connection, str) and connection.position: + position = connection.position + if position.start > position.end or position.end > len(chunk.content): + _fail( + "chunks.json", + f"chunks.{index}.metadata.connect_to.{connection_index}.position", + "invalid_span", + ) + + expected = { + "total_chunks": len(parsed_chunks.chunks), + **{f"{kind}_chunks": count for kind, count in counts.items()}, + } + for artifact, stats in [ + ("manifest.json", parsed_manifest.statistics), + ("doc_nav.json", parsed_nav.stats if parsed_nav else None), + ]: + if stats is None: + continue + for field, count in expected.items(): + if getattr(stats, field) != count: + _fail( + artifact, + f"{'statistics' if artifact == 'manifest.json' else 'stats'}.{field}", + "chunk_count_mismatch", + ) + return ValidatedParseResult( + parsed_manifest, parsed_chunks, parsed_nav, tuple(warnings) + ) + + +def _safe_member(name: str) -> bool: + parts = name.rstrip("/").split("/") + return ( + bool(name) + and not name.startswith("/") + and "\\" not in name + and "\x00" not in name + and all(part not in {"", ".", ".."} for part in parts) + and ":" not in parts[0] + ) + + +def _validate_asset_path(path: str, artifact: str, field: str) -> None: + if ( + not _safe_member(path) + or path.endswith("/") + or PurePosixPath(path).parts[0] + not in {"images", "tables", "page_citation_assets"} + ): + _fail(artifact, field, "invalid_asset_path") + + +def validate_parse_result_archive( + path: str, + *, + allow_legacy: bool = True, + max_json_bytes: int | None = 128 * 1024 * 1024, +) -> ValidatedParseResult: + """Read ZIP metadata without extraction; reject missing assets/unsafe members. + + max_json_bytes bounds each JSON member before decompression for SDK consumers. + The producer may use None for its own locally generated package. + """ + try: + with zipfile.ZipFile(path) as archive: + infos = archive.infolist() + names = [info.filename for info in infos] + if len(set(names)) != len(names): + _fail("ZIP", "members", "duplicate_member") + # ZipInfo.filename truncates at NUL; validate the original name. + if any(not _safe_member(info.orig_filename) for info in infos): + _fail("ZIP", "members", "unsafe_member") + files = {info.filename for info in infos if not info.is_dir()} + payloads: dict[str, Any] = {} + for name in ("manifest.json", "chunks.json", "doc_nav.json"): + if name not in files: + if name == "doc_nav.json" and allow_legacy: + payloads[name] = None + continue + _fail(name, "$", "required_artifact_missing") + if ( + max_json_bytes is not None + and archive.getinfo(name).file_size > max_json_bytes + ): + _fail(name, "$", "json_size_limit") + try: + payloads[name] = json.loads( + archive.read(name), + parse_constant=_invalid_json_constant, + parse_float=_finite_json_float, + object_pairs_hook=_unique_json_object, + ) + except (ValueError, UnicodeError): + _fail(name, "$", "invalid_json") + validated = validate_parse_result( + payloads["manifest.json"], + payloads["chunks.json"], + payloads["doc_nav.json"], + allow_legacy=allow_legacy, + ) + for index, chunk in enumerate(validated.chunks.chunks): + paths = [ + (chunk.metadata.file_path, f"chunks.{index}.metadata.file_path") + ] + paths.extend( + ( + asset.artifact_ref, + f"chunks.{index}.metadata.page_assets.{asset_index}.artifact_ref", + ) + for asset_index, asset in enumerate( + chunk.metadata.page_assets or [] + ) + ) + for asset_path, field in paths: + if asset_path and asset_path not in files: + _fail("chunks.json", field, "referenced_asset_missing") + return validated + except ( + zipfile.BadZipFile, + OSError, + RuntimeError, + NotImplementedError, + zlib.error, + EOFError, + ): + _fail("ZIP", "$", "unreadable_archive") + + +def _invalid_json_constant(value: str) -> NoReturn: + raise ValueError("Non-finite JSON number") + + +def _unique_json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError("Duplicate JSON object key") + result[key] = value + return result + + +def _finite_json_float(value: str) -> float: + number = float(value) + if not math.isfinite(number): + raise ValueError("Non-finite JSON number") + return number diff --git a/packages/shared-python/shared/core/exceptions/__init__.py b/packages/shared-python/shared/core/exceptions/__init__.py index 8cfe04a96..85542f8c4 100644 --- a/packages/shared-python/shared/core/exceptions/__init__.py +++ b/packages/shared-python/shared/core/exceptions/__init__.py @@ -5,6 +5,7 @@ ConflictException, FileSystemException, LibreOfficeServiceException, + ParseResultContractException, NotFoundException, PermissionDeniedException, QuotaExceededException, @@ -39,6 +40,7 @@ "UnknownException", "FileSystemException", "LibreOfficeServiceException", + "ParseResultContractException", # Webhook Exceptions "WebhookException", "WebhookConfigException", diff --git a/packages/shared-python/shared/core/exceptions/domain_exceptions.py b/packages/shared-python/shared/core/exceptions/domain_exceptions.py index 389daa9ad..1683c8481 100644 --- a/packages/shared-python/shared/core/exceptions/domain_exceptions.py +++ b/packages/shared-python/shared/core/exceptions/domain_exceptions.py @@ -686,6 +686,24 @@ def __init__( ) +class ParseResultContractException(KnowhereException): + """A producer/consumer ZIP contract violation; details never contain input values.""" + + log_traceback = False + + def __init__(self, *, violations: List[Dict[str, str]], schema_version: int): + super().__init__( + code=ErrorCode.INTERNAL_ERROR, + internal_message="Parse-result ZIP contract validation failed", + user_message="Parsed result does not match the supported ZIP contract.", + details={ + "reason": "PARSE_RESULT_CONTRACT_VIOLATION", + "schema_version": schema_version, + "violations": violations, + }, + ) + + class LibreOfficeServiceException(KnowhereException): """ LibreOffice document conversion failed. diff --git a/packages/shared-python/shared/core/exceptions/knowhere_exception.py b/packages/shared-python/shared/core/exceptions/knowhere_exception.py index 424431a0b..2ee1d99e2 100644 --- a/packages/shared-python/shared/core/exceptions/knowhere_exception.py +++ b/packages/shared-python/shared/core/exceptions/knowhere_exception.py @@ -114,6 +114,10 @@ class KnowhereException(Exception): original_exception: Wrapped exception for logging (NOT sent to client) """ + # Sensitive domain exceptions can keep structured fields without allowing + # diagnostic log sinks to render the original traceback's local variables. + log_traceback: bool = True + def __init__( self, code: ErrorCode, @@ -241,7 +245,7 @@ def logging(self, **extra_context): - Reads current log context (request_id, task_id, job_id, etc.) - Merges exception fields from to_log() - Logs at appropriate level (ERROR for 5xx, WARNING for 4xx) - - Includes stacktrace for 5xx errors + - Includes stacktrace for 5xx errors unless the domain opts out - Ensures user_message is present in exception logs Usage: @@ -274,9 +278,9 @@ def logging(self, **extra_context): # Log at appropriate level with appropriate event if self.http_status_code >= 500: - # 5xx: ERROR level with stacktrace + # 5xx: ERROR level; sensitive domains suppress diagnostic locals. logger.bind(event=LogEvent.EXCEPTION_SYSTEM.value, **log_data).opt( - exception=self + exception=self if self.log_traceback else None ).error(self.internal_message) else: # 4xx: WARNING level without stacktrace diff --git a/packages/shared-python/shared/services/storage/zip_package_writer.py b/packages/shared-python/shared/services/storage/zip_package_writer.py index 9d2361834..2bebd91ed 100644 --- a/packages/shared-python/shared/services/storage/zip_package_writer.py +++ b/packages/shared-python/shared/services/storage/zip_package_writer.py @@ -12,6 +12,8 @@ from loguru import logger +from shared.contracts.parse_result import SCHEMA_VERSION, validate_parse_result_archive +from shared.core.exceptions.domain_exceptions import ParseResultContractException from shared.services.storage.zip_result_resources import ZipResourceFileInfo @@ -46,52 +48,85 @@ def write(self, request: ZipPackageWriteRequest) -> ZipPackageArtifact: os.makedirs(effective_temp_dir, exist_ok=True) zip_file_path = os.path.join(effective_temp_dir, f"result_{request.job_id}.zip") - with zipfile.ZipFile(zip_file_path, "w", zipfile.ZIP_DEFLATED) as zip_file: - chunks_json = json.dumps( - {"chunks": request.formatted_chunks}, - ensure_ascii=False, - indent=2, - ) - zip_file.writestr("chunks.json", chunks_json.encode("utf-8")) - - self._write_optional_file(zip_file, request.add_dir, "full.md") - if request.include_toc_hierarchies and self._write_optional_file( - zip_file, - request.add_dir, - "toc_hierarchies.json", - ): - logger.info("Added toc_hierarchies.json to ZIP") - if self._write_optional_json_file( - zip_file, - request.add_dir, - "_doc_agent/trace.json", - "debug/trace.json", - compact_trace=True, - ): - logger.info("Added debug/trace.json to ZIP") - if self._write_optional_json_file( - zip_file, - request.add_dir, - "_doc_agent/anatomy_map.json", - "debug/anatomy_map.json", - ): - logger.info("Added debug/anatomy_map.json to ZIP") - - self._write_resource_files(zip_file, request.image_files, label="Image") - self._write_resource_files(zip_file, request.table_files, label="Table") - self._write_resource_files( - zip_file, - request.page_citation_files, - label="Page citation asset", + written_resources: dict[str, str] = {} + try: + with zipfile.ZipFile(zip_file_path, "w", zipfile.ZIP_DEFLATED) as zip_file: + chunks_json = json.dumps( + { + "schema_version": SCHEMA_VERSION, + "chunks": request.formatted_chunks, + }, + ensure_ascii=False, + indent=2, + ) + zip_file.writestr("chunks.json", chunks_json.encode("utf-8")) + + self._write_optional_file(zip_file, request.add_dir, "full.md") + if request.include_toc_hierarchies and self._write_optional_file( + zip_file, + request.add_dir, + "toc_hierarchies.json", + ): + logger.info("Added toc_hierarchies.json to ZIP") + if self._write_optional_json_file( + zip_file, + request.add_dir, + "_doc_agent/trace.json", + "debug/trace.json", + compact_trace=True, + ): + logger.info("Added debug/trace.json to ZIP") + if self._write_optional_json_file( + zip_file, + request.add_dir, + "_doc_agent/anatomy_map.json", + "debug/anatomy_map.json", + ): + logger.info("Added debug/anatomy_map.json to ZIP") + + self._write_resource_files( + zip_file, + request.image_files, + label="Image", + written=written_resources, + ) + self._write_resource_files( + zip_file, + request.table_files, + label="Table", + written=written_resources, + ) + self._write_resource_files( + zip_file, + request.page_citation_files, + label="Page citation asset", + written=written_resources, + ) + + if request.doc_nav is not None: + doc_nav_json = json.dumps( + {"schema_version": SCHEMA_VERSION, **request.doc_nav}, + ensure_ascii=False, + indent=2, + ) + zip_file.writestr("doc_nav.json", doc_nav_json.encode("utf-8")) + logger.info("Added doc_nav.json") + + manifest_json = json.dumps( + {"schema_version": SCHEMA_VERSION, **request.manifest}, + ensure_ascii=False, + indent=2, + ) + zip_file.writestr("manifest.json", manifest_json.encode("utf-8")) + + validate_parse_result_archive( + zip_file_path, allow_legacy=False, max_json_bytes=None ) - - if request.doc_nav is not None: - doc_nav_json = json.dumps(request.doc_nav, ensure_ascii=False, indent=2) - zip_file.writestr("doc_nav.json", doc_nav_json.encode("utf-8")) - logger.info("Added doc_nav.json") - - manifest_json = json.dumps(request.manifest, ensure_ascii=False, indent=2) - zip_file.writestr("manifest.json", manifest_json.encode("utf-8")) + except Exception: + # Invalid packages must never be returned to the upload/finalization path. + if os.path.exists(zip_file_path): + os.remove(zip_file_path) + raise checksum_value = _calculate_zip_checksum(zip_file_path) zip_size = os.path.getsize(zip_file_path) @@ -145,11 +180,30 @@ def _write_resource_files( resources: tuple[ZipResourceFileInfo, ...], *, label: str, + written: dict[str, str], ) -> None: for resource in resources: source_path = resource["source_path"] if os.path.exists(source_path): - zip_file.write(source_path, resource["zip_path"]) + zip_path = resource["zip_path"] + previous = written.get(zip_path) + if previous is not None: + if previous != source_path and _calculate_zip_checksum( + previous + ) != _calculate_zip_checksum(source_path): + raise ParseResultContractException( + violations=[ + { + "artifact": "ZIP", + "field": "members", + "reason": "conflicting_asset_path", + } + ], + schema_version=SCHEMA_VERSION, + ) + continue + zip_file.write(source_path, zip_path) + written[zip_path] = source_path else: logger.warning(f"{label} file not found: {source_path}") diff --git a/packages/shared-python/shared/services/storage/zip_result_service.py b/packages/shared-python/shared/services/storage/zip_result_service.py index 160223a9e..8238a0078 100644 --- a/packages/shared-python/shared/services/storage/zip_result_service.py +++ b/packages/shared-python/shared/services/storage/zip_result_service.py @@ -14,8 +14,11 @@ import pandas as pd +from shared.contracts.parse_result import SCHEMA_VERSION, validate_parse_result + from shared.core.exceptions.domain_exceptions import ( KnowhereException, + ParseResultContractException, StorageServiceException, ) from shared.services.storage.zip_package_writer import ( @@ -85,6 +88,15 @@ def generate_zip_package( hierarchy=hierarchy, ) manifest = strip_manifest_cost_fields(manifest) + manifest = {"schema_version": SCHEMA_VERSION, **manifest} + if doc_nav is not None: + doc_nav = {"schema_version": SCHEMA_VERSION, **doc_nav} + validate_parse_result( + manifest, + {"schema_version": SCHEMA_VERSION, "chunks": formatted_chunks}, + doc_nav, + allow_legacy=False, + ) parse_track = str((job_metadata or {}).get("parse_track") or "") artifact = self._writer.write( ZipPackageWriteRequest( @@ -144,6 +156,8 @@ def _build_navigation_outputs( doc_nav = self._schema.build_doc_nav(formatted_chunks, source_file_name) hierarchy = self._schema.build_hierarchy_dict(doc_nav.get("sections", [])) return doc_nav, hierarchy + except ParseResultContractException: + raise except Exception as exc: logger.warning(f"generate doc_nav.json fail {exc}") return None, {} @@ -163,6 +177,17 @@ def _load_existing_doc_nav(add_dir: str) -> dict[str, Any] | None: return None if not isinstance(payload, dict): return None + if "schema_version" in payload: + version = payload["schema_version"] + if type(version) is not int or version != SCHEMA_VERSION: + raise ParseResultContractException( + violations=[{ + "artifact": "doc_nav.json", + "field": "schema_version", + "reason": "unsupported_schema_version", + }], + schema_version=SCHEMA_VERSION, + ) from None if not isinstance(payload.get("sections"), list): return None return payload diff --git a/packages/shared-python/shared/tests/fixtures/parse_result/chunks.json b/packages/shared-python/shared/tests/fixtures/parse_result/chunks.json new file mode 100644 index 000000000..56d639efc --- /dev/null +++ b/packages/shared-python/shared/tests/fixtures/parse_result/chunks.json @@ -0,0 +1,77 @@ +{ + "schema_version": 1, + "chunks": [ + { + "chunk_id": "text-1", + "type": "text", + "content": "hello", + "path": "sample.pdf/Intro", + "metadata": { + "length": 5, + "summary": "hello", + "page_nums": [ + 1 + ], + "tokens": 0, + "keywords": [], + "connect_to": [] + } + }, + { + "chunk_id": "image-1", + "type": "image", + "content": "A picture", + "path": "images/picture.png", + "metadata": { + "length": 9, + "summary": "picture", + "page_nums": [ + 1 + ], + "file_path": "images/picture.png", + "tokens": [], + "keywords": [] + } + }, + { + "chunk_id": "table-1", + "type": "table", + "content": "tables/table.html", + "path": "tables/table.html", + "metadata": { + "length": 17, + "summary": "table", + "page_nums": [ + 2 + ], + "file_path": "tables/table.html", + "tokens": [], + "keywords": [], + "connect_to": [] + } + }, + { + "chunk_id": "page-1", + "type": "page", + "content": "page text", + "path": "sample.pdf/Chapter", + "metadata": { + "length": 9, + "summary": "page", + "page_nums": [ + 3 + ], + "keywords": [], + "connect_to": [], + "page_assets": [ + { + "page_num": 3, + "artifact_ref": "page_citation_assets/page-3.webp", + "source": "render", + "content_type": "image/webp" + } + ] + } + } + ] +} diff --git a/packages/shared-python/shared/tests/fixtures/parse_result/doc_nav.json b/packages/shared-python/shared/tests/fixtures/parse_result/doc_nav.json new file mode 100644 index 000000000..3fd6b4d6a --- /dev/null +++ b/packages/shared-python/shared/tests/fixtures/parse_result/doc_nav.json @@ -0,0 +1,38 @@ +{ + "schema_version": 1, + "version": "1.0", + "file_name": "sample.pdf", + "top_summary": "Preserve enriched extra fields", + "stats": { + "total_chunks": 4, + "text_chunks": 1, + "image_chunks": 1, + "table_chunks": 1, + "page_chunks": 1, + "max_depth": 1 + }, + "sections": [ + { + "title": "Intro", + "path": "sample.pdf/Intro", + "level": 1, + "summary": "hello", + "chunk_count": 1, + "children": [] + } + ], + "resources": { + "images": [ + { + "path": "images/picture.png", + "summary": "picture" + } + ], + "tables": [ + { + "path": "tables/table.html", + "summary": "table" + } + ] + } +} diff --git a/packages/shared-python/shared/tests/fixtures/parse_result/manifest.json b/packages/shared-python/shared/tests/fixtures/parse_result/manifest.json new file mode 100644 index 000000000..4c33feb9a --- /dev/null +++ b/packages/shared-python/shared/tests/fixtures/parse_result/manifest.json @@ -0,0 +1,34 @@ +{ + "schema_version": 1, + "version": "2.0", + "job_id": "job-fixture", + "data_id": null, + "source_file_name": "sample.pdf", + "processing_date": "2026-10-07T00:00:00Z", + "processing": { + "page_count": 3, + "billing_status": "skipped", + "cost": { + "micro_dollars": null, + "credits": null + }, + "timing": { + "started_at": null, + "completed_at": null, + "duration_ms": null + }, + "stages": {} + }, + "statistics": { + "total_chunks": 4, + "text_chunks": 1, + "image_chunks": 1, + "table_chunks": 1, + "page_chunks": 1, + "total_pages": null + }, + "HIERARCHY": { + "Intro": {}, + "Chapter": {} + } +} diff --git a/packages/shared-python/shared/tests/test_parse_result_contract.py b/packages/shared-python/shared/tests/test_parse_result_contract.py new file mode 100644 index 000000000..285b037d5 --- /dev/null +++ b/packages/shared-python/shared/tests/test_parse_result_contract.py @@ -0,0 +1,299 @@ +from __future__ import annotations + +import copy +import json +import pickle +import traceback +from pathlib import Path +import warnings +import zipfile + +import pytest +from jsonschema import Draft202012Validator +from loguru import logger + +from shared.contracts.parse_result import ( + validate_parse_result, + validate_parse_result_archive, +) +from shared.core.exceptions.domain_exceptions import ParseResultContractException + +FIXTURES = Path(__file__).parent / "fixtures" / "parse_result" +ROOT = Path(__file__).resolve().parents[4] + + +@pytest.fixture +def payloads(): + return [ + json.loads((FIXTURES / f"{name}.json").read_text()) + for name in ("manifest", "chunks", "doc_nav") + ] + + +def _archive(path, payloads, extras=None): + with zipfile.ZipFile(path, "w") as archive: + for name, payload in zip( + ("manifest.json", "chunks.json", "doc_nav.json"), payloads + ): + if payload is not None: + archive.writestr(name, json.dumps(payload)) + for name in ( + "images/picture.png", + "tables/table.html", + "page_citation_assets/page-3.webp", + ): + archive.writestr(name, b"synthetic asset") + for name, content in extras or []: + archive.writestr(name, content) + return str(path) + + +def test_real_v1_shapes_and_generated_schemas(payloads, tmp_path): + result = validate_parse_result(*payloads, allow_legacy=False) + assert [chunk.type for chunk in result.chunks.chunks] == [ + "text", + "image", + "table", + "page", + ] + assert result.chunks.chunks[2].content == "tables/table.html" + assert result.doc_nav.model_extra["top_summary"].startswith("Preserve") + assert result.warnings == () + for name, payload in zip(("manifest", "chunks", "doc_nav"), payloads): + schema = json.loads( + ( + ROOT / "packages" / "contracts" / "parse_result" / f"{name}.schema.json" + ).read_text() + ) + Draft202012Validator.check_schema(schema) + Draft202012Validator(schema).validate(payload) + assert ( + validate_parse_result_archive( + _archive(tmp_path / "result.zip", payloads), allow_legacy=False + ).warnings + == () + ) + + +def test_legacy_versions_and_navigation_omission(payloads): + for payload in payloads: + payload.pop("schema_version") + result = validate_parse_result(*payloads) + assert len(result.warnings) == 3 + assert all("schema_version" not in payload for payload in payloads) + assert validate_parse_result(payloads[0], payloads[1], None).doc_nav is None + with pytest.raises(ParseResultContractException): + validate_parse_result(*payloads, allow_legacy=False) + + +@pytest.mark.parametrize("version", [2, "1", True, None]) +def test_unsupported_versions_do_not_coerce(payloads, version): + payloads[1]["schema_version"] = version + with pytest.raises(ParseResultContractException) as error: + validate_parse_result(*payloads) + assert error.value.details["violations"] == [ + { + "artifact": "chunks.json", + "field": "schema_version", + "reason": "unsupported_schema_version", + } + ] + + +@pytest.mark.parametrize( + "field,value", + [ + ("content", {"secret": "private customer content"}), + ("type", "unknown"), + ("metadata", None), + ("chunk_id", ""), + ], +) +def test_malformed_chunk_fields_are_sanitized(payloads, field, value): + payloads[1]["chunks"][0][field] = value + with pytest.raises(ParseResultContractException) as error: + validate_parse_result(*payloads) + client = error.value.to_client("request") + assert client["error"]["details"]["schema_version"] == 1 + assert "private customer content" not in json.dumps(client) + assert "private customer content" not in "".join(traceback.format_exception(error.value)) + assert "input" not in json.dumps(client) + assert pickle.loads(pickle.dumps(error.value)).details == error.value.details + + +@pytest.mark.parametrize("field", ["length", "summary", "page_nums"]) +def test_required_metadata_and_strict_integer(payloads, field): + del payloads[1]["chunks"][0]["metadata"][field] + with pytest.raises(ParseResultContractException): + validate_parse_result(*payloads) + payloads = [ + json.loads((FIXTURES / f"{name}.json").read_text()) + for name in ("manifest", "chunks", "doc_nav") + ] + payloads[1]["chunks"][0]["metadata"]["length"] = "5" + with pytest.raises(ParseResultContractException): + validate_parse_result(*payloads) + + +def test_counts_and_connection_span_are_consistent(payloads): + invalid = copy.deepcopy(payloads) + invalid[0]["statistics"]["total_chunks"] = 3 + with pytest.raises(ParseResultContractException, match="validation failed"): + validate_parse_result(*invalid) + payloads[1]["chunks"][0]["metadata"]["connect_to"] = [ + {"target": "image-1", "relation": "embeds", "position": {"start": 7, "end": 3}} + ] + with pytest.raises(ParseResultContractException): + validate_parse_result(*payloads) + + +@pytest.mark.parametrize( + "extra,reason", + [ + (("../escape", "x"), "unsafe_member"), + (("chunks.json", "{}"), "duplicate_member"), + (("/absolute", "x"), "unsafe_member"), + (("images\\escape", "x"), "unsafe_member"), + ], +) +def test_invalid_archive_members(payloads, tmp_path, extra, reason): + with warnings.catch_warnings(): + warnings.simplefilter("ignore", UserWarning) + path = _archive(tmp_path / "bad.zip", payloads, [extra]) + with pytest.raises(ParseResultContractException) as error: + validate_parse_result_archive(path) + assert error.value.details["violations"][0]["reason"] == reason + + +def test_archive_requires_json_and_referenced_assets(payloads, tmp_path): + path = tmp_path / "bad.zip" + with zipfile.ZipFile(path, "w") as archive: + archive.writestr("manifest.json", json.dumps(payloads[0])) + archive.writestr("chunks.json", "{invalid") + with pytest.raises(ParseResultContractException) as error: + validate_parse_result_archive(str(path)) + assert error.value.details["violations"][0]["reason"] == "invalid_json" + payloads[1]["chunks"][1]["metadata"]["file_path"] = "images/absent.png" + with pytest.raises(ParseResultContractException) as error: + validate_parse_result_archive(_archive(path, payloads)) + assert error.value.details["violations"][0]["reason"] == "referenced_asset_missing" + payloads[1]["chunks"][1]["metadata"]["file_path"] = "images/../secret" + with pytest.raises(ParseResultContractException): + validate_parse_result(*payloads) + + +def test_raw_nul_member_name_is_not_hidden_by_zip_normalization(payloads, tmp_path): + path = tmp_path / "nul-name.zip" + with zipfile.ZipFile(path, "w") as archive: + for name, payload in zip(("manifest", "chunks", "doc_nav"), payloads): + member = "chunks.json!hidden" if name == "chunks" else f"{name}.json" + archive.writestr(member, json.dumps(payload)) + for name in ( + "images/picture.png", "tables/table.html", "page_citation_assets/page-3.webp" + ): + archive.writestr(name, b"synthetic asset") + path.write_bytes( + path.read_bytes().replace(b"chunks.json!hidden", b"chunks.json\0hidden") + ) + with zipfile.ZipFile(path) as archive: + info = archive.getinfo("chunks.json") + assert "\0" in info.orig_filename and "\0" not in info.filename + with pytest.raises(ParseResultContractException) as error: + validate_parse_result_archive(str(path), allow_legacy=False) + assert error.value.details["violations"][0]["reason"] == "unsafe_member" + + +def test_new_versioned_archive_requires_nav_and_bounded_json(payloads, tmp_path): + with pytest.raises(ParseResultContractException): + validate_parse_result_archive( + _archive(tmp_path / "missing.zip", [*payloads[:2], None]) + ) + with pytest.raises(ParseResultContractException) as error: + validate_parse_result_archive( + _archive(tmp_path / "oversized.zip", payloads), max_json_bytes=1 + ) + assert error.value.details["violations"][0]["reason"] == "json_size_limit" + + +def test_malformed_hierarchy_does_not_expose_customer_headings(payloads): + payloads[0]["HIERARCHY"] = {"private customer heading": []} + with pytest.raises(ParseResultContractException) as error: + validate_parse_result(*payloads) + assert "private customer heading" not in json.dumps( + error.value.to_client("request") + ) + assert error.value.details["violations"][0]["field"] == "HIERARCHY" + + +def test_canonical_logging_does_not_render_document_locals(payloads): + secret = "private-document-content-in-diagnostic-locals" + payloads[1]["chunks"][0]["content"] = {"secret": secret} + chunks = {"customer_content": secret, **payloads[1]} + messages = [] + sink = logger.add( + messages.append, + format="{extra} {message}", + diagnose=True, + backtrace=True, + ) + try: + with pytest.raises(ParseResultContractException) as error: + validate_parse_result(payloads[0], chunks, payloads[2]) + error.value.logging(job_id="synthetic-job") + finally: + logger.remove(sink) + + assert len(messages) == 1 + assert secret not in str(messages[0]) + assert messages[0].record["exception"] is None + extra = messages[0].record["extra"] + assert extra["event"] == "exception.system" + assert extra["job_id"] == "synthetic-job" + assert extra["details"]["reason"] == "PARSE_RESULT_CONTRACT_VIOLATION" + + +@pytest.mark.parametrize( + "content", ['{"schema_version": 1, "schema_version": 2}', '{"schema_version": NaN}'] +) +def test_ambiguous_or_nonfinite_json_rejected(payloads, tmp_path, content): + path = tmp_path / "invalid.zip" + with zipfile.ZipFile(path, "w") as archive: + archive.writestr("manifest.json", content) + archive.writestr("chunks.json", json.dumps(payloads[1])) + with pytest.raises(ParseResultContractException) as error: + validate_parse_result_archive(str(path)) + assert error.value.details["violations"][0]["reason"] == "invalid_json" + + +def test_unsupported_compression_has_structured_error(payloads, tmp_path): + import struct + + path = tmp_path / "compression.zip" + _archive(path, payloads) + data = bytearray(path.read_bytes()) + local = data.index(b"PK\x03\x04") + central = data.index(b"PK\x01\x02") + struct.pack_into(""), + ("page_citation_assets/page-3.webp", b"page"), + ]: + source = tmp_path / name + source.parent.mkdir() + source.write_bytes(content) + resources.append({"source_path": str(source), "zip_path": name}) + return ZipPackageWriteRequest( + job_id="writer", + add_dir=str(tmp_path), + formatted_chunks=chunks, + image_files=(resources[0],), + table_files=(resources[1],), + page_citation_files=(resources[2],), + manifest=manifest, + doc_nav=nav, + temp_dir=str(tmp_path), + ) + + +def test_writer_adds_versions_preserves_extensions_and_repeated_asset(package_request): + request = package_request + request = replace(package_request, image_files=request.image_files * 2) + artifact = ZipPackageWriter().write(request) + result = validate_parse_result_archive(artifact.zip_file_path, allow_legacy=False) + assert result.doc_nav.model_extra["top_summary"] == request.doc_nav["top_summary"] + assert [chunk.type for chunk in result.chunks.chunks] == [ + "text", + "image", + "table", + "page", + ] + assert "schema_version" not in request.manifest + assert "schema_version" not in request.doc_nav + with zipfile.ZipFile(artifact.zip_file_path) as archive: + assert archive.namelist().count("images/picture.png") == 1 + assert artifact.zip_size == Path(artifact.zip_file_path).stat().st_size + assert artifact.checksum["algorithm"] == "sha256" + + +def test_writer_discards_conflicting_duplicate_assets(package_request, tmp_path): + request = package_request + source = tmp_path / "conflicting.png" + source.write_bytes(b"different content") + request = replace( + request, + image_files=( + *request.image_files, + {"source_path": str(source), "zip_path": "images/picture.png"}, + ), + ) + with pytest.raises(ParseResultContractException) as error: + ZipPackageWriter().write(request) + assert error.value.details["violations"][0]["reason"] == "conflicting_asset_path" + assert not (tmp_path / "result_writer.zip").exists() + + +@pytest.mark.parametrize( + "case,reason", + [ + ("missing_nav", "required_artifact_missing"), + ("unsupported_version", "unsupported_schema_version"), + ("missing_asset", "referenced_asset_missing"), + ("unsafe_member", "unsafe_member"), + ], +) +def test_writer_never_returns_invalid_archive(package_request, tmp_path, case, reason): + request = package_request + if case == "missing_nav": + request = replace(package_request, doc_nav=None) + elif case == "unsupported_version": + request = replace(package_request, manifest={**request.manifest, "schema_version": 2}) + elif case == "missing_asset": + request = replace(package_request, image_files=()) + else: + request = replace( + request, + image_files=({**request.image_files[0], "zip_path": "../picture.png"},), + ) + with pytest.raises(ParseResultContractException) as error: + ZipPackageWriter().write(request) + assert error.value.details["violations"][0]["reason"] == reason + assert not (tmp_path / "result_writer.zip").exists() diff --git a/packages/shared-python/shared/tests/test_zip_result_contract.py b/packages/shared-python/shared/tests/test_zip_result_contract.py new file mode 100644 index 000000000..bb23be8fb --- /dev/null +++ b/packages/shared-python/shared/tests/test_zip_result_contract.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from unittest.mock import Mock +import zipfile + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-contract-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +import pytest + +from shared.contracts.parse_result import validate_parse_result_archive +from shared.core.exceptions.domain_exceptions import ParseResultContractException +from shared.services.storage.zip_result_schema import ZipResultSchemaBuilder +from shared.services.storage.zip_result_service import ZipResultService + + +def test_actual_writer_produces_valid_text_table_image_and_page_archive(tmp_path): + for directory in ("images", "tables", "page_citation_assets"): + (tmp_path / directory).mkdir() + (tmp_path / "images" / "picture.png").write_bytes(b"synthetic image bytes") + (tmp_path / "tables" / "table.html").write_text( + "
cell
" + ) + (tmp_path / "page_citation_assets" / "page-3.webp").write_bytes( + b"synthetic page bytes" + ) + fixtures = Path(__file__).parent / "fixtures" / "parse_result" + chunks = json.loads((fixtures / "chunks.json").read_text())["chunks"] + nav = json.loads((fixtures / "doc_nav.json").read_text()) + nav.pop("schema_version") + (tmp_path / "doc_nav.json").write_text(json.dumps(nav)) + path, checksum, statistics, size = ZipResultService().generate_zip_package( + job_id="job-test", + chunks=chunks, + add_dir=str(tmp_path), + source_file_name="sample.pdf", + data_id=None, + job_metadata={}, + temp_dir=str(tmp_path), + ) + validated = validate_parse_result_archive(path, allow_legacy=False) + assert validated.doc_nav.model_extra["top_summary"] == nav["top_summary"] + assert statistics["page_chunks"] == 1 + assert checksum["algorithm"] == "sha256" + assert size == os.path.getsize(path) + with zipfile.ZipFile(path) as archive: + assert all( + json.loads(archive.read(name))["schema_version"] == 1 + for name in ("chunks.json", "manifest.json", "doc_nav.json") + ) + assert "schema_version" not in nav + + +def test_empty_document_still_has_complete_navigation(tmp_path): + path, _, stats, _ = ZipResultService().generate_zip_package( + job_id="empty", + chunks=[], + add_dir=str(tmp_path), + source_file_name="empty.txt", + data_id=None, + job_metadata={}, + temp_dir=str(tmp_path), + ) + result = validate_parse_result_archive(path, allow_legacy=False) + assert result.chunks.chunks == [] + assert result.doc_nav.sections == [] + assert stats["total_chunks"] == 0 + + +def test_invalid_formatted_shape_prevents_writer_call(tmp_path): + schema = ZipResultSchemaBuilder() + schema.format_chunks = Mock(return_value=[{"chunk_id": "bad", "type": "unknown"}]) + writer = Mock() + with pytest.raises(ParseResultContractException): + ZipResultService( + schema_builder=schema, package_writer=writer + ).generate_zip_package( + job_id="bad", + chunks=[], + add_dir=str(tmp_path), + source_file_name="sample.txt", + data_id=None, + job_metadata={}, + temp_dir=str(tmp_path), + ) + writer.write.assert_not_called() + + +def test_navigation_failure_prevents_writer_call(tmp_path): + schema = ZipResultSchemaBuilder() + schema.build_doc_nav = Mock(side_effect=ValueError("navigation failure")) + writer = Mock() + with pytest.raises(ParseResultContractException) as error: + ZipResultService( + schema_builder=schema, package_writer=writer + ).generate_zip_package( + job_id="bad", + chunks=[], + add_dir=str(tmp_path), + source_file_name="sample.txt", + data_id=None, + job_metadata={}, + temp_dir=str(tmp_path), + ) + assert error.value.details["violations"][0]["artifact"] == "doc_nav.json" + writer.write.assert_not_called() + + +def test_missing_referenced_asset_discards_archive(tmp_path): + chunks = [ + { + "chunk_id": "table-missing", + "type": "table", + "content": "tables/missing.html", + "path": "tables/missing.html", + "metadata": {}, + } + ] + with pytest.raises(ParseResultContractException): + ZipResultService().generate_zip_package( + job_id="missing", + chunks=chunks, + add_dir=str(tmp_path), + source_file_name="sample.txt", + data_id=None, + job_metadata={}, + temp_dir=str(tmp_path), + ) + assert not (tmp_path / "result_missing.zip").exists() + + +def test_repeated_image_reference_writes_one_asset_member(tmp_path): + (tmp_path / "images").mkdir() + (tmp_path / "images" / "logo.png").write_bytes(b"synthetic logo bytes") + chunks = [ + { + "chunk_id": chunk_id, + "type": "image", + "content": "logo", + "path": "images/logo.png", + "metadata": {"file_path": "images/logo.png"}, + } + for chunk_id in ("logo-1", "logo-2") + ] + path, _, stats, _ = ZipResultService().generate_zip_package( + job_id="repeated", + chunks=chunks, + add_dir=str(tmp_path), + source_file_name="sample.docx", + data_id=None, + job_metadata={}, + temp_dir=str(tmp_path), + ) + with zipfile.ZipFile(path) as archive: + assert archive.namelist().count("images/logo.png") == 1 + assert stats["image_chunks"] == 2 + validate_parse_result_archive(path, allow_legacy=False) + + +@pytest.mark.parametrize("sections", ["existing", None, "missing"]) +def test_explicit_unsupported_navigation_version_is_not_relabelled(tmp_path, sections): + fixtures = Path(__file__).parent / "fixtures" / "parse_result" + nav = json.loads((fixtures / "doc_nav.json").read_text()) + nav["schema_version"] = 2 + if sections is None: + nav["sections"] = None + elif sections == "missing": + nav.pop("sections") + (tmp_path / "doc_nav.json").write_text(json.dumps(nav)) + writer = Mock() + with pytest.raises(ParseResultContractException) as error: + ZipResultService(package_writer=writer).generate_zip_package( + job_id="unsupported", + chunks=[], + add_dir=str(tmp_path), + source_file_name="sample.pdf", + data_id=None, + job_metadata={}, + temp_dir=str(tmp_path), + ) + assert ( + error.value.details["violations"][0]["reason"] == "unsupported_schema_version" + ) + writer.write.assert_not_called() diff --git a/scripts/sync_parse_result_contracts.py b/scripts/sync_parse_result_contracts.py new file mode 100644 index 000000000..3d13f9447 --- /dev/null +++ b/scripts/sync_parse_result_contracts.py @@ -0,0 +1,49 @@ +"""Regenerate SDK-discoverable JSON Schema from the Python source of truth.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "packages" / "shared-python")) + +from shared.contracts.parse_result.models import Chunks, DocNav, Manifest # noqa: E402 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--check", action="store_true") + args = parser.parse_args() + output = ROOT / "packages" / "contracts" / "parse_result" + mismatches = [] + for name, model in [ + ("chunks", Chunks), + ("doc_nav", DocNav), + ("manifest", Manifest), + ]: + schema = model.model_json_schema() + schema["$schema"] = "https://json-schema.org/draft/2020-12/schema" + schema["$id"] = ( + f"https://knowhereto.ai/contracts/parse_result/v1/{name}.schema.json" + ) + content = ( + json.dumps(schema, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + ) + path = output / f"{name}.schema.json" + if args.check: + if not path.exists() or path.read_text(encoding="utf-8") != content: + mismatches.append(str(path.relative_to(ROOT))) + else: + output.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + if mismatches: + print("Schema drift; run make sync-contracts: " + ", ".join(mismatches)) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())