diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6a4b6f66f..5dcc814f1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -52,6 +52,8 @@ All notable changes to this project are documented here. Format follows
### Fixed
+- Structure adjudication now rejects malformed or duplicate unit indexes before
+ calling the orchestrator.
- Event Lineage's DAG no longer leaves a linear (no-branch) reconstruct
chain unexplained. `is_branch_point` is only `true` when a post has 2+
children -- correct reconstruct behavior, not a bug -- but the graph
diff --git a/lineageweave/http_client.py b/lineageweave/http_client.py
index 5edc3e696..0f4be13df 100644
--- a/lineageweave/http_client.py
+++ b/lineageweave/http_client.py
@@ -31,6 +31,22 @@ class HttpClientError(RuntimeError):
"""The remote endpoint failed, returned a non-success status, or invalid JSON."""
+def json_request_body(payload: dict) -> bytes:
+ """Serialize the exact JSON body sent by :func:`post_json`."""
+ request_payload = payload
+ request_metadata = current_llm_metadata()
+ if request_metadata:
+ request_payload = dict(payload)
+ existing_metadata = request_payload.get("metadata")
+ if existing_metadata is None:
+ request_payload["metadata"] = request_metadata
+ elif isinstance(existing_metadata, dict):
+ request_payload["metadata"] = {**existing_metadata, **request_metadata}
+ else:
+ raise ValueError("metadata must be an object")
+ return json.dumps(request_payload).encode("utf-8")
+
+
def _validated_response_limit(value: int | None) -> int | None:
"""Return a positive byte limit or reject ambiguous numeric values."""
@@ -237,25 +253,10 @@ def post_json(
ValueError: ``url`` is not an ``http`` / ``https`` URL with a host.
HttpClientError: the server responded with HTTP >= 400 or non-JSON.
"""
-
- request_payload = payload
- request_metadata = current_llm_metadata()
- if request_metadata:
- request_payload = dict(payload)
- existing_metadata = request_payload.get("metadata")
- if existing_metadata is None:
- request_payload["metadata"] = request_metadata
- elif isinstance(existing_metadata, dict):
- request_payload["metadata"] = {
- **existing_metadata,
- **request_metadata,
- }
- else:
- raise ValueError("metadata must be an object")
status, raw = _request(
"POST",
url,
- body=json.dumps(request_payload).encode("utf-8"),
+ body=json_request_body(payload),
headers={"content-type": "application/json", **headers},
timeout=timeout,
)
diff --git a/lineageweave/post_content_persistence.py b/lineageweave/post_content_persistence.py
index 86dcfd4de..c070a17b0 100644
--- a/lineageweave/post_content_persistence.py
+++ b/lineageweave/post_content_persistence.py
@@ -9,6 +9,7 @@
import asyncio
import hashlib
+import json
import logging
import math
from typing import Any, TypeVar
@@ -16,8 +17,10 @@
from .chunking import Chunk, chunk_by_source_body
from .embedding_client import EmbeddingClient
from .image_content import ImageContentClient, ImageDescription
+from .http_client import HttpClientError, json_request_body
from .post_content_normalization import ImageContentResult, normalize_post_body
from .post_structure import (
+ ContextualOrchestratorPostStructureClient,
NullPostStructureClient,
PostStructureClient,
StructureDecision,
@@ -31,14 +34,19 @@
def _bounded_unit_batches( # noqa: UP047 - retain Python 3.10 compatibility.
- units: list[tuple[_BatchKey, str]],
-) -> list[list[tuple[_BatchKey, str]]]:
+ units: list[tuple[_BatchKey, str | dict[str, object]]],
+) -> list[list[tuple[_BatchKey, str | dict[str, object]]]]:
"""Keep provider requests bounded without changing persisted source units."""
- batches: list[list[tuple[_BatchKey, str]]] = []
- batch: list[tuple[_BatchKey, str]] = []
+ batches: list[list[tuple[_BatchKey, str | dict[str, object]]]] = []
+ batch: list[tuple[_BatchKey, str | dict[str, object]]] = []
batch_chars = 0
for unit in units:
- unit_chars = len(unit[1])
+ payload = unit[1]
+ unit_chars = (
+ len(payload)
+ if isinstance(payload, str)
+ else len(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))
+ )
if batch and (
len(batch) >= _LLM_BATCH_MAX_UNITS
or batch_chars + unit_chars > _LLM_BATCH_MAX_CHARS
@@ -53,6 +61,32 @@ def _bounded_unit_batches( # noqa: UP047 - retain Python 3.10 compatibility.
return batches
+def _bounded_structure_batches(
+ units: list[tuple[int, dict[str, object]]], post_title: str
+) -> list[list[tuple[int, dict[str, object]]]]:
+ """Bound structure batches by their exact serialized HTTP request body."""
+ batches: list[list[tuple[int, dict[str, object]]]] = []
+ batch: list[tuple[int, dict[str, object]]] = []
+ for unit in units:
+ candidate = [*batch, unit]
+ candidate_body = json_request_body(
+ ContextualOrchestratorPostStructureClient.request_payload(
+ post_title, [payload for _index, payload in candidate]
+ )
+ )
+ if batch and (
+ len(batch) >= _LLM_BATCH_MAX_UNITS
+ or len(candidate_body) > _LLM_BATCH_MAX_CHARS
+ ):
+ batches.append(batch)
+ batch = [unit]
+ else:
+ batch = candidate
+ if batch:
+ batches.append(batch)
+ return batches
+
+
def _render_description(description: ImageDescription | None) -> str:
"""Render one image or visual-region description as searchable text."""
if description is None:
@@ -151,21 +185,38 @@ async def persist_post_content(
structure_units = [
(
chunk.index,
- chunk.text[:_STRUCTURE_UNIT_MAX_CHARS]
- + (
- "\n[truncated for structure adjudication]"
- if len(chunk.text) > _STRUCTURE_UNIT_MAX_CHARS
- else ""
- ),
+ {
+ "unit_index": chunk.index,
+ "text": chunk.text[:_STRUCTURE_UNIT_MAX_CHARS]
+ + (
+ "\n[truncated for structure adjudication]"
+ if len(chunk.text) > _STRUCTURE_UNIT_MAX_CHARS
+ else ""
+ ),
+ "label": chunk.label,
+ "style": formatting.get(chunk.index),
+ "source_indent_width": max(
+ 0,
+ int(chunk.indent_width) - int(chunk.declared_indent_width),
+ ),
+ "declared_indent_width": int(chunk.declared_indent_width),
+ },
)
for chunk in unresolved
]
- for batch in _bounded_unit_batches(structure_units):
+ for batch in _bounded_structure_batches(structure_units, post_title):
try:
+ request_body = json_request_body(
+ ContextualOrchestratorPostStructureClient.request_payload(
+ post_title, [payload for _index, payload in batch]
+ )
+ )
+ if len(request_body) > _LLM_BATCH_MAX_CHARS:
+ raise HttpClientError("structure adjudication request exceeds size limit")
decisions = await asyncio.to_thread(
client.infer,
post_title,
- [{"unit_index": index, "text": text} for index, text in batch],
+ [payload for _index, payload in batch],
)
for decision in decisions:
if decision.unit_index in unresolved_indexes:
diff --git a/lineageweave/post_structure.py b/lineageweave/post_structure.py
index 17dc5f025..ae77e2dcc 100644
--- a/lineageweave/post_structure.py
+++ b/lineageweave/post_structure.py
@@ -5,7 +5,7 @@
import json
import math
from dataclasses import dataclass
-from typing import Any, Protocol
+from typing import Any, ClassVar, Protocol
from .http_client import post_json
@@ -50,7 +50,7 @@ class ContextualOrchestratorPostStructureClient:
available = True
- _DECISION_ITEM_SCHEMA = {
+ _DECISION_ITEM_SCHEMA: ClassVar[dict[str, object]] = {
"type": "object",
"properties": {
"unit_index": {"type": "integer", "minimum": 0},
@@ -61,7 +61,7 @@ class ContextualOrchestratorPostStructureClient:
"required": ["unit_index", "indent_level", "confidence", "evidence"],
"additionalProperties": False,
}
- _DECISION_SCHEMA = {
+ _DECISION_SCHEMA: ClassVar[dict[str, object]] = {
"type": "object",
"properties": {
"decisions": {
@@ -78,55 +78,80 @@ def __init__(self, base_url: str, api_key: str, timeout: float = 600.0):
self.api_key = api_key
self.timeout = timeout
+ @classmethod
+ def request_payload(
+ cls, post_title: str, units: list[dict[str, object]]
+ ) -> dict[str, object]:
+ """Return the canonical orchestrator request for structure inference."""
+ return {
+ "messages": [
+ {
+ "role": "system",
+ "content": (
+ "Adjudicate the indentation level of every supplied document unit. "
+ "Return one JSON object with a decisions array, with one decision for "
+ "each unit_index. Determine indentation from ordering, numbering, "
+ "bullets, paragraph semantics, and visible explicit formatting. The "
+ "input also reports source_indent_width from leading spaces or NBSP "
+ "and declared_indent_width from HTML/CSS/OOXML. Treat declared "
+ "formatting as explicit evidence and source whitespace as supporting "
+ "evidence only. Do not mistake continuation-line alignment after a "
+ "bullet or number for a new hierarchy level. Do not invent nesting. "
+ "If evidence conflicts or is insufficient, use level 0 and low "
+ "confidence."
+ ),
+ },
+ {
+ "role": "user",
+ "content": json.dumps(
+ {"post_title": post_title, "ordered_units": units},
+ ensure_ascii=False,
+ ),
+ },
+ ],
+ "response_format": {
+ "type": "json_schema",
+ "json_schema": {
+ "name": "post_structure_decisions",
+ "strict": True,
+ "schema": cls._DECISION_SCHEMA,
+ },
+ },
+ "mode": "auto",
+ "reasoning_effort": "auto",
+ "max_tokens": 4096,
+ }
+
def infer(
self, post_title: str, units: list[dict[str, object]]
) -> tuple[StructureDecision, ...]:
"""Ask the orchestrator to adjudicate an indent level for each unit."""
if not units:
return ()
+ expected_indexes: set[int] = set()
+ for unit in units:
+ unit_index = unit.get("unit_index") if isinstance(unit, dict) else None
+ if (
+ type(unit_index) is not int
+ or unit_index < 0
+ or unit_index in expected_indexes
+ ):
+ raise ValueError(
+ "structure adjudication units require unique non-negative integer indexes"
+ )
+ expected_indexes.add(unit_index)
response = post_json(
f"{self.base_url}/v1/chat/completions",
- {
- "messages": [
- {
- "role": "system",
- "content": (
- "Adjudicate the indentation level of every supplied document unit. "
- "Return one JSON object with a decisions array, with one decision for "
- "each unit_index. Determine indentation from ordering, numbering, "
- "bullets, paragraph semantics, and visible explicit formatting. Do not "
- "invent nesting. If evidence is insufficient, use level 0 and low "
- "confidence."
- ),
- },
- {
- "role": "user",
- "content": json.dumps(
- {"post_title": post_title, "ordered_units": units},
- ensure_ascii=False,
- ),
- },
- ],
- "response_format": {
- "type": "json_schema",
- "json_schema": {
- "name": "post_structure_decisions",
- "strict": True,
- "schema": self._DECISION_SCHEMA,
- },
- },
- "mode": "auto",
- "reasoning_effort": "auto",
- "max_tokens": 4096,
- },
+ self.request_payload(post_title, units),
headers={"authorization": f"Bearer {self.api_key}"},
timeout=self.timeout,
)
parsed = json.loads(_response_content(response))
if not isinstance(parsed, dict) or not isinstance(parsed.get("decisions"), list):
- raise ValueError("structure adjudication response has no decisions array")
+ raise ValueError( # noqa: TRY004 - invalid provider shape is a retriable channel error.
+ "structure adjudication response has no decisions array"
+ )
- expected_indexes = {int(unit["unit_index"]) for unit in units}
decisions: list[StructureDecision] = []
for item in parsed["decisions"]:
if not isinstance(item, dict):
diff --git a/tests/test_post_content_persistence_edges.py b/tests/test_post_content_persistence_edges.py
index 24bec9e6d..bca6aa495 100644
--- a/tests/test_post_content_persistence_edges.py
+++ b/tests/test_post_content_persistence_edges.py
@@ -15,6 +15,7 @@
NormalizedPostContent,
)
from lineageweave.post_content_persistence import (
+ _bounded_structure_batches,
_bounded_unit_batches,
_render_image_text,
persist_post_content,
@@ -91,6 +92,17 @@ def infer(
raise ValueError("synthetic invalid structure response")
+class _NeverCalledStructure:
+ """Reject provider calls when the serialized request is already oversized."""
+
+ available = True
+
+ def infer(
+ self, _post_title: str, _units: list[dict[str, object]]
+ ) -> tuple[StructureDecision, ...]:
+ raise AssertionError("oversized structure request must not be sent")
+
+
class _UnexpectedChannelFailure:
"""Represent a programming defect that persistence must expose."""
@@ -112,10 +124,14 @@ class _ResolvedStructure:
available = True
+ def __init__(self) -> None:
+ self.units: list[dict[str, object]] = []
+
def infer(
self, _post_title: str, units: list[dict[str, object]]
) -> tuple[StructureDecision, ...]:
"""Return bounded synthetic decisions for persistence filtering."""
+ self.units = units
return (
StructureDecision(
unit_index=int(units[0]["unit_index"]),
@@ -310,21 +326,81 @@ def test_bounded_batches_cover_empty_count_and_character_limits() -> None:
len(batch)
for batch in _bounded_unit_batches([(str(i), "x" * 12_001) for i in range(3)])
] == [1, 1, 1]
+ assert [
+ len(batch)
+ for batch in _bounded_unit_batches([("x", {"text": "x" * 12_001}) for _ in range(2)])
+ ] == [1, 1]
+ metadata_bounded = _bounded_unit_batches(
+ [(str(i), {"text": "x" * 11_900, "label": "y" * 200}) for i in range(2)]
+ )
+ assert [len(batch) for batch in metadata_bounded] == [1, 1]
+
+
+def test_structure_batches_measure_the_complete_serialized_request() -> None:
+ """Envelope, schema, JSON escaping, and UTF-8 bytes all count toward the limit."""
+ units = [
+ (
+ index,
+ {
+ "unit_index": index,
+ "text": "가" * 4_000,
+ "label": "p",
+ "style": None,
+ "source_indent_width": 0,
+ "declared_indent_width": 0,
+ },
+ )
+ for index in range(2)
+ ]
+
+ assert [len(batch) for batch in _bounded_structure_batches(units, "Synthetic title")] == [1, 1]
+
+
+def test_oversized_structure_request_remains_unresolved_without_transport() -> None:
+ """An oversized title fails closed before the provider call and preserves source units."""
+ conn = _Connection()
+
+ assert (
+ _persist(
+ conn,
+ "post-oversized",
+ "plain text",
+ structure_client=_NeverCalledStructure(),
+ post_title="x" * 24_000,
+ )
+ == 1
+ )
+ assert any(
+ args[2] == "unresolved"
+ for query, args in conn.executed
+ if "insert into post_content_unit_structure" in query
+ )
def test_explicit_and_adjudicated_structure_are_persisted_by_unit() -> None:
"""Persist explicit depth and only in-scope orchestrator decisions."""
conn = _Connection()
+ structure_client = _ResolvedStructure()
assert (
_persist(
conn,
"post-7",
- 'Explicit
Semantic
',
- structure_client=_ResolvedStructure(),
+ 'Explicit
Semantic
',
+ structure_client=structure_client,
)
== 2
)
+ assert structure_client.units == [
+ {
+ "unit_index": 1,
+ "text": "Semantic",
+ "label": "p",
+ "style": None,
+ "source_indent_width": 2,
+ "declared_indent_width": 0,
+ }
+ ]
structure_rows = [
args
for query, args in conn.executed
diff --git a/tests/test_post_structure.py b/tests/test_post_structure.py
index 6e07242c8..50d03f925 100644
--- a/tests/test_post_structure.py
+++ b/tests/test_post_structure.py
@@ -41,15 +41,58 @@ def fake_post_json(*args, **kwargs):
client = ContextualOrchestratorPostStructureClient("http://orchestrator", "test-key")
assert client.timeout == 600.0
- assert client.infer("Title", [{"unit_index": 0, "text": "1. Heading"}])[0].indent_level == 0
+ assert client.infer(
+ "Title",
+ [
+ {
+ "unit_index": 0,
+ "text": "1. Heading",
+ "label": "p",
+ "style": "margin-left: 16px",
+ "source_indent_width": 2,
+ "declared_indent_width": 2,
+ }
+ ],
+ )[0].indent_level == 0
assert len(captured) == 1
response_format = captured[0]["response_format"]
assert response_format["type"] == "json_schema"
assert response_format["json_schema"]["strict"] is True
assert response_format["json_schema"]["schema"]["required"] == ["decisions"]
+ ordered_unit = json.loads(captured[0]["messages"][1]["content"])["ordered_units"][0]
+ assert ordered_unit["source_indent_width"] == 2
+ assert ordered_unit["declared_indent_width"] == 2
assert captured[0]["max_tokens"] == 4096
+@pytest.mark.parametrize(
+ "units",
+ [
+ [{}],
+ [{"unit_index": "0"}],
+ [{"unit_index": None}],
+ [{"unit_index": -1}],
+ [{"unit_index": True}],
+ [{"unit_index": 0}, {"unit_index": 0}],
+ ],
+)
+def test_structure_client_rejects_invalid_unit_indexes_before_transport(
+ monkeypatch, units
+) -> None:
+ """Malformed or duplicate indexes fail before reaching the orchestrator."""
+
+ def unexpected_post(*args, **kwargs):
+ raise AssertionError("invalid units must not cross the orchestrator boundary")
+
+ monkeypatch.setattr("lineageweave.post_structure.post_json", unexpected_post)
+ client = ContextualOrchestratorPostStructureClient(
+ "http://orchestrator", "test-key"
+ )
+
+ with pytest.raises(ValueError, match="unique non-negative integer indexes"):
+ client.infer("Title", units)
+
+
@pytest.mark.parametrize(
"response",
[{"choices": ["provider secret"]}, {"choices": [{"message": "provider secret"}]}],