fix: resolve critical bugs from code review

docker-compose.yml

@@ -25,6 +25,8 @@ services:
       - patentflow-net
     ports:
       - "8000:8000"
+    volumes:
+      - patentflow-data:/app/data
     depends_on:
       redis:
         condition: service_healthy
@@ -50,6 +52,8 @@ services:
       - CELERY_CONCURRENCY=1
     networks:
       - patentflow-net
+    volumes:
+      - patentflow-data:/app/data
     depends_on:
       redis:
         condition: service_healthy
@@ -74,3 +78,4 @@ networks:
 
 volumes:
   redis-data:
+  patentflow-data:

src/api.py

@@ -115,6 +115,8 @@ def _redis_url() -> str:
     return os.getenv("REDIS_URL", "redis://localhost:6379/0")
 
 
+
+
 def _redis_client() -> redis.Redis:
     return redis.Redis.from_url(_redis_url(), decode_responses=True)
 
@@ -346,14 +348,19 @@ def add_memory(req: MemoryAddRequest) -> Dict[str, str]:
 @app.exception_handler(Exception)
 async def global_exception_handler(request, exc):
     """Global exception handler for all unhandled exceptions."""
+    import logging
     from fastapi.responses import JSONResponse
+
+    logger = logging.getLogger("patentflow.api")
+    logger.error("Unhandled exception on %s %s: %s", request.method, request.url.path, exc, exc_info=True)
+
     return JSONResponse(
         status_code=500,
         content={
             "status": "error",
-            "error": f"INTERNAL_SERVER_ERROR: {str(exc)}",
-            "detail": "An unexpected error occurred. Please try again or contact support."
-        }
+            "error": "INTERNAL_SERVER_ERROR",
+            "detail": "An unexpected error occurred. Please try again or contact support.",
+        },
     )
 
 

src/skills/verifier.py

@@ -243,5 +243,5 @@ def execute(
                 "markdown_table": "\n".join(md_lines),
                 "overall_risk": overall_risk
             },
-            warnings=[r["warnings"] for r in analysis_result if r["warnings"]]
+            warnings=[w for r in analysis_result for w in r["warnings"] if w]
         )

src/tasks.py

