fix(api): keep invoice catalog searchable on stream and DATA: URIs

.github/workflows/fuzz.yml

@@ -83,6 +83,9 @@ jobs:
       - name: Fuzz orchestration engine
         run: python fuzz/fuzz_orchestration.py -max_total_time=${FUZZ_SECONDS} -artifact_prefix=crash- fuzz/corpus/orchestration
 
+      - name: Fuzz image placement catalog
+        run: python fuzz/fuzz_image_catalog.py -max_total_time=${FUZZ_SECONDS} -artifact_prefix=crash- fuzz/corpus/image_catalog
+
       - name: Upload crash artifacts
         if: failure()
         uses: actions/upload-artifact@330a01c490aca151604b8cf639adc76d48f6c5d4 # actions/upload-artifact@v5

CHANGELOG.md

@@ -0,0 +1,37 @@
+# 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
+
+- Chat messages accept OpenAI `text` + `image_url` content parts. The gateway
+  records a 3NF `image_content_catalog` (`image_payload` / `image_placement` /
+  `image_recognition_event`) so an invoice PNG stays next to
+  `Please pay invoice 1042`. Raw base64 is hashed, not stored. Next action:
+  send the figure as `data:image/png;base64,...` or `https://...` and read
+  `orchestration.image_content_catalog` to find it.
+- Catalog honesty: `DATA:` / `HTTPS:` schemes and RFC 2397 whitespace in
+  base64 stay searchable; each placement carries `placement_id`; streamed
+  completions and `--state-db` restarts keep the catalog; credential shapes
+  in `adjacent_text` are redacted while invoice numbers and AP emails stay.
+  Next action: POST `stream: true` with a wrapped `DATA:image/png;base64,`
+  invoice and read the stop-chunk catalog.
+
+### References
+
+- Faysse, M., Sibille, H., Wu, T., Omrani, B., Viaud, G., Hudelot, C., &
+  Colombo, P. (2024). *ColPali: Efficient document retrieval with vision
+  language models* (arXiv:2407.01449). arXiv.
+  https://doi.org/10.48550/arXiv.2407.01449
+- Xu, Y., Li, M., Cui, L., Huang, S., Wei, F., & Zhou, M. (2020). LayoutLM:
+  Pre-training of text and layout for document image understanding. In
+  *Proceedings of the 26th ACM SIGKDD International Conference on Knowledge
+  Discovery & Data Mining* (pp. 1192–1200). Association for Computing
+  Machinery. https://doi.org/10.1145/3394486.3403172
+- Masinter, L. (1998). *The "data" URL scheme* (RFC 2397). Internet
+  Engineering Task Force. https://doi.org/10.17487/RFC2397

CLAUDE.md

@@ -85,7 +85,7 @@ A stdlib-Python lab implementing a single OpenAI-compatible API that routes, del
 
 ### Modules (`contextual_orchestrator/`)
 
-- `orchestrator.py` — the domain heart: `ModelAgent`, `WorkflowStep`, `OrchestrationPolicy`, `ModelClient`, `TaskOrchestrator`, secret/PII redaction, budget enforcement, spend analytics, and the commercial-readiness report generators behind `/api/v1/*`. Domain code stays here until a second implementation forces extraction (see `docs/code_conventions.md`).
+- `orchestrator.py` — the domain heart: `ModelAgent`, `WorkflowStep`, `OrchestrationPolicy`, `ModelClient`, `TaskOrchestrator`, `collect_image_catalog` (3NF figure placement), secret/PII redaction, budget enforcement, spend analytics, and the commercial-readiness report generators behind `/api/v1/*`. Domain code stays here until a second implementation forces extraction (see `docs/code_conventions.md`).
 - `server.py` — HTTP delivery adapter and `SecurityConfig`; all request validation lives here.
 - `admin.py` — static HTML/CSS/JS for the `/admin` operator console (stays inline while the product is dependency-free).
 - `credentials.py` / `kv_config.py` — the KV seam: `get_credential`/`register_credential` over pluggable backends (`InMemoryCredentialBackend` default; pgcrypto-encrypted `PostgresCredentialBackend`, selected via `CONTEXTUAL_ORCHESTRATOR_KV_BACKEND`).

