From a6656bbcc08802232af32195ae281ab8cac26108 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Sat, 29 Aug 2026 20:18:38 +0300 Subject: [PATCH 1/8] fix: harden parsing and version correctness --- migrations/0009_correctness_hardening.sql | 31 ++++++ schemas/article.schema.json | 2 + schemas/citation.schema.json | 1 + schemas/common.schema.json | 1 + schemas/decision.schema.json | 1 + schemas/legislation.schema.json | 1 + src/mesa_legal_data/catalog.py | 54 ++++++++-- src/mesa_legal_data/harvest/service_bridge.py | 5 +- src/mesa_legal_data/parsers/legislation.py | 3 + src/mesa_legal_data/pipeline.py | 100 ++++++++++++------ src/mesa_legal_data/quality.py | 46 ++++++-- src/mesa_legal_data/sources/manual.py | 11 ++ tests/unit/test_correctness_hardening.py | 95 +++++++++++++++++ 13 files changed, 303 insertions(+), 48 deletions(-) create mode 100644 migrations/0009_correctness_hardening.sql create mode 100644 tests/unit/test_correctness_hardening.py diff --git a/migrations/0009_correctness_hardening.sql b/migrations/0009_correctness_hardening.sql new file mode 100644 index 0000000..b6c3745 --- /dev/null +++ b/migrations/0009_correctness_hardening.sql @@ -0,0 +1,31 @@ +-- Preserve state truth: a local polling timeout is pending, not failure. +PRAGMA foreign_keys = OFF; + +CREATE TABLE mesa_delivery_items_v2 ( + item_id TEXT PRIMARY KEY, + delivery_id TEXT NOT NULL, + document_id TEXT NOT NULL, + version_id TEXT NOT NULL, + chunk_id TEXT NOT NULL, + content_hash TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + remote_mutation_id TEXT, + remote_state TEXT NOT NULL CHECK (remote_state IN ('PLANNED', 'SENDING', 'QUEUED', 'PROCESSING', 'AWAITING_MUTATION', 'COMMITTED', 'FAILED', 'REJECTED', 'SKIPPED')), + payload_json TEXT NOT NULL, + last_error TEXT, + updated_at TEXT NOT NULL, + created_at TEXT NOT NULL, + FOREIGN KEY (delivery_id) REFERENCES mesa_deliveries(delivery_id) +); +INSERT INTO mesa_delivery_items_v2 SELECT * FROM mesa_delivery_items; +DROP TABLE mesa_delivery_items; +ALTER TABLE mesa_delivery_items_v2 RENAME TO mesa_delivery_items; +CREATE INDEX idx_mesa_items_delivery_state ON mesa_delivery_items(delivery_id, remote_state); +CREATE INDEX idx_mesa_items_doc_ver_chunk ON mesa_delivery_items(document_id, version_id, chunk_id, content_hash); +CREATE INDEX idx_mesa_items_idempotency ON mesa_delivery_items(idempotency_key); +CREATE INDEX idx_mesa_items_remote_state ON mesa_delivery_items(remote_state); + +-- Do not silently merge historical collisions: applying this migration fails +-- visibly if existing data violates the invariant. +CREATE UNIQUE INDEX idx_versions_document_revision_unique ON versions(document_id, revision_number); +PRAGMA foreign_keys = ON; diff --git a/schemas/article.schema.json b/schemas/article.schema.json index cd1c50c..0e09096 100644 --- a/schemas/article.schema.json +++ b/schemas/article.schema.json @@ -22,6 +22,7 @@ "article_number": { "type": "string" }, "article_kind": { "type": "string" }, "heading": { "type": ["string", "null"] }, + "ordinal": { "type": ["integer", "null"], "minimum": 1 }, "text": { "type": "string" }, "structure": { "type": ["object", "null"], @@ -55,6 +56,7 @@ "source_id": { "type": "string" }, "source_url": { "type": "string" }, "retrieved_at": { "type": "string" }, + "publication_date": { "type": ["string", "null"] }, "artifact_sha256": { "type": "string", "pattern": "^[a-fA-F0-9]{64}$" }, "artifact_path": { "type": ["string", "null"] } }, diff --git a/schemas/citation.schema.json b/schemas/citation.schema.json index da333e1..b2eeab6 100644 --- a/schemas/citation.schema.json +++ b/schemas/citation.schema.json @@ -42,6 +42,7 @@ "source_id": { "type": "string" }, "source_url": { "type": "string" }, "retrieved_at": { "type": "string" }, + "publication_date": { "type": ["string", "null"] }, "artifact_sha256": { "type": "string", "pattern": "^[a-fA-F0-9]{64}$" }, "artifact_path": { "type": ["string", "null"] } }, diff --git a/schemas/common.schema.json b/schemas/common.schema.json index 2f49477..d0c7f79 100644 --- a/schemas/common.schema.json +++ b/schemas/common.schema.json @@ -25,6 +25,7 @@ "source_id": { "type": "string" }, "source_url": { "type": "string" }, "retrieved_at": { "type": "string" }, + "publication_date": { "type": ["string", "null"] }, "artifact_sha256": { "type": "string" }, "artifact_path": { "type": ["string", "null"] } }, diff --git a/schemas/decision.schema.json b/schemas/decision.schema.json index 115fef0..585fcfe 100644 --- a/schemas/decision.schema.json +++ b/schemas/decision.schema.json @@ -35,6 +35,7 @@ "source_id": { "type": "string" }, "source_url": { "type": "string" }, "retrieved_at": { "type": "string" }, + "publication_date": { "type": ["string", "null"] }, "artifact_sha256": { "type": "string", "pattern": "^[a-fA-F0-9]{64}$" }, "artifact_path": { "type": ["string", "null"] } }, diff --git a/schemas/legislation.schema.json b/schemas/legislation.schema.json index c28abff..103257d 100644 --- a/schemas/legislation.schema.json +++ b/schemas/legislation.schema.json @@ -57,6 +57,7 @@ "source_id": { "type": "string" }, "source_url": { "type": "string" }, "retrieved_at": { "type": "string" }, + "publication_date": { "type": ["string", "null"] }, "artifact_sha256": { "type": "string", "pattern": "^[a-fA-F0-9]{64}$" }, "artifact_path": { "type": ["string", "null"] } }, diff --git a/src/mesa_legal_data/catalog.py b/src/mesa_legal_data/catalog.py index ed71040..c3057de 100644 --- a/src/mesa_legal_data/catalog.py +++ b/src/mesa_legal_data/catalog.py @@ -462,6 +462,47 @@ def insert_version( ) +def replace_derived_version_output( + conn: sqlite3.Connection, + *, + version_id: str, + document_id: str, + artifact_id: str, + canonical_path: str, + canonical_line: int, + canonical_sha256: str, + parser_name: str, + parser_version: str, + validation_status: str, + privacy_status: str, + quality_status: str, + quality_json: str, +) -> None: + """Replace only reproducible output for one immutable legal version.""" + current = get_version(conn, version_id) + if not current: + raise CatalogError(f"Cannot reprocess unknown version {version_id}") + if current["document_id"] != document_id or current["artifact_id"] != artifact_id: + raise CatalogError(f"Immutable version identity mismatch for {version_id}") + + with transaction(conn): + conn.execute( + """UPDATE versions + SET canonical_path = ?, canonical_line = ?, canonical_sha256 = ?, + parser_name = ?, parser_version = ?, validation_status = ?, + privacy_status = ?, approval_status = 'pending', + quality_status = ?, quality_json = ?, auto_approved = 0, + is_audit_sample = 0, audit_sample_reason = NULL + WHERE version_id = ?""", + ( + canonical_path, canonical_line, canonical_sha256, parser_name, + parser_version, validation_status, privacy_status, quality_status, + quality_json, version_id, + ), + ) + conn.execute("DELETE FROM records WHERE version_id = ?", (version_id,)) + + def get_version(conn: sqlite3.Connection, version_id: str) -> dict[str, Any] | None: cursor = conn.cursor() cursor.execute( @@ -649,20 +690,17 @@ def iter_records_for_release( cursor = conn.cursor() cursor.execute(""" WITH eligible_versions AS ( - SELECT v.version_id, - ROW_NUMBER() OVER ( - PARTITION BY v.document_id - ORDER BY COALESCE(v.revision_number, 1) DESC, v.created_at DESC - ) AS version_rank - FROM versions v + SELECT v.version_id + FROM documents d + JOIN versions v ON v.version_id = d.current_version_id WHERE v.approval_status = 'approved' AND v.validation_status = 'valid' - AND (v.quality_status IS NULL OR v.quality_status != 'BLOCK') + AND v.quality_status = 'PASS' AND v.privacy_status IN ('clean', 'approved') ) SELECT r.record_id, r.record_type, r.record_sha256, r.canonical_path, r.canonical_line, r.version_id FROM records r - JOIN eligible_versions ev ON r.version_id = ev.version_id AND ev.version_rank = 1 + JOIN eligible_versions ev ON r.version_id = ev.version_id WHERE r.approval_status = 'approved' AND r.validation_status = 'valid' ORDER BY r.canonical_path ASC, r.canonical_line ASC diff --git a/src/mesa_legal_data/harvest/service_bridge.py b/src/mesa_legal_data/harvest/service_bridge.py index de1eb7c..518150e 100644 --- a/src/mesa_legal_data/harvest/service_bridge.py +++ b/src/mesa_legal_data/harvest/service_bridge.py @@ -4,7 +4,7 @@ from mesa_legal_data.harvest.models import CollectResult, HarvestItem, PipelineResult from mesa_legal_data.pipeline import InvalidStateTransition, process_artifact_pipeline from mesa_legal_data.schema_validation import SchemaValidationError -from mesa_legal_data.sources.manual import import_manual_url +from mesa_legal_data.sources.manual import ArtifactDocumentCollisionError, import_manual_url from mesa_legal_data.sources.url_fetcher import ( SourcePolicyError, SSRFError, @@ -34,6 +34,7 @@ def collect_url_item(item: HarvestItem, sources_yaml_path: Path | None = None) - jurisdiction="TR", title=item.title, stable_key=item.document_id, + publication_date=item.publication_date, sources_yaml_path=sources_yaml_path, ) return CollectResult( @@ -70,6 +71,8 @@ def collect_url_item(item: HarvestItem, sources_yaml_path: Path | None = None) - err_code = "HTTP_429" elif any(s in err_msg for s in ("500", "502", "503", "504")): err_code = "HTTP_SERVER_ERROR" + elif isinstance(e, ArtifactDocumentCollisionError): + err_code = "ARTIFACT_DOCUMENT_COLLISION" else: if "Host" in err_msg and "not allowed" in err_msg: err_code = "SOURCE_HOST_NOT_ALLOWED" diff --git a/src/mesa_legal_data/parsers/legislation.py b/src/mesa_legal_data/parsers/legislation.py index 41dcb12..89df9d7 100644 --- a/src/mesa_legal_data/parsers/legislation.py +++ b/src/mesa_legal_data/parsers/legislation.py @@ -4,6 +4,9 @@ from mesa_legal_data.parsers.text_normalizer import normalize_text +PARSER_NAME = "legislation_parser" +PARSER_VERSION = "1.0.0" + class ParsedArticle(BaseModel): model_config = ConfigDict(frozen=True) diff --git a/src/mesa_legal_data/pipeline.py b/src/mesa_legal_data/pipeline.py index 3745ccf..d99d01f 100644 --- a/src/mesa_legal_data/pipeline.py +++ b/src/mesa_legal_data/pipeline.py @@ -17,6 +17,7 @@ insert_record, insert_version, open_issue, + replace_derived_version_output, transaction, update_artifact_transport_status, update_document_status, @@ -37,6 +38,8 @@ parse_pdf, ) from mesa_legal_data.parsers.coverage import compute_parsing_coverage +from mesa_legal_data.parsers.legislation import PARSER_NAME as LEGISLATION_PARSER_NAME +from mesa_legal_data.parsers.legislation import PARSER_VERSION as LEGISLATION_PARSER_VERSION from mesa_legal_data.parsers.text_normalizer import normalize_text from mesa_legal_data.quality import evaluate_quality from mesa_legal_data.schema_validation import validate_record @@ -84,6 +87,7 @@ def process_artifact_pipeline( sha256: str | None = None, byte_size: int | None = None, detected_mime: str | None = None, + force_reprocess: bool = False, ) -> str: """ Orchestrates end-to-end processing of an artifact: @@ -234,17 +238,20 @@ def process_artifact_pipeline( "source_id": art_row["source_id"], "source_url": art_row["source_url"], "retrieved_at": art_row["retrieved_at"], + "publication_date": meta_dict.get("publication_date"), "artifact_sha256": art_row["sha256"], "artifact_path": art_row["raw_path"], } + parser_name = LEGISLATION_PARSER_NAME if fam == "legislation" else f"{fam}_parser" + parser_version = LEGISLATION_PARSER_VERSION if fam == "legislation" else "1.0.0" provenance_obj = { - "parser_name": f"{fam}_parser", - "parser_version": "1.0.0", - "pipeline_run_id": f"artifact:{artifact_id}:{fam}_parser:1.0.0", + "parser_name": parser_name, + "parser_version": parser_version, + "pipeline_run_id": f"artifact:{artifact_id}:{parser_name}:{parser_version}", } existing_version = get_version(conn, version_id) - if existing_version: + if existing_version and not force_reprocess: if existing_version["artifact_id"] != artifact_id or existing_version["document_id"] != doc_id: finish_run( conn, @@ -344,6 +351,7 @@ def process_artifact_pipeline( "article_number": a.article_number, "article_kind": a.article_kind, "heading": a.heading, + "ordinal": a.ordinal, "text": a.text, "structure": None, "status": "active", @@ -496,8 +504,8 @@ def process_artifact_pipeline( canonical_text=canonical_text, coverage=coverage_result, privacy_issues=privacy_issues, - parser_name=f"{fam}_parser", - parser_version="1.0.0", + parser_name=parser_name, + parser_version=parser_version, ) quality_status = quality_report.decision @@ -528,7 +536,7 @@ def process_artifact_pipeline( records_by_type.setdefault(rt, []).append(r) canonical_locations = [] - canonical_write_id = "version-" + hashlib.sha256(f"{version_id}:parser:1.0.0".encode()).hexdigest()[:20] + canonical_write_id = "version-" + hashlib.sha256(f"{version_id}:parser:{parser_version}".encode()).hexdigest()[:20] for rt, r_list in records_by_type.items(): locs = write_canonical_part(r_list, rt, canonical_write_id) canonical_locations.extend(locs) @@ -539,28 +547,45 @@ def process_artifact_pipeline( c_path = first_loc.relative_path if first_loc else "" c_sha = first_loc.record_sha256 if first_loc else expected_sha - insert_version( - conn=conn, - version_id=version_id, - document_id=doc_id or "tr:legislation:unknown", - artifact_id=artifact_id, - version_kind=v_kind, - snapshot_date=ver_date, - effective_from=None, - effective_to=None, - canonical_path=c_path, - canonical_line=1, - canonical_sha256=c_sha, - parser_name=f"{fam}_parser", - parser_version="1.0.0", - schema_version="1.0.0", - validation_status=val_status, - privacy_status=privacy_status, - approval_status="pending", - quality_status=quality_status, - quality_json=quality_json, - supersedes_version_id=meta_dict.get("supersedes_version_id"), - ) + if existing_version: + replace_derived_version_output( + conn, + version_id=version_id, + document_id=doc_id or "tr:legislation:unknown", + artifact_id=artifact_id, + canonical_path=c_path, + canonical_line=1, + canonical_sha256=c_sha, + parser_name=parser_name, + parser_version=parser_version, + validation_status=val_status, + privacy_status=privacy_status, + quality_status=quality_status, + quality_json=quality_json, + ) + else: + insert_version( + conn=conn, + version_id=version_id, + document_id=doc_id or "tr:legislation:unknown", + artifact_id=artifact_id, + version_kind=v_kind, + snapshot_date=ver_date, + effective_from=None, + effective_to=None, + canonical_path=c_path, + canonical_line=1, + canonical_sha256=c_sha, + parser_name=parser_name, + parser_version=parser_version, + schema_version="1.0.0", + validation_status=val_status, + privacy_status=privacy_status, + approval_status="pending", + quality_status=quality_status, + quality_json=quality_json, + supersedes_version_id=meta_dict.get("supersedes_version_id"), + ) for loc in canonical_locations: insert_record( @@ -587,8 +612,8 @@ def process_artifact_pipeline( conn, version_id=version_id, source_id=art_row.get("source_id", "manual"), - parser_name=f"{fam}_parser", - parser_version="1.0.0", + parser_name=parser_name, + parser_version=parser_version, quality_decision=quality_status, has_privacy_blocker=(privacy_status == "flagged"), schema_valid=True, @@ -597,8 +622,17 @@ def process_artifact_pipeline( final_status = "approved" if doc_id: update_document_status(conn, doc_id, "approved", current_version_id=version_id) - except Exception: - pass + except Exception as exc: + open_issue( + conn, + issue_id=f"iss-{uuid.uuid4().hex[:8]}", + subject_type="version", + subject_id=version_id, + severity="error", + code="AUTO_APPROVAL_EVALUATION_FAILED", + message=str(exc), + details_json=json.dumps({"parser_name": parser_name, "parser_version": parser_version}), + ) # Step 10: Finish Run with Real Counters counters = { diff --git a/src/mesa_legal_data/quality.py b/src/mesa_legal_data/quality.py index d1802a4..2618fdd 100644 --- a/src/mesa_legal_data/quality.py +++ b/src/mesa_legal_data/quality.py @@ -7,6 +7,11 @@ QualityDecision = Literal["PASS", "REVIEW", "BLOCK"] +# Kept deliberately small and central: these are structural plausibility guards, +# not a configurable scoring system. +MIN_LEGISLATION_COVERAGE = 0.30 +MAX_EXPLAINED_UNCOVERED_RATIO = 0.70 + @dataclass(frozen=True) class CheckResult: @@ -156,6 +161,7 @@ def evaluate_quality( leg_recs = [r for r in canonical_records if r.get("record_type") == "legislation"] art_recs = [r for r in canonical_records if r.get("record_type") == "article"] + is_legislation = bool(leg_recs) if not canonical_records: checks.append( CheckResult("STRUCTURE", "records_generated", "BLOCK", "No canonical records generated by parser") @@ -217,30 +223,58 @@ def evaluate_quality( else: checks.append(CheckResult("STRUCTURE", "ordinal_sequence", "PASS", "Article sequence valid")) + if is_legislation and not art_recs: + checks.append( + CheckResult( + "STRUCTURE", + "article_plausibility", + "BLOCK", + "Legislation parsing produced zero articles; it cannot be auto-approved or released", + ) + ) + elif is_legislation: + checks.append(CheckResult("STRUCTURE", "article_plausibility", "PASS", f"Parsed {len(art_recs)} articles")) + # Coverage evaluation cov_dict = None if coverage: cov_dict = coverage.to_dict() - unexplained_gaps = [ + large_uncovered = [ gap for gap in coverage.uncovered_ranges - if gap.get("candidate_type") == "gap" + if gap.get("candidate_type") in {"gap", "preamble", "annex_trailing", "unknown", "other"} and gap.get("length", 0) >= max(1000, int(max(1, coverage.canonical_chars) * 0.05)) ] - if coverage.coverage_ratio < 0.20 and leg_recs and art_recs: + explained_uncovered_ratio = ( + coverage.uncovered_chars / coverage.canonical_chars if coverage.canonical_chars else 0.0 + ) + if is_legislation and coverage.coverage_ratio < MIN_LEGISLATION_COVERAGE: checks.append( CheckResult( - "STRUCTURE", "coverage", "REVIEW", f"Low parser coverage: {coverage.coverage_ratio * 100:.1f}%" + "STRUCTURE", + "coverage", + "BLOCK" if not art_recs else "REVIEW", + f"Low legislation parser coverage: {coverage.coverage_ratio * 100:.1f}%", + ) + ) + elif is_legislation and large_uncovered and explained_uncovered_ratio > MAX_EXPLAINED_UNCOVERED_RATIO: + checks.append( + CheckResult( + "STRUCTURE", + "coverage", + "REVIEW", + "Most of the legislation text is uncovered, even if classified as preamble or annex", + {"uncovered_ranges": large_uncovered}, ) ) - elif unexplained_gaps: + elif large_uncovered: checks.append( CheckResult( "STRUCTURE", "coverage", "REVIEW", "Large unexplained gaps exist between parsed legal units", - {"gaps": unexplained_gaps}, + {"gaps": large_uncovered}, ) ) else: diff --git a/src/mesa_legal_data/sources/manual.py b/src/mesa_legal_data/sources/manual.py index 6ecdea4..09d3212 100644 --- a/src/mesa_legal_data/sources/manual.py +++ b/src/mesa_legal_data/sources/manual.py @@ -21,6 +21,10 @@ from mesa_legal_data.storage_paths import build_raw_path, secure_slug +class ArtifactDocumentCollisionError(ValueError): + """The immutable payload already belongs to another logical document.""" + + def import_manual_file( file_path: Path, source_id: str, @@ -185,6 +189,7 @@ def import_manual_url( document_type: str = "law", jurisdiction: str = "TR", title: str | None = None, + publication_date: str | None = None, stable_key: str | None = None, sources_yaml_path: Path | None = None, ) -> FetchedArtifact: @@ -239,6 +244,11 @@ def import_manual_url( except OSError: pass + if existing.get("document_id") and existing["document_id"] != document_id: + conn.close() + raise ArtifactDocumentCollisionError( + "ARTIFACT_DOCUMENT_COLLISION: identical payload is already bound to a different document" + ) doc_key = stable_key if stable_key else secure_slug(document_id) upsert_document( conn=conn, @@ -305,6 +315,7 @@ def import_manual_url( "sha256": artifact_sha256, "etag": headers.get("etag"), "last_modified": headers.get("last-modified"), + "publication_date": publication_date, "collector_version": "1.0.0", "access_policy_version": policy.policy_version, } diff --git a/tests/unit/test_correctness_hardening.py b/tests/unit/test_correctness_hardening.py new file mode 100644 index 0000000..3f0d5e5 --- /dev/null +++ b/tests/unit/test_correctness_hardening.py @@ -0,0 +1,95 @@ +import respx + +from mesa_legal_data.catalog import get_connection, insert_artifact, insert_version, migrate, upsert_document +from mesa_legal_data.parsers.coverage import compute_parsing_coverage +from mesa_legal_data.publisher.client import MesaClient +from mesa_legal_data.publisher.engine import get_ready_versions_and_content +from mesa_legal_data.publisher.models import MesaTargetSettings +from mesa_legal_data.quality import evaluate_quality + + +def _quality(text: str, spans: list[tuple[int, int]]): + provenance = {"pipeline_run_id": "test-run"} + source = {"artifact_sha256": "a" * 64} + records = [{"id": "doc", "record_type": "legislation", "title": "Kanun", "source": source, "provenance": provenance}] + records.extend( + { + "id": f"article-{index}", + "record_type": "article", + "ordinal": index, + "article_number": str(index), + "text": text[start:end], + "source_span": {"char_start": start, "char_end": end}, + "source": source, + "provenance": provenance, + } + for index, (start, end) in enumerate(spans, 1) + ) + return evaluate_quality( + source_info={"source_id": "resmi_gazete", "source_url": "https://example.test/law"}, + raw_info={"byte_size": 1, "sha256": "a" * 64, "file_exists": True}, + canonical_records=records, + canonical_text=text, + coverage=compute_parsing_coverage(text, spans), + ) + + +def test_legislation_zero_or_anomalous_coverage_cannot_pass(): + text = "Başlık ve açıklama\n" * 200 + assert _quality(text, []).decision != "PASS" + + article = "MADDE 1- Kısa hüküm" + anomalous = ("Uzun başlangıç metni\n" * 250) + article + assert _quality(anomalous, [(len(anomalous) - len(article), len(anomalous))]).decision != "PASS" + + healthy = "Önsöz\n\nMADDE 1- Birinci hüküm\n\nMADDE 2- İkinci hüküm\n\nEK CETVEL\nTablo" + first = healthy.index("MADDE 1") + second = healthy.index("MADDE 2") + annex = healthy.index("EK CETVEL") + assert _quality(healthy, [(first, second - 2), (second, annex - 2)]).decision == "PASS" + + +@respx.mock +def test_target_authentication_and_allowlist_are_fail_closed(monkeypatch): + settings = MesaTargetSettings( + base_url="https://mesa.example.test", + contract_source="configured", + health_path="/health", + publish_path="/publish", + mutation_status_path_template="/mutations/{mutation_id}", + ) + client = MesaClient(settings, api_key="not-logged") + assert client.test_connection()["connected"] is False + + monkeypatch.setenv("MESA_DATA_MESA_ALLOWED_HOST", "mesa.example.test") + respx.get("https://mesa.example.test/health").respond(401) + result = client.test_connection() + assert result["reachable"] is True + assert result["authenticated"] is False + assert result["connected"] is False + + +def test_only_current_version_is_publishable(tmp_path): + db_path = tmp_path / "catalog.sqlite" + migrate(None, db_path) + conn = get_connection(db_path) + doc_id = "doc-current" + upsert_document(conn, doc_id, "legislation", "law", "TR", "Kanun", "current", "fetched") + for version_id, revision, approval, quality in (("v1", 1, "approved", "PASS"), ("v2", 2, "pending", "REVIEW")): + artifact_id = f"art-{version_id}" + insert_artifact( + conn, artifact_id, doc_id, "resmi_gazete", "https://example.test/law", "2026-08-10T00:00:00Z", + "http", 200, "text/html", "text/html", 1, ("a" if version_id == "v1" else "b") * 64, + f"raw/{version_id}.html", None, None, + "verified", None, "{}", + ) + insert_version( + conn, version_id, doc_id, artifact_id, "original_publication", "2026-08-10", None, None, + "canonical/law.jsonl", 1, "a" * 64, "legislation_parser", "1.0.0", "1.0.0", "valid", + "clean", approval, revision_number=revision, quality_status=quality, + ) + conn.execute("UPDATE documents SET current_version_id = 'v2' WHERE document_id = ?", (doc_id,)) + ready, blocked = get_ready_versions_and_content(conn) + assert ready == [] + assert blocked == 0 + conn.close() From c20aa1799a1a553ef4c271ca98534d5f7006d790 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Sat, 29 Aug 2026 20:18:50 +0300 Subject: [PATCH 2/8] fix: secure current-version MESA publishing --- src/mesa_legal_data/publisher/client.py | 69 ++++++++++++++-- src/mesa_legal_data/publisher/engine.py | 82 +++++++++++++------ src/mesa_legal_data/publisher/ledger.py | 25 +++--- src/mesa_legal_data/publisher/models.py | 1 + src/mesa_legal_data/web/api.py | 9 +- src/mesa_legal_data/web/static/app.js | 4 +- src/mesa_legal_data/web/static/index.html | 2 +- .../integration/test_mesa_v4_publisher_e2e.py | 1 + .../unit/test_publisher_client_and_ledger.py | 7 +- 9 files changed, 142 insertions(+), 58 deletions(-) diff --git a/src/mesa_legal_data/publisher/client.py b/src/mesa_legal_data/publisher/client.py index a43bfa2..519b77e 100644 --- a/src/mesa_legal_data/publisher/client.py +++ b/src/mesa_legal_data/publisher/client.py @@ -1,6 +1,7 @@ import os import time from typing import Any +from urllib.parse import urlparse import httpx @@ -50,6 +51,29 @@ def _get_headers(self, idempotency_key: str | None = None) -> dict[str, str]: headers["Idempotency-Key"] = idempotency_key return headers + def target_safety_error(self) -> str | None: + """Validate the target before a request can carry Authorization.""" + try: + parsed = urlparse(self.settings.base_url) + except ValueError: + return "MESA target URL is malformed" + host = (parsed.hostname or "").lower().rstrip(".") + if not host or parsed.username or parsed.password or parsed.query or parsed.fragment: + return "MESA target URL is malformed or contains forbidden userinfo/query data" + local_hosts = {"localhost", "127.0.0.1", "::1"} + if host in local_hosts: + if parsed.scheme != "http" and parsed.scheme != "https": + return "Local MESA target must use HTTP or HTTPS" + return None + if parsed.scheme != "https": + return "Non-local MESA target must use HTTPS" + allowed_host = os.environ.get("MESA_DATA_MESA_ALLOWED_HOST", "").lower().rstrip(".") + if not allowed_host: + return "MESA_DATA_MESA_ALLOWED_HOST must explicitly allow the HTTPS target host" + if host != allowed_host: + return "MESA target host is not the explicitly allowed host" + return None + @property def is_contract_configured(self) -> bool: return bool( @@ -77,9 +101,14 @@ def test_connection(self) -> dict[str, Any]: """ Tests connectivity to the configured MESA base_url. """ + target_error = self.target_safety_error() + if target_error: + return {"connected": False, "reachable": False, "authenticated": False, "latency_ms": 0.0, "details": target_error} if not self.is_contract_configured: return { "connected": False, + "reachable": False, + "authenticated": False, "latency_ms": 0.0, "details": "MESA HTTP contract routes are not configured or verified", } @@ -95,18 +124,24 @@ def test_connection(self) -> dict[str, Any]: if resp.status_code in (200, 204): return { "connected": True, + "reachable": True, + "authenticated": True, "latency_ms": round(latency_ms, 2), "details": f"Connected ({resp.status_code})", } elif resp.status_code in (401, 403): return { - "connected": True, + "connected": False, + "reachable": True, + "authenticated": False, "latency_ms": round(latency_ms, 2), "details": "Server reachable, authentication required", } else: return { "connected": False, + "reachable": True, + "authenticated": False, "latency_ms": round(latency_ms, 2), "details": f"Server responded with status {resp.status_code}", } @@ -114,6 +149,8 @@ def test_connection(self) -> dict[str, Any]: latency_ms = (time.perf_counter() - start_time) * 1000.0 return { "connected": False, + "reachable": False, + "authenticated": False, "latency_ms": round(latency_ms, 2), "details": f"Connection error: {e}", } @@ -133,14 +170,15 @@ def run_preflight_checks( checks: list[PreflightCheckItem] = [] # 1. Target URL Check - if self.settings.base_url and self.settings.base_url.startswith(("http://", "https://")): + target_error = self.target_safety_error() + if not target_error: checks.append( PreflightCheckItem( name="target_url", status="PASS", message=f"Target URL configured: {self.settings.base_url}" ) ) else: - checks.append(PreflightCheckItem(name="target_url", status="FAIL", message="Invalid or missing target URL")) + checks.append(PreflightCheckItem(name="target_url", status="FAIL", message=target_error)) if self.is_contract_configured: checks.append( @@ -205,6 +243,14 @@ def run_preflight_checks( message=f"Server reachable ({conn_res['details']}, {conn_res['latency_ms']}ms)", ) ) + elif conn_res.get("reachable") and not conn_res.get("authenticated"): + checks.append( + PreflightCheckItem( + name="connectivity", + status="FAIL", + message=f"Server reachable but authentication failed ({conn_res['details']})", + ) + ) else: checks.append( PreflightCheckItem( @@ -230,12 +276,12 @@ def run_preflight_checks( ) ) - if blocked_versions_count > 0: + if blocked_versions_count > 0 and ready_versions_count == 0: checks.append( PreflightCheckItem( name="quality_guard", status="FAIL", - message=f"{blocked_versions_count} versions have quality status BLOCK (excluded from delivery)", + message=f"No current eligible versions; {blocked_versions_count} current versions are BLOCK and excluded", ) ) elif ready_versions_count > 0: @@ -243,7 +289,10 @@ def run_preflight_checks( PreflightCheckItem( name="quality_guard", status="PASS", - message=f"{ready_versions_count} approved versions ready for publishing ({estimated_chunks_count} chunks)", + message=( + f"{ready_versions_count} approved current versions ready for publishing ({estimated_chunks_count} chunks)" + + (f"; {blocked_versions_count} current BLOCK versions excluded" if blocked_versions_count else "") + ), ) ) else: @@ -299,6 +348,10 @@ def publish_source_chunk( "message": "MESA HTTP contract is unknown; refusing to guess a publish route", } + target_error = self.target_safety_error() + if target_error: + return {"mutation_id": None, "state": MutationState.FAILED.value, "message": target_error} + url = f"{self.settings.base_url.rstrip('/')}{self.settings.publish_path}" headers = self._get_headers(idempotency_key=idempotency_key) @@ -356,6 +409,10 @@ def get_mutation_status(self, mutation_id: str) -> dict[str, Any]: "error": "MESA HTTP contract is unknown", } + target_error = self.target_safety_error() + if target_error: + return {"mutation_id": mutation_id, "state": MutationState.FAILED.value, "error": target_error} + mutation_path = self.settings.mutation_status_path_template.replace("{mutation_id}", mutation_id) url = f"{self.settings.base_url.rstrip('/')}{mutation_path}" headers = self._get_headers() diff --git a/src/mesa_legal_data/publisher/engine.py b/src/mesa_legal_data/publisher/engine.py index 63e066c..cab8adb 100644 --- a/src/mesa_legal_data/publisher/engine.py +++ b/src/mesa_legal_data/publisher/engine.py @@ -32,29 +32,24 @@ def get_ready_versions_and_content(conn) -> tuple[list[dict[str, Any]], int]: Fetches approved versions ready for publishing, and count of blocked versions. """ cursor = conn.cursor() - # 1. Count blocked versions - cursor.execute("SELECT count(*) FROM versions WHERE quality_status = 'BLOCK'") + # Only the true current version can make a document publishable. Historical + # blocked or approved versions must not affect another current document. + cursor.execute( + """SELECT count(*) FROM documents d JOIN versions v ON v.version_id = d.current_version_id + WHERE v.quality_status = 'BLOCK'""" + ) blocked_count = cursor.fetchone()[0] # 2. Fetch approved versions - cursor.execute("""WITH eligible AS ( - SELECT v.*, - ROW_NUMBER() OVER ( - PARTITION BY v.document_id - ORDER BY COALESCE(v.revision_number, 1) DESC, v.created_at DESC - ) AS version_rank - FROM versions v - WHERE v.approval_status = 'approved' - AND v.validation_status = 'valid' - AND v.privacy_status IN ('clean', 'approved') - AND (v.quality_status IS NULL OR v.quality_status != 'BLOCK') - ) - SELECT v.version_id, v.document_id, v.canonical_path, v.canonical_sha256, - v.revision_number, d.family - FROM eligible v - JOIN documents d ON v.document_id = d.document_id - WHERE v.version_rank = 1 - ORDER BY v.created_at ASC""") + cursor.execute("""SELECT v.version_id, v.document_id, v.canonical_path, v.canonical_sha256, + v.revision_number, d.family + FROM documents d + JOIN versions v ON v.version_id = d.current_version_id + WHERE v.approval_status = 'approved' + AND v.validation_status = 'valid' + AND v.privacy_status IN ('clean', 'approved') + AND v.quality_status = 'PASS' + ORDER BY v.created_at ASC""") rows = cursor.fetchall() version_items = [ { @@ -317,6 +312,8 @@ def execute_publish_delivery( final_item_state = polled_state break poll_attempts += 1 + if final_item_state in (MutationState.QUEUED.value, MutationState.PROCESSING.value): + final_item_state = MutationState.AWAITING_MUTATION.value if final_item_state == MutationState.COMMITTED.value: committed_count += 1 @@ -336,6 +333,14 @@ def execute_publish_delivery( remote_mutation_id=remote_mutation_id, last_error=last_err, ) + elif final_item_state == MutationState.AWAITING_MUTATION.value: + update_delivery_item_state( + conn, + item_id=item_id, + remote_state=MutationState.AWAITING_MUTATION.value, + remote_mutation_id=remote_mutation_id, + last_error="Local poll window elapsed; remote mutation is still pending", + ) else: failed_count += 1 last_err = pub_res.get("message") or "Mutation failed to commit" @@ -360,7 +365,12 @@ def execute_publish_delivery( ) # 3. Compute final delivery status - if total_items == 0: + cursor = conn.cursor() + cursor.execute("SELECT count(*) FROM mesa_delivery_items WHERE delivery_id = ? AND remote_state = 'AWAITING_MUTATION'", (delivery_id,)) + awaiting_count = cursor.fetchone()[0] + if awaiting_count: + final_delivery_status = DeliveryStatus.AWAITING_MUTATION.value + elif total_items == 0: final_delivery_status = DeliveryStatus.COMMITTED.value elif failed_count == 0: final_delivery_status = DeliveryStatus.COMMITTED.value @@ -377,7 +387,7 @@ def execute_publish_delivery( failed_items=failed_count, skipped_items=skipped_count, last_error=last_err, - finished=True, + finished=final_delivery_status != DeliveryStatus.AWAITING_MUTATION.value, ) conn.close() @@ -429,9 +439,14 @@ def retry_delivery_failures( remote_state=MutationState.SENDING.value, ) - pub_res = client.publish_source_chunk(chunk, idempotency_key=idemp_key) - remote_mutation_id = pub_res.get("mutation_id") - final_state = pub_res.get("state", MutationState.FAILED.value) + remote_mutation_id = item.get("remote_mutation_id") + if item.get("remote_state") == MutationState.AWAITING_MUTATION.value and remote_mutation_id: + pub_res = client.get_mutation_status(remote_mutation_id) + final_state = pub_res.get("state", MutationState.FAILED.value) + else: + pub_res = client.publish_source_chunk(chunk, idempotency_key=idemp_key) + remote_mutation_id = pub_res.get("mutation_id") + final_state = pub_res.get("state", MutationState.FAILED.value) if final_state in (MutationState.QUEUED.value, MutationState.PROCESSING.value): update_delivery_item_state( @@ -453,6 +468,8 @@ def retry_delivery_failures( if polled_state != MutationState.COMMITTED.value: last_err = poll_res.get("error") or last_err break + if final_state in (MutationState.QUEUED.value, MutationState.PROCESSING.value): + final_state = MutationState.AWAITING_MUTATION.value if final_state == MutationState.COMMITTED.value: retried_success += 1 @@ -462,6 +479,14 @@ def retry_delivery_failures( remote_state=MutationState.COMMITTED.value, remote_mutation_id=remote_mutation_id, ) + elif final_state == MutationState.AWAITING_MUTATION.value: + update_delivery_item_state( + conn, + item_id=item_id, + remote_state=MutationState.AWAITING_MUTATION.value, + remote_mutation_id=remote_mutation_id, + last_error="Local poll window elapsed; remote mutation is still pending", + ) else: retried_failed += 1 last_err = pub_res.get("message") @@ -495,7 +520,10 @@ def retry_delivery_failures( failed = counts.get("FAILED", 0) + counts.get("REJECTED", 0) skipped = counts.get("SKIPPED", 0) - if failed == 0: + awaiting = counts.get("AWAITING_MUTATION", 0) + if awaiting: + new_status = DeliveryStatus.AWAITING_MUTATION.value + elif failed == 0: new_status = DeliveryStatus.COMMITTED.value elif committed > 0 or skipped > 0: new_status = DeliveryStatus.PARTIAL.value @@ -510,7 +538,7 @@ def retry_delivery_failures( failed_items=failed, skipped_items=skipped, last_error=last_err, - finished=True, + finished=new_status != DeliveryStatus.AWAITING_MUTATION.value, ) conn.close() diff --git a/src/mesa_legal_data/publisher/ledger.py b/src/mesa_legal_data/publisher/ledger.py index 13742ed..39855c4 100644 --- a/src/mesa_legal_data/publisher/ledger.py +++ b/src/mesa_legal_data/publisher/ledger.py @@ -352,23 +352,15 @@ def get_document_mesa_status(conn: sqlite3.Connection, document_id: str) -> dict """ cursor = conn.cursor() cursor.execute( - """WITH delivered_versions AS ( - SELECT i.version_id, - ROW_NUMBER() OVER ( - ORDER BY COALESCE(v.revision_number, 1) DESC, v.created_at DESC - ) AS version_rank - FROM mesa_delivery_items i - LEFT JOIN versions v ON v.version_id = i.version_id - WHERE i.document_id = ? - GROUP BY i.version_id - ), latest_items AS ( + """WITH latest_items AS ( SELECT i.remote_state, ROW_NUMBER() OVER ( PARTITION BY i.chunk_id ORDER BY i.updated_at DESC, i.created_at DESC ) AS attempt_rank FROM mesa_delivery_items i - JOIN delivered_versions dv ON dv.version_id = i.version_id AND dv.version_rank = 1 + JOIN documents d ON d.document_id = i.document_id + WHERE i.document_id = ? AND i.version_id = d.current_version_id ) SELECT remote_state, count(*) FROM latest_items @@ -378,12 +370,17 @@ def get_document_mesa_status(conn: sqlite3.Connection, document_id: str) -> dict ) counts = dict((r[0], r[1]) for r in cursor.fetchall()) if not counts: - return {"status": "Not Sent", "committed_count": 0, "failed_count": 0, "total_chunks": 0} + return {"status": "Update Pending", "committed_count": 0, "failed_count": 0, "total_chunks": 0} total = sum(counts.values()) committed = counts.get("COMMITTED", 0) + counts.get("SKIPPED", 0) failed = counts.get("FAILED", 0) + counts.get("REJECTED", 0) - sending = counts.get("SENDING", 0) + counts.get("QUEUED", 0) + counts.get("PROCESSING", 0) + sending = ( + counts.get("SENDING", 0) + + counts.get("QUEUED", 0) + + counts.get("PROCESSING", 0) + + counts.get("AWAITING_MUTATION", 0) + ) if sending > 0: status = "Sending" @@ -394,7 +391,7 @@ def get_document_mesa_status(conn: sqlite3.Connection, document_id: str) -> dict elif committed == total and total > 0: status = "Committed" else: - status = "Not Sent" + status = "Update Pending" return { "status": status, diff --git a/src/mesa_legal_data/publisher/models.py b/src/mesa_legal_data/publisher/models.py index 91c8de5..53f5752 100644 --- a/src/mesa_legal_data/publisher/models.py +++ b/src/mesa_legal_data/publisher/models.py @@ -19,6 +19,7 @@ class MutationState(str, Enum): SENDING = "SENDING" QUEUED = "QUEUED" PROCESSING = "PROCESSING" + AWAITING_MUTATION = "AWAITING_MUTATION" COMMITTED = "COMMITTED" FAILED = "FAILED" REJECTED = "REJECTED" diff --git a/src/mesa_legal_data/web/api.py b/src/mesa_legal_data/web/api.py index 8ab9a48..91347fc 100644 --- a/src/mesa_legal_data/web/api.py +++ b/src/mesa_legal_data/web/api.py @@ -157,10 +157,11 @@ def get_dashboard(): c.execute(""" SELECT count(DISTINCT r.record_instance_id) FROM records r JOIN versions v ON r.version_id = v.version_id + JOIN documents d ON d.current_version_id = v.version_id WHERE r.approval_status = 'approved' AND r.validation_status = 'valid' AND v.validation_status = 'valid' - AND (v.quality_status IS NULL OR v.quality_status != 'BLOCK') + AND v.quality_status = 'PASS' """) mesa_ready_count = c.fetchone()[0] @@ -1019,7 +1020,7 @@ def get_document_text_content(document_id: str): @router.post("/documents/{document_id:path}/reprocess") async def reprocess_document(document_id: str): - return await process_document_pipeline(document_id=document_id) + return await process_document_pipeline(document_id=document_id, force_reprocess=True) @router.get("/documents/{document_id:path}/versions") @@ -1202,7 +1203,7 @@ async def process_artifact(artifact_id: str): @router.post("/documents/{document_id:path}/pipeline") -async def process_document_pipeline(document_id: str): +async def process_document_pipeline(document_id: str, force_reprocess: bool = False): async with write_lock.acquire_write(): conn = get_connection() c = conn.cursor() @@ -1216,7 +1217,7 @@ async def process_document_pipeline(document_id: str): error_response("ARTIFACT_NOT_FOUND", f"No artifact found for document {document_id}", status_code=404) artifact_id = row[0] try: - pipeline_status = process_artifact_pipeline(artifact_id=artifact_id) + pipeline_status = process_artifact_pipeline(artifact_id=artifact_id, force_reprocess=force_reprocess) return ok_response( {"document_id": document_id, "artifact_id": artifact_id, "pipeline_status": pipeline_status} ) diff --git a/src/mesa_legal_data/web/static/app.js b/src/mesa_legal_data/web/static/app.js index 1cc9d96..606ade6 100644 --- a/src/mesa_legal_data/web/static/app.js +++ b/src/mesa_legal_data/web/static/app.js @@ -2167,7 +2167,7 @@ async function loadReleasesView() { ${rel.status === "draft" ? `` : ""} ${rel.status === "verified" ? `` : ""} - ${rel.status === "published" ? `` : ""} + ${rel.status === "published" ? `` : ""} `; tbody.appendChild(tr); @@ -2209,7 +2209,7 @@ async function importRelease(releaseId) { setBusy(true); try { await apiRequest(`/api/releases/${releaseId}/import-staging`, { method: "POST" }); - showToast("Release MESA staging ortamına aktarıldı.", "success"); + showToast("Release yerel development staging ortamına aktarıldı.", "success"); await loadReleasesView(); } catch (err) { console.error(err); diff --git a/src/mesa_legal_data/web/static/index.html b/src/mesa_legal_data/web/static/index.html index cdc7432..7f345cf 100644 --- a/src/mesa_legal_data/web/static/index.html +++ b/src/mesa_legal_data/web/static/index.html @@ -1003,7 +1003,7 @@

Manuel Olarak Çözüldü Kabul Et