diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml index cfa62adad..49321c8c4 100644 --- a/.github/workflows/fuzz.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 000000000..e78e101da --- /dev/null +++ b/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 diff --git a/CLAUDE.md b/CLAUDE.md index f893b5f7b..4a83d6599 100644 --- a/CLAUDE.md +++ b/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`). diff --git a/README.md b/README.md index 65f57dd4c..905c918a8 100644 --- a/README.md +++ b/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 diff --git a/conductor/product.md b/conductor/product.md index 3092bda2a..277311564 100644 --- a/conductor/product.md +++ b/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 diff --git a/conductor/tracks.md b/conductor/tracks.md index 968c08ef8..4916b0644 100644 --- a/conductor/tracks.md +++ b/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. | diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722e..898790dc6 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/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) diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index c58d4cb79..5389f4431 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -183,16 +183,55 @@ def _validate_mode(mode: Any) -> str: return mode -def _validate_messages(messages: Any) -> list[dict[str, str]]: +def _validate_content_parts(content: list[Any]) -> list[dict[str, Any]]: + """Accept OpenAI text + image_url parts; keep the figure at its offset.""" + if not content: + raise RequestError(400, "invalid_message_content", "content parts must be a non-empty array") + cleaned: list[dict[str, Any]] = [] + for part in content: + if not isinstance(part, dict): + raise RequestError(400, "invalid_message_content", "each content part must be an object") + part_type = part.get("type") + if part_type == "text": + text = part.get("text") + if not isinstance(text, str) or not text.strip(): + raise RequestError(400, "invalid_message_content", "text content part requires non-empty text") + cleaned.append({"type": "text", "text": text}) + continue + if part_type == "image_url": + 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: + raise RequestError(400, "invalid_message_content", "image_url content part requires a url") + if not isinstance(url, str) or not url.strip(): + raise RequestError(400, "invalid_message_content", "image_url content part requires a non-empty url") + url = url.strip() + lowered = url.lower() + if not (lowered.startswith("https://") or lowered.startswith("data:image/")): + raise RequestError(400, "invalid_message_content", "image_url must be https or data:image") + cleaned.append({"type": "image_url", "image_url": {"url": url}}) + continue + raise RequestError(400, "invalid_message_content", "content part type must be text or image_url") + return cleaned + + +def _validate_messages(messages: Any) -> list[dict[str, Any]]: if not isinstance(messages, list) or not messages: raise RequestError(400, "invalid_message", "messages must be a non-empty array") - validated: list[dict[str, str]] = [] + validated: list[dict[str, Any]] = [] for message in messages: if not isinstance(message, dict): raise RequestError(400, "invalid_message", "each message must be an object") role = message.get("role") content = message.get("content") - if not isinstance(role, str) or role not in ALLOWED_MESSAGE_ROLES or not isinstance(content, str): + if not isinstance(role, str) or role not in ALLOWED_MESSAGE_ROLES: + raise RequestError(400, "invalid_message", "message role or content is invalid") + if isinstance(content, list): + content = _validate_content_parts(content) + elif not isinstance(content, str): raise RequestError(400, "invalid_message", "message role or content is invalid") validated.append({"role": role, "content": content}) return validated @@ -1042,7 +1081,23 @@ def frame(delta: dict[str, Any], finish: str | None = None) -> str: try: for delta in orchestrator.stream_route(messages, workflow_run_id=run_id): self._write_sse(frame({"content": delta})) - self._write_sse(frame({}, finish="stop")) + stop = { + "id": completion_id, + "object": "chat.completion.chunk", + "created": created, + "model": model_name, + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + } + try: + record = orchestrator.get_workflow_run(run_id) + except KeyError: + record = {} + orchestration = {"workflow_run_id": run_id, "mode": "route"} + catalog = record.get("image_content_catalog") + if catalog: + orchestration["image_content_catalog"] = catalog + stop["orchestration"] = orchestration + self._write_sse(f"data: {json.dumps(stop, ensure_ascii=False)}\n\n") except Exception: # noqa: BLE001 - headers already sent; surface as a terminal error frame self._write_sse(frame({}, finish="error")) self._write_sse("data: [DONE]\n\n") diff --git a/docs/architecture.md b/docs/architecture.md index c0f63a81e..f77b6a618 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -2,10 +2,15 @@ ## Sources Read -- Sakana AI launch article, "Sakana Fugu: One Model to Command Them All" (June 22, 2026): https://sakana.ai/fugu-release/ -- Sakana Fugu Technical Report: https://github.com/SakanaAI/fugu/blob/main/Fugu_technical_report.pdf -- TRINITY: An Evolved LLM Coordinator: https://arxiv.org/abs/2512.04695 -- Learning to Orchestrate Agents in Natural Language with the Conductor: https://arxiv.org/abs/2512.04388 +APA 7th citations (titles retained for paper-contract search): + +- Sakana AI. (2026, June 22). *Sakana Fugu: One model to command them all*. https://sakana.ai/fugu-release/ +- Tang, Y., Cetin, E., Xu, J., Sun, Q., Nielsen, S., Richard, V., Goda, H., Tymchenko, I., Nguyen, N., Lee, H., Ashiga, M., Kotyan, S., Kuroki, S., & Clanuwat, T. (2026). *Sakana Fugu technical report* (arXiv:2606.21228). arXiv. https://doi.org/10.48550/arXiv.2606.21228 +- Xu, J., Sun, Q., Schwendeman, P., Nielsen, S., Cetin, E., & Tang, Y. (2026). *TRINITY: An evolved LLM coordinator* (arXiv:2512.04695). arXiv. https://doi.org/10.48550/arXiv.2512.04695 +- Nielsen, S., Cetin, E., Schwendeman, P., Sun, Q., Xu, J., & Tang, Y. (2026). *Learning to orchestrate agents in natural language with the Conductor* (arXiv:2512.04388). arXiv. https://doi.org/10.48550/arXiv.2512.04388 +- 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 ## What The Architecture Is @@ -31,12 +36,40 @@ The Fugu report combines these ideas into production constraints: This repository implements the interface and control plane, not the trained coordinator. -- `contextual_orchestrator.orchestrator.Agent`: one configured worker model. -- `Orchestrator.route_once`: the low-latency routing path. -- `Orchestrator.conduct`: the workflow path with planner, worker, verifier, and synthesizer steps. +- `contextual_orchestrator.orchestrator.ModelAgent`: one configured worker model. +- `TaskOrchestrator.route_once`: the low-latency routing path. +- `TaskOrchestrator.conduct`: the workflow path with planner, worker, verifier, and synthesizer steps. - `WorkflowStep.access`: Conductor-style visibility control. +- `collect_image_catalog`: 3NF image payload / placement / recognition-event split so a figure stays next to its pay line. + +```mermaid +erDiagram + IMAGE_PAYLOAD ||--o{ IMAGE_PLACEMENT : appears_on + IMAGE_PAYLOAD ||--o{ IMAGE_RECOGNITION_EVENT : recognized_as + WORKFLOW_RUN ||--o{ IMAGE_PLACEMENT : contains + IMAGE_PAYLOAD { + text payload_digest PK + text mime_type + int byte_length + } + IMAGE_PLACEMENT { + text placement_id PK + text payload_digest FK + int message_index + int part_index + text source_kind + text adjacent_text + } + IMAGE_RECOGNITION_EVENT { + text recognition_event_id PK + text payload_digest FK + text recognized_text + text object_tags + timestamptz observed_at + } +``` - `ModelClient`: OpenAI-compatible HTTP client, with `mock://` for local checks. -- `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server. +- `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server. Buyer next action: send OpenAI `text` + `image_url` parts (`DATA:` and wrapped base64 are accepted); read `orchestration.image_content_catalog` on the JSON body or the SSE stop chunk. Each `image_placement` has a `placement_id` that matches `docs/database_design.sql`. 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. diff --git a/docs/database_design.sql b/docs/database_design.sql index fb5892479..914c021da 100644 --- a/docs/database_design.sql +++ b/docs/database_design.sql @@ -81,6 +81,39 @@ create index audit_event_retention_idx on audit_event (retention_expires_at) where deleted_at is null; +create table image_payload ( + payload_digest text primary key, + mime_type text not null, + byte_length integer not null, + created_at timestamptz not null default now() +); + +create table image_placement ( + placement_id text primary key, + payload_digest text not null references image_payload(payload_digest), + workflow_run_id text references workflow_run(workflow_run_id), + message_index integer not null, + part_index integer not null, + source_kind text not null, + adjacent_text text not null, + created_at timestamptz not null default now() +); + +create table image_recognition_event ( + recognition_event_id text primary key, + payload_digest text not null references image_payload(payload_digest), + recognized_text text not null default '', + object_tags text not null default '', + model_name text not null default '', + observed_at timestamptz not null default now() +); + +create index image_placement_payload_idx + on image_placement (payload_digest); + +create index image_recognition_payload_idx + on image_recognition_event (payload_digest, observed_at); + create view workflow_run_safe_view as select workflow_run_id, diff --git a/docs/fuzzing.md b/docs/fuzzing.md index 9897b2bd2..19705cd36 100644 --- a/docs/fuzzing.md +++ b/docs/fuzzing.md @@ -31,6 +31,10 @@ 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. **Image content catalog** — `orchestrator.collect_image_catalog`. Arbitrary + multimodal message lists must yield the 3NF catalog and never echo raw + `data:image` payloads. Seeds include uppercase `DATA:` and RFC 2397 + whitespace so those client shapes stay in the corpus. ## Running locally @@ -49,6 +53,7 @@ python fuzz/fuzz_request_body.py -max_total_time=60 fuzz/corpus/request_body python fuzz/fuzz_agent_config.py -max_total_time=60 fuzz/corpus/agent_config python fuzz/fuzz_redaction.py -max_total_time=60 fuzz/corpus/redaction python fuzz/fuzz_orchestration.py -max_total_time=60 fuzz/corpus/orchestration +python fuzz/fuzz_image_catalog.py -max_total_time=60 fuzz/corpus/image_catalog ``` Seed corpora live in `fuzz/corpus//`. diff --git a/docs/library_research.md b/docs/library_research.md index 42c7fa95c..617890a6f 100644 --- a/docs/library_research.md +++ b/docs/library_research.md @@ -53,6 +53,17 @@ Extraction triggers: Until those triggers exist, Ponytail recommends strengthening the current single-repo product instead of splitting it. +## Image placement catalog (2026-08-16) + +Invoice and email figures must stay searchable at the text they sat next to. +This is a parsing and 3NF identity problem, not a new vision library. + +| Area | Researched | Decision | Skipped | +|---|---|---|---| +| Inline image parse | RFC 2397 `data:` URLs (case-insensitive scheme; ignore whitespace in the data portion); OpenAI `image_url` content parts; stdlib `base64` + `hashlib` | Keep identity as SHA-256 of decoded bytes. Accept `DATA:` / wrapped base64. Never persist the raw payload on the catalog. | Pillow, pypdfium, Tesseract, ColPali runtime, a second OCR service. | +| Layout-aware retrieval | Faysse et al. (2024) ColPali; Xu et al. (2020) LayoutLM | Record `image_placement.message_index` / `part_index` / `adjacent_text` so a later embedder can retrieve the figure with the pay line. | Shipping a VLM retriever in this slice. | +| Temporal recognition | Separate `image_recognition_event` rows | OCR/object tags arrive later and must not be attributes of `image_payload`. | Writing fake OCR text in the first catalog pass. | + ## 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/papers/README.md b/docs/papers/README.md index 65a89d2af..1f11cd3f7 100644 --- a/docs/papers/README.md +++ b/docs/papers/README.md @@ -43,6 +43,36 @@ 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. +## Visually rich document retrieval (image placement) + +Invoice and email figures are retrieval objects, not decoration. Text-only +chunking drops the picture that sat under ``Please pay invoice 1042``. + +- 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 + `colpali-2407.01449.pdf` + Grounds **keeping the figure as a first-class retrieval unit** next to its + source offset. Distributed under the arXiv non-exclusive license. + +- 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 + Grounds **2-D / sequential position** as a retrieval signal. ACM copyright; + cited and summarized, not vendored. + +- Masinter, L. (1998). *The "data" URL scheme* (RFC 2397). Internet + Engineering Task Force. https://doi.org/10.17487/RFC2397 + Grounds **inline `data:image/...;base64,` identity** without storing the + payload on the catalog. IETF RFC; public. + +Buyer next action: send the PNG as an OpenAI `image_url` part under the pay +line, then search `orchestration.image_content_catalog.image_placements` +for `invoice 1042`. + > 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/colpali-2407.01449.pdf b/docs/papers/colpali-2407.01449.pdf new file mode 100644 index 000000000..cda37b38b Binary files /dev/null and b/docs/papers/colpali-2407.01449.pdf differ diff --git a/docs/user_stories.md b/docs/user_stories.md index 7f476d41f..042d64c4d 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 an invoice PNG under ``Please pay invoice 1042`` to stay searchable at that line so that I can retrieve the figure, not only the words. - 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/corpus/image_catalog/invoice_png.json b/fuzz/corpus/image_catalog/invoice_png.json new file mode 100644 index 000000000..ac0b1fcd7 --- /dev/null +++ b/fuzz/corpus/image_catalog/invoice_png.json @@ -0,0 +1 @@ +[{"role":"user","content":[{"type":"text","text":"Please pay invoice 1042 shown below."},{"type":"image_url","image_url":{"url":"data:image/png;base64,iVBORw0KGgo="}}]}] diff --git a/fuzz/corpus/image_catalog/uppercase_wrapped.json b/fuzz/corpus/image_catalog/uppercase_wrapped.json new file mode 100644 index 000000000..4e3e5c776 --- /dev/null +++ b/fuzz/corpus/image_catalog/uppercase_wrapped.json @@ -0,0 +1 @@ +[{"role":"user","content":[{"type":"text","text":"Please pay invoice 1042 shown below."},{"type":"image_url","image_url":{"url":"DATA:IMAGE/PNG;BASE64,iVBORw0KGgo=\nAAA="}}]}] diff --git a/fuzz/fuzz_image_catalog.py b/fuzz/fuzz_image_catalog.py new file mode 100644 index 000000000..4f2b8c568 --- /dev/null +++ b/fuzz/fuzz_image_catalog.py @@ -0,0 +1,41 @@ +#!/usr/bin/env python3 +"""Atheris coverage-guided harness: image placement catalog. + +Surface: ``orchestrator.collect_image_catalog`` -- untrusted multimodal +message lists. The catalog must stay 3NF-shaped and must never echo raw +base64 image payloads (see ``fuzz.targets``). + +Run locally:: + + python fuzz/fuzz_image_catalog.py -max_total_time=60 fuzz/corpus/image_catalog +""" + +import json +import sys +from pathlib import Path + +import atheris + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +with atheris.instrument_imports(): + from fuzz.targets import exercise_image_catalog + + +def one_input(data: bytes) -> None: + fdp = atheris.FuzzedDataProvider(data) + raw = fdp.ConsumeUnicodeNoSurrogates(fdp.remaining_bytes()) + try: + value = json.loads(raw) + except json.JSONDecodeError: + value = raw + exercise_image_catalog(value) + + +def main() -> None: + atheris.Setup(sys.argv, one_input) + atheris.Fuzz() + + +if __name__ == "__main__": + main() diff --git a/fuzz/targets.py b/fuzz/targets.py index d0c344462..b9958d5ca 100644 --- a/fuzz/targets.py +++ b/fuzz/targets.py @@ -8,7 +8,7 @@ ``AttributeError``, ``RecursionError``, ``SystemError`` or a hang; and * structural invariants on any successful result (shape, types, idempotence). -CodeGraph (``codegraph explore``) surfaced these four surfaces as the ones that +CodeGraph (``codegraph explore``) surfaced these surfaces as the ones that consume untrusted bytes/JSON: 1. ``server._coerce_json`` / ``_validate_mode`` / ``_validate_messages`` / @@ -18,6 +18,9 @@ over arbitrary trace payloads (regex + recursion). 4. ``orchestrator.TaskOrchestrator.run`` (+ ``sse_stream_body``) -- end-to-end prompt processing on a mock (offline) provider. +5. ``orchestrator.collect_image_catalog`` -- multimodal content-part parsing. + Arbitrary message lists must yield the 3NF catalog shape and never echo + raw ``data:image`` payloads. No network, no secrets, no filesystem: every target runs fully offline. """ @@ -32,6 +35,7 @@ ModelAgent, TaskOrchestrator, chat_completion_chunks, + collect_image_catalog, redact_text, redact_value, sse_stream_body, @@ -125,6 +129,21 @@ def exercise_agent_config(value: Any) -> None: assert isinstance(agent.disabled, bool) +def exercise_image_catalog(messages: Any) -> None: + """Drive image-placement parsing over arbitrary message lists. + + Invariants: never crashes, always returns the three 3NF collections, never + embeds a ``data:image`` payload in the catalog JSON, and is JSON-serialisable. + """ + catalog = collect_image_catalog(messages) + assert isinstance(catalog, dict) + assert isinstance(catalog.get("image_payloads"), list) + assert isinstance(catalog.get("image_placements"), list) + assert isinstance(catalog.get("image_recognition_events"), list) + blob = json.dumps(catalog) + assert "data:image" not in blob + + def exercise_redaction(text: str) -> None: """Drive secret/PII redaction over arbitrary text and structures. diff --git a/tests/fuzz/test_fuzz_properties.py b/tests/fuzz/test_fuzz_properties.py index 7e7b3f347..068ededd8 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_image_catalog, exercise_orchestration, exercise_redaction, exercise_request_body, @@ -101,6 +102,12 @@ def test_redaction_never_crashes_and_is_idempotent(text: str) -> None: exercise_redaction(text) +@_SETTINGS +@given(_json_values) +def test_image_catalog_never_crashes_on_arbitrary_messages(value: object) -> None: + exercise_image_catalog(value) + + @settings(max_examples=100, deadline=None) @given( st.text(max_size=2048), diff --git a/tests/test_image_catalog_honesty.py b/tests/test_image_catalog_honesty.py new file mode 100644 index 000000000..7c195703c --- /dev/null +++ b/tests/test_image_catalog_honesty.py @@ -0,0 +1,267 @@ +"""Catalog honesty for invoice figures that real clients actually send. + +#663 records a 3NF image catalog. Buyers still lose the figure when: + +- the data URI scheme is ``DATA:`` (RFC 2397 is case-insensitive) +- the base64 payload is wrapped with whitespace (RFC 2397 §3) +- they stream (``stream: true``) and look for the catalog on the stop chunk +- they persist a streamed run to ``--state-db`` +- an API key sat next to the pay line (credential shapes must not leak) + +Operational emails and invoice numbers stay searchable. Masking those +paralyzes AP/AR retrieval; only credential shapes are redacted. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import tempfile +import threading +import urllib.error +import urllib.request +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.orchestrator import ( # noqa: E402 + chat_completion_chunks, + chat_completion_response, + collect_image_catalog, +) +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + +_PNG_B64 = ( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQ" + "AAAABJRU5ErkJggg==" +) +_PNG_DIGEST = hashlib.sha256(base64.b64decode(_PNG_B64, validate=True)).hexdigest() +_TEST_AUTH_TOKEN = "image_catalog_honesty_http_token" # noqa: S105 + + +def _vision_message(url: str, text: str) -> dict[str, object]: + return { + "role": "user", + "content": [ + {"type": "text", "text": text}, + {"type": "image_url", "image_url": {"url": url}}, + ], + } + + +def test_uppercase_data_uri_keeps_invoice_figure() -> None: + """A client that emits DATA:IMAGE/PNG;BASE64 must still find invoice 1042.""" + url = f"DATA:IMAGE/PNG;BASE64,{_PNG_B64}" + catalog = collect_image_catalog( + [_vision_message(url, "Please pay invoice 1042 shown below.")] + ) + assert catalog["image_payloads"][0]["payload_digest"] == _PNG_DIGEST + assert catalog["image_placements"][0]["adjacent_text"].find("invoice 1042") >= 0 + assert catalog["image_placements"][0]["placement_id"] == "image_placement_0_1" + + +def test_rfc2397_whitespace_in_base64_keeps_invoice_figure() -> None: + """Wrapped data URIs (mail clients, JSON pretty-print) must still hash.""" + wrapped = f"data:image/png;base64,{_PNG_B64[:40]}\n{_PNG_B64[40:]}" + catalog = collect_image_catalog( + [_vision_message(wrapped, "Please pay invoice 1042 shown below.")] + ) + assert catalog["image_payloads"][0]["payload_digest"] == _PNG_DIGEST + assert _PNG_B64 not in json.dumps(catalog) + + +def test_https_uppercase_scheme_is_placed() -> None: + catalog = collect_image_catalog( + [ + _vision_message( + "HTTPS://example.com/packing-slip.png", + "See the packing slip.", + ) + ] + ) + assert catalog["image_placements"][0]["source_kind"] == "remote_https" + assert "packing slip" in catalog["image_placements"][0]["adjacent_text"] + + +def test_http_accepts_uppercase_https_and_data_schemes() -> None: + """Validator and parser must agree so the figure is not silently dropped.""" + orchestrator = TaskOrchestrator( + [ModelAgent("general_agent", "mock-planner", tags=("reasoning", "writing"))] + ) + server = build_server( + orchestrator, + port=0, + security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN, rate_limit_requests=10_000), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/chat/completions", + data=json.dumps( + { + "model": "mock-planner", + "messages": [ + _vision_message( + f"DATA:image/png;base64,{_PNG_B64}", + "Please pay invoice 1042 shown below.", + ) + ], + "include_orchestration_trace": True, + } + ).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {_TEST_AUTH_TOKEN}", + "connection": "close", + }, + method="POST", + ) + with urllib.request.urlopen(request, timeout=15) as response: + body = json.loads(response.read().decode("utf-8")) + catalog = body["orchestration"]["image_content_catalog"] + assert catalog["image_placements"][0]["adjacent_text"].find("invoice 1042") >= 0 + assert catalog["image_payloads"][0]["payload_digest"] == _PNG_DIGEST + except urllib.error.HTTPError as exc: + raise AssertionError(exc.read().decode("utf-8")) from exc + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_adjacent_text_redacts_credentials_keeps_invoice_and_email() -> None: + """AP search needs the pay line and the AP mailbox; keys must not leak.""" + text = ( + "Please pay invoice 1042 to ap@acme.com " + "api_key=sk-abcdefghijklmnopqrstuvwxyz" + ) + result = { + "mode": "route", + "answer": "ok", + "image_content_catalog": collect_image_catalog( + [_vision_message(f"data:image/png;base64,{_PNG_B64}", text)] + ), + } + catalog = chat_completion_response(result)["orchestration"]["image_content_catalog"] + adjacent = catalog["image_placements"][0]["adjacent_text"] + assert "invoice 1042" in adjacent + assert "ap@acme.com" in adjacent + assert "sk-abcdefghijklmnopqrstuvwxyz" not in adjacent + assert "[REDACTED]" in adjacent + assert _PNG_B64 not in json.dumps(catalog) + + +def test_stream_stop_chunk_includes_searchable_catalog() -> None: + """Framed SSE buyers read the catalog from the terminal stop chunk.""" + result = { + "mode": "route", + "answer": "ok", + "workflow_run_id": "run_stream_catalog", + "image_content_catalog": collect_image_catalog( + [_vision_message(f"data:image/png;base64,{_PNG_B64}", "Please pay invoice 1042.")] + ), + } + chunks = chat_completion_chunks(result) + final = chunks[-1] + assert final["choices"][0]["finish_reason"] == "stop" + catalog = final["orchestration"]["image_content_catalog"] + assert catalog["image_placements"][0]["adjacent_text"].find("invoice 1042") >= 0 + assert catalog["image_placements"][0]["placement_id"] == "image_placement_0_1" + + +def test_stream_route_persists_catalog_to_state_db() -> None: + """A streamed invoice must survive restart the same way a JSON completion does.""" + messages = [ + _vision_message(f"data:image/png;base64,{_PNG_B64}", "Please pay invoice 1042 shown below.") + ] + with tempfile.TemporaryDirectory() as directory: + db_path = os.path.join(directory, "state.db") + first = TaskOrchestrator( + [ModelAgent("general_agent", "mock-planner", tags=("reasoning", "writing"))], + state_db=db_path, + ) + list(first.stream_route(messages, workflow_run_id="run_stream_invoice")) + first.close() + second = TaskOrchestrator( + [ModelAgent("general_agent", "mock-planner", tags=("reasoning", "writing"))], + state_db=db_path, + ) + try: + record = second.get_workflow_run("run_stream_invoice") + catalog = record["image_content_catalog"] + assert catalog["image_placements"][0]["adjacent_text"].find("invoice 1042") >= 0 + assert catalog["image_payloads"][0]["payload_digest"] == _PNG_DIGEST + finally: + second.close() + + +def test_http_route_stream_stop_frame_has_invoice_catalog() -> None: + """True-stream route path must still return the figure at the pay line.""" + orchestrator = TaskOrchestrator( + [ModelAgent("general_agent", "mock-planner", tags=("reasoning", "writing"))] + ) + server = build_server( + orchestrator, + port=0, + security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN, rate_limit_requests=10_000), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/chat/completions", + data=json.dumps( + { + "model": "mock-planner", + "messages": [ + _vision_message( + f"data:image/png;base64,{_PNG_B64}", + "Please pay invoice 1042 shown below.", + ) + ], + "mode": "route", + "stream": True, + } + ).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {_TEST_AUTH_TOKEN}", + "connection": "close", + }, + method="POST", + ) + with urllib.request.urlopen(request, timeout=15) as response: + sse = response.read().decode("utf-8") + except urllib.error.HTTPError as exc: + raise AssertionError(exc.read().decode("utf-8")) from exc + finally: + server.shutdown() + thread.join(timeout=5) + stop_catalog = None + for frame in sse.split("\n\n"): + frame = frame.strip() + if not frame.startswith("data: ") or frame == "data: [DONE]": + continue + chunk = json.loads(frame[len("data: ") :]) + if chunk["choices"][0].get("finish_reason") == "stop": + stop_catalog = chunk.get("orchestration", {}).get("image_content_catalog") + assert stop_catalog is not None + assert stop_catalog["image_placements"][0]["adjacent_text"].find("invoice 1042") >= 0 + assert _PNG_B64 not in sse + + +if __name__ == "__main__": + test_uppercase_data_uri_keeps_invoice_figure() + test_rfc2397_whitespace_in_base64_keeps_invoice_figure() + test_https_uppercase_scheme_is_placed() + test_http_accepts_uppercase_https_and_data_schemes() + test_adjacent_text_redacts_credentials_keeps_invoice_and_email() + test_stream_stop_chunk_includes_searchable_catalog() + test_stream_route_persists_catalog_to_state_db() + test_http_route_stream_stop_frame_has_invoice_catalog() + print("ok") diff --git a/tests/test_image_placement_catalog.py b/tests/test_image_placement_catalog.py new file mode 100644 index 000000000..6ef714257 --- /dev/null +++ b/tests/test_image_placement_catalog.py @@ -0,0 +1,162 @@ +"""Invoice figures stay searchable at the text they sat next to. + +Buyers paste a PNG under ``Please pay invoice 1042``. Text-only chunking +drops the figure, so retrieval cannot find the picture that belongs to that +line. ColPali (Faysse et al., 2024) and LayoutLM (Xu et al., 2020) treat +page layout as retrieval signal: the image must keep its source offset. + +3NF split: one ``image_payload`` (digest identity), many ``image_placement`` +rows (the same bytes can appear on a reminder thread), and time-varying +``image_recognition_event`` rows (OCR/tags arrive later). +""" + +from __future__ import annotations + +import hashlib +import json +import threading +import urllib.error +import urllib.request +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.orchestrator import collect_image_catalog # noqa: E402 +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + +# 1x1 PNG (real raster, not a stub string). +_PNG_B64 = ( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQ" + "AAAABJRU5ErkJggg==" +) +_PNG_DATA_URI = f"data:image/png;base64,{_PNG_B64}" +_PNG_DIGEST = hashlib.sha256( + __import__("base64").b64decode(_PNG_B64, validate=True) +).hexdigest() +_TEST_AUTH_TOKEN = "image_placement_catalog_http_honesty_token" # noqa: S105 + + +def _invoice_messages() -> list[dict]: + return [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Please pay invoice 1042 shown below."}, + {"type": "image_url", "image_url": {"url": _PNG_DATA_URI}}, + ], + }, + { + "role": "user", + "content": [ + {"type": "text", "text": "Same figure attached to the reminder thread."}, + {"type": "image_url", "image_url": {"url": _PNG_DATA_URI}}, + ], + }, + ] + + +def test_invoice_png_keeps_one_payload_and_two_placements() -> None: + """The same invoice PNG on a reminder thread is one payload, two placements.""" + catalog = collect_image_catalog(_invoice_messages()) + payloads = catalog["image_payloads"] + placements = catalog["image_placements"] + assert len(payloads) == 1 + assert payloads[0]["payload_digest"] == _PNG_DIGEST + assert payloads[0]["mime_type"] == "image/png" + assert payloads[0]["byte_length"] == 70 + assert len(placements) == 2 + assert placements[0]["payload_digest"] == _PNG_DIGEST + assert placements[0]["message_index"] == 0 + assert placements[0]["part_index"] == 1 + assert placements[0]["source_kind"] == "inline_data_uri" + assert "invoice 1042" in placements[0]["adjacent_text"] + assert placements[1]["message_index"] == 1 + assert "reminder thread" in placements[1]["adjacent_text"] + blob = json.dumps(catalog) + assert _PNG_B64 not in blob + assert "image_recognition_events" in catalog + assert catalog["image_recognition_events"] == [] + + +def test_remote_https_image_is_placed_without_fetching_bytes() -> None: + catalog = collect_image_catalog( + [ + { + "role": "user", + "content": [ + {"type": "text", "text": "See the packing slip."}, + { + "type": "image_url", + "image_url": {"url": "https://example.com/packing-slip.png"}, + }, + ], + } + ] + ) + assert catalog["image_payloads"][0]["byte_length"] == 0 + assert catalog["image_placements"][0]["source_kind"] == "remote_https" + assert "packing slip" in catalog["image_placements"][0]["adjacent_text"] + + +def test_http_chat_returns_invoice_figure_next_to_pay_line() -> None: + """A buyer sending a vision invoice must get the figure back at that line.""" + orchestrator = TaskOrchestrator( + [ModelAgent("general_agent", "mock-planner", tags=("reasoning", "writing"))] + ) + server = build_server( + orchestrator, + port=0, + security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN, rate_limit_requests=10_000), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/chat/completions", + data=json.dumps( + { + "model": "mock-planner", + "messages": _invoice_messages()[:1], + "include_orchestration_trace": True, + } + ).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {_TEST_AUTH_TOKEN}", + "connection": "close", + }, + method="POST", + ) + with urllib.request.urlopen(request, timeout=15) as response: + status = response.status + body = json.loads(response.read().decode("utf-8")) + assert status == 200, body + catalog = body["orchestration"]["image_content_catalog"] + assert catalog["image_placements"][0]["adjacent_text"].find("invoice 1042") >= 0 + assert _PNG_B64 not in json.dumps(body) + except urllib.error.HTTPError as exc: + raise AssertionError(exc.read().decode("utf-8")) from exc + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_docs_cite_colpali_and_layoutlm() -> None: + root = Path(__file__).resolve().parents[1] + papers = (root / "docs" / "papers" / "README.md").read_text(encoding="utf-8") + architecture = (root / "docs" / "architecture.md").read_text(encoding="utf-8") + assert "ColPali" in papers and "2407.01449" in papers + assert "LayoutLM" in papers + assert "collect_image_catalog" in architecture + assert (root / "docs" / "papers" / "colpali-2407.01449.pdf").is_file() + + +if __name__ == "__main__": + test_invoice_png_keeps_one_payload_and_two_placements() + test_remote_https_image_is_placed_without_fetching_bytes() + test_http_chat_returns_invoice_figure_next_to_pay_line() + test_docs_cite_colpali_and_layoutlm() + print("ok")