README.md

@@ -256,6 +256,8 @@ python tests/test_admin_contract.py
 python tests/test_conventions.py
 python tests/test_api_contract.py
 python tests/test_security_hardening.py
+python tests/test_image_placement_catalog.py
+python tests/test_image_catalog_honesty.py
 python tests/test_repository_security_metadata.py
 python tests/test_product_planning_contract.py
 python tests/test_plugin_driven_artifacts.py

conductor/product.md

@@ -33,6 +33,7 @@ Provide one API and one domain model:
 - TRINITY: make thinker, worker, and verifier roles visible in the trace.
 - Conductor: show natural-language subtasks and access lists as first-class audit objects.
 - Enterprise operations: treat provider exclusion, locale bundles, and replayable workflow evidence as product surfaces.
+- Image placement: keep invoice and email figures searchable at the text they sat next to (ColPali / LayoutLM).
 
 ## Non-Goals
 

conductor/tracks.md

@@ -4,3 +4,4 @@
 |---|---|---|
 | 001-paper-grounded-orchestrator | active | Implement the source-backed orchestration contract with TDD, DDD, and CDD |
 | 002-enterprise-design-foundation | active | Add paper-grounded screen design, user stories, REST API, code/DB conventions, and i18n |
+| 005-image-placement-catalog | active | Keep invoice/email figures searchable at their source offset (ColPali / LayoutLM). 3NF payload + placement + later recognition events. |

contextual_orchestrator/orchestrator.py

@@ -4,6 +4,8 @@
 
 from collections import Counter, deque, OrderedDict
 from contextvars import ContextVar
+import base64
+import binascii
 import copy
 from dataclasses import dataclass, replace
 from functools import wraps
@@ -29,7 +31,7 @@
 from .credentials import NotConfigured, get_credential
 
 
-ChatMessage = dict[str, str]
+ChatMessage = dict[str, Any]
 
 class BudgetExceededError(RuntimeError):
     """Raised when an operator-configured spend budget is already exhausted."""
@@ -39,6 +41,114 @@ def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
         self.detail = detail or {}
 
 
