From 3ff537b9807e9cd7b7f7275fee0d252f895205aa Mon Sep 17 00:00:00 2001 From: corlin Date: Sun, 4 Oct 2026 17:40:07 +0800 Subject: [PATCH] feat(legal-status): global append-only store for public legal status (ADR 0006) Legal status of a published patent is public and the same for every organization, so it is stored once in two global tables without organization_id or tenant RLS. Which organization looks at which patent is sensitive, so no requester, case or user field is allowed in these tables; tenant references will live in tenant-scoped (FORCE RLS) tables. - 0018_legal_status_store: legal_status_source_records (each lookup, including not_found, with request_ref, retrieved_at, the source's own data time, the rule-relevant extract, and the SHA-256, size and storage key of the exact response bytes) and legal_status_assessments (status, term, countries, evidence, review reasons, rules_version, content_sha256; unique per record/as_of/rules_version). Runtime roles: worker SELECT+INSERT, app SELECT, platform nothing; no UPDATE/DELETE, so both are append-only. - Raw responses are content-addressed in object storage (public/legal-status//.json). The DB keeps only fields the rules read. Raw retention involves public personal data and needs compliance sign-off (ADR 0006). - Worker LegalStatusRefresher: fetch -> archive exact bytes -> record -> assess from the stored extract -> record. A source failure writes nothing; not_found records the lookup but no status. - Clients now expose the exact response bytes (SourceResponse) so the stored hash proves what the source sent. - RULES_VERSION and canonical serialization in modules/legal_status. Tests: 8 Postgres integration tests (exact grants, no tenant columns, app cannot write, worker cannot UPDATE/DELETE, found rows must carry extract and hash, US/EP end-to-end with each stored assessment reproduced from its stored extract to the same content hash, not_found, failure writes nothing) plus 3 unit tests. test-postgres.sh now runs the new file. Co-Authored-By: Claude Opus 5.5 --- adapters/legal_status/clients.py | 35 ++- .../versions/0018_legal_status_store.py | 114 +++++++++ .../integration/test_legal_status_store.py | 216 ++++++++++++++++++ .../tests/unit/test_legal_status_serialize.py | 49 ++++ .../legal_status_refresh.py | 200 ++++++++++++++++ ...0006-public-patent-legal-status-storage.md | 101 ++++++++ modules/legal_status/serialize.py | 99 ++++++++ scripts/test-postgres.sh | 3 +- 8 files changed, 810 insertions(+), 7 deletions(-) create mode 100644 apps/api/migrations/versions/0018_legal_status_store.py create mode 100644 apps/api/tests/integration/test_legal_status_store.py create mode 100644 apps/api/tests/unit/test_legal_status_serialize.py create mode 100644 apps/worker/src/patent_evidence_worker/legal_status_refresh.py create mode 100644 docs/adr/0006-public-patent-legal-status-storage.md create mode 100644 modules/legal_status/serialize.py diff --git a/adapters/legal_status/clients.py b/adapters/legal_status/clients.py index 0f9eb40..00dbdb9 100644 --- a/adapters/legal_status/clients.py +++ b/adapters/legal_status/clients.py @@ -11,11 +11,22 @@ import base64 import os import time +from dataclasses import dataclass from typing import Any +from urllib.parse import urlencode import httpx +@dataclass(frozen=True) +class SourceResponse: + """A source answer with the exact bytes received, for hashing and archiving.""" + + data: dict[str, Any] | None # None means "not found" (unknown, not "not in force") + raw: bytes | None + request_ref: str # method + URL + parameters, never credentials + + class LegalStatusSourceError(Exception): """The source could not answer (transport, server, auth or quota).""" @@ -76,21 +87,28 @@ async def aclose(self) -> None: await self._client.aclose() async def file_wrapper(self, patent_number: str) -> dict[str, Any] | None: + return (await self.file_wrapper_response(patent_number)).data + + async def file_wrapper_response(self, patent_number: str) -> SourceResponse: + params = {"q": f"applicationMetaData.patentNumber:{patent_number}", "limit": 1} + request_ref = f"GET {self.SEARCH_URL}?{urlencode(params)}" response = await _get_with_retries( self._client, self.SEARCH_URL, headers={"X-API-KEY": self.api_key or "", "Accept": "application/json"}, - params={"q": f"applicationMetaData.patentNumber:{patent_number}", "limit": 1}, + params=params, backoff_seconds=self._backoff, ) if response.status_code == 404: - return None + return SourceResponse(None, None, request_ref) if response.status_code in (401, 403): raise LegalStatusSourceError(f"USPTO ODP refused the request (HTTP {response.status_code})") if response.status_code != 200: raise LegalStatusSourceError(f"USPTO ODP HTTP {response.status_code}") bag = response.json().get("patentFileWrapperDataBag") or [] - return bag[0] if bag else None + if not bag: + return SourceResponse(None, None, request_ref) + return SourceResponse(bag[0], response.content, request_ref) class EpoOpsLegalClient: @@ -143,15 +161,20 @@ async def _access_token(self) -> str: return self._token async def legal(self, country: str, number: str, kind: str) -> dict[str, Any] | None: + return (await self.legal_response(country, number, kind)).data + + async def legal_response(self, country: str, number: str, kind: str) -> SourceResponse: + url = f"{self.BASE_URL}/rest-services/legal/publication/docdb/{country}.{number}.{kind}" + request_ref = f"GET {url}" token = await self._access_token() response = await _get_with_retries( self._client, - f"{self.BASE_URL}/rest-services/legal/publication/docdb/{country}.{number}.{kind}", + url, headers={"Authorization": f"Bearer {token}", "Accept": "application/json"}, backoff_seconds=self._backoff, ) if response.status_code == 404: - return None + return SourceResponse(None, None, request_ref) if response.status_code == 403: text = response.text if "quota" in text.lower() or "throttl" in text.lower(): @@ -159,4 +182,4 @@ async def legal(self, country: str, number: str, kind: str) -> dict[str, Any] | raise LegalStatusSourceError("EPO OPS refused the request (HTTP 403)") if response.status_code != 200: raise LegalStatusSourceError(f"EPO OPS HTTP {response.status_code}") - return response.json() + return SourceResponse(response.json(), response.content, request_ref) diff --git a/apps/api/migrations/versions/0018_legal_status_store.py b/apps/api/migrations/versions/0018_legal_status_store.py new file mode 100644 index 0000000..14e6475 --- /dev/null +++ b/apps/api/migrations/versions/0018_legal_status_store.py @@ -0,0 +1,114 @@ +"""Store public patent legal-status observations and assessments (ADR 0006). + +Revision ID: 0018_legal_status_store +Revises: 0017_source_doc_findings + +Legal status of a published patent is public and identical for every +organization, so it lives in global tables without ``organization_id`` and +without per-organization RLS. Which organization looks at which patent is +sensitive and must never be stored here; tenant references belong in +tenant-scoped tables. + +Both tables are append-only for runtime roles (no UPDATE/DELETE grants): a new +lookup or a new rules version adds rows and never rewrites history. Only the +worker writes; the API role can only read. +""" + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import JSONB + +revision = "0018_legal_status_store" +down_revision = "0017_source_doc_findings" +branch_labels = None +depends_on = None + +SOURCES = ("uspto_odp", "epo_ops_inpadoc") +STATUSES = ("presumed_in_force", "lapsed", "expired", "undetermined") + + +def _in(column: str, values: tuple[str, ...]) -> str: + return f"{column} IN ({', '.join(repr(v) for v in values)})" + + +def upgrade() -> None: + op.create_table( + "legal_status_source_records", + sa.Column("id", sa.Uuid(), primary_key=True, nullable=False), + sa.Column("publication_number", sa.String(32), nullable=False), + sa.Column("source", sa.String(32), nullable=False), + # Endpoint and parameters of the request, never credentials. + sa.Column("request_ref", sa.Text(), nullable=False), + sa.Column("retrieved_at", sa.DateTime(timezone=True), nullable=False), + # The source's own data time (ODP lastIngestionDateTime); may be months old. + sa.Column("source_data_as_of", sa.DateTime(timezone=True), nullable=True), + # "not_found" is recorded too: it proves the lookup happened, not that the patent is dead. + sa.Column("outcome", sa.String(16), nullable=False), + # Only the fields the rules read; full responses go to object storage. + sa.Column("extract", JSONB(), nullable=True), + sa.Column("raw_sha256", sa.String(64), nullable=True), + sa.Column("raw_size_bytes", sa.BigInteger(), nullable=True), + sa.Column("raw_storage_key", sa.String(256), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.CheckConstraint(_in("source", SOURCES), name="ck_legal_status_source_records_source"), + sa.CheckConstraint("outcome IN ('found', 'not_found')", name="ck_legal_status_source_records_outcome"), + sa.CheckConstraint( + "(outcome = 'found' AND extract IS NOT NULL AND raw_sha256 IS NOT NULL)" + " OR (outcome = 'not_found' AND extract IS NULL)", + name="ck_legal_status_source_records_found_has_data", + ), + sa.CheckConstraint( + "raw_sha256 IS NULL OR raw_sha256 ~ '^[0-9a-f]{64}$'", + name="ck_legal_status_source_records_sha256", + ), + ) + op.create_index( + "ix_legal_status_source_records_lookup", + "legal_status_source_records", + ["publication_number", "source", sa.text("retrieved_at DESC")], + ) + + op.create_table( + "legal_status_assessments", + sa.Column("id", sa.Uuid(), primary_key=True, nullable=False), + sa.Column("source_record_id", sa.Uuid(), nullable=False), + sa.Column("publication_number", sa.String(32), nullable=False), + sa.Column("jurisdiction", sa.String(2), nullable=False), + sa.Column("as_of", sa.Date(), nullable=False), + sa.Column("rules_version", sa.String(64), nullable=False), + sa.Column("status", sa.String(24), nullable=False), + sa.Column("term", JSONB(), nullable=True), + sa.Column("countries", JSONB(), nullable=False, server_default=sa.text("'[]'::jsonb")), + sa.Column("evidence", JSONB(), nullable=False), + sa.Column("review_reasons", JSONB(), nullable=False, server_default=sa.text("'[]'::jsonb")), + # SHA-256 of the canonical assessment JSON, bound when an evidence snapshot is sealed. + sa.Column("content_sha256", sa.String(64), nullable=False), + sa.Column("assessed_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["source_record_id"], ["legal_status_source_records.id"], ondelete="RESTRICT" + ), + sa.CheckConstraint("jurisdiction IN ('US', 'EP')", name="ck_legal_status_assessments_jurisdiction"), + sa.CheckConstraint(_in("status", STATUSES), name="ck_legal_status_assessments_status"), + sa.CheckConstraint( + "content_sha256 ~ '^[0-9a-f]{64}$'", name="ck_legal_status_assessments_sha256" + ), + ) + op.create_unique_constraint( + "uq_legal_status_assessments_input", + "legal_status_assessments", + ["source_record_id", "as_of", "rules_version"], + ) + op.create_index( + "ix_legal_status_assessments_lookup", + "legal_status_assessments", + ["publication_number", sa.text("as_of DESC")], + ) + + for table in ("legal_status_source_records", "legal_status_assessments"): + op.execute(f'GRANT SELECT, INSERT ON "{table}" TO patent_evidence_worker') + op.execute(f'GRANT SELECT ON "{table}" TO patent_evidence_app') + + +def downgrade() -> None: + op.drop_table("legal_status_assessments") + op.drop_table("legal_status_source_records") diff --git a/apps/api/tests/integration/test_legal_status_store.py b/apps/api/tests/integration/test_legal_status_store.py new file mode 100644 index 0000000..1e064b7 --- /dev/null +++ b/apps/api/tests/integration/test_legal_status_store.py @@ -0,0 +1,216 @@ +"""Global legal-status store (ADR 0006): grants, append-only, reproducible assessments.""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Iterator +from datetime import UTC, date, datetime +from pathlib import Path +from uuid import uuid4 + +import httpx +import psycopg +import pytest +from test_support import async_postgres_url, postgres_url + +from adapters.legal_status.clients import EpoOpsLegalClient, LegalStatusSourceError, UsptoOdpClient +from adapters.object_storage.client import ObjectStorageClient +from modules.legal_status.ep import assess_ep +from modules.legal_status.serialize import RULES_VERSION, content_sha256 +from modules.legal_status.us import assess_us +from patent_evidence_api.core.database import create_engine, create_worker_session_factory +from patent_evidence_worker.legal_status_refresh import LegalStatusRefresher + +FIXTURES = Path(__file__).resolve().parents[4] / "fixtures" / "legal-status" +TABLES = ("legal_status_source_records", "legal_status_assessments") +AS_OF = date(2026, 10, 4) +NOW = datetime(2026, 10, 4, 8, 0, tzinfo=UTC) + + +@pytest.fixture +def migration() -> Iterator[psycopg.Connection]: + with psycopg.connect( + postgres_url("PE_TEST_MIGRATION_DATABASE_URL", "patent_evidence_migration"), autocommit=True + ) as connection: + connection.execute("DELETE FROM legal_status_assessments") + connection.execute("DELETE FROM legal_status_source_records") + yield connection + + +def _runtime(variable: str, role: str) -> psycopg.Connection: + return psycopg.connect(postgres_url(variable, role), autocommit=True) + + +def _insert_record(connection: psycopg.Connection, **overrides: object) -> None: + row = { + "id": uuid4(), + "pub": "US-1-B2", + "source": "uspto_odp", + "ref": "GET test", + "outcome": "not_found", + "extract": None, + "sha": None, + } + row.update(overrides) + connection.execute( + """INSERT INTO legal_status_source_records + (id, publication_number, source, request_ref, retrieved_at, outcome, extract, raw_sha256) + VALUES (%(id)s, %(pub)s, %(source)s, %(ref)s, now(), %(outcome)s, %(extract)s, %(sha)s)""", + row, + ) + + +def test_tables_are_global_with_exact_grants(migration: psycopg.Connection) -> None: + for table in TABLES: + rls, force, owner = migration.execute( + """SELECT c.relrowsecurity, c.relforcerowsecurity, o.rolname + FROM pg_class c JOIN pg_roles o ON o.oid = c.relowner WHERE c.oid = %s::regclass""", + (table,), + ).fetchone() + # Public data shared by all organizations: no tenant RLS, owned by the migration role. + assert (rls, force, owner) == (False, False, "patent_evidence_migration") + columns = {r[0] for r in migration.execute( + "SELECT column_name FROM information_schema.columns WHERE table_name = %s", (table,) + )} + assert not columns & {"organization_id", "case_id", "requested_by"} + grants = migration.execute( + """SELECT table_name, grantee, privilege_type FROM information_schema.role_table_grants + WHERE table_name IN ('legal_status_source_records', 'legal_status_assessments') + AND grantee IN ('patent_evidence_app', 'patent_evidence_worker', 'patent_evidence_platform') + ORDER BY 1, 2, 3""" + ).fetchall() + assert grants == [ + ("legal_status_assessments", "patent_evidence_app", "SELECT"), + ("legal_status_assessments", "patent_evidence_worker", "INSERT"), + ("legal_status_assessments", "patent_evidence_worker", "SELECT"), + ("legal_status_source_records", "patent_evidence_app", "SELECT"), + ("legal_status_source_records", "patent_evidence_worker", "INSERT"), + ("legal_status_source_records", "patent_evidence_worker", "SELECT"), + ] + + +def test_app_reads_but_cannot_write(migration: psycopg.Connection) -> None: + with _runtime("PE_TEST_APPLICATION_DATABASE_URL", "patent_evidence_app") as app: + assert app.execute("SELECT count(*) FROM legal_status_source_records").fetchone() == (0,) + with pytest.raises(psycopg.errors.InsufficientPrivilege): + _insert_record(app) + + +def test_worker_appends_but_cannot_rewrite(migration: psycopg.Connection) -> None: + record_id = uuid4() + with _runtime("PE_TEST_WORKER_DATABASE_URL", "patent_evidence_worker") as worker: + _insert_record(worker, id=record_id) + with pytest.raises(psycopg.errors.InsufficientPrivilege): + worker.execute("UPDATE legal_status_source_records SET outcome = 'found' WHERE id = %s", (record_id,)) + with pytest.raises(psycopg.errors.InsufficientPrivilege): + worker.execute("DELETE FROM legal_status_source_records WHERE id = %s", (record_id,)) + + +def test_found_records_must_carry_extract_and_hash(migration: psycopg.Connection) -> None: + with _runtime("PE_TEST_WORKER_DATABASE_URL", "patent_evidence_worker") as worker: + with pytest.raises(psycopg.errors.CheckViolation): + _insert_record(worker, outcome="found") + with pytest.raises(psycopg.errors.CheckViolation): + _insert_record(worker, outcome="found", extract=json.dumps({}), sha="not-a-sha") + + +# ------------------------------------------------------------- refresher end to end + + +def _odp_transport(body: dict | None, status: int = 200) -> httpx.MockTransport: + def handler(request: httpx.Request) -> httpx.Response: + if status != 200: + return httpx.Response(status) + return httpx.Response(200, json=body) + + return httpx.MockTransport(handler) + + +def _ops_transport(payload: dict) -> httpx.MockTransport: + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/auth/accesstoken"): + return httpx.Response(200, json={"access_token": "t", "expires_in": "1199"}) + return httpx.Response(200, json=payload) + + return httpx.MockTransport(handler) + + +async def _refresh(publication: str, *, odp=None, ops=None, ep_filing_date=None, storage_dir: Path): + engine = create_engine(async_postgres_url("PE_TEST_WORKER_DATABASE_URL", "patent_evidence_worker")) + try: + refresher = LegalStatusRefresher( + create_worker_session_factory(engine), + ObjectStorageClient(storage_dir), + odp=odp, + ops=ops, + clock=lambda: NOW, + ) + return await refresher.refresh(publication, as_of=AS_OF, ep_filing_date=ep_filing_date) + finally: + await engine.dispose() + + +def test_refresh_us_stores_reproducible_assessment(migration: psycopg.Connection, tmp_path: Path) -> None: + fixture = json.loads((FIXTURES / "us" / "US-7252747.json").read_text()) + odp = UsptoOdpClient("k", transport=_odp_transport({"count": 1, "patentFileWrapperDataBag": [fixture]}), backoff_seconds=0) + result = asyncio.run(_refresh("US-7252747-B2", odp=odp, storage_dir=tmp_path)) + assert (result.outcome, result.status) == ("found", "expired") + + record = migration.execute( + """SELECT source, outcome, extract, raw_sha256, raw_storage_key, raw_size_bytes, source_data_as_of + FROM legal_status_source_records WHERE id = %s""", + (result.source_record_id,), + ).fetchone() + source, outcome, extract, raw_sha, raw_key, raw_size, data_as_of = record + assert (source, outcome) == ("uspto_odp", "found") + # Exact response bytes are archived, content-addressed. + raw = (tmp_path / raw_key).read_bytes() + assert len(raw) == raw_size and raw_key.endswith(f"{raw_sha}.json") + assert data_as_of is not None + + stored = migration.execute( + "SELECT status, rules_version, content_sha256, term FROM legal_status_assessments WHERE id = %s", + (result.assessment_id,), + ).fetchone() + assert stored[:2] == ("expired", RULES_VERSION) + assert stored[3]["expiry_date"] == "2023-07-25" + # Reproducible from the database alone: re-assessing the stored extract gives the same hash. + again = assess_us(extract, publication_number="US-7252747-B2", as_of=AS_OF) + assert content_sha256(again) == stored[2] + + +def test_refresh_ep_stores_country_evidence(migration: psycopg.Connection, tmp_path: Path) -> None: + payload = json.loads((FIXTURES / "ep" / "EP-1801605-B1.json").read_text()) + ops = EpoOpsLegalClient("id", "s", transport=_ops_transport(payload), backoff_seconds=0) + result = asyncio.run( + _refresh("EP-1801605-B1", ops=ops, ep_filing_date=date(2006, 12, 22), storage_dir=tmp_path) + ) + assert result.status == "lapsed" + extract, countries, sha = migration.execute( + """SELECT r.extract, a.countries, a.content_sha256 FROM legal_status_assessments a + JOIN legal_status_source_records r ON r.id = a.source_record_id WHERE a.id = %s""", + (result.assessment_id,), + ).fetchone() + assert {c["country"] for c in countries} >= {"DE", "FR", "GB"} + again = assess_ep(extract, publication_number="EP-1801605-B1", filing_date=date(2006, 12, 22), as_of=AS_OF) + assert content_sha256(again) == sha + + +def test_not_found_is_recorded_without_any_status(migration: psycopg.Connection, tmp_path: Path) -> None: + odp = UsptoOdpClient("k", transport=_odp_transport({"count": 0}), backoff_seconds=0) + result = asyncio.run(_refresh("US-99999999-B2", odp=odp, storage_dir=tmp_path)) + assert (result.outcome, result.assessment_id, result.status) == ("not_found", None, None) + assert migration.execute( + "SELECT outcome, extract, raw_sha256 FROM legal_status_source_records WHERE id = %s", + (result.source_record_id,), + ).fetchone() == ("not_found", None, None) + assert migration.execute("SELECT count(*) FROM legal_status_assessments").fetchone() == (0,) + + +def test_source_failure_writes_nothing(migration: psycopg.Connection, tmp_path: Path) -> None: + odp = UsptoOdpClient("k", transport=_odp_transport(None, status=500), backoff_seconds=0) + with pytest.raises(LegalStatusSourceError): + asyncio.run(_refresh("US-7252747-B2", odp=odp, storage_dir=tmp_path)) + for table in TABLES: + assert migration.execute(f"SELECT count(*) FROM {table}").fetchone() == (0,) diff --git a/apps/api/tests/unit/test_legal_status_serialize.py b/apps/api/tests/unit/test_legal_status_serialize.py new file mode 100644 index 0000000..700eb1c --- /dev/null +++ b/apps/api/tests/unit/test_legal_status_serialize.py @@ -0,0 +1,49 @@ +"""Extract trimming and canonical hashing for stored assessments (ADR 0006).""" + +from __future__ import annotations + +import json +from datetime import date +from pathlib import Path + +from modules.legal_status.serialize import ( + RULES_VERSION, + assessment_to_dict, + content_sha256, + ep_extract, + us_extract, +) +from modules.legal_status.us import assess_us + +FIXTURES = Path(__file__).resolve().parents[4] / "fixtures" / "legal-status" + + +def test_extracts_are_idempotent_on_the_approved_fixtures() -> None: + """The fixtures were produced by the same trimming, so extracting them again is a no-op.""" + for path in (FIXTURES / "us").glob("*.json"): + record = json.loads(path.read_text()) + assert us_extract(record) == record, path.name + for path in (FIXTURES / "ep").glob("*.json"): + payload = json.loads(path.read_text()) + assert ep_extract(payload) == payload, path.name + + +def test_us_extract_drops_fields_the_rules_do_not_read() -> None: + record = json.loads((FIXTURES / "us" / "US-7252747.json").read_text()) + record["inventorBag"] = [{"firstName": "A", "lastName": "B"}] + record["eventDataBag"].append({"eventCode": "CTNF", "eventDescriptionText": "Non-Final Rejection"}) + extract = us_extract(record) + assert "inventorBag" not in extract + assert all(e["eventCode"] != "CTNF" for e in extract["eventDataBag"]) + + +def test_content_hash_is_stable_and_sensitive() -> None: + record = json.loads((FIXTURES / "us" / "US-7252747.json").read_text()) + a = assess_us(record, publication_number="US-7252747-B2", as_of=date(2026, 10, 4)) + b = assess_us(record, publication_number="US-7252747-B2", as_of=date(2026, 10, 4)) + assert content_sha256(a) == content_sha256(b) + c = assess_us(record, publication_number="US-7252747-B2", as_of=date(2020, 1, 1)) + assert content_sha256(a) != content_sha256(c) + body = assessment_to_dict(a) + assert body["rules_version"] == RULES_VERSION + assert body["term"]["expiry_date"] == "2023-07-25" diff --git a/apps/worker/src/patent_evidence_worker/legal_status_refresh.py b/apps/worker/src/patent_evidence_worker/legal_status_refresh.py new file mode 100644 index 0000000..f33dd93 --- /dev/null +++ b/apps/worker/src/patent_evidence_worker/legal_status_refresh.py @@ -0,0 +1,200 @@ +"""Fetch, archive, record and assess public patent legal status (ADR 0006). + +Flow for one publication: + +1. query the source (USPTO ODP for US, EPO OPS INPADOC for EP); +2. archive the exact response bytes in object storage, content-addressed by SHA-256; +3. insert a ``legal_status_source_records`` row (``not_found`` is recorded too); +4. assess **from the stored extract**, so the result is reproducible from the database; +5. insert a ``legal_status_assessments`` row with the canonical content hash. + +If the source fails, the error propagates and nothing is written: no status +is ever fabricated. These tables are global and carry no tenant data, so the +worker needs no organization context here. +""" + +from __future__ import annotations + +import hashlib +import json +import uuid +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, date, datetime +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from adapters.legal_status.clients import EpoOpsLegalClient, SourceResponse, UsptoOdpClient +from adapters.object_storage.client import ObjectStorageClient +from modules.legal_status.ep import assess_ep +from modules.legal_status.models import LegalStatusAssessment +from modules.legal_status.serialize import ( + RULES_VERSION, + assessment_to_dict, + content_sha256, + ep_extract, + us_extract, +) +from modules.legal_status.us import assess_us + + +@dataclass(frozen=True) +class RefreshResult: + publication_number: str + source_record_id: uuid.UUID + outcome: str # "found" | "not_found" + assessment_id: uuid.UUID | None + status: str | None + + +def parse_publication_number(publication_number: str) -> tuple[str, str, str]: + """``US-7252747-B2`` -> ("US", "7252747", "B2"); only US and EP are supported.""" + parts = publication_number.split("-") + if len(parts) != 3 or parts[0] not in ("US", "EP") or not parts[1].isdigit(): + raise ValueError(f"unsupported publication number: {publication_number!r}") + return parts[0], parts[1], parts[2] + + +def raw_storage_key(source: str, sha256: str) -> str: + return f"public/legal-status/{source}/{sha256}.json" + + +class LegalStatusRefresher: + def __init__( + self, + session_factory: async_sessionmaker[AsyncSession], + storage: ObjectStorageClient, + *, + odp: UsptoOdpClient | None = None, + ops: EpoOpsLegalClient | None = None, + clock: Callable[[], datetime] | None = None, + ) -> None: + self._session_factory = session_factory + self._storage = storage + self._odp = odp + self._ops = ops + self._clock = clock or (lambda: datetime.now(UTC)) + + async def refresh( + self, publication_number: str, *, as_of: date, ep_filing_date: date | None = None + ) -> RefreshResult: + country, number, kind = parse_publication_number(publication_number) + if country == "US": + if self._odp is None: + raise RuntimeError("USPTO ODP client is not configured") + source = "uspto_odp" + response = await self._odp.file_wrapper_response(number) + extract = us_extract(response.data) if response.data is not None else None + data_as_of = extract.get("lastIngestionDateTime") if extract else None + else: + if self._ops is None: + raise RuntimeError("EPO OPS client is not configured") + if ep_filing_date is None: + raise ValueError("EP assessment needs the EP filing date (term start)") + source = "epo_ops_inpadoc" + response = await self._ops.legal_response(country, number, kind) + extract = ep_extract(response.data) if response.data is not None else None + data_as_of = None # OPS reports no data snapshot time + + retrieved_at = self._clock() + raw_sha, raw_key = await self._archive(source, response) + + assessment: LegalStatusAssessment | None = None + if extract is not None: + if country == "US": + assessment = assess_us(extract, publication_number=publication_number, as_of=as_of) + else: + assessment = assess_ep( + extract, + publication_number=publication_number, + filing_date=ep_filing_date, # type: ignore[arg-type] + as_of=as_of, + ) + + record_id = uuid.uuid4() + assessment_id: uuid.UUID | None = None + async with self._session_factory() as session, session.begin(): + await session.execute( + text( + """INSERT INTO legal_status_source_records + (id, publication_number, source, request_ref, retrieved_at, source_data_as_of, + outcome, extract, raw_sha256, raw_size_bytes, raw_storage_key) + VALUES (:id, :pub, :source, :request_ref, :retrieved_at, :data_as_of, + :outcome, CAST(:extract AS jsonb), :raw_sha, :raw_size, :raw_key)""" + ), + { + "id": record_id, + "pub": publication_number, + "source": source, + "request_ref": response.request_ref, + "retrieved_at": retrieved_at, + "data_as_of": _parse_timestamp(data_as_of), + "outcome": "found" if extract is not None else "not_found", + "extract": json.dumps(extract, ensure_ascii=False) if extract is not None else None, + "raw_sha": raw_sha, + "raw_size": len(response.raw) if response.raw is not None else None, + "raw_key": raw_key, + }, + ) + if assessment is not None: + assessment_id = uuid.uuid4() + body = assessment_to_dict(assessment) + await session.execute( + text( + """INSERT INTO legal_status_assessments + (id, source_record_id, publication_number, jurisdiction, as_of, rules_version, + status, term, countries, evidence, review_reasons, content_sha256, assessed_at) + VALUES (:id, :record, :pub, :jurisdiction, :as_of, :rules_version, :status, + CAST(:term AS jsonb), CAST(:countries AS jsonb), CAST(:evidence AS jsonb), + CAST(:reasons AS jsonb), :sha, :assessed_at)""" + ), + { + "id": assessment_id, + "record": record_id, + "pub": publication_number, + "jurisdiction": assessment.jurisdiction, + "as_of": as_of, + "rules_version": RULES_VERSION, + "status": assessment.status, + "term": _dumps(body["term"]) if body["term"] is not None else None, + "countries": _dumps(body["countries"]), + "evidence": _dumps(body["evidence"]), + "reasons": _dumps(body["review_reasons"]), + "sha": content_sha256(assessment), + "assessed_at": retrieved_at, + }, + ) + return RefreshResult( + publication_number, + record_id, + "found" if extract is not None else "not_found", + assessment_id, + assessment.status if assessment else None, + ) + + async def _archive(self, source: str, response: SourceResponse) -> tuple[str | None, str | None]: + if response.raw is None: + return None, None + sha = hashlib.sha256(response.raw).hexdigest() + key = raw_storage_key(source, sha) + await self._storage.put_object(key, response.raw, "application/json") + return sha, key + + +def _dumps(value: Any) -> str: + return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + + +def _parse_timestamp(value: Any) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value)) + except ValueError: + return None + # ODP gives no UTC offset and documents none that we have verified. The column + # needs one, so UTC is assumed here; the original string is kept verbatim in + # ``extract.lastIngestionDateTime`` and is the authoritative value. + return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) diff --git a/docs/adr/0006-public-patent-legal-status-storage.md b/docs/adr/0006-public-patent-legal-status-storage.md new file mode 100644 index 0000000..4d84f9e --- /dev/null +++ b/docs/adr/0006-public-patent-legal-status-storage.md @@ -0,0 +1,101 @@ +# ADR 0006: 公开专利法律状态数据的存储方案 + +- Status: Accepted（产品负责人于 2026-10-04 授权工程方设计） +- Date: 2026-10-04 + +## Context + +`modules/legal_status`（#11）可以判定美国、欧洲专利在某一分析日期的法律状态。判定结果需要持久化， +供 FTO 分析引用和封存。这些数据与现有业务数据有三点不同： + +1. **公开、跨租户共享**：同一件专利的法律状态对所有机构都一样，按机构重复抓取既浪费配额， + 也会让不同机构看到的“同一事实”不一致。 +2. **会随时间变化**：同一件专利今天有效、明年可能失效；数据源本身也有延迟（ODP 记录的 + `lastIngestionDateTime` 实测为数月之前）。 +3. **要作为证据**：FTO 结论引用某个状态时，必须能说明“依据哪次查询、哪份原始响应、哪版规则”。 + +现有表都按机构隔离（FORCE RLS + `app.current_organization_id`），不能直接容纳跨租户共享的公开数据。 +仓库已有不按机构隔离的全局表（如 `global_identities`），靠精确授权保护；集成测试只要求机构表启用 RLS。 + +## Decision + +### 1. 两张全局表，只追加 + +公开数据放在不带 `organization_id` 的全局表中。运行时角色只授予 `SELECT` / `INSERT`， +不授予 `UPDATE` / `DELETE`（与 `source_result_snapshots`、审计表相同的只追加方式）。 + +**`legal_status_source_records`：每次查询数据源的记录** + +| 列 | 说明 | +|---|---| +| `id` | UUID 主键 | +| `publication_number` | 规范化公开号，如 `US-7252747-B2`、`EP-1801605-B1` | +| `source` | `uspto_odp` / `epo_ops_inpadoc` | +| `request_ref` | 请求的端点与参数（不含凭据） | +| `retrieved_at` | 本平台取数时间 | +| `source_data_as_of` | 数据源自报的数据时间（ODP `lastIngestionDateTime`；OPS 无此字段时为空） | +| `outcome` | `found` / `not_found`。“查不到”也记录：它证明查过，但不代表失效 | +| `extract` | 规则实际读取的字段（与 fixtures 的裁剪口径相同），`not_found` 时为空 | +| `raw_sha256`、`raw_size_bytes`、`raw_storage_key` | 原始响应字节的哈希、大小与对象存储位置 | + +**`legal_status_assessments`：规则对某次查询记录的判定** + +| 列 | 说明 | +|---|---| +| `id` | UUID 主键 | +| `source_record_id` | 外键，指向所依据的查询记录 | +| `publication_number`、`jurisdiction` | `US` / `EP` | +| `as_of` | 分析日期 | +| `rules_version` | 规则版本（如 `legal-status-rules/1`）。规则改变时产生新版本的新行，旧判定保留 | +| `status` | `presumed_in_force` / `lapsed` / `expired` / `undetermined` | +| `term`、`countries`、`evidence`、`review_reasons` | 判定的完整内容（JSONB） | +| `content_sha256` | 判定内容规范化 JSON 的 SHA-256，供封存时绑定 | +| `assessed_at` | 判定时间 | + +唯一约束：`(source_record_id, as_of, rules_version)`，同一输入、同一日期、同一规则只判定一次。 + +### 2. 原始响应存对象存储，数据库只存规则所需字段 + +原始响应按内容寻址存放：`public/legal-status//.json`（相同内容只存一份）。 +数据库保存其哈希，可证明判定所依据的原始字节；`extract` 只含规则读取的字段。 + +原始响应可能包含发明人、代理人等公开的个人信息。是否保留原始全文、保留多久，涉及个人信息处理的 +合法性与最小必要原则（中国《个人信息保护法》对已公开个人信息的处理有专门规定；涉及欧洲个人时还需考虑 GDPR）。 +**该保留策略须由合规人员确认**。在确认之前，实现只把原始响应存入对象存储、不复制进数据库， +并保留按前缀整体删除的能力（删除原始文件不影响 `raw_sha256` 作为事后核对依据）。 + +### 3. 不启用 RLS，但机构关联留在机构表里 + +两张全局表不含任何机构信息，所以不启用按机构的 RLS。 + +“哪个机构在关注哪件专利”本身是敏感信息（会暴露客户的产品方向和竞争对手）。因此： + +- 全局表中**不得**出现请求方机构、案件或用户字段； +- 机构对某个判定的引用，放在将来的机构表中（FORCE RLS），只保存 `assessment_id` 与其 + `content_sha256`，并在证据快照封存时一并绑定。 + +### 4. 角色与权限 + +| 角色 | `legal_status_source_records` | `legal_status_assessments` | +|---|---|---| +| `patent_evidence_worker`（抓取与判定） | SELECT, INSERT | SELECT, INSERT | +| `patent_evidence_app`（API 读取） | SELECT | SELECT | +| `patent_evidence_platform` | 无 | 无 | +| `patent_evidence_migration` | 所有者 | 所有者 | + +API 不直接写入公开数据，抓取与判定只在 worker 中进行，避免请求路径中的外部调用与配额消耗。 + +### 5. 刷新策略 + +- 每次判定绑定一条查询记录，并同时记录 `retrieved_at` 与 `source_data_as_of`。两者含义不同： + 刚取到的数据，其数据源时间可能是数月之前。 +- 某件专利的最新查询记录早于可配置的刷新间隔时，重新查询；封存 FTO 证据快照前，对涉及的专利强制重新查询。 + 刷新间隔的默认值在首次全量运行、测得配额消耗之后确定，本 ADR 不预设数字。 +- 新查询不覆盖旧记录：历史状态可追溯，“当时依据的是什么”始终可答。 + +## Consequences + +- 新增迁移 `0018_legal_status_store`：两张表、索引、授权；集成测试验证授权边界与只追加。 +- `modules/legal_status` 增加规则版本常量与判定的规范化序列化（用于 `content_sha256`）。 +- worker 增加“查询 → 存原始响应 → 记录 → 判定 → 记录”的流程；数据源失败时不写入判定（不伪造状态）。 +- 尚未包含：任务队列与触发方式、机构侧引用表、API 读取接口、刷新间隔默认值、原始响应保留期限。 diff --git a/modules/legal_status/serialize.py b/modules/legal_status/serialize.py new file mode 100644 index 0000000..8121d01 --- /dev/null +++ b/modules/legal_status/serialize.py @@ -0,0 +1,99 @@ +"""Extracts, canonical serialization and content hashing for stored assessments (ADR 0006). + +Assessments are always computed from the stored ``extract`` (never from the raw +response), so any stored assessment can be reproduced from the database alone. +The extract keeps exactly the fields the rules read and drops personal data; +it is the same trimming that produced ``fixtures/legal-status`` and was +checked to give identical results to the untrimmed responses. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict +from datetime import date +from typing import Any + +from modules.legal_status.models import LegalStatusAssessment + +# Bump when any rule in us.py / ep.py changes meaning; old assessments stay as recorded. +RULES_VERSION = "legal-status-rules/1" + +_US_META_FIELDS = ( + "filingDate", + "grantDate", + "patentNumber", + "applicationStatusDescriptionText", + "applicationStatusDate", + "effectiveFilingDate", +) +_US_EVENT_PREFIXES = ("M155", "M255", "M355", "EXP", "REM", "DIST", "P574") +_EP_KEEP_CODES = {"AK", "PGFP", "PGRI", "26N", "P01"} +_EP_EVENT_FIELDS = ("@code", "@desc", "@infl", "ops:pre", "ops:L001EP", "ops:L007EP") + + +def us_extract(record: dict[str, Any]) -> dict[str, Any]: + meta = record.get("applicationMetaData") or {} + pta = record.get("patentTermAdjustmentData") + return { + "applicationNumberText": record.get("applicationNumberText"), + "lastIngestionDateTime": record.get("lastIngestionDateTime"), + "applicationMetaData": {k: meta.get(k) for k in _US_META_FIELDS}, + "eventDataBag": [ + e for e in record.get("eventDataBag") or [] if (e.get("eventCode") or "").startswith(_US_EVENT_PREFIXES) + ], + "parentContinuityBag": record.get("parentContinuityBag") or [], + "patentTermAdjustmentData": ( + {"adjustmentTotalQuantity": pta.get("adjustmentTotalQuantity")} if pta is not None else None + ), + } + + +def _legal_events(node: Any, out: list[dict[str, Any]]) -> None: + if isinstance(node, dict): + if "ops:legal" in node: + items = node["ops:legal"] + out.extend(items if isinstance(items, list) else [items]) + for value in node.values(): + _legal_events(value, out) + elif isinstance(node, list): + for value in node: + _legal_events(value, out) + + +def ep_extract(payload: dict[str, Any]) -> dict[str, Any]: + events: list[dict[str, Any]] = [] + _legal_events(payload, events) + kept = [ + {k: v for k, v in e.items() if k in _EP_EVENT_FIELDS} + for e in events + if str(e.get("@code", "")).strip() in _EP_KEEP_CODES + or str(e.get("@code", "")).strip().startswith("U") + or str(e.get("@infl", "")).strip() == "-" + ] + return {"ops:legal": kept} + + +def _jsonable(value: Any) -> Any: + if isinstance(value, date): + return value.isoformat() + if isinstance(value, dict): + return {k: _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(v) for v in value] + return value + + +def assessment_to_dict(assessment: LegalStatusAssessment) -> dict[str, Any]: + data = _jsonable(asdict(assessment)) + data["rules_version"] = RULES_VERSION + return data + + +def canonical_json(data: dict[str, Any]) -> str: + return json.dumps(data, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + + +def content_sha256(assessment: LegalStatusAssessment) -> str: + return hashlib.sha256(canonical_json(assessment_to_dict(assessment)).encode("utf-8")).hexdigest() diff --git a/scripts/test-postgres.sh b/scripts/test-postgres.sh index 8d3f75e..6ca98f8 100755 --- a/scripts/test-postgres.sh +++ b/scripts/test-postgres.sh @@ -47,6 +47,7 @@ uv sync --frozen --no-install-project apps/api/tests/integration/test_database_context.py \ apps/api/tests/integration/test_authentication_api.py \ apps/api/tests/integration/test_platform_provisioning_api.py \ - apps/api/tests/integration/test_organization_administration_api.py + apps/api/tests/integration/test_organization_administration_api.py \ + apps/api/tests/integration/test_legal_status_store.py .venv/bin/alembic -c apps/api/alembic.ini downgrade 0001_identity_tenancy .venv/bin/alembic -c apps/api/alembic.ini upgrade head