@@ -140,17 +140,14 @@ def parse_docs(
         except Exception:
             pass
 
-    claim_text = (
-        "A method for wireless communication, comprising: transmitting a downlink control information (DCI) format; "
-        "determining a timing offset K0; and receiving a physical downlink shared channel (PDSCH) based on the timing offset."
-    )
+    claim_text = ""
     if specification_text and len(specification_text.strip()) > 20:
         claim_text = specification_text.strip()
-    
-    prior_art_text = "D1 discloses a wireless communication system with fixed timing relations."
+
+    prior_art_text = ""
     if specification_text and len(specification_text) > 50:
         prior_art_text = (specification_text[:200] + "...").replace("\n", " ")
-    
+
     # Load mock office action if none provided (for demo purposes)
     if not office_action_text:
         try:

tests/test_skills.py

@@ -1,76 +1,177 @@
 from __future__ import annotations
 
-from src.skills import classify_claim, generate_claim_chart
-
-
-def test_classify_claim_method_independent() -> None:
-    claim = "A method for wireless communication comprising: receiving DCI; wherein the DCI indicates a timing offset."
-    out = classify_claim(claim)
-    assert out["category"] == "Method"
-    assert out["is_independent"] is True
-
-
-def test_classify_claim_apparatus_dependent() -> None:
-    claim = "A device according to claim 1, wherein the UE is configured to transmit HARQ-ACK."
-    out = classify_claim(claim)
-    assert out["category"] == "Apparatus"
-    assert out["is_independent"] is False
-
-
-def test_generate_claim_chart_basic() -> None:
-    claim = "receiving DCI; determining an offset; transmitting HARQ feedback"
-    d1 = "D1 describes receiving DCI and transmitting HARQ feedback."
-    out = generate_claim_chart(claim, d1)
-
-    assert out["status"] == "success"
-    chart = out["claim_chart"]
-    assert isinstance(chart, list)
-    assert len(chart) == 3
-    assert chart[0]["feature_id"] == "1.1"
-    assert chart[1]["feature_id"] == "1.2"
-    assert chart[2]["feature_id"] == "1.3"
-    assert chart[0]["claim_limitation"] == "receiving DCI"
-    assert chart[0]["assessment"] in {"✅ Yes", "⚠️ Partial", "❌ No (Difference)"}
-    assert "disclosure" in chart[0]
-    assert "attorney_remarks" in chart[0]
-
-
-def test_generate_claim_chart_supports_multiple_prior_arts() -> None:
-    claim = "receiving DCI; determining K0; transmitting HARQ feedback"
-    oa = (
-        "D1 discloses receiving DCI in paragraph [0052]. "
-        "D2 discloses K0 timing offset for PDSCH scheduling. "
-        "D3 discloses HARQ feedback signaling behavior."
-    )
-    out = generate_claim_chart(claim, "fallback prior art", office_action_text=oa)
-    assert out["status"] == "success"
-    assert out["cited_docs"] == ["D1", "D2", "D3"]
-    chart = out["claim_chart"]
-    assert len(chart) == 3
-    assert all("disclosure" in row for row in chart)
-    assert all("assessment" in row for row in chart)
-    assert all("[N/A]" not in str(row["disclosure"]) for row in chart)
-
-
-def test_generate_claim_chart_not_disclosed_is_english_only() -> None:
-    claim = "dynamic timing offset field"
-    oa = "D1 discloses static RRC timing configuration."
-    out = generate_claim_chart(claim, "", office_action_text=oa)
-    row = out["claim_chart"][0]
-    assert "Not disclosed." in row["disclosure"]
-    assert "未公开" not in row["disclosure"]
-
-
-def test_disclosed_without_marker_can_still_match() -> None:
-    claim = "base station transmits control info to schedule downlink data"
-    oa_no_marker = "D1 discloses receiving DCI in control channel signaling."
-    out_no_marker = generate_claim_chart(claim, "", office_action_text=oa_no_marker)
-    row_no_marker = out_no_marker["claim_chart"][0]
-    assert row_no_marker["disclosure"].startswith("D1 ")
-    assert row_no_marker["assessment"] in {"✅ Yes", "⚠️ Partial"}
-
-    oa_with_marker = 'D1 [0045] "The base station transmits control info to schedule downlink data."'
-    out_with_marker = generate_claim_chart(claim, "", office_action_text=oa_with_marker)
-    row_with_marker = out_with_marker["claim_chart"][0]
-    assert "[0045]" in row_with_marker["disclosure"]
-    assert row_with_marker["assessment"] in {"✅ Yes", "⚠️ Partial"}
+from src.skills import ClaimChartGenerator, TranslationVerifier
+
+
+# ---------------------------------------------------------------------------
+# ClaimChartGenerator — deterministic tokenization (no LLM needed)
+# ---------------------------------------------------------------------------
+
+class TestClaimTokenizer:
+    """Test the deterministic claim splitting logic."""
+
+    def _tokenize(self, claim_text: str):
+        gen = ClaimChartGenerator()
+        return gen._tokenize_claim(claim_text)
+
+    def test_comprising_split(self) -> None:
+        claim = (
+            "A method for wireless communication, comprising: "
+            "receiving DCI; determining an offset; transmitting HARQ feedback."
+        )
+        features = self._tokenize(claim)
+        assert len(features) == 4
+        assert features[0]["feature_id"] == "1.1"
+        assert "comprising" in features[0]["limitation"].lower()
+        assert features[1]["feature_id"] == "1.2"
+        assert "receiving DCI" in features[1]["limitation"]
+
+    def test_no_comprising_fallback(self) -> None:
+        claim = "receiving DCI; determining an offset; transmitting HARQ feedback"
+        features = self._tokenize(claim)
+        assert len(features) == 3
+        assert features[0]["feature_id"] == "1.1"
+        assert features[0]["limitation"] == "receiving DCI"
+
+    def test_empty_claim(self) -> None:
+        assert self._tokenize("") == []
+        assert self._tokenize("   ") == []
+
+    def test_single_feature(self) -> None:
+        claim = "A method for wireless communication, comprising: receiving DCI."
+        features = self._tokenize(claim)
+        assert len(features) == 2
+        assert "comprising" in features[0]["limitation"].lower()
+
+
+# ---------------------------------------------------------------------------
+# ClaimChartGenerator — cited document extraction
+# ---------------------------------------------------------------------------
+
+class TestCitedDocExtraction:
+    def _extract(self, oa_text: str):
+        gen = ClaimChartGenerator()
+        return gen._extract_cited_docs(oa_text)
+
+    def test_basic_d1_d2(self) -> None:
+        oa = "Document D1 discloses... Document D2 describes..."
+        assert self._extract(oa) == ["D1", "D2"]
+
+    def test_multiple_docs(self) -> None:
+        oa = "D1 discloses receiving DCI. D2 discloses K0 timing. D3 discloses HARQ."
+        assert self._extract(oa) == ["D1", "D2", "D3"]
+
+    def test_no_docs(self) -> None:
+        assert self._extract("") == []
+        assert self._extract("no documents here") == []
+
+
+# ---------------------------------------------------------------------------
+# ClaimChartGenerator — snippet extraction
+# ---------------------------------------------------------------------------
+
+class TestSnippetExtraction:
+    def _extract_snippets(self, oa_text: str, doc_id: str):
+        gen = ClaimChartGenerator()
+        return gen._extract_snippets_for_doc(oa_text, doc_id)
+
+    def test_basic_snippet(self) -> None:
+        oa = "D1 discloses receiving DCI in paragraph [0052]. D2 discloses K0."
+        snippets = self._extract_snippets(oa, "D1")
+        assert len(snippets) > 0
+        assert any("DCI" in s for s in snippets)
+
+    def test_empty_oa(self) -> None:
+        assert self._extract_snippets("", "D1") == []
+
+
+# ---------------------------------------------------------------------------
+# ClaimChartGenerator — full execute (mock LLM, deterministic fallback)
+# ---------------------------------------------------------------------------
+
+class TestClaimChartExecute:
+    def test_basic_chart(self) -> None:
+        gen = ClaimChartGenerator()
+        result = gen.execute(
+            claim_text="receiving DCI; determining an offset; transmitting HARQ feedback",
+            prior_art_text="D1 describes receiving DCI and transmitting HARQ feedback.",
+            office_action_text="D1 discloses receiving DCI in paragraph [0052].",
+        )
+        assert result.status in {"success", "partial"}
+        chart = result.data.get("chart", [])
+        assert len(chart) == 3
+        assert chart[0]["feature_id"] == "1.1"
+        assert chart[0]["assessment"] in {"Yes", "No", "Partial"}
+
+    def test_multiple_prior_arts(self) -> None:
+        gen = ClaimChartGenerator()
+        oa = (
+            "D1 discloses receiving DCI in paragraph [0052]. "
+            "D2 discloses K0 timing offset for PDSCH scheduling. "
+            "D3 discloses HARQ feedback signaling behavior."
+        )
+        result = gen.execute(
+            claim_text="receiving DCI; determining K0; transmitting HARQ feedback",
+            prior_art_text="fallback prior art",
+            office_action_text=oa,
+        )
+        assert result.status in {"success", "partial"}
+        assert set(result.data.get("cited_docs", [])) >= {"D1", "D2", "D3"}
+        chart = result.data.get("chart", [])
+        assert len(chart) == 3
+        for row in chart:
+            assert "d1_disclosure" in row or "d1_disclosure" in row
+
+
+# ---------------------------------------------------------------------------
+# TranslationVerifier — glossary-based risk detection
+# ---------------------------------------------------------------------------
+
+class TestTranslationVerifier:
+    def test_safe_translation(self) -> None:
+        verifier = TranslationVerifier()
+        result = verifier.execute(
+            original_cn="一种无线通信方法，包括发送下行控制信息",
+            target_en="A method for wireless communication, comprising transmitting Downlink Control Information",
+            back_cn="一种无线通信方法，包括发送下行控制信息",
+        )
+        assert result.status == "success"
+        assert result.data["overall_risk"] in {"Safe", "Warning"}
+
+    def test_lethal_mismatch(self) -> None:
+        """'包括' should never map to 'consisting of' — that's a lethal mismatch."""
+        verifier = TranslationVerifier()
+        result = verifier.execute(
+            original_cn="一种方法，包括发送DCI",
+            target_en="A method, consisting of transmitting DCI",
+            back_cn="一种方法，组成发送DCI",
+        )
+        assert result.data["overall_risk"] == "CRITICAL"
+
+    def test_missing_comprising(self) -> None:
+        """Missing 'comprising' when CN has '包括' is critical."""
+        verifier = TranslationVerifier()
+        result = verifier.execute(
+            original_cn="一种方法，包括发送DCI",
+            target_en="A method, including transmitting DCI",
+            back_cn="一种方法，包括发送DCI",
+        )
+        # 'including' is not 'comprising' — should flag
+        rows = result.data.get("rows", [])
+        assert len(rows) > 0
+        # At least one row should have warnings
+        has_warning = any(r.get("warnings") for r in rows)
+        has_critical = result.data["overall_risk"] == "CRITICAL"
+        assert has_warning or has_critical
+
+    def test_empty_input(self) -> None:
+        verifier = TranslationVerifier()
+        result = verifier.execute(original_cn="", target_en="", back_cn="")
+        assert result.status == "success"
+
+    def test_highlight_cn_terms(self) -> None:
+        verifier = TranslationVerifier()
+        highlighted = verifier._highlight_cn_terms("一种方法，包括发送DCI，其中所述终端被配置为接收")
+        assert "**包括**" in highlighted
+        assert "**其中**" in highlighted
+        assert "**配置为**" in highlighted or "**被配置为**" in highlighted