Minor updates and fixes for split doc collection

compass/extraction/apply.py

@@ -81,9 +81,9 @@ async def check_for_relevant_text(
     legal_text_validator = (
         LegalTextValidator(
             tech=tech,
+            doc=doc,
             llm_service=model_config.llm_service,
             usage_tracker=usage_tracker,
-            doc_is_from_ocr=doc.attrs.get("from_ocr", False),
             **model_config.llm_call_kwargs,
         )
         if doc.attrs.get("check_if_legal_doc", True)

compass/pipeline/collection/persistence.py

@@ -23,7 +23,9 @@
 COLLECTION_MANIFEST_FILENAME = "collection_manifest.json"
 
 
-def build_collection_manifest(tech, jurisdictions):
+def build_collection_manifest(
+    tech, jurisdictions, time_start_utc, num_jurisdictions_searched
+):
     """Build the serialized collection manifest payload
 
     Parameters
@@ -35,16 +37,29 @@ def build_collection_manifest(tech, jurisdictions):
         Dictionary mapping jurisdiction full names to serialized
         collection metadata for each jurisdiction, including
         jurisdiction identifiers and the persisted document records.
+    time_start_utc : datetime.datetime
+        UTC datetime when the collection process started, used to
+        calculate elapsed time for the manifest metadata.
+    num_jurisdictions_searched : int
+        Number of jurisdictions that were searched during the collection
+        process, included in the manifest for informational purposes.
 
     Returns
     -------
     dict
         Collection manifest as a dictionary, ready to be serialized and
         written to disk.
     """
+    time_end_utc = datetime.now(UTC)
+    time_elapsed = time_end_utc - time_start_utc
     return {
         "tech": tech,
-        "created_at": datetime.now(UTC).isoformat(),
+        "time_start_utc": time_start_utc.isoformat(),
+        "time_end_utc": time_end_utc.isoformat(),
+        "total_time": time_elapsed.total_seconds(),
+        "total_time_string": str(time_elapsed),
+        "num_jurisdictions_searched": num_jurisdictions_searched,
+        "num_jurisdictions_found": len(jurisdictions),
         "jurisdictions": jurisdictions,
     }
 
@@ -344,4 +359,6 @@ def _load_collection_manifest_from_shards(manifest_fp, expected_tech):
         )
         jurisdictions.append(resolve_all_paths(collection_info, manifest_dir))
 
-    return build_collection_manifest(expected_tech, jurisdictions)
+    return build_collection_manifest(
+        expected_tech, jurisdictions, datetime.now(UTC), len(jurisdictions)
+    )

compass/pipeline/coordinator.py

@@ -268,16 +268,16 @@ async def run(self, jurisdictions_df):
 
         collection_infos = await asyncio.gather(*tasks)
         manifest = build_collection_manifest(
-            self.runtime.tech, collection_infos
+            self.runtime.tech,
+            list(filter(None, collection_infos)),
+            start_date,
+            len(jurisdictions_df),
         )
         manifest_fp = await write_collection_manifest(
             self.runtime.dirs.out, manifest
         )
-        time_elapsed = datetime.now(UTC) - start_date
         collection_msg = compile_collection_summary_message(
-            manifest_fp,
-            manifest,
-            total_seconds=time_elapsed.total_seconds(),
+            manifest_fp, manifest
         )
         for sub_msg in collection_msg.split("\n"):
             logger.info(sub_msg)

compass/pipeline/jurisdiction.py

@@ -154,14 +154,13 @@ async def collect(self, *, relative_to=None):
         collection_info = await self.collection_workflow.execute(
             eager_extract=False, relative_to=relative_to
         )
-        shard_fp = await write_collection_manifest_shard(
-            self.runtime.dirs.jurisdiction_dbs, collection_info
-        )
-        logger.info(
-            "Collection manifest shard for %s stored here: '%s'",
+
+        await _safe_shard_write(
+            self.runtime.dirs.jurisdiction_dbs,
+            collection_info,
             self.jurisdiction.full_name,
-            shard_fp,
         )
+
         logger.info(
             "Completed collection for jurisdiction: %s",
             self.jurisdiction.full_name,
@@ -323,3 +322,20 @@ async def _record_jurisdiction_info(
     await JurisdictionUpdater.call(
         jurisdiction, extraction_context, seconds_elapsed, usage_tracker
     )
+
+
+async def _safe_shard_write(shard_dir, collection_info, jur_name):
+    """Safely write a collection manifest shard"""
+    try:
+        shard_fp = await write_collection_manifest_shard(
+            shard_dir, collection_info
+        )
+        logger.info(
+            "Collection manifest shard for %s stored here: '%s'",
+            jur_name,
+            shard_fp,
+        )
+    except Exception:
+        logger.exception(
+            "Failed to write collection manifest shard for %s", jur_name
+        )

compass/scripts/download.py

@@ -836,14 +836,26 @@ async def _contains_relevant_text(
         usage_tracker=usage_tracker,
         **kwargs,
     )
+    doc.attrs["found_any_extraction_text"] = found_text
     if found_text:
-        logger.debug("Detected relevant text; parsing date...")
+        logger.info(
+            "Detected some relevant extraction text for document from "
+            "source: %s ; parsing date...",
+            doc.attrs.get("source", "Unknown"),
+        )
         date_model_config = model_configs.get(
             LLMTasks.DATE_EXTRACTION, model_configs[LLMTasks.DEFAULT]
         )
         doc = await extract_date(
             doc, date_model_config, usage_tracker=usage_tracker
         )
+    else:
+        logger.info(
+            "Did not detect relevant extraction text for document from "
+            "source: %s",
+            doc.attrs.get("source", "Unknown"),
+        )
+
     return found_text
 
 

compass/services/threaded.py

@@ -197,7 +197,7 @@ def release_resources(self):
         self._td.cleanup()
         super().release_resources()
 
-    async def process(self, doc, file_content, make_name_unique=False):
+    async def process(self, doc, file_content, make_name_unique=True):
         """Write URL doc to file asynchronously
 
         Parameters
@@ -212,7 +212,7 @@ async def process(self, doc, file_content, make_name_unique=False):
             for PDF file.
         make_name_unique : bool, optional
             Option to make file name unique by adding a UUID at the end
-            of the file name. By default, ``False``.
+            of the file name. By default, ``True``.
 
         Returns
         -------
@@ -237,7 +237,7 @@ async def process(self, doc, file_content, make_name_unique=False):
 class TempFileCachePB(TempFileCache):
     """Service that locally caches files downloaded from the internet"""
 
-    async def process(self, doc, file_content, make_name_unique=False):
+    async def process(self, doc, file_content, make_name_unique=True):
         """Write URL doc to file asynchronously
 
         Parameters
@@ -252,7 +252,7 @@ async def process(self, doc, file_content, make_name_unique=False):
             for PDF file.
         make_name_unique : bool, optional
             Option to make file name unique by adding a UUID at the end
-            of the file name. By default, ``False``.
+            of the file name. By default, ``True``.
 
         Returns
         -------

compass/utilities/costs.py

@@ -13,6 +13,10 @@
     "gpt-5-mini": {"prompt": 0.25, "response": 2},
     "gpt-5-nano": {"prompt": 0.05, "response": 0.4},
     "gpt-5-chat-latest": {"prompt": 1.25, "response": 10},
+    "gpt-5.4": {"prompt": 2.50, "response": 15},
+    "gpt-5.4-mini": {"prompt": 0.75, "response": 4.5},
+    "gpt-5.4-nano": {"prompt": 0.20, "response": 1.25},
+    "gpt-5.5": {"prompt": 5, "response": 30},
     "compassop-gpt-4o": {"prompt": 2.5, "response": 10},
     "compassop-gpt-4o-mini": {"prompt": 0.15, "response": 0.6},
     "compassop-gpt-4.1": {"prompt": 2, "response": 8},
@@ -22,6 +26,10 @@
     "compassop-gpt-5-mini": {"prompt": 0.25, "response": 2},
     "compassop-gpt-5-nano": {"prompt": 0.05, "response": 0.4},
     "compassop-gpt-5-chat-latest": {"prompt": 1.25, "response": 10},
+    "compassop-gpt-5.4": {"prompt": 2.50, "response": 15},
+    "compassop-gpt-5.4-mini": {"prompt": 0.75, "response": 4.5},
+    "compassop-gpt-5.4-nano": {"prompt": 0.20, "response": 1.25},
+    "compassop-gpt-5.5": {"prompt": 5, "response": 30},
     "egswaterord-gpt4.1-mini": {"prompt": 0.4, "response": 1.6},
     "wetosa-gpt-4o": {"prompt": 2.5, "response": 10},
     "wetosa-gpt-4o-mini": {"prompt": 0.15, "response": 0.6},

compass/utilities/finalize.py

@@ -333,9 +333,7 @@ def compile_run_summary_message(
     )
 
 
-def compile_collection_summary_message(
-    manifest_fp, collection_manifest, total_seconds
-):
+def compile_collection_summary_message(manifest_fp, collection_manifest):
     """Compile a short collection summary message
 
     Parameters
@@ -363,7 +361,7 @@ def compile_collection_summary_message(
         len((info or {}).get("documents") or [])
         for info in collection_manifest.get("jurisdictions", [])
     )
-    runtime = _elapsed_time_as_str(total_seconds)
+    runtime = _elapsed_time_as_str(collection_manifest["total_time"])
     locs = "jurisdiction" if num_jurisdictions == 1 else "jurisdictions"
     docs = "document" if num_documents == 1 else "documents"
     return (

compass/utilities/logs.py

@@ -574,7 +574,11 @@ def _prepare_mp_queue_safe_record(record, formatter):
     if prepared.exc_info:
         exc_type, exc_value, __ = prepared.exc_info
         prepared.exc_type = getattr(exc_type, "__name__", None)
-        prepared.exc_message = getattr(exc_value, "args", [None])[0]
+        msg = getattr(exc_value, "args", [None])
+        if len(msg) > 0:
+            prepared.exc_message = msg[0]
+        else:
+            prepared.exc_message = ""
         prepared.exc_text = formatter.formatException(prepared.exc_info)
         prepared.exc_info = None
 

compass/validation/content.py

@@ -315,21 +315,19 @@ class LegalTextValidator(TextKindValidator, JSONFromTextLLMCaller):
     )
     """System message for legal text validation LLM calls"""
 
-    def __init__(
-        self,
-        tech,
-        *args,
-        score_threshold=None,
-        doc_is_from_ocr=False,
-        **kwargs,
-    ):
+    def __init__(self, tech, doc, *args, score_threshold=None, **kwargs):
         """
 
         Parameters
         ----------
         tech : str
             Technology of interest (e.g. "solar", "wind", etc). This is
             used to set up some document validation decision trees.
+        doc : Document
+            The document being validated. This is used to set up some
+            document validation decision trees and should contain
+            metadata about the document (e.g. whether or not it was
+            extracted from OCR).
         score_threshold : float, optional
             Minimum fraction of text chunks that have to pass the legal
             check for the whole document to be considered legal text.
@@ -340,17 +338,31 @@ def __init__(
         """
         super().__init__(*args, **kwargs)
         self.tech = tech
+        self.doc = doc
         self._user_input_score_threshold = score_threshold
         self._legal_text_mem = []
-        self.doc_is_from_ocr = doc_is_from_ocr
+        self._doc_is_from_ocr = doc.attrs.get("from_ocr", False)
 
     @property
     def is_correct_kind_of_text(self):
         """bool: ``True`` if text was found to be from a legal source"""
         if not self._legal_text_mem:
             return False
         score = sum(self._legal_text_mem) / len(self._legal_text_mem)
-        return score >= self.score_threshold
+        is_legal_text = score >= self.score_threshold
+
+        logger.info(
+            "Document %s legal text check (score %.2f; threshold %.2f "
+            "source: %s)",
+            "passed" if is_legal_text else "failed",
+            score,
+            self.score_threshold,
+            self.doc.attrs.get("source", "Unknown"),
+        )
+        self.doc.attrs["legal_text_score"] = score
+        self.doc.attrs["is_legal_text"] = is_legal_text
+
+        return is_legal_text
 
     @property
     def score_threshold(self):
@@ -402,7 +414,7 @@ async def _check_chunk_for_legal_text(self, key, text_chunk):
             key=key,
             text=text_chunk,
             chat_llm_caller=chat_llm_caller,
-            doc_is_from_ocr=self.doc_is_from_ocr,
+            doc_is_from_ocr=self._doc_is_from_ocr,
         )
         out = await run_async_tree(tree, response_as_json=True)
         logger.debug("LLM response: %s", out)

tests/python/unit/services/test_services_threaded.py

@@ -55,6 +55,33 @@ async def test_temp_file_cache_service():
     assert not out_fp.exists()
 
 
+@pytest.mark.asyncio
+async def test_temp_file_cache_uses_unique_names_for_same_source():
+    """Repeated downloads from one source should not share a temp path"""
+
+    first_doc = HTMLDocument(["first"])
+    first_doc.attrs["source"] = "http://www.example.com/shared"
+
+    second_doc = HTMLDocument(["second"])
+    second_doc.attrs["source"] = "http://www.example.com/shared"
+
+    cache = TempFileCache()
+    cache.acquire_resources()
+
+    first_fp = await cache.process(first_doc, first_doc.text)
+    second_fp = await cache.process(second_doc, second_doc.text)
+
+    assert first_fp != second_fp
+    assert first_fp.exists()
+    assert second_fp.exists()
+    assert first_fp.read_text().startswith("first")
+    assert second_fp.read_text().startswith("second")
+
+    cache.release_resources()
+    assert not first_fp.exists()
+    assert not second_fp.exists()
+
+
 @pytest.mark.asyncio
 async def test_file_move_service(tmp_path):
     """Test base implementation of `FileMover` class"""
@@ -216,7 +243,7 @@ def test_write_cleaned_file_with_debug(tmp_path):
     }
 
     CLEANED_FP_REGISTRY["cleaned_file_test"] = fp_names
-    try:
+    try:  # noqa
         outputs = threaded._write_cleaned_file(
             doc,
             tmp_path,

tests/python/unit/validation/test_validation_content.py

@@ -97,11 +97,16 @@ async def test_legal_text_validation(
 ):
     """Test using `LegalTextValidator` instance on documents"""
 
+    doc = doc_loader(file_name)
     legal_text_validator = LegalTextValidator(
-        llm_service=oai_llm_service, temperature=0, seed=42, timeout=30
+        tech="wind",
+        doc=doc,
+        llm_service=oai_llm_service,
+        temperature=0,
+        seed=42,
+        timeout=30,
     )
 
-    doc = doc_loader(file_name)
     chunks = text_splitter.split_text(doc.text)
     chunk_parser = ParseChunksWithMemory(chunks, num_to_recall=2)
 
@@ -134,8 +139,15 @@ async def test_legal_text_validation_ocr(
         pages = read_pdf_ocr(fh.read())
         doc = PDFDocument(pages)
 
+    doc.attrs["from_ocr"] = True
+
     legal_text_validator = LegalTextValidator(
-        llm_service=oai_llm_service, temperature=0, seed=42, timeout=30
+        tech="wind",
+        doc=doc,
+        llm_service=oai_llm_service,
+        temperature=0,
+        seed=42,
+        timeout=30,
     )
 
     chunks = text_splitter.split_text(doc.text)