diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 000000000..37940d6b7
--- /dev/null
+++ b/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.
diff --git a/README.md b/README.md
index 65f57dd4c..efb36cd28 100644
--- a/README.md
+++ b/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
```
diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py
index 5f68c3b74..6de61a574 100644
--- a/contextual_orchestrator/__main__.py
+++ b/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,
diff --git a/contextual_orchestrator/api_contract.py b/contextual_orchestrator/api_contract.py
index fae9fba0b..c565dc8a1 100644
--- a/contextual_orchestrator/api_contract.py
+++ b/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}"},
diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py
index d3943c5be..54a3c1d03 100644
--- a/contextual_orchestrator/cost_ledger.py
+++ b/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()]
diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py
index bfbe159db..066ea74e8 100644
--- a/contextual_orchestrator/cost_router.py
+++ b/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
diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py
index 0097b722e..9126d607a 100644
--- a/contextual_orchestrator/orchestrator.py
+++ b/contextual_orchestrator/orchestrator.py
@@ -200,6 +200,27 @@ def is_transient_error(exc: BaseException) -> bool:
return False
+class _RejectProviderRedirects(urllib.request.HTTPRedirectHandler):
+ """Refuse redirects so a validated provider host cannot redirect egress."""
+
+ def redirect_request(
+ self,
+ request: urllib.request.Request,
+ file_pointer: Any,
+ code: int,
+ message: str,
+ headers: Any,
+ new_url: str,
+ ) -> None:
+ raise urllib.error.HTTPError(
+ request.full_url,
+ code,
+ f"provider redirects are not allowed: {new_url}",
+ headers,
+ file_pointer,
+ )
+
+
class ModelClient:
"""Small chat-completions client with retry, backoff, and mock support."""
@@ -222,15 +243,21 @@ def __init__(
self._sleep = time.sleep
# Per-thread usage from the most recent chat() (the server is threaded).
self._local = threading.local()
- # TLS trust for provider egress. Default verifies against the system trust store;
- # ca_bundle points at a custom CA (corporate gateways); verify_tls=False is an
- # explicit dev-only opt-out (insecure) for self-signed endpoints.
+ # Provider egress always verifies TLS. Corporate gateways must supply
+ # an explicit CA bundle rather than disabling certificate verification.
self._ssl_context = self._build_ssl_context(ca_bundle, verify_tls)
+ self._provider_opener = urllib.request.build_opener(
+ urllib.request.HTTPSHandler(context=self._ssl_context),
+ _RejectProviderRedirects(),
+ )
@staticmethod
def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext:
if not verify_tls:
- return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out.
+ raise ValueError(
+ "provider TLS verification cannot be disabled; configure ca_bundle "
+ "for a private or corporate certificate authority"
+ )
if ca_bundle:
if not os.path.isfile(ca_bundle):
raise ValueError(f"provider CA bundle does not exist: {ca_bundle}")
@@ -306,12 +333,28 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str:
return data["choices"][0]["message"]["content"]
def _open_provider(self, request: urllib.request.Request) -> Any:
- """Open a provider request built from a validated provider URL."""
- return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation.
- request,
- timeout=self.timeout,
- context=self._ssl_context,
- )
+ """Open HTTPS, or loopback HTTP for isolated transport tests, without redirects."""
+ parsed = urlparse(request.full_url)
+ if not parsed.hostname:
+ raise RuntimeError("provider transport requires an absolute URL")
+ if parsed.scheme == "https":
+ pass
+ elif parsed.scheme == "http":
+ try:
+ peer = ipaddress.ip_address(parsed.hostname)
+ except ValueError as exc:
+ raise RuntimeError(
+ "provider transport requires https or numeric loopback http"
+ ) from exc
+ if not peer.is_loopback:
+ raise RuntimeError(
+ "provider transport requires https or numeric loopback http"
+ )
+ else:
+ raise RuntimeError(
+ "provider transport requires https or numeric loopback http"
+ )
+ return self._provider_opener.open(request, timeout=self.timeout)
def stream_chat(self, agent: ModelAgent, messages: list[ChatMessage], temperature: float = 0.2):
"""Yield content deltas from a mock or OpenAI-compatible streaming endpoint.
diff --git a/contextual_orchestrator/semantic_chunking.py b/contextual_orchestrator/semantic_chunking.py
new file mode 100644
index 000000000..e6c8c26c8
--- /dev/null
+++ b/contextual_orchestrator/semantic_chunking.py
@@ -0,0 +1,537 @@
+"""Split embedding inputs into searchable meaning units.
+
+Token-budget splitting in :mod:`contextual_orchestrator.cost_router` keeps a
+provider call under a ceiling and then averages parts back into one vector.
+That is the wrong grain for retrieval: a naruon invoice email then embeds the
+greeting, the balance line, and the signature as one point.
+
+This module cuts at linguistic meaning units — email parties, HTML blocks,
+embedded images, paragraphs, and sentences — so a later lexical or neural
+search can recover the invoice line without the greeting (Zhao et al., 2024;
+Qu et al., 2025; Unicode Consortium, 2024). Similarity-breakpoint “semantic
+chunking” is deliberately not used: Qu et al. (2025) found that cost is not
+justified by consistent gains. Sentence cuts follow the Unicode text
+segmentation intent in UAX #29 without adding a Unicode dependency.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+from typing import Any
+
+_EMAIL_HEADER = re.compile(
+ r"^(From|To|Cc|Bcc|Subject|Reply-To|Date):\s*.+$",
+ re.MULTILINE | re.IGNORECASE,
+)
+_IMAGE_HEAD = re.compile(
+ r"data:image/[A-Za-z0-9.+-]+(?:;[A-Za-z0-9!#$&^_.+-]+(?:=[^;,\s]+)?)*;base64,",
+ re.IGNORECASE,
+)
+_B64_CHAR = frozenset(
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/-_"
+)
+_HTML_BLOCK_TAGS = frozenset(
+ {
+ "p",
+ "div",
+ "li",
+ "h1",
+ "h2",
+ "h3",
+ "h4",
+ "h5",
+ "h6",
+ "tr",
+ "td",
+ "section",
+ "article",
+ "blockquote",
+ }
+)
+_ASCII_CASE_FOLD = str.maketrans(
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ",
+ "abcdefghijklmnopqrstuvwxyz",
+)
+_PARAGRAPH_BREAK = re.compile(r"\n\s*\n+")
+_SENTENCE_CUT = re.compile(r"(?<=[.!?。！？])(?=\s+(?:[A-Z\"'(가-힣]))")
+_TOKEN = re.compile(r"[A-Za-z0-9]+(?:[.-][A-Za-z0-9]+)*|[가-힣]+")
+_ABBREVIATIONS = frozenset(
+ {"mr", "mrs", "ms", "dr", "prof", "sr", "jr", "inc", "ltd", "vs", "etc"}
+)
+_EMAIL_KIND = {
+ "from": "email_sender",
+ "to": "email_recipient",
+ "cc": "email_copy",
+ "bcc": "email_copy",
+ "subject": "email_subject",
+ "reply-to": "email_header",
+ "date": "email_header",
+}
+
+
+@dataclass(frozen=True)
+class MeaningUnit:
+ """One retrieval grain cut from a source document.
+
+ ``chunk_text`` is always the exact source slice
+ ``source[source_offset:source_offset + source_length]`` so a buyer can
+ put the unit back into the original email, HTML, or image position.
+ """
+
+ chunk_kind: str
+ source_offset: int
+ source_length: int
+ chunk_text: str
+ input_index: int = 0
+ chunk_index: int = 0
+
+ def to_dict(self) -> dict[str, Any]:
+ """Serialize the unit for the embeddings batch document."""
+ return {
+ "chunk_kind": self.chunk_kind,
+ "source_offset": self.source_offset,
+ "source_length": self.source_length,
+ "chunk_text": self.chunk_text,
+ "input_index": self.input_index,
+ "chunk_index": self.chunk_index,
+ }
+
+ def with_index(self, input_index: int, chunk_index: int) -> MeaningUnit:
+ """Return a copy stamped with batch input and unit indexes."""
+ return MeaningUnit(
+ self.chunk_kind,
+ self.source_offset,
+ self.source_length,
+ self.chunk_text,
+ input_index,
+ chunk_index,
+ )
+
+
+def meaning_unit_chunks(
+ text: str,
+ *,
+ input_index: int = 0,
+ unit_grain: str = "body_paragraph",
+) -> list[MeaningUnit]:
+ """Cut ``text`` into non-overlapping meaning units in source order.
+
+ Empty or whitespace-only input becomes a single ``source_document`` unit
+ so a batch slot is never dropped. Image data-URLs keep their original
+ offset so OCR or object tags added later can point at the same span.
+ ``unit_grain`` is ``body_paragraph`` (default retrieval grain) or
+ ``body_sentence`` (UAX #29-style sentence cuts inside leftover prose).
+ """
+ if not isinstance(text, str):
+ text = str(text)
+ if not text.strip():
+ return [MeaningUnit("source_document", 0, len(text), text, input_index, 0)]
+
+ reserved: list[tuple[int, int, str, str]] = []
+ for start, end, piece in _iter_embedded_images(text):
+ reserved.append((start, end, "embedded_image", piece))
+ if _looks_like_email(text):
+ header_end = _leading_email_header_end(text)
+ for match in _EMAIL_HEADER.finditer(text, 0, header_end):
+ if _overlaps(reserved, match.start(), match.end()):
+ continue
+ kind = _EMAIL_KIND.get(match.group(1).lower(), "email_header")
+ reserved.append((match.start(), match.end(), kind, match.group(0)))
+ if "<" in text and ">" in text:
+ for start, end, piece in _html_leaf_spans(text):
+ if _overlaps(reserved, start, end):
+ continue
+ reserved.append((start, end, "html_block", piece))
+
+ reserved.sort(key=lambda item: (item[0], item[1]))
+ units: list[MeaningUnit] = []
+ cursor = 0
+ for start, end, kind, piece in reserved:
+ if start > cursor:
+ units.extend(_split_plain(text[cursor:start], cursor, input_index, unit_grain))
+ units.append(MeaningUnit(kind, start, end - start, piece, input_index, 0))
+ cursor = max(cursor, end)
+ if cursor < len(text):
+ units.extend(_split_plain(text[cursor:], cursor, input_index, unit_grain))
+ units = [
+ unit
+ for unit in units
+ if unit.chunk_text.strip() and not _is_tag_only(unit.chunk_text)
+ ]
+ if not units:
+ return [MeaningUnit("source_document", 0, len(text), text, input_index, 0)]
+ return [unit.with_index(input_index, index) for index, unit in enumerate(units)]
+
+
+def expand_embedding_inputs(
+ inputs: list[str],
+ *,
+ chunking_strategy: str | None,
+) -> tuple[list[str], list[MeaningUnit]]:
+ """Expand batch inputs according to ``chunking_strategy``.
+
+ ``None`` / empty / ``source_document`` keeps one unit per input so the
+ naruon contract (one vector per submitted string) does not change.
+ ``meaning_units`` emits one embeddable string per meaning unit.
+ """
+ if chunking_strategy in (None, "", "source_document"):
+ units = [
+ MeaningUnit("source_document", 0, len(text), text, input_index, 0)
+ for input_index, text in enumerate(inputs)
+ ]
+ return list(inputs), units
+ if chunking_strategy != "meaning_units":
+ raise ValueError(
+ "chunking_strategy must be omitted, source_document, or meaning_units"
+ )
+ units: list[MeaningUnit] = []
+ for input_index, text in enumerate(inputs):
+ parts = meaning_unit_chunks(text, input_index=input_index)
+ if not parts:
+ parts = [MeaningUnit("source_document", 0, len(text), text, input_index, 0)]
+ for chunk_index, part in enumerate(parts):
+ units.append(part.with_index(input_index, chunk_index))
+ return [unit.chunk_text for unit in units], units
+
+
+def rank_meaning_units(query: str, units: list[MeaningUnit]) -> list[MeaningUnit]:
+ """Rank units by query-token overlap for retrieval-accuracy tests.
+
+ The standalone embedding backend is a SHA-256 heuristic and is not
+ semantically meaningful. Lexical overlap is the honest offline proof that
+ isolating the invoice line makes that line retrievable.
+ """
+ query_tokens = _tokens(query)
+ scored: list[tuple[float, int, MeaningUnit]] = []
+ for unit in units:
+ unit_tokens = _tokens(unit.chunk_text)
+ overlap = (len(query_tokens & unit_tokens) / len(query_tokens)) if query_tokens else 0.0
+ scored.append((overlap, -unit.source_offset, unit))
+ scored.sort(reverse=True)
+ matched = [unit for score, _offset, unit in scored if score > 0]
+ return matched or [unit for _score, _offset, unit in scored]
+
+
+def _tokens(text: str) -> set[str]:
+ return {match.group(0).lower() for match in _TOKEN.finditer(text)}
+
+
+def _looks_like_email(text: str) -> bool:
+ headers = []
+ for line in text.splitlines():
+ stripped = line.strip()
+ if not stripped:
+ if headers:
+ break
+ continue
+ match = _EMAIL_HEADER.match(stripped)
+ if match is None:
+ break
+ headers.append(match.group(1).lower())
+ if len(headers) >= 8:
+ break
+ return bool({"from", "to", "subject"} & set(headers))
+
+
+
+def _leading_email_header_end(text: str) -> int:
+ """Return the exclusive end of the leading RFC 5322-style header block.
+
+ Header recognition and reservation share this bound so a body paragraph
+ beginning with ``Subject:`` is never promoted to ``email_subject``.
+ """
+ offset = 0
+ saw_header = False
+ header_count = 0
+ for line in text.splitlines(keepends=True):
+ stripped = line.strip()
+ if not stripped:
+ if saw_header:
+ return offset
+ offset += len(line)
+ continue
+ if _EMAIL_HEADER.match(stripped) is None:
+ return offset
+ saw_header = True
+ header_count += 1
+ offset += len(line)
+ if header_count >= 8:
+ return offset
+ return offset
+
+
+def _iter_embedded_images(text: str) -> list[tuple[int, int, str]]:
+ """Return exact RFC 2397 ``data:image`` spans including MIME folds."""
+ found: list[tuple[int, int, str]] = []
+ for head in _IMAGE_HEAD.finditer(text):
+ if found and head.start() < found[-1][1]:
+ continue
+ end = _consume_base64_payload(text, head.end())
+ if end <= head.end():
+ continue
+ found.append((head.start(), end, text[head.start() : end]))
+ return found
+
+
+def _is_b64_line(line: str) -> bool:
+ """Return True for a complete physical base64/base64url payload line."""
+ stripped = line.rstrip(" \t")
+ if not stripped:
+ return False
+ padding = 0
+ for char in stripped:
+ if char == "=":
+ padding += 1
+ if padding > 2:
+ return False
+ continue
+ if padding or char not in _B64_CHAR:
+ return False
+ return True
+
+
+def _is_mime_continuation(line: str, previous_payload_len: int) -> bool:
+ """Recognize MIME-folded payload without swallowing plain body prose."""
+ stripped = line.rstrip(" \t")
+ if not _is_b64_line(stripped):
+ return False
+ if "=" in stripped:
+ return True
+ if previous_payload_len == 76:
+ return True
+ has_b64_mark = any(char in stripped for char in "+/_-")
+ return has_b64_mark and previous_payload_len >= 16
+
+
+def _consume_base64_payload(text: str, start_index: int) -> int:
+ """Return the exclusive end of a data-URL payload."""
+ length = len(text)
+ cursor = start_index
+ end = start_index
+ padding = 0
+ line_payload_len = 0
+ while cursor < length:
+ char = text[cursor]
+ if char in " \t\"'<>":
+ break
+ if char in "\r\n":
+ if padding:
+ break
+ next_index = cursor + 1
+ if char == "\r" and next_index < length and text[next_index] == "\n":
+ next_index += 1
+ line_end = next_index
+ while line_end < length and text[line_end] not in "\r\n":
+ line_end += 1
+ peek = text[next_index:line_end]
+ if _is_mime_continuation(peek, line_payload_len):
+ cursor = next_index
+ line_payload_len = 0
+ continue
+ break
+ if char == "=":
+ padding += 1
+ if padding > 2:
+ break
+ end = cursor + 1
+ line_payload_len += 1
+ cursor += 1
+ continue
+ if char in _B64_CHAR:
+ if padding:
+ break
+ end = cursor + 1
+ line_payload_len += 1
+ cursor += 1
+ continue
+ break
+ return end
+
+
+def _is_tag_only(text: str) -> bool:
+ """Return True when ``text`` is only HTML tags and whitespace."""
+ index = 0
+ length = len(text)
+ saw_tag = False
+ while index < length:
+ while index < length and text[index].isspace():
+ index += 1
+ if index >= length:
+ return saw_tag
+ if text[index] != "<":
+ return False
+ name_at = index + 1
+ if name_at < length and text[name_at] == "/":
+ name_at += 1
+ if (
+ name_at >= length
+ or not text[name_at].isascii()
+ or not text[name_at].isalpha()
+ ):
+ return False
+ close_at = text.find(">", name_at + 1)
+ if close_at < 0:
+ return False
+ index = close_at + 1
+ saw_tag = True
+ return saw_tag
+
+
+def _fold_ascii_case(text: str) -> str:
+ """Lower ASCII tag names without changing source-string length."""
+ return text.translate(_ASCII_CASE_FOLD)
+
+
+def _is_ascii_tag_name_char(char: str) -> bool:
+ return "A" <= char <= "Z" or "a" <= char <= "z" or "0" <= char <= "9"
+
+
+def _is_ascii_word_char(char: str) -> bool:
+ return _is_ascii_tag_name_char(char) or char == "_"
+
+
+def _html_leaf_spans(text: str) -> list[tuple[int, int, str]]:
+ """Return innermost block elements with one left-to-right stack walk."""
+ leaves: list[tuple[int, int, str]] = []
+ stack: list[tuple[str, int, bool]] = []
+ index = 0
+ length = len(text)
+ while index < length:
+ lt = text.find("<", index)
+ if lt < 0 or lt + 1 >= length:
+ break
+ closing = text[lt + 1] == "/"
+ name_at = lt + 2 if closing else lt + 1
+ if name_at >= length:
+ break
+ name_end = name_at
+ while name_end < length and _is_ascii_tag_name_char(text[name_end]):
+ name_end += 1
+ if name_end == name_at:
+ index = lt + 1
+ continue
+ if name_end < length and _is_ascii_word_char(text[name_end]):
+ index = lt + 1
+ continue
+ tag = _fold_ascii_case(text[name_at:name_end])
+ gt = text.find(">", name_end)
+ if gt < 0:
+ break
+ if tag in _HTML_BLOCK_TAGS:
+ if closing:
+ for depth in range(len(stack) - 1, -1, -1):
+ if stack[depth][0] != tag:
+ continue
+ _name, start, has_child = stack[depth]
+ del stack[depth:]
+ if not has_child:
+ end = gt + 1
+ leaves.append((start, end, text[start:end]))
+ break
+ else:
+ if stack:
+ parent_name, parent_start, _has_child = stack[-1]
+ stack[-1] = (parent_name, parent_start, True)
+ stack.append((tag, lt, False))
+ index = gt + 1
+ return leaves
+
+def _overlaps(reserved: list[tuple[int, int, str, str]], start: int, end: int) -> bool:
+ for existing_start, existing_end, _kind, _piece in reserved:
+ if start < existing_end and end > existing_start:
+ return True
+ return False
+
+
+def _split_plain(
+ text: str,
+ base_offset: int,
+ input_index: int,
+ unit_grain: str = "body_paragraph",
+) -> list[MeaningUnit]:
+ if not text.strip():
+ return []
+ units: list[MeaningUnit] = []
+ cursor = 0
+ breaks = list(_PARAGRAPH_BREAK.finditer(text))
+ spans: list[tuple[int, int]] = []
+ for match in breaks:
+ if match.start() > cursor:
+ spans.append((cursor, match.start()))
+ cursor = match.end()
+ if cursor < len(text):
+ spans.append((cursor, len(text)))
+ if not spans:
+ spans = [(0, len(text))]
+ for start, end in spans:
+ piece = text[start:end]
+ leading = len(piece) - len(piece.lstrip())
+ trailing = len(piece) - len(piece.rstrip())
+ inner_start = start + leading
+ inner_end = end - trailing
+ if inner_end <= inner_start:
+ continue
+ inner = text[inner_start:inner_end]
+ if unit_grain == "body_sentence":
+ units.extend(_split_sentences(inner, base_offset + inner_start, input_index))
+ continue
+ units.append(
+ MeaningUnit(
+ "body_paragraph",
+ base_offset + inner_start,
+ inner_end - inner_start,
+ inner,
+ input_index,
+ 0,
+ )
+ )
+ return units
+
+
+def _split_sentences(text: str, base_offset: int, input_index: int) -> list[MeaningUnit]:
+ cuts = [0]
+ for match in _SENTENCE_CUT.finditer(text):
+ cuts.append(match.start())
+ cuts.append(len(text))
+ raw_spans = [
+ (left, right) for left, right in zip(cuts, cuts[1:]) if right > left
+ ]
+ merged = _merge_abbreviations(text, raw_spans)
+ units: list[MeaningUnit] = []
+ for start, end in merged:
+ piece = text[start:end]
+ leading = len(piece) - len(piece.lstrip())
+ trailing = len(piece) - len(piece.rstrip())
+ inner_start = start + leading
+ inner_end = end - trailing
+ if inner_end <= inner_start:
+ continue
+ inner = text[inner_start:inner_end]
+ kind = "body_sentence" if len(merged) > 1 else "body_paragraph"
+ units.append(
+ MeaningUnit(
+ kind,
+ base_offset + inner_start,
+ inner_end - inner_start,
+ inner,
+ input_index,
+ 0,
+ )
+ )
+ return units
+
+
+def _merge_abbreviations(text: str, spans: list[tuple[int, int]]) -> list[tuple[int, int]]:
+ if len(spans) < 2:
+ return spans
+ merged: list[tuple[int, int]] = [spans[0]]
+ for start, end in spans[1:]:
+ previous_start, previous_end = merged[-1]
+ tail = text[previous_start:previous_end].rstrip()
+ word = re.search(r"([A-Za-z]+)\.$", tail)
+ if word and word.group(1).lower() in _ABBREVIATIONS:
+ merged[-1] = (previous_start, end)
+ continue
+ merged.append((start, end))
+ return merged
diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py
index c58d4cb79..b20fe52f1 100644
--- a/contextual_orchestrator/server.py
+++ b/contextual_orchestrator/server.py
@@ -17,6 +17,7 @@
from .cost_ledger import ATTRIBUTION_DIMENSIONS, dimension_catalog
from .cost_router import CostRoutingCoordinator
from .batch_routing import BatchRequest
+from .semantic_chunking import expand_embedding_inputs
from .orchestrator import (
BudgetExceededError,
TaskOrchestrator,
@@ -45,7 +46,15 @@
"model", "input", "instructions", "stream", "metadata", "reasoning",
} | OPENAI_PASSTHROUGH_PARAM_KEYS
ALLOWED_BATCH_KEYS = {"requests", "attribution", "routing", "model"}
-ALLOWED_EMBEDDINGS_BATCH_KEYS = {"model", "input", "inputs", "endpoint", "metadata", "attribution"}
+ALLOWED_EMBEDDINGS_BATCH_KEYS = {
+ "model",
+ "input",
+ "inputs",
+ "endpoint",
+ "metadata",
+ "attribution",
+ "chunking_strategy",
+}
ALLOWED_MESSAGE_ROLES = {"system", "user", "assistant", "tool"}
ALLOWED_MODES = {"auto", "route", "conduct"}
ALLOWED_SIMULATE_KEYS = {"prompt", "mode", "include_orchestration_trace"}
@@ -247,6 +256,35 @@ def _validate_batch_requests(body: dict[str, Any], expose_trace: bool) -> list[B
return batch
+def _validate_chunking_strategy(body: dict[str, Any]) -> str | None:
+ """Return ``meaning_units`` or omit-equivalent ``None``.
+
+ Unknown values fail closed so a buyer cannot believe the gateway split a
+ document when it actually embedded the whole string. JSON null,
+ empty/whitespace strings, and ``source_document`` are treat-as-omit.
+ """
+ if "chunking_strategy" not in body:
+ return None
+ value = body.get("chunking_strategy")
+ if value is None or (isinstance(value, str) and not value.strip()):
+ return None
+ if not isinstance(value, str):
+ raise RequestError(
+ 400,
+ "invalid_chunking_strategy",
+ "chunking_strategy must be a string",
+ )
+ if value == "source_document":
+ return None
+ if value == "meaning_units":
+ return value
+ raise RequestError(
+ 400,
+ "invalid_chunking_strategy",
+ "chunking_strategy must be omitted, null, source_document, or meaning_units; send meaning_units to embed email, HTML, and paragraph units separately",
+ )
+
+
def _validate_embeddings_inputs(body: dict[str, Any]) -> list[str]:
"""Validate the embeddings batch inputs (accepts ``inputs`` or ``input``)."""
raw = body.get("inputs")
@@ -800,12 +838,18 @@ def do_POST(self) -> None: # noqa: N802
if path == "/v1/batch/embeddings":
_reject_unknown_keys(body, ALLOWED_EMBEDDINGS_BATCH_KEYS)
inputs = _validate_embeddings_inputs(body)
+ chunking_strategy = _validate_chunking_strategy(body)
+ inputs, chunk_units = expand_embedding_inputs(
+ inputs, chunking_strategy=chunking_strategy
+ )
model_name = str(body.get("model", "contextual-orchestrator"))
attribution = _embeddings_attribution(body)
submit_metadata: dict[str, Any] = {"actor_scope": "inference"}
endpoint_alias = body.get("endpoint")
if endpoint_alias:
submit_metadata["endpoint_alias"] = str(endpoint_alias)
+ if chunking_strategy == "meaning_units":
+ submit_metadata["chunk_units"] = [unit.to_dict() for unit in chunk_units]
document = self._run(lambda: coordinator.complete_embeddings_batch(
inputs,
model=model_name,
diff --git a/docs/architecture.md b/docs/architecture.md
index c0f63a81e..4c31ce347 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -36,6 +36,7 @@ This repository implements the interface and control plane, not the trained coor
- `Orchestrator.conduct`: the workflow path with planner, worker, verifier, and synthesizer steps.
- `WorkflowStep.access`: Conductor-style visibility control.
- `ModelClient`: OpenAI-compatible HTTP client, with `mock://` for local checks.
+- `contextual_orchestrator.semantic_chunking`: meaning-unit cuts for `/v1/batch/embeddings` when `chunking_strategy=meaning_units` (email parties, HTML blocks, embedded-image offsets, paragraphs). Token-budget map/reduce stays the provider-safety path and still averages parts of *one* unit.
- `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server.
The deliberate simplification is the policy. The paper systems learn routing and topology from rewards; this lab uses deterministic keyword scoring so the repo runs without training data, GPUs, or vendor credentials.
@@ -53,3 +54,14 @@ The product is not a Fugu clone. It is a control-plane prototype for the same pu
- replayable evaluation runs before any learned coordinator replaces the deterministic policy.
See [product_planning.md](product_planning.md) for the product reboot.
+
+## References
+
+Zhao, J., Ji, Z., Ye, Y., Feng, X., Zhang, X., & Rong, C. (2024). *Meta-chunking: Learning text segmentation and semantic completion via logical perception*. arXiv. https://doi.org/10.48550/arXiv.2410.12788
+
+Qu, R., Tu, R., & Bao, F. (2025). Is semantic chunking worth the computational cost? In *Findings of the Association for Computational Linguistics: NAACL 2025* (pp. 2012–2027). Association for Computational Linguistics. https://aclanthology.org/2025.findings-naacl.114/
+
+Unicode Consortium. (2024). *Unicode Standard Annex #29: Unicode text segmentation*. https://www.unicode.org/reports/tr29/
+
+Lewis, P., Perez, E., Piktus, A., Petroni, F., Karpukhin, V., Goyal, N., Küttler, H., Lewis, M., Yih, W., Rocktäschel, T., Riedel, S., & Kiela, D. (2020). Retrieval-augmented generation for knowledge-intensive NLP tasks. In *Advances in Neural Information Processing Systems, 33*. https://doi.org/10.48550/arXiv.2005.11401
+
diff --git a/docs/database_conventions.md b/docs/database_conventions.md
index 24ef02bb9..41ff75e35 100644
--- a/docs/database_conventions.md
+++ b/docs/database_conventions.md
@@ -16,6 +16,24 @@
- Breaking changes use expand/backfill/contract phases.
- Large data changes are batched and monitored.
+## Meaning-unit embeddings (3NF)
+
+These objects are the persistence target when a buyer stores meaning-unit
+vectors (naruon import, Clearfolio attachment search). They are not required
+for the in-memory batch path.
+
+| Object | Keys | Purpose |
+|---|---|---|
+| `source_document` | `document_id`, `account_id`, `received_at`, `media_type` | One imported email, HTML body, or file. |
+| `meaning_unit` | `unit_id`, `document_id`, `input_index`, `chunk_index`, `chunk_kind`, `source_offset`, `source_length`, `chunk_text` | One retrieval grain. `chunk_text` equals the source slice. |
+| `unit_embedding` | `unit_id`, `model_name`, `embedding_vector`, `prompt_tokens` | One vector per unit per model. Not stored on `meaning_unit` (3NF). |
+| `embedded_image` | `image_id`, `document_id`, `source_offset`, `source_length`, `media_type` | Position of a `data:image` span. OCR text and object tags belong on child tables, not here. |
+| `image_text_span` | `image_id`, `span_index`, `ocr_text` | Recognized text for one image (future NIM/OCR job). |
+| `image_object_tag` | `image_id`, `tag_index`, `object_label` | Detected objects for image search (future). |
+
+Do not store greeting text on `unit_embedding`. Do not collapse invoice and
+greeting into one `source_document` row and call it searched.
+
## Persistence Stack
- PostgreSQL for the primary store.
diff --git a/docs/fuzzing.md b/docs/fuzzing.md
index 9897b2bd2..57115ae6a 100644
--- a/docs/fuzzing.md
+++ b/docs/fuzzing.md
@@ -31,6 +31,9 @@ deserialize request config validate untrusted input"`):
4. **End-to-end orchestration** — `orchestrator.TaskOrchestrator.run` against
`mock://` providers (fully offline). Arbitrary prompt text and mode must
produce a JSON-serialisable record whose SSE framing round-trips.
+5. **Meaning-unit chunking** — `semantic_chunking.meaning_unit_chunks`.
+ Arbitrary email/HTML/image text must yield non-overlapping source spans
+ whose `chunk_text` equals the original slice.
## Running locally
diff --git a/docs/library_research.md b/docs/library_research.md
index 42c7fa95c..1460e2737 100644
--- a/docs/library_research.md
+++ b/docs/library_research.md
@@ -53,6 +53,19 @@ Extraction triggers:
Until those triggers exist, Ponytail recommends strengthening the current
single-repo product instead of splitting it.
+## Meaning-unit embedding chunking
+
+Researched before adding `contextual_orchestrator/semantic_chunking.py`:
+
+| Library / standard | Decision | Evidence |
+|---|---|---|
+| [Unicode Standard Annex #29](https://www.unicode.org/reports/tr29/) | Cite as the sentence/word boundary authority; implement a stdlib subset (`.?!。！？` + capital/Hangul continuation, abbreviation merge). | UAX #29 is the current international text-segmentation standard. A full UCD dependency is not justified for email/HTML retrieval grain. |
+| [LangChain RecursiveCharacterTextSplitter](https://python.langchain.com/docs/how_to/recursive_text_splitter/) | Skip. | Adds a runtime dependency and splits on character budgets, then the coordinator already token-splits and averages parts back to one vector. |
+| [LlamaIndex SentenceSplitter](https://docs.llamaindex.ai/) | Skip. | Same dependency and token-window grain. Does not isolate email parties or `data:image` offsets. |
+| Similarity-breakpoint “semantic chunking” (LangChain / LlamaIndex) | Skip. | Qu et al. (2025) found computational cost is not justified by consistent retrieval gains. |
+
+Selected: stdlib regular expressions plus paragraph/email/HTML/image detectors. Custom code that was deliberately skipped: LLM-as-chunker, perplexity chunking, and any new pip dependency.
+
## Required For New Designs
Every new subsystem design must update this file before implementation starts. The entry must name the existing libraries researched, the selected library or stdlib alternative, and the custom code that was deliberately skipped.
diff --git a/docs/meaning_unit_chunking.md b/docs/meaning_unit_chunking.md
new file mode 100644
index 000000000..338f4e5db
--- /dev/null
+++ b/docs/meaning_unit_chunking.md
@@ -0,0 +1,84 @@
+# Meaning-unit embedding chunking
+
+```mermaid
+flowchart LR
+ email[Invoice email] --> chunker[meaning_unit_chunks]
+ chunker --> sender[email_sender]
+ chunker --> greet[greeting paragraph]
+ chunker --> invoice[invoice paragraph]
+ invoice --> search["Search INV-20260816"]
+```
+
+## Buyer next action
+
+On `POST /v1/batch/embeddings`, send `"chunking_strategy": "meaning_units"` with
+the raw email, HTML, or mixed image body. Read `chunk_units` in the completed
+document. Each `chunk_units[i]` is the source slice behind `embeddings[i]`.
+Search for an invoice id against those units — not against one averaged
+document vector.
+
+Omit `chunking_strategy` or send JSON null to keep the existing naruon
+one-vector-per-input contract. `"source_document"` is an explicit alias for
+the same behavior; it does not request chunk expansion or alter token-budget
+map/reduce boundaries.
+
+## Why this exists
+
+`CostRoutingCoordinator` already splits oversized inputs so a provider call
+stays under a token/character ceiling, then **averages those parts back into
+one vector**. That is a transport safety valve. It is not retrieval.
+
+A naruon invoice email that begins with “Good morning” and later says
+`INV-20260816` / `1840.00 USD` must not become a single point. The greeting
+and the balance line are different meaning units (Zhao et al., 2024).
+
+Similarity-breakpoint “semantic chunking” is not used. Qu et al. (2025) found
+that cost is not justified by consistent gains over simpler splits. This
+gateway cuts at linguistic units:
+
+- email parties (`email_sender`, `email_recipient`, `email_subject`, `email_copy`)
+- innermost HTML block leaves (`html_block`), so Gmail wrapper `div` elements
+ do not glue sibling paragraphs together
+- RFC 2397 `data:image` spans (`embedded_image`), including parameters,
+ base64url characters, and RFC 2045 line wrapping, with the original
+ `source_offset` so a later OCR/object-tag job can attach to the same place
+- remaining prose as `body_paragraph` (default retrieval grain)
+
+`unit_grain=body_sentence` is available in-process for UAX #29-style sentence
+cuts. The HTTP field stays `meaning_units` so buyers get paragraph-level
+invoice isolation by default.
+
+## Standalone and as a module
+
+- Standalone: `meaning_unit_chunks(text)` and `expand_embedding_inputs(...)`
+ have no HTTP or provider dependency.
+- As a module: the batch embeddings handler expands inputs before
+ `complete_embeddings_batch`. Token-budget map/reduce still runs **per unit**.
+
+## Embedded images
+
+A base64 image in the body is a first-class unit. The 3NF target is
+`embedded_image` plus child `image_text_span` / `image_object_tag` tables
+(see `docs/database_conventions.md`). This slice records position and media
+type in `chunk_units`; it does not invent OCR text. Live OCR/object tags
+belong on an opt-in NIM job (`NVIDIA_NIM_API_KEY`), not on the default path.
+
+## References
+
+Zhao, J., Ji, Z., Ye, Y., Feng, X., Zhang, X., & Rong, C. (2024).
+*Meta-chunking: Learning text segmentation and semantic completion via
+logical perception*. arXiv. https://doi.org/10.48550/arXiv.2410.12788
+
+Qu, R., Tu, R., & Bao, F. (2025). Is semantic chunking worth the computational
+cost? In *Findings of the Association for Computational Linguistics: NAACL
+2025* (pp. 2012–2027). Association for Computational Linguistics.
+https://aclanthology.org/2025.findings-naacl.114/
+
+Unicode Consortium. (2024). *Unicode Standard Annex #29: Unicode text
+segmentation*. https://www.unicode.org/reports/tr29/
+
+Lewis, P., Perez, E., Piktus, A., Petroni, F., Karpukhin, V., Goyal, N.,
+Küttler, H., Lewis, M., Yih, W., Rocktäschel, T., Riedel, S., & Kiela, D.
+(2020). Retrieval-augmented generation for knowledge-intensive NLP tasks.
+In *Advances in Neural Information Processing Systems, 33*.
+https://doi.org/10.48550/arXiv.2005.11401
diff --git a/docs/papers/README.md b/docs/papers/README.md
index 65a89d2af..f8a76c27c 100644
--- a/docs/papers/README.md
+++ b/docs/papers/README.md
@@ -43,6 +43,37 @@ motivate throughput-oriented **batched** inference and the load-balancing that
makes the latency-tolerant batch route economical. Those sources are referenced
but not vendored here so this repository remains one deployable control plane.
+## Meaning-unit retrieval chunking
+
+Embedding search is only useful when each vector is a meaning unit a buyer can
+ask for (invoice line, sender, HTML block), not a token-budget fragment that is
+later averaged away.
+
+- Zhao, J., Ji, Z., Ye, Y., Feng, X., Zhang, X., & Rong, C. (2024). *Meta-chunking: Learning text segmentation and semantic completion via logical perception*. arXiv. https://doi.org/10.48550/arXiv.2410.12788
+ `meta-chunking-2410.12788.pdf` when the arXiv PDF is vendored. Grounds
+ paragraph-level meta-chunks: sequential sentences inside a paragraph that
+ share a logical relation (here, invoice identifier + balance due) stay
+ together, while the greeting is a separate unit.
+- Qu, R., Tu, R., & Bao, F. (2025). Is semantic chunking worth the computational
+ cost? In *Findings of the Association for Computational Linguistics: NAACL
+ 2025* (pp. 2012–2027). Association for Computational Linguistics.
+ https://aclanthology.org/2025.findings-naacl.114/
+ Cite + link + summary only (ACL anthology HTML/PDF redistribution is not
+ assumed). Similarity-breakpoint chunking did not consistently beat fixed-size
+ splits; this gateway therefore uses linguistic meaning units, not embedding
+ distance cuts.
+- Unicode Consortium. (2024). *Unicode Standard Annex #29: Unicode text
+ segmentation*. https://www.unicode.org/reports/tr29/
+ Cite + link (Unicode copyright). Sentence cuts in leftover prose follow UAX
+ #29 intent (terminator + continuation) without vendoring the Unicode database.
+- Lewis, P., Perez, E., Piktus, A., Petroni, F., Karpukhin, V., Goyal, N.,
+ Küttler, H., Lewis, M., Yih, W., Rocktäschel, T., Riedel, S., & Kiela, D.
+ (2020). Retrieval-augmented generation for knowledge-intensive NLP tasks. In
+ *Advances in Neural Information Processing Systems, 33*.
+ https://doi.org/10.48550/arXiv.2005.11401
+ Grounds why the retrieved *chunk*, not the whole document, is the generation
+ context. `rag-2005.11401.pdf` when vendored.
+
> Citations are provided for scholarly attribution. Redistribution here relies
> on the arXiv non-exclusive distribution license each author granted; no
> GPL/AGPL-licensed material is vendored anywhere in this repository.
diff --git a/docs/papers/meta-chunking-2410.12788.pdf b/docs/papers/meta-chunking-2410.12788.pdf
new file mode 100644
index 000000000..215760303
Binary files /dev/null and b/docs/papers/meta-chunking-2410.12788.pdf differ
diff --git a/docs/papers/rag-2005.11401.pdf b/docs/papers/rag-2005.11401.pdf
new file mode 100644
index 000000000..39218379c
Binary files /dev/null and b/docs/papers/rag-2005.11401.pdf differ
diff --git a/docs/rest_api_design.md b/docs/rest_api_design.md
index 9378e5a37..61c796732 100644
--- a/docs/rest_api_design.md
+++ b/docs/rest_api_design.md
@@ -15,7 +15,7 @@
|---|---|---|
| `GET` | `/openapi.json` | API contract |
| `POST` | `/v1/chat/completions` | Compatibility chat endpoint |
-| `POST` | `/v1/batch/embeddings` | Submit a bulk, latency-tolerant embeddings batch; oversized inputs are token-split before routing via pg-llm-batch |
+| `POST` | `/v1/batch/embeddings` | Submit a bulk, latency-tolerant embeddings batch; oversized inputs are token-split before routing via pg-llm-batch. Send `chunking_strategy=meaning_units` to embed email, HTML, image, and paragraph units separately and read `chunk_units` for source offsets. |
| `GET` | `/v1/batch/embeddings/{batch_id}` | Poll an embeddings batch; returns reduced vectors + recorded cost once completed |
| `GET` | `/api/v1/agent_pools` | List model agents |
| `GET` | `/api/v1/orchestration_policies/default_policy` | Read active policy |
diff --git a/docs/user_stories.md b/docs/user_stories.md
index 7f476d41f..2456950cd 100644
--- a/docs/user_stories.md
+++ b/docs/user_stories.md
@@ -29,6 +29,7 @@ These stories are derived from the product planning reboot, not from generic adm
## API Consumer
- As an API consumer, I want a single chat-completion compatible endpoint so that I can adopt orchestration without rewriting client code.
+- As an API consumer, I want `/v1/batch/embeddings` to optionally embed email, HTML, image, and paragraph meaning units so that a search for an invoice number returns the balance line instead of the greeting.
- As an API consumer, I want resource-oriented REST endpoints so that enterprise integrations can manage pools, policies, workflow runs, and locales.
## Localization Manager
diff --git a/fuzz/targets.py b/fuzz/targets.py
index d0c344462..7854d3ea8 100644
--- a/fuzz/targets.py
+++ b/fuzz/targets.py
@@ -28,6 +28,7 @@
from typing import Any
from contextual_orchestrator import server
+from contextual_orchestrator.semantic_chunking import meaning_unit_chunks
from contextual_orchestrator.orchestrator import (
ModelAgent,
TaskOrchestrator,
@@ -188,3 +189,26 @@ def exercise_orchestration(prompt: str, mode: str) -> None:
continue
assert frame.startswith("data: ")
json.loads(frame[len("data: "):])
+
+
+def exercise_meaning_unit_chunks(text: str) -> None:
+ """Meaning-unit parser must keep exact source spans and never overlap.
+
+ Arbitrary prompt/email/HTML text is a retrieval input. Units are either a
+ well-formed non-overlapping cover of slices or a single source_document
+ fallback — never a crash or a span that does not match the source.
+ """
+ units = meaning_unit_chunks(text)
+ assert isinstance(units, list)
+ seen: list[tuple[int, int]] = []
+ for unit in units:
+ start = unit.source_offset
+ end = start + unit.source_length
+ assert unit.chunk_text == text[start:end]
+ assert unit.source_length == len(unit.chunk_text)
+ assert start >= 0
+ assert end <= len(text)
+ for other_start, other_end in seen:
+ assert end <= other_start or start >= other_end
+ seen.append((start, end))
+ meaning_unit_chunks(text, unit_grain="body_sentence")
diff --git a/tests/fuzz/test_fuzz_properties.py b/tests/fuzz/test_fuzz_properties.py
index 7e7b3f347..11403a473 100644
--- a/tests/fuzz/test_fuzz_properties.py
+++ b/tests/fuzz/test_fuzz_properties.py
@@ -18,6 +18,7 @@
from fuzz.targets import (
exercise_agent_config,
+ exercise_meaning_unit_chunks,
exercise_orchestration,
exercise_redaction,
exercise_request_body,
@@ -108,3 +109,9 @@ def test_redaction_never_crashes_and_is_idempotent(text: str) -> None:
)
def test_orchestration_on_arbitrary_prompt(prompt: str, mode: str) -> None:
exercise_orchestration(prompt, mode)
+
+
+@_SETTINGS
+@given(st.text(max_size=2048))
+def test_meaning_unit_chunks_keep_source_spans(text: str) -> None:
+ exercise_meaning_unit_chunks(text)
diff --git a/tests/test_embeddings_meaning_units_http_honesty.py b/tests/test_embeddings_meaning_units_http_honesty.py
new file mode 100644
index 000000000..308c8e577
--- /dev/null
+++ b/tests/test_embeddings_meaning_units_http_honesty.py
@@ -0,0 +1,204 @@
+"""HTTP honesty: meaning-unit chunking on /v1/batch/embeddings.
+
+Omit keeps the naruon one-vector-per-input contract. ``meaning_units`` embeds
+each email/HTML/paragraph unit separately and returns ``chunk_units`` so a
+buyer can map vectors back to the invoice line. Unknown strategies fail closed.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sys
+import threading
+import urllib.error
+import urllib.request
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from contextual_orchestrator import ( # noqa: E402
+ CostRoutingCoordinator,
+ InMemoryConfigStore,
+ ModelAgent,
+ PriceBook,
+ TaskOrchestrator,
+)
+from contextual_orchestrator.semantic_chunking import MeaningUnit, rank_meaning_units # noqa: E402
+from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402
+
+INVOICE_EMAIL = """From: alice.billing@acme.example
+To: ap@buyer.example
+Subject: Invoice INV-20260816 is due
+
+Good morning. Thank you for your continued partnership this quarter.
+
+Please remit payment for invoice INV-20260816. The balance due is 1840.00 USD by 2026-08-30.
+
+Kind regards,
+Alice Billing
+"""
+INVOICE_QUERY = "invoice INV-20260816 balance due 1840.00 USD"
+
+_TEST_AUTH_TOKEN = "meaning_units_http_honesty_token" # noqa: S105
+
+
+def _serve():
+ orchestrator = TaskOrchestrator(
+ [
+ ModelAgent(
+ id="mock_worker",
+ model="mock-a",
+ base_url="mock://a",
+ provider_name="mock",
+ tags=("reasoning", "writing"),
+ priority=1,
+ )
+ ]
+ )
+ coordinator = CostRoutingCoordinator(
+ orchestrator, InMemoryConfigStore(), price_book=PriceBook(InMemoryConfigStore())
+ )
+ server = build_server(
+ orchestrator,
+ port=0,
+ security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN),
+ coordinator=coordinator,
+ )
+ thread = threading.Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+ return server, thread, server.server_address[1]
+
+
+def _post(port: int, payload: dict) -> tuple[int, dict]:
+ request = urllib.request.Request(
+ f"http://127.0.0.1:{port}/v1/batch/embeddings",
+ data=json.dumps(payload).encode("utf-8"),
+ headers={
+ "content-type": "application/json",
+ "authorization": f"Bearer {_TEST_AUTH_TOKEN}",
+ "connection": "close",
+ },
+ method="POST",
+ )
+ try:
+ with urllib.request.urlopen(request, timeout=15) as response:
+ return response.status, json.loads(response.read().decode("utf-8"))
+ except urllib.error.HTTPError as exc:
+ return exc.code, json.loads(exc.read().decode("utf-8"))
+
+
+def test_http_omit_keeps_one_vector_per_input() -> None:
+ server, thread, port = _serve()
+ try:
+ status, body = _post(
+ port,
+ {"model": "mock-a", "inputs": ["alpha body", "beta attachment"]},
+ )
+ assert status == 200, body
+ assert "chunk_units" not in body
+ assert len(body["embeddings"]) == 2
+ finally:
+ server.shutdown()
+ thread.join(timeout=5)
+
+
+def test_http_meaning_units_exposes_invoice_chunk_for_search() -> None:
+ server, thread, port = _serve()
+ try:
+ status, body = _post(
+ port,
+ {
+ "model": "mock-a",
+ "inputs": [INVOICE_EMAIL],
+ "chunking_strategy": "meaning_units",
+ },
+ )
+ assert status == 200, body
+ units = [MeaningUnit(**item) for item in body["chunk_units"]]
+ assert len(body["embeddings"]) == len(units)
+ assert len(units) > 1
+ ranked = rank_meaning_units(INVOICE_QUERY, units)
+ assert "INV-20260816" in ranked[0].chunk_text
+ assert "1840.00" in ranked[0].chunk_text
+ assert "Good morning" not in ranked[0].chunk_text
+ poll_status, poll_body = _get(port, body["batch_id"])
+ assert poll_status == 200, poll_body
+ assert poll_body["chunk_units"] == body["chunk_units"]
+ finally:
+ server.shutdown()
+ thread.join(timeout=5)
+
+
+def _get(port: int, batch_id: str) -> tuple[int, dict]:
+ request = urllib.request.Request(
+ f"http://127.0.0.1:{port}/v1/batch/embeddings/{batch_id}",
+ headers={"authorization": f"Bearer {_TEST_AUTH_TOKEN}", "connection": "close"},
+ method="GET",
+ )
+ with urllib.request.urlopen(request, timeout=15) as response:
+ return response.status, json.loads(response.read().decode("utf-8"))
+
+
+def test_http_unknown_chunking_strategy_fails_closed() -> None:
+ server, thread, port = _serve()
+ try:
+ status, body = _post(
+ port,
+ {
+ "model": "mock-a",
+ "inputs": [INVOICE_EMAIL],
+ "chunking_strategy": "tokens",
+ },
+ )
+ assert status == 400, body
+ assert body["error_code"] == "invalid_chunking_strategy"
+ assert "meaning_units" in body["error_message"]
+ finally:
+ server.shutdown()
+ thread.join(timeout=5)
+
+
+def test_http_null_chunking_strategy_is_omit() -> None:
+ server, thread, port = _serve()
+ try:
+ status, body = _post(
+ port,
+ {
+ "model": "mock-a",
+ "inputs": ["one document"],
+ "chunking_strategy": None,
+ },
+ )
+ assert status == 200, body
+ assert "chunk_units" not in body
+ assert len(body["embeddings"]) == 1
+ finally:
+ server.shutdown()
+ thread.join(timeout=5)
+
+
+def test_http_non_string_chunking_strategy_fails_closed() -> None:
+ server, thread, port = _serve()
+ try:
+ status, body = _post(
+ port,
+ {
+ "model": "mock-a",
+ "inputs": ["one document"],
+ "chunking_strategy": True,
+ },
+ )
+ assert status == 400, body
+ assert body["error_code"] == "invalid_chunking_strategy"
+ finally:
+ server.shutdown()
+ thread.join(timeout=5)
+
+
+if __name__ == "__main__":
+ test_http_omit_keeps_one_vector_per_input()
+ test_http_meaning_units_exposes_invoice_chunk_for_search()
+ test_http_unknown_chunking_strategy_fails_closed()
+ test_http_null_chunking_strategy_is_omit()
+ test_http_non_string_chunking_strategy_fails_closed()
+ print("ok")
diff --git a/tests/test_meaning_unit_chunking.py b/tests/test_meaning_unit_chunking.py
new file mode 100644
index 000000000..5dd622e5a
--- /dev/null
+++ b/tests/test_meaning_unit_chunking.py
@@ -0,0 +1,221 @@
+"""Meaning-unit chunking isolates searchable facts from real documents.
+
+Buyers embed naruon-style email, HTML, and mixed image+text bodies. Token-budget
+splits keep provider calls under a ceiling and then average parts back into one
+vector, which hides the invoice line inside the greeting. These tests require
+the chunker to emit linguistic meaning units so a lexical retriever can rank the
+invoice unit first for an invoice query (Zhao et al., 2024; Unicode, 2024).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import re
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from contextual_orchestrator.semantic_chunking import ( # noqa: E402
+ expand_embedding_inputs,
+ meaning_unit_chunks,
+ rank_meaning_units,
+)
+
+INVOICE_EMAIL = """From: alice.billing@acme.example
+To: ap@buyer.example
+Subject: Invoice INV-20260816 is due
+
+Good morning. Thank you for your continued partnership this quarter.
+
+Please remit payment for invoice INV-20260816. The balance due is 1840.00 USD by 2026-08-30.
+
+Kind regards,
+Alice Billing
+"""
+
+INVOICE_HTML = (
+ "Good morning from support.
"
+ "Invoice INV-20260816 balance due is 1840.00 USD.
"
+)
+
+INVOICE_WITH_IMAGE = (
+ "See the scanned invoice below.\n"
+ "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII=\n"
+ "The amount on that scan is 1840.00 USD for INV-20260816."
+)
+
+INVOICE_QUERY = "invoice INV-20260816 balance due 1840.00 USD"
+
+
+def _assert_span(text: str, unit) -> None:
+ assert unit.chunk_text == text[unit.source_offset : unit.source_offset + unit.source_length]
+ assert unit.source_length == len(unit.chunk_text)
+ assert unit.source_offset >= 0
+
+
+def test_invoice_email_isolates_sender_subject_and_balance_line() -> None:
+ units = meaning_unit_chunks(INVOICE_EMAIL)
+ assert units, "an invoice email must produce at least one meaning unit"
+ for unit in units:
+ _assert_span(INVOICE_EMAIL, unit)
+
+ kinds = {unit.chunk_kind for unit in units}
+ assert "email_sender" in kinds
+ assert "email_recipient" in kinds
+ assert "email_subject" in kinds
+
+ sender = next(unit for unit in units if unit.chunk_kind == "email_sender")
+ assert "alice.billing@acme.example" in sender.chunk_text
+ assert "INV-20260816" not in sender.chunk_text
+
+ invoice_units = [unit for unit in units if "INV-20260816" in unit.chunk_text]
+ assert invoice_units, "the invoice identifier must land in a dedicated unit"
+ greeting_only = [
+ unit
+ for unit in units
+ if "Good morning" in unit.chunk_text and "INV-20260816" not in unit.chunk_text
+ ]
+ assert greeting_only, "the greeting must not be glued to the invoice line"
+
+
+def test_invoice_query_ranks_the_balance_unit_first() -> None:
+ units = meaning_unit_chunks(INVOICE_EMAIL)
+ ranked = rank_meaning_units(INVOICE_QUERY, units)
+ assert ranked, "ranking must return the invoice units"
+ top = ranked[0]
+ assert "INV-20260816" in top.chunk_text
+ assert "1840.00" in top.chunk_text
+ assert "Good morning" not in top.chunk_text
+
+
+def test_html_blocks_keep_invoice_out_of_the_greeting() -> None:
+ units = meaning_unit_chunks(INVOICE_HTML)
+ for unit in units:
+ _assert_span(INVOICE_HTML, unit)
+ greeting = next(unit for unit in units if "Good morning" in unit.chunk_text)
+ invoice = next(unit for unit in units if "INV-20260816" in unit.chunk_text)
+ assert greeting is not invoice
+ assert "INV-20260816" not in greeting.chunk_text
+ assert "Good morning" not in invoice.chunk_text
+
+
+def test_embedded_image_keeps_source_offset_and_neighbors() -> None:
+ units = meaning_unit_chunks(INVOICE_WITH_IMAGE)
+ image = next(unit for unit in units if unit.chunk_kind == "embedded_image")
+ _assert_span(INVOICE_WITH_IMAGE, image)
+ assert INVOICE_WITH_IMAGE[image.source_offset :].startswith("data:image/png;base64,")
+ invoice = next(unit for unit in units if "INV-20260816" in unit.chunk_text)
+ assert invoice.chunk_kind != "embedded_image"
+ assert image.source_offset < invoice.source_offset
+
+
+def test_expand_embedding_inputs_preserves_input_index() -> None:
+ texts, units = expand_embedding_inputs(
+ [INVOICE_EMAIL, "single note"],
+ chunking_strategy="meaning_units",
+ )
+ assert len(texts) == len(units)
+ assert texts == [unit.chunk_text for unit in units]
+ assert {unit.input_index for unit in units} == {0, 1}
+ assert any(unit.input_index == 1 and unit.chunk_text == "single note" for unit in units)
+
+
+def test_expand_omitted_strategy_keeps_one_document_unit() -> None:
+ texts, units = expand_embedding_inputs([INVOICE_EMAIL], chunking_strategy=None)
+ assert texts == [INVOICE_EMAIL]
+ assert len(units) == 1
+ assert units[0].chunk_kind == "source_document"
+
+
+def test_units_do_not_overlap() -> None:
+ units = meaning_unit_chunks(INVOICE_EMAIL)
+ spans = sorted((unit.source_offset, unit.source_offset + unit.source_length) for unit in units)
+ for previous, current in zip(spans, spans[1:]):
+ assert previous[1] <= current[0]
+
+
+def test_korean_invoice_sentence_is_its_own_unit() -> None:
+ body = (
+ "안녕하세요. 이번 분기 정산 안내입니다.\n\n"
+ "청구서 INV-20260816의 미납 잔액은 1,840.00 USD입니다."
+ )
+ units = meaning_unit_chunks(body)
+ invoice = next(unit for unit in units if "INV-20260816" in unit.chunk_text)
+ assert "안녕하세요" not in invoice.chunk_text
+ ranked = rank_meaning_units("INV-20260816 미납 잔액", units)
+ assert "INV-20260816" in ranked[0].chunk_text
+
+
+def test_sentence_grain_keeps_greeting_as_its_own_unit() -> None:
+ units = meaning_unit_chunks(INVOICE_EMAIL, unit_grain="body_sentence")
+ greeting = next(unit for unit in units if unit.chunk_text == "Good morning.")
+ assert greeting.chunk_kind == "body_sentence"
+ assert "INV-20260816" not in greeting.chunk_text
+
+
+def test_copy_and_reply_headers_are_their_own_units() -> None:
+ text = (
+ "From: alice.billing@acme.example\n"
+ "To: ap@buyer.example\n"
+ "Cc: audit@acme.example\n"
+ "Reply-To: alice.billing@acme.example\n"
+ "Subject: Invoice INV-20260816 is due\n\n"
+ "Please remit payment for invoice INV-20260816."
+ )
+ units = meaning_unit_chunks(text)
+ kinds = {unit.chunk_kind for unit in units}
+ assert "email_copy" in kinds
+ assert "email_header" in kinds
+
+
+def test_source_document_strategy_is_omit_equivalent() -> None:
+ texts, units = expand_embedding_inputs(["note"], chunking_strategy="source_document")
+ assert texts == ["note"]
+ assert units[0].chunk_kind == "source_document"
+
+
+def test_whitespace_input_is_one_source_document() -> None:
+ units = meaning_unit_chunks(" ")
+ assert len(units) == 1
+ assert units[0].chunk_kind == "source_document"
+ assert units[0].chunk_text == " "
+
+
+def test_unknown_chunking_strategy_is_rejected() -> None:
+ try:
+ expand_embedding_inputs(["note"], chunking_strategy="tokens")
+ except ValueError:
+ return
+ raise AssertionError("expected ValueError for an unknown chunking_strategy")
+
+
+def test_sentence_grain_keeps_title_abbreviation_with_the_invoice() -> None:
+ text = "Dr. Smith paid invoice INV-20260816. Goodbye later."
+ units = meaning_unit_chunks(text, unit_grain="body_sentence")
+ invoice = next(unit for unit in units if "INV-20260816" in unit.chunk_text)
+ assert "Dr. Smith" in invoice.chunk_text
+
+
+def test_token_overlap_ignores_punctuation() -> None:
+ score = rank_meaning_units("INV-20260816", meaning_unit_chunks("Pay INV-20260816."))
+ assert score[0].chunk_text
+ assert re.search(r"INV-20260816", score[0].chunk_text)
+
+
+if __name__ == "__main__":
+ test_invoice_email_isolates_sender_subject_and_balance_line()
+ test_invoice_query_ranks_the_balance_unit_first()
+ test_html_blocks_keep_invoice_out_of_the_greeting()
+ test_embedded_image_keeps_source_offset_and_neighbors()
+ test_expand_embedding_inputs_preserves_input_index()
+ test_expand_omitted_strategy_keeps_one_document_unit()
+ test_units_do_not_overlap()
+ test_korean_invoice_sentence_is_its_own_unit()
+ test_sentence_grain_keeps_greeting_as_its_own_unit()
+ test_copy_and_reply_headers_are_their_own_units()
+ test_source_document_strategy_is_omit_equivalent()
+ test_whitespace_input_is_one_source_document()
+ test_unknown_chunking_strategy_is_rejected()
+ test_sentence_grain_keeps_title_abbreviation_with_the_invoice()
+ test_token_overlap_ignores_punctuation()
+ print("ok")
diff --git a/tests/test_pr652_review_regressions.py b/tests/test_pr652_review_regressions.py
new file mode 100644
index 000000000..c8b3183a8
--- /dev/null
+++ b/tests/test_pr652_review_regressions.py
@@ -0,0 +1,140 @@
+"""RED regressions for the unresolved review threads on PR #652.
+
+These cases model the inputs that buyers actually submit: Gmail wrapper HTML,
+RFC 2397 image data URLs emitted by MIME-aware clients, RFC 5322-like mail
+bodies, and the public batch-embeddings contract. The timing comparison locks
+linear scanning rather than a machine-specific absolute duration.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import statistics
+import sys
+import time
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from contextual_orchestrator.semantic_chunking import meaning_unit_chunks # noqa: E402
+from tests.test_embeddings_meaning_units_http_honesty import _post, _serve # noqa: E402
+
+
+def _assert_exact_span(source: str, unit) -> None:
+ assert unit.chunk_text == source[
+ unit.source_offset : unit.source_offset + unit.source_length
+ ]
+
+
+def test_gmail_wrapper_html_emits_leaf_units() -> None:
+ source = (
+ ''
+ "
Good morning from support.
"
+ "
Invoice INV-20260816 balance due is 1840.00 USD.
"
+ "
")
+ assert invoice.chunk_text.startswith("
")
+ assert "INV-20260816" not in greeting.chunk_text
+ assert "Good morning" not in invoice.chunk_text
+
+
+def test_rfc2397_image_parameters_stay_in_one_exact_unit() -> None:
+ data_url = (
+ "data:image/png;charset=utf-8;name=invoice.png;base64,"
+ "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8="
+ )
+ source = f"Scan follows.\n{data_url}\nInvoice INV-20260816 remains open."
+ units = meaning_unit_chunks(source)
+ image = next(unit for unit in units if unit.chunk_kind == "embedded_image")
+ _assert_exact_span(source, image)
+ assert image.chunk_text == data_url
+ invoice = next(unit for unit in units if "INV-20260816" in unit.chunk_text)
+ assert invoice.chunk_kind != "embedded_image"
+ assert "data:image" not in invoice.chunk_text
+
+
+def test_rfc2397_urlsafe_payload_is_not_truncated() -> None:
+ data_url = "data:image/png;base64,QUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVo-_w=="
+ source = f"{data_url}\nInvoice INV-20260816 remains open."
+ units = meaning_unit_chunks(source)
+ image = next(unit for unit in units if unit.chunk_kind == "embedded_image")
+ _assert_exact_span(source, image)
+ assert image.chunk_text == data_url
+ assert image.chunk_text.endswith("-_w==")
+
+
+def test_rfc2397_mime_wrapped_payload_is_one_exact_unit() -> None:
+ payload = (
+ "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8x8AAwMCAO"
+ "ip1sAAAAASUVORK5CYII="
+ )
+ first, second = payload[:76], payload[76:]
+ data_url = f"data:image/png;base64,{first}\r\n{second}"
+ source = f"Scan follows.\r\n{data_url}\r\nInvoice INV-20260816 remains open."
+ units = meaning_unit_chunks(source)
+ image = next(unit for unit in units if unit.chunk_kind == "embedded_image")
+ _assert_exact_span(source, image)
+ assert image.chunk_text == data_url
+ invoice = next(unit for unit in units if "INV-20260816" in unit.chunk_text)
+ assert second not in invoice.chunk_text
+ assert invoice.chunk_kind != "embedded_image"
+
+
+def _median_unclosed_html_seconds(openers: int) -> float:
+ source = "
" * openers + "Invoice INV-20260816 remains open."
+ samples: list[float] = []
+ for _ in range(3):
+ started = time.perf_counter()
+ meaning_unit_chunks(source)
+ samples.append(time.perf_counter() - started)
+ return statistics.median(samples)
+
+
+def test_unclosed_html_scan_scales_near_linearly() -> None:
+ small = _median_unclosed_html_seconds(1_500)
+ large = _median_unclosed_html_seconds(3_000)
+ assert large < small * 3.0, (
+ "doubling unclosed wrapper input must not approach quadratic scanning: "
+ f"small={small:.6f}s large={large:.6f}s ratio={large / small:.2f}"
+ )
+
+
+def test_subject_line_inside_body_is_not_an_email_header() -> None:
+ source = (
+ "From: alice.billing@acme.example\n"
+ "To: ap@buyer.example\n"
+ "Subject: Invoice INV-20260816 is due\n"
+ "\n"
+ "Subject: see attached SKU-77 packing list.\n"
+ "\n"
+ "Invoice INV-20260816 remains open."
+ )
+ units = meaning_unit_chunks(source)
+ body_subject = next(unit for unit in units if "SKU-77" in unit.chunk_text)
+ assert body_subject.chunk_kind == "body_paragraph"
+ assert len([unit for unit in units if unit.chunk_kind == "email_subject"]) == 1
+
+
+def test_http_source_document_strategy_is_omit_alias() -> None:
+ server, thread, port = _serve()
+ try:
+ status, body = _post(
+ port,
+ {
+ "model": "mock-a",
+ "inputs": ["one document"],
+ "chunking_strategy": "source_document",
+ },
+ )
+ assert status == 200, body
+ assert "chunk_units" not in body
+ assert len(body["embeddings"]) == 1
+ finally:
+ server.shutdown()
+ thread.join(timeout=5)
diff --git a/tests/test_pr652_semgrep_regressions.py b/tests/test_pr652_semgrep_regressions.py
new file mode 100644
index 000000000..b8cf072dd
--- /dev/null
+++ b/tests/test_pr652_semgrep_regressions.py
@@ -0,0 +1,33 @@
+"""Regression tests for the exact Semgrep root causes on PR #652."""
+
+from __future__ import annotations
+
+import sqlite3
+import urllib.request
+
+from contextual_orchestrator.cost_ledger import SqlLedgerStore
+from contextual_orchestrator.orchestrator import ModelClient
+
+
+def test_sql_ledger_rejects_unknown_paramstyle() -> None:
+ connection = sqlite3.connect(":memory:")
+ try:
+ try:
+ SqlLedgerStore(connection, paramstyle="format")
+ except ValueError as exc:
+ assert "unsupported ledger paramstyle" in str(exc)
+ else:
+ raise AssertionError("unrecognized placeholder grammar must fail closed")
+ finally:
+ connection.close()
+
+
+def test_provider_transport_rejects_local_file_before_io() -> None:
+ client = ModelClient()
+ request = urllib.request.Request("file:///etc/passwd")
+ try:
+ client._open_provider(request)
+ except RuntimeError as exc:
+ assert "absolute URL" in str(exc) or "requires https" in str(exc)
+ else:
+ raise AssertionError("local-file provider transport must fail before I/O")
diff --git a/tests/test_provider_tls.py b/tests/test_provider_tls.py
index e78e90f42..d06556a8d 100644
--- a/tests/test_provider_tls.py
+++ b/tests/test_provider_tls.py
@@ -24,10 +24,14 @@ def test_default_verifies_against_system_store() -> None:
assert context.check_hostname is True
-def test_insecure_skip_verify_disables_checks() -> None:
- context = ModelClient(verify_tls=False)._ssl_context
- assert context.verify_mode == ssl.CERT_NONE
- assert context.check_hostname is False
+def test_insecure_skip_verify_is_rejected() -> None:
+ try:
+ ModelClient(verify_tls=False)
+ except ValueError as exc:
+ assert "cannot be disabled" in str(exc)
+ assert "ca_bundle" in str(exc)
+ else:
+ raise AssertionError("provider TLS verification must remain mandatory")
def test_ca_bundle_is_loaded() -> None: