feat(embeddings): embed email and paragraph meaning units

CHANGELOG.md

@@ -0,0 +1,19 @@
+# Changelog
+
+All notable changes to this project are documented in this file.
+
+The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [Unreleased]
+
+### Added
+
+- `chunking_strategy=meaning_units` on `POST /v1/batch/embeddings` embeds email
+  parties, HTML blocks, embedded images, and paragraphs as separate vectors and
+  returns `chunk_units` with source offsets. Omit the field to keep the naruon
+  one-vector-per-input contract. Next action: send the raw invoice email and
+  search `chunk_units` for the invoice id. Gmail wrapper HTML now emits
+  innermost leaves; RFC 2397 image parameters, base64url payloads, and MIME
+  line wraps remain exact image units. `source_document` is an explicit omit
+  alias.

README.md

@@ -187,7 +187,10 @@ is read from a **KV config store**, never `os.getenv`.
   embedding parts (`routing.embedding_max_tokens_per_request`, default 280,000;
   `routing.embedding_max_chars_per_part`, default 240,000) and reduces part
   vectors with a token-weighted average, so Azure/LiteLLM over-limit embedding
-  requests are split internally instead of surfacing as caller errors. It routes
+  requests are split internally instead of surfacing as caller errors. Send
+  `chunking_strategy=meaning_units` to embed email, HTML, image, and paragraph
+  units separately and read `chunk_units` for source offsets; omit it to keep
+  one vector per submitted string. It routes
   through the same RoutingPolicy/cost optimiser and `pg-llm-batch` embeddings
   backend (local in-process backend standalone), and records one usage-ledger row
   per original vector with the full attribution dimensions (service, team,
@@ -204,6 +207,7 @@ Grounding papers (LLM cost, routing, load balancing) live in
 
 ## Design Artifacts
 
+- [Meaning-unit chunking](docs/meaning_unit_chunking.md)
 - [Library research](docs/library_research.md)
 - [Product planning](docs/product_planning.md)
 - [Screen design](docs/screen_design.md)
@@ -284,4 +288,6 @@ python tests/test_commercial_proposal_packet.py
 python tests/test_commercial_purchase_approval_packet.py
 python tests/test_commercial_due_diligence_room.py
 python tests/test_commercial_investment_committee_memo.py
+python tests/test_meaning_unit_chunking.py
+python tests/test_embeddings_meaning_units_http_honesty.py
 ```

contextual_orchestrator/__main__.py

@@ -94,7 +94,12 @@ def main() -> None:
                         help="Measure orchestration vs a single-worker baseline on these prompts and print the report.")
     args = parser.parse_args()
 
-    client = ModelClient(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify)
+    if args.insecure_skip_tls_verify:
+        parser.error(
+            "--insecure-skip-tls-verify is no longer supported; "
+            "configure --provider-ca-bundle for a private certificate authority"
+        )
+    client = ModelClient(ca_bundle=args.provider_ca_bundle)
     orchestrator = TaskOrchestrator(
         load_agents(args.agents),
         client=client,

contextual_orchestrator/api_contract.py

@@ -440,6 +440,16 @@
                                         "description": "observability + attribution dims (service, team, group, company, provider)",
                                     },
                                     "attribution": {"type": "object"},
+                                    "chunking_strategy": {
+                                        "type": "string",
+                                        "enum": ["source_document", "meaning_units"],
+                                        "description": (
+                                            "Omit to keep one vector per input (naruon contract). "
+                                            "meaning_units embeds email parties, HTML blocks, "
+                                            "embedded images, and paragraphs separately and "
+                                            "returns chunk_units with source offsets."
+                                        ),
+                                    },
                                 },
                             }
                         }
@@ -451,7 +461,7 @@
                             "Batch completed synchronously: "
                             "{batch_id, status, embeddings:[{index, embedding}], "
                             "cost_micro_usd, token_counts, total_tokens, part_count, "
-                            "input_part_counts, map_reduce}"
+                            "input_part_counts, map_reduce, optional chunk_units}"
                         )
                     },
                     "202": {"description": "Batch accepted; poll GET /v1/batch/embeddings/{batch_id}"},

contextual_orchestrator/cost_ledger.py

@@ -564,14 +564,13 @@ class SqlLedgerStore:
     """
 
     def __init__(self, connection: Any, paramstyle: str = "qmark") -> None:
+        if paramstyle not in {"qmark", "pyformat"}:
+            raise ValueError(f"unsupported ledger paramstyle: {paramstyle!r}")
         self._conn = connection
         self._paramstyle = paramstyle
         self._create_schema()
         self._seed_dimension_catalog()
 
-    def _placeholder(self) -> str:
-        return "?" if self._paramstyle == "qmark" else "%s"
-
     def _create_schema(self) -> None:
         cur = self._conn.cursor()
         for statement in SCHEMA_SQL.strip().split(";"):
@@ -580,49 +579,139 @@ def _create_schema(self) -> None:
         self._conn.commit()
 
     def _seed_dimension_catalog(self) -> None:
-        ph = self._placeholder()
+        """Insert the fixed attribution-dimension catalog rows once."""
         cur = self._conn.cursor()
         for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG):
-            cur.execute(
-                f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
-                (name,),
-            )
-            if cur.fetchone() is None:
+            if self._paramstyle == "qmark":
+                cur.execute(
+                    "SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = ?",
+                    (name,),
+                )
+            else:
+                cur.execute(
+                    "SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = %s",
+                    (name,),
+                )
+            if cur.fetchone() is not None:
+                continue
+            if self._paramstyle == "qmark":
                 cur.execute(
                     "INSERT INTO cost_attribution_dimensions "
-                    f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})",  # nosec B608 - ph is a DB-API placeholder.
+                    "(dimension_name, dimension_label, dimension_order) "
+                    "VALUES (?, ?, ?)",
+                    (name, label, order),
+                )
+            else:
+                cur.execute(
+                    "INSERT INTO cost_attribution_dimensions "
+                    "(dimension_name, dimension_label, dimension_order) "
+                    "VALUES (%s, %s, %s)",
                     (name, label, order),
                 )
         self._conn.commit()
 
     def append(self, record: UsageRecord) -> None:
-        """Insert a usage record row."""
+        """Insert a usage record row with driver-appropriate bound values."""
         row = record.as_dict()
-        ph = self._placeholder()
-        placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
-        columns = ", ".join(_USAGE_COLUMNS)
+        values = tuple(row.get(column) for column in _USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(
-            f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are fixed _USAGE_COLUMNS.
-            tuple(row.get(column) for column in _USAGE_COLUMNS),
-        )
+        if self._paramstyle == "qmark":
+            cur.execute(
+                "INSERT INTO llm_usage_records ("
+                "usage_record_id, created_at, workflow_run_id, request_channel, "
+                "route_mode, provider_name, model_name, account_name, service_name, "
+                "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                "completion_tokens, total_tokens, cost_amount, currency_code"
+                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                values,
+            )
+        else:
+            cur.execute(
+                "INSERT INTO llm_usage_records ("
+                "usage_record_id, created_at, workflow_run_id, request_channel, "
+                "route_mode, provider_name, model_name, account_name, service_name, "
+                "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                "completion_tokens, total_tokens, cost_amount, currency_code"
+                ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
+                values,
+            )
         self._conn.commit()
 
     def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[Dict[str, Any]]:
         """Return record rows in the optional half-open window."""
-        ph = self._placeholder()
-        clauses: List[str] = []
-        params: List[Any] = []
-        if start is not None:
-            clauses.append(f"created_at >= {ph}")
-            params.append(start)
-        if end is not None:
-            clauses.append(f"created_at < {ph}")
-            params.append(end)
-        where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
-        columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.
+        if self._paramstyle == "qmark":
+            if start is not None and end is not None:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records WHERE created_at >= ? AND created_at < ?",
+                    (start, end),
+                )
+            elif start is not None:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records WHERE created_at >= ?",
+                    (start,),
+                )
+            elif end is not None:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records WHERE created_at < ?",
+                    (end,),
+                )
+            else:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records"
+                )
+        else:
+            if start is not None and end is not None:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records WHERE created_at >= %s AND created_at < %s",
+                    (start, end),
+                )
+            elif start is not None:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records WHERE created_at >= %s",
+                    (start,),
+                )
+            elif end is not None:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records WHERE created_at < %s",
+                    (end,),
+                )
+            else:
+                cur.execute(
+                    "SELECT usage_record_id, created_at, workflow_run_id, request_channel, "
+                    "route_mode, provider_name, model_name, account_name, service_name, "
+                    "upstream_api, team_name, group_name, company_name, prompt_tokens, "
+                    "completion_tokens, total_tokens, cost_amount, currency_code "
+                    "FROM llm_usage_records"
+                )
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]
 
 

contextual_orchestrator/cost_router.py

@@ -85,6 +85,7 @@ def __init__(
         self._embedding_part_counts: Dict[str, List[int]] = {}
         self._embedding_part_limits: Dict[str, Dict[str, int]] = {}
         self._embedding_documents: Dict[str, Dict[str, Any]] = {}
+        self._embedding_chunk_units: Dict[str, List[Dict[str, Any]]] = {}
 
     # ------------------------------------------------------------------
     # Provider / model resolution
@@ -286,12 +287,16 @@ def submit_embeddings_batch(
         requests, part_counts, part_limits = self._build_embedding_requests(
             inputs, model=model, attribution=shared_attribution
         )
-        job = self.embedding_batch_backend.submit(requests, metadata=metadata)
+        submit_metadata = dict(metadata or {})
+        chunk_units = submit_metadata.pop("chunk_units", None)
+        job = self.embedding_batch_backend.submit(requests, metadata=submit_metadata or None)
         self._embedding_jobs[job.job_id] = job
         self._embedding_requests[job.job_id] = requests
         self._embedding_input_counts[job.job_id] = len(inputs)
         self._embedding_part_counts[job.job_id] = part_counts
         self._embedding_part_limits[job.job_id] = part_limits
+        if chunk_units:
+            self._embedding_chunk_units[job.job_id] = list(chunk_units)
         return job
 
     def _build_embedding_requests(
@@ -471,12 +476,16 @@ def embeddings_batch_document(self, batch_id: str) -> Dict[str, Any]:
         job = self._require_embedding_job(batch_id)
         status = self.embedding_batch_backend.poll(job)
         if not status.get("is_complete"):
-            return {
+            pending = {
                 "batch_id": batch_id,
                 "status": status.get("status") or job.status,
                 "backend": job.backend,
                 "embeddings": None,
             }
+            pending_units = self._embedding_chunk_units.get(batch_id)
+            if pending_units:
+                pending["chunk_units"] = pending_units
+            return pending
 
         items: List[EmbeddingBatchResultItem] = self.embedding_batch_backend.retrieve(job)
         requests = self._embedding_requests.get(batch_id, [])
@@ -561,6 +570,9 @@ def embeddings_batch_document(self, batch_id: str) -> Dict[str, Any]:
             "currency_code": currency_code,
             "cost_micro_usd": int(round(total_cost_amount * 1_000_000)),
         }
+        chunk_units = self._embedding_chunk_units.get(batch_id)
+        if chunk_units:
+            document["chunk_units"] = chunk_units
         self._embedding_documents[batch_id] = document
         return document