+def flatten_message_text(content: Any) -> str:
+    """Return concatenated text parts from a chat ``content`` value.
+
+    OpenAI vision callers send a list of ``text`` and ``image_url`` parts.
+    Routing and adjacent-text anchors need the words that sat next to the
+    figure, not the base64 payload.
+    """
+    if isinstance(content, str):
+        return content
+    if not isinstance(content, list):
+        return ""
+    texts: list[str] = []
+    for part in content:
+        if isinstance(part, dict) and isinstance(part.get("text"), str):
+            texts.append(part["text"])
+    return " ".join(texts)
+
+
+def _parse_image_source(url: str) -> tuple[str, str, int, str] | None:
+    """Return ``(payload_digest, mime_type, byte_length, source_kind)`` or None.
+
+    RFC 2397 treats the ``data:`` scheme as case-insensitive and says
+    whitespace in the data portion should be ignored. Real invoice clients
+    emit ``DATA:IMAGE/PNG;BASE64,`` and wrap long payloads.
+    """
+    stripped = url.strip()
+    if stripped.lower().startswith("data:"):
+        header, separator, payload = stripped.partition(",")
+        if not separator or ";base64" not in header.lower():
+            return None
+        media = header.split(":", 1)[-1].split(";", 1)[0].strip().lower()
+        if not media.startswith("image/"):
+            return None
+        compact = "".join(payload.split())
+        try:
+            raw = base64.b64decode(compact, validate=True)
+        except (ValueError, binascii.Error):
+            return None
+        if not raw:
+            return None
+        return hashlib.sha256(raw).hexdigest(), media, len(raw), "inline_data_uri"
+    parsed = urlparse(stripped)
+    if parsed.scheme.lower() != "https" or not parsed.hostname:
+        return None
+    return hashlib.sha256(stripped.encode("utf-8")).hexdigest(), "image/remote", 0, "remote_https"
+
+
+def collect_image_catalog(messages: list[Any]) -> dict[str, Any]:
+    """Build a 3NF image catalog that keeps each figure at its source offset.
+
+    ``image_payload`` is identity by digest so the same invoice PNG on a
+    reminder thread is one payload with two ``image_placement`` rows.
+    ``image_recognition_event`` stays empty until a later vision/OCR pass
+    (temporal modeling: tags are not attributes of the bytes).
+    """
+    payloads: dict[str, dict[str, Any]] = {}
+    placements: list[dict[str, Any]] = []
+    if not isinstance(messages, list):
+        return {
+            "image_payloads": [],
+            "image_placements": [],
+            "image_recognition_events": [],
+        }
+    for message_index, message in enumerate(messages):
+        if not isinstance(message, dict):
+            continue
+        content = message.get("content")
+        adjacent_text = flatten_message_text(content)
+        if not isinstance(content, list):
+            continue
+        for part_index, part in enumerate(content):
+            if not isinstance(part, dict) or part.get("type") != "image_url":
+                continue
+            image_url = part.get("image_url")
+            if isinstance(image_url, str):
+                url = image_url
+            elif isinstance(image_url, dict):
+                url = image_url.get("url")
+            else:
+                continue
+            if not isinstance(url, str) or not url.strip():
+                continue
+            parsed = _parse_image_source(url.strip())
+            if parsed is None:
+                continue
+            payload_digest, mime_type, byte_length, source_kind = parsed
+            payloads[payload_digest] = {
+                "payload_digest": payload_digest,
+                "mime_type": mime_type,
+                "byte_length": byte_length,
+            }
+            placements.append(
+                {
+                    "placement_id": f"image_placement_{message_index}_{part_index}",
+                    "payload_digest": payload_digest,
+                    "message_index": message_index,
+                    "part_index": part_index,
+                    "source_kind": source_kind,
+                    "adjacent_text": adjacent_text,
+                }
+            )
+    return {
+        "image_payloads": list(payloads.values()),
+        "image_placements": placements,
+        "image_recognition_events": [],
+    }
+
+
 def estimate_tokens(text: str) -> int:
     """Rough token estimate (~4 chars/token). ponytail: heuristic, not a real tokenizer.
 
@@ -492,9 +602,13 @@ def _provider_url(self, agent: ModelAgent, path: str) -> str:
         return f"{agent.base_url.rstrip('/')}{path}"
 
     def _mock(self, agent: ModelAgent, messages: list[ChatMessage]) -> str:
-        last = next((m["content"] for m in reversed(messages) if m.get("role") == "user"), "")
+        last = next(
+            (flatten_message_text(m.get("content", "")) for m in reversed(messages) if m.get("role") == "user"),
+            "",
+        )
         role = "worker"
-        system = messages[0]["content"] if messages and messages[0].get("role") == "system" else ""
+        system_content = messages[0].get("content", "") if messages and messages[0].get("role") == "system" else ""
+        system = flatten_message_text(system_content)
         match = re.search(r"Role: ([a-z]+)", system)
         if match:
             role = match.group(1)
@@ -924,8 +1038,11 @@ def complete(self, messages: list[ChatMessage], mode: str = "auto") -> dict[str,
     def _dispatch(self, messages: list[ChatMessage], mode: str) -> dict[str, Any]:
         text = self._latest_user_text(messages)
         if mode == "route" or (mode == "auto" and not self._needs_workflow(text)):
-            return self.route_once(messages)
-        return self.conduct(messages)
+            result = self.route_once(messages)
+        else:
+            result = self.conduct(messages)
+        result["image_content_catalog"] = collect_image_catalog(messages)
+        return result
 
     def would_route(self, messages: list[ChatMessage], mode: str = "auto") -> bool:
         """True when this request takes the single-worker route path (vs the conduct workflow)."""
@@ -958,9 +1075,12 @@ def stream_route(self, messages: list[ChatMessage], workflow_run_id: str | None
             ],
             "policy_snapshot": self.policy.as_dict(),
             "verification": {"accepted": True, "reason": "single route path", "verifier_output": ""},
+            "image_content_catalog": _emit_image_catalog(collect_image_catalog(messages)),
         }
         self._workflow_runs[record["workflow_run_id"]] = record
         self._run_order.appendleft(record["workflow_run_id"])
+        if self._store is not None:
+            self._store.save("workflow_run", record["workflow_run_id"], record)
         self._append_audit_event(
             "workflow_run_created",
             {"workflow_run_id": record["workflow_run_id"], "mode": "route", "agent_count": 1},
@@ -993,6 +1113,9 @@ def run(self, messages: list[ChatMessage], mode: str = "auto", workflow_run_id:
             "trace": result["trace"],
             "policy_snapshot": self.policy.as_dict(),
             "verification": result.get("verification"),
+            "image_content_catalog": _emit_image_catalog(
+                result.get("image_content_catalog") or collect_image_catalog(messages)
+            ),
         }
         self._workflow_runs[record["workflow_run_id"]] = record
         self._run_order.appendleft(record["workflow_run_id"])
@@ -1600,7 +1723,10 @@ def _needs_workflow(self, text: str) -> bool:
         return hits >= self.policy.conduct_hint_threshold or len(text) > 700
 
     def _latest_user_text(self, messages: list[ChatMessage]) -> str:
-        return next((m.get("content", "") for m in reversed(messages) if m.get("role") == "user"), "")  # pragma: no cover
+        return next(
+            (flatten_message_text(m.get("content", "")) for m in reversed(messages) if m.get("role") == "user"),
+            "",
+        )
 
     def _model_judge_verification(self, task: str, fallback: dict[str, Any]) -> dict[str, Any]:
         """Ask a model to judge the verifier report (fixes term-matching false negatives).
