feat(ocr): add vnext figure ownership contracts

paperforge/worker/ocr_figure_vnext_state.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from .ocr_figure_vnext_types import ClaimProposal, OwnershipConflict, ResourceRef
+
+
+class OwnershipLedger:
+    def __init__(self) -> None:
+        self._owners: dict[ResourceRef, ResourceRef] = {}
+        self._journal: list[dict[str, object]] = []
+
+    def claim_assets(self, assets: list[ResourceRef], *, owner: ResourceRef, reason: str) -> None:
+        conflict = self.try_claim_assets(assets, owner=owner, reason=reason)
+        if conflict is not None:
+            raise ValueError(f"asset already owned: {conflict.resource}")
+
+    def try_claim_assets(
+        self, assets: list[ResourceRef], *, owner: ResourceRef, reason: str
+    ) -> OwnershipConflict | None:
+        for asset in assets:
+            current = self._owners.get(asset)
+            if current is not None and current != owner:
+                conflict = OwnershipConflict(
+                    resource=asset,
+                    current_owner=current,
+                    attempted_owner=owner,
+                    reason=reason,
+                )
+                self._journal.append({
+                    "action": "conflict",
+                    "resource": asset,
+                    "current_owner": current,
+                    "attempted_owner": owner,
+                    "reason": reason,
+                })
+                return conflict
+        for asset in assets:
+            self._owners[asset] = owner
+            self._journal.append({"action": "claim", "resource": asset, "owner": owner, "reason": reason})
+        return None
+
+    def owner_of(self, resource: ResourceRef) -> ResourceRef | None:
+        return self._owners.get(resource)
+
+    def owner_of_asset(self, *, page: int, block_id: int | str) -> ResourceRef | None:
+        return self.owner_of(ResourceRef(kind="asset", page=page, block_id=block_id))
+
+    def snapshot(self) -> list[dict[str, object]]:
+        return list(self._journal)
+
+
+@dataclass
+class FigurePipelineState:
+    corpus: object | None
+    candidate_index: object | None
+    ledger: OwnershipLedger
+    matches: list[dict] = field(default_factory=list)
+    unresolved: list[dict] = field(default_factory=list)
+    hypotheses: list[dict] = field(default_factory=list)
+    diagnostics: list[dict] = field(default_factory=list)
+
+    def accept_match(self, proposal: ClaimProposal, match_record: dict) -> None:
+        self.matches.append(match_record)
+        self.diagnostics.append({
+            "event": "match_accepted",
+            "pass_name": proposal.pass_name,
+            "figure_no": proposal.figure_no,
+            "reason": proposal.reason,
+            "resources": {
+                "legends": proposal.legends,
+                "assets": proposal.assets,
+                "groups": proposal.groups,
+            },
+        })

paperforge/worker/ocr_figure_vnext_types.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+
+@dataclass(frozen=True)
+class ResourceRef:
+    kind: Literal["legend", "asset", "group"]
+    page: int | None
+    block_id: str | None
+    group_id: str | None = None
+    figure_no: int | None = None
+    origin: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.page is not None:
+            object.__setattr__(self, "page", int(self.page))
+        if self.block_id is not None:
+            object.__setattr__(self, "block_id", str(self.block_id))
+        if self.group_id is not None:
+            object.__setattr__(self, "group_id", str(self.group_id))
+
+        if self.kind == "asset" and (self.page is None or self.block_id is None):
+            raise ValueError("asset ResourceRef requires page + block_id")
+        if self.kind == "legend" and (self.page is None or self.block_id is None):
+            raise ValueError("legend ResourceRef requires page + block_id")
+        if self.kind == "group" and (self.page is None or self.group_id is None):
+            raise ValueError("group ResourceRef requires page + group_id")
+
+
+@dataclass(frozen=True)
+class OwnershipConflict:
+    resource: ResourceRef
+    current_owner: ResourceRef | None
+    attempted_owner: ResourceRef | None
+    reason: str
+
+
+@dataclass
+class ClaimProposal:
+    pass_name: str
+    figure_no: int | None
+    claim_type: Literal["match", "reserve", "block", "unresolved_cluster", "composite_parent"]
+    legends: list[ResourceRef]
+    assets: list[ResourceRef]
+    groups: list[ResourceRef]
+    confidence: float
+    evidence_rank: int
+    reason: str
+    diagnostics: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class PassReport:
+    pass_name: str
+    proposals: list[ClaimProposal] = field(default_factory=list)
+    accepted: list[ClaimProposal] = field(default_factory=list)
+    rejected: list[ClaimProposal] = field(default_factory=list)
+    conflicts: list[OwnershipConflict] = field(default_factory=list)
+    invariant_errors: list[str] = field(default_factory=list)

tests/test_ocr_figure_vnext_state.py

@@ -0,0 +1,42 @@
+"""Ownership ledger and pipeline state tests for vnext figure contracts."""
+
+from __future__ import annotations
+
+from paperforge.worker.ocr_figure_vnext_types import ClaimProposal, ResourceRef
+from paperforge.worker.ocr_figure_vnext_state import FigurePipelineState, OwnershipLedger
+
+
+def test_ledger_rejects_double_ownership_for_same_asset_and_records_conflict() -> None:
+    ledger = OwnershipLedger()
+    asset = ResourceRef(kind="asset", page=3, block_id="42")
+    legend_a = ResourceRef(kind="legend", page=3, block_id="77", figure_no=1)
+    legend_b = ResourceRef(kind="legend", page=3, block_id="78", figure_no=2)
+
+    ledger.claim_assets([asset], owner=legend_a, reason="same_page_primary")
+    conflict = ledger.try_claim_assets([asset], owner=legend_b, reason="conflicting_match")
+
+    assert conflict is not None
+    assert conflict.resource == asset
+    assert conflict.current_owner == legend_a
+    assert any(entry["action"] == "conflict" for entry in ledger.snapshot())
+
+
+def test_pipeline_state_accept_match_records_diagnostic() -> None:
+    state = FigurePipelineState(corpus=None, candidate_index=None, ledger=OwnershipLedger())
+    proposal = ClaimProposal(
+        pass_name="primary_same_page",
+        figure_no=1,
+        claim_type="match",
+        legends=[ResourceRef(kind="legend", page=1, block_id="c1", figure_no=1)],
+        assets=[ResourceRef(kind="asset", page=1, block_id="a1")],
+        groups=[],
+        confidence=0.9,
+        evidence_rank=1,
+        reason="same_page_primary",
+        diagnostics={"evidence": ["test"]},
+    )
+
+    state.accept_match(proposal, {"figure_id": "Figure 1"})
+
+    assert state.matches == [{"figure_id": "Figure 1"}]
+    assert state.diagnostics[-1]["event"] == "match_accepted"

tests/test_ocr_figure_vnext_types.py

@@ -0,0 +1,21 @@
+"""Identity and normalization tests for vnext figure contracts."""
+
+from __future__ import annotations
+
+import pytest
+
+from paperforge.worker.ocr_figure_vnext_types import ResourceRef
+
+
+def test_resource_ref_rejects_page_agnostic_asset() -> None:
+    with pytest.raises(ValueError):
+        ResourceRef(kind="asset", page=None, block_id="42")
+
+
+def test_resource_ref_normalizes_block_id_type() -> None:
+    assert ResourceRef(kind="asset", page=1, block_id=42) == ResourceRef(kind="asset", page=1, block_id="42")
+
+
+def test_resource_ref_rejects_group_without_group_id() -> None:
+    with pytest.raises(ValueError):
+        ResourceRef(kind="group", page=1, block_id=None)