@@ -8213,6 +8339,36 @@ def redact_text(text: str) -> str:
     return redacted
 
 
+def redact_credential_text(text: str) -> str:
+    """Mask API keys, tokens, and bearer secrets. Keep operational emails.
+
+    Invoice and AP retrieval need the mailbox next to the figure. Full PII
+    masking would hide ``ap@acme.com`` and paralyze search; credential
+    shapes are the only values that must not leave the catalog.
+    """
+    redacted = text
+    for pattern in SECRET_PATTERNS:
+        marker = pattern.pattern.lower()
+        if marker.startswith("(?i)(api"):
+            redacted = pattern.sub(lambda match: f"{match.group(1)}{match.group(2)}[REDACTED]", redacted)
+        elif marker.startswith("(?i)(bearer"):
+            redacted = pattern.sub(lambda match: f"{match.group(1)}[REDACTED]", redacted)
+    return redacted
+
+
+def _emit_image_catalog(catalog: Any) -> dict[str, Any] | None:
+    """Copy a catalog and redact credential shapes in adjacent text."""
+    if not isinstance(catalog, dict):
+        return None
+    emitted = copy.deepcopy(catalog)
+    placements = emitted.get("image_placements")
+    if isinstance(placements, list):
+        for placement in placements:
+            if isinstance(placement, dict) and isinstance(placement.get("adjacent_text"), str):
+                placement["adjacent_text"] = redact_credential_text(placement["adjacent_text"])
+    return emitted
+
+
 def redact_value(value: Any) -> Any:
     """Recursively redact string values while preserving response shape."""
     if isinstance(value, str):
@@ -8500,6 +8656,9 @@ def chat_completion_response(
     }
     if include_trace:
         orchestration["trace"] = redact_value(result["trace"])
+    catalog = _emit_image_catalog(result.get("image_content_catalog"))
+    if catalog:
+        orchestration["image_content_catalog"] = catalog
     return {
         "id": f"chatcmpl-{int(time.time() * 1000)}",
         "object": "chat.completion",
@@ -8550,6 +8709,9 @@ def chat_completion_chunks(
     }
     if include_trace and "trace" in result:
         orchestration["trace"] = redact_value(result["trace"])
+    catalog = _emit_image_catalog(result.get("image_content_catalog"))
+    if catalog:
+        orchestration["image_content_catalog"] = catalog
     final = {**base, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
     final["orchestration"] = {key: value for key, value in orchestration.items() if value is not None}
     chunks.append(final)