diff --git a/apps/api/tests/contract/test_evidence_renderer_contract.py b/apps/api/tests/contract/test_evidence_renderer_contract.py
new file mode 100644
index 000000000..8188408b9
--- /dev/null
+++ b/apps/api/tests/contract/test_evidence_renderer_contract.py
@@ -0,0 +1,45 @@
+from shared.services.retrieval.execution.routes import _render_rows_evidence
+
+
+def test_render_rows_evidence_should_group_by_traceable_path() -> None:
+ rows = [
+ {
+ "chunk_id": "c2",
+ "content": "second section content",
+ "sort_order": 2,
+ "source": {
+ "source_file_name": "alpha.pdf",
+ "section_path": "Alpha / Two",
+ },
+ },
+ {
+ "chunk_id": "c1",
+ "content": "first section content\nwith more detail",
+ "sort_order": 1,
+ "source": {
+ "source_file_name": "alpha.pdf",
+ "section_path": "Alpha / One",
+ },
+ },
+ {
+ "chunk_id": "c3",
+ "content": "",
+ "source_file_name": "beta.pdf",
+ "section_path": "Beta / Table",
+ },
+ ]
+
+ evidence_text = _render_rows_evidence(rows)
+
+ assert "[E1]" in evidence_text
+ assert "[E2]" in evidence_text
+ assert "[E3]" in evidence_text
+ assert "[§ alpha.pdf / Alpha / One]" in evidence_text
+ assert "[§ alpha.pdf / Alpha / Two]" in evidence_text
+ assert "[§ beta.pdf / Beta / Table]" in evidence_text
+ assert "first section content" in evidence_text
+ assert "second section content" in evidence_text
+ assert "" in evidence_text
+ assert "[Document]" not in evidence_text
+ assert "▸" not in evidence_text
+ assert "┈" not in evidence_text
diff --git a/apps/api/tests/contract/test_legacy_evidence_renderer_contract.py b/apps/api/tests/contract/test_legacy_evidence_renderer_contract.py
deleted file mode 100644
index dcf4c086f..000000000
--- a/apps/api/tests/contract/test_legacy_evidence_renderer_contract.py
+++ /dev/null
@@ -1,43 +0,0 @@
-from shared.services.retrieval.hydration.legacy_evidence import render_legacy_evidence_text
-
-
-def test_render_legacy_evidence_text_should_group_documents_and_sections() -> None:
- rows = [
- {
- "chunk_id": "c2",
- "content": "second section content",
- "sort_order": 2,
- "source": {
- "source_file_name": "alpha.pdf",
- "section_path": "Alpha / Two",
- },
- },
- {
- "chunk_id": "c1",
- "content": "first section content\nwith more detail",
- "sort_order": 1,
- "source": {
- "source_file_name": "alpha.pdf",
- "section_path": "Alpha / One",
- },
- },
- {
- "chunk_id": "c3",
- "content": "",
- "source_file_name": "beta.pdf",
- "section_path": "Beta / Table",
- },
- ]
-
- evidence_text = render_legacy_evidence_text(rows)
-
- assert "[Document] alpha.pdf" in evidence_text
- assert "[Document] beta.pdf" in evidence_text
- assert "▸ Alpha / One" in evidence_text
- assert "▸ Alpha / Two" in evidence_text
- assert " ┈ first section content" in evidence_text
- assert " ┈ with more detail" in evidence_text
- assert " ┈ " in evidence_text
- assert "\u3010\u6587\u6863\u3011" not in evidence_text
- assert "[\u8868\u683c\u5185\u5bb9]" not in evidence_text
- assert "[\u56fe\u7247" not in evidence_text
diff --git a/apps/api/tests/contract/test_retrieval_lazy_tree_contract.py b/apps/api/tests/contract/test_retrieval_lazy_tree_contract.py
index a920bb396..9571f151f 100644
--- a/apps/api/tests/contract/test_retrieval_lazy_tree_contract.py
+++ b/apps/api/tests/contract/test_retrieval_lazy_tree_contract.py
@@ -70,7 +70,7 @@ def node_meta(self, section_id: str) -> NodeMeta:
budget_chars=100,
)
- assert result.evidence_text == "[E1]\n[§ Child]\nevidence"
+ assert result.evidence_text == "[E1]\n[§ Root / Child]\nevidence"
assert provider.metadata_calls == 0
@@ -102,4 +102,4 @@ def test_evidence_pack_identifies_header_owners_from_parent_chain() -> None:
)
assert result.kept_chunks == [chunks[1]]
- assert result.evidence_text == "[E1]\n[§ Child]\nchild evidence"
+ assert result.evidence_text == "[E1]\n[§ Root / Parent / Child]\nchild evidence"
diff --git a/apps/worker/tests/contract/test_page_memory_retrieval_contract.py b/apps/worker/tests/contract/test_page_memory_retrieval_contract.py
index 922aff4c5..1319db3fb 100644
--- a/apps/worker/tests/contract/test_page_memory_retrieval_contract.py
+++ b/apps/worker/tests/contract/test_page_memory_retrieval_contract.py
@@ -144,6 +144,8 @@ async def test_table_result_assembly_uses_summary_not_html() -> None:
assert "企业名称;统一社会信用代码" in content
assert "SHOULD NOT LEAK" not in content
assert " AbstractAsyncContextManager[AsyncSession]:
return get_db_context()
+def _evidence_path_header(row: dict) -> str:
+ source = row.get("source")
+ if not isinstance(source, dict):
+ source = row
+ file_name = str(source.get("source_file_name") or "").strip()
+ section_path = str(source.get("section_path") or "").strip()
+ if file_name and section_path:
+ return f"{file_name} / {section_path}"
+ return file_name or section_path
+
+
+def _render_rows_evidence(rows: list[dict]) -> str:
+ groups: dict[str, list[str]] = {}
+ for row in rows:
+ header = _evidence_path_header(row)
+ content = str(row.get("content") or "").strip()
+ if not content:
+ continue
+ groups.setdefault(header, []).append(content)
+ return render_evidence_blocks(list(groups.items()))
+
+
async def run_retrieval_route(
context: RetrievalRouteContext,
) -> RetrievalRouteOutcome:
@@ -102,7 +122,7 @@ async def _try_run_small_corpus_route(
"namespace": context.namespace,
"query": context.query,
"router_used": "small_corpus_all",
- "evidence_text": render_legacy_evidence_text(results),
+ "evidence_text": _render_rows_evidence(results),
"answer_text": "",
"results": results,
}
@@ -157,7 +177,7 @@ async def _run_classic_topk_route(
"namespace": context.namespace,
"query": context.query,
"router_used": "classic_topk",
- "evidence_text": render_legacy_evidence_text(results),
+ "evidence_text": _render_rows_evidence(results),
"answer_text": "",
"results": results,
}
diff --git a/packages/shared-python/shared/services/retrieval/hydration/asset_inline.py b/packages/shared-python/shared/services/retrieval/hydration/asset_inline.py
new file mode 100644
index 000000000..4c49f00d8
--- /dev/null
+++ b/packages/shared-python/shared/services/retrieval/hydration/asset_inline.py
@@ -0,0 +1,85 @@
+"""Insert connected image/table bodies at text placeholders.
+
+Replaces ``[images/...]`` / ``[tables/...]`` (or ``connect_to.ref``) with the
+asset display body, with a newline before and after. Targets not found at a
+placeholder are appended once. Leftover path placeholders are stripped.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+_PATH_REF_RE = re.compile(r"\[(?:images|tables)/[^\]\n]+\]")
+_SAME_AS_RE = re.compile(r"\[SAME-AS [^\]]+\]")
+
+
+def strip_path_placeholders(content: str) -> str:
+ text = _PATH_REF_RE.sub("", content)
+ text = _SAME_AS_RE.sub("", text)
+ return text.strip()
+
+
+def inline_assets_at_placeholders(
+ host_text: str,
+ *,
+ connections: Sequence[Mapping[str, Any]] | Sequence[Any],
+ display_by_target: Mapping[str, str],
+) -> tuple[str, set[str]]:
+ """Return (body, embedded_target_ids).
+
+ ``display_by_target`` maps chunk_id → display body. Only targets present in
+ that map are inserted. Each target is inserted at most once.
+ """
+ text = str(host_text or "")
+ embedded: set[str] = set()
+ pending_append: list[tuple[str, str]] = []
+
+ for item in connections or ():
+ if not isinstance(item, Mapping):
+ continue
+ target_id = str(item.get("target") or "").strip()
+ if not target_id or target_id in embedded:
+ continue
+ body = str(display_by_target.get(target_id) or "").strip()
+ if not body:
+ continue
+
+ ref = str(item.get("ref") or "").strip()
+ placed = False
+ for candidate in _ref_candidates(ref):
+ if candidate and candidate in text:
+ text = text.replace(candidate, f"\n{body}\n", 1)
+ embedded.add(target_id)
+ placed = True
+ break
+ if not placed:
+ pending_append.append((target_id, body))
+
+ for target_id, body in pending_append:
+ if target_id in embedded:
+ continue
+ if text.strip():
+ text = f"{text.rstrip()}\n\n{body}"
+ else:
+ text = body
+ embedded.add(target_id)
+
+ return strip_path_placeholders(text), embedded
+
+
+def _ref_candidates(ref: str) -> list[str]:
+ raw = str(ref or "").strip()
+ if not raw:
+ return []
+ out: list[str] = [raw]
+ if raw.startswith("[") and raw.endswith("]"):
+ inner = raw[1:-1].strip()
+ if inner and inner not in out:
+ out.append(inner)
+ else:
+ bracketed = f"[{raw}]"
+ if bracketed not in out:
+ out.append(bracketed)
+ return out
diff --git a/packages/shared-python/shared/services/retrieval/hydration/evidence_text.py b/packages/shared-python/shared/services/retrieval/hydration/evidence_text.py
new file mode 100644
index 000000000..a868302e7
--- /dev/null
+++ b/packages/shared-python/shared/services/retrieval/hydration/evidence_text.py
@@ -0,0 +1,45 @@
+"""Shared evidence_text rendering: one ``[E#]`` block per group.
+
+Each block is ``[E#]`` + ``[§ path]`` (full traceable path) + body lines.
+Groups are caller-provided; bodies in one group stay in that group.
+"""
+
+from __future__ import annotations
+
+from typing import Sequence
+
+
+def render_evidence_blocks(
+ groups: Sequence[tuple[str, Sequence[str]]],
+ *,
+ start_index: int = 1,
+) -> str:
+ """Render (path, bodies) groups into evidence_text.
+
+ ``path`` is the full traceable header (e.g. file / section chain).
+ Bodies are joined with newlines; multiple bodies in one group are indented.
+ ``start_index`` sets the first ``[E#]`` number (default 1).
+ """
+ parts: list[str] = []
+ index = max(1, int(start_index or 1))
+ for path, bodies in groups:
+ texts = [str(t or "").strip() for t in bodies]
+ texts = [t for t in texts if t]
+ if not texts:
+ continue
+ block: list[str] = [f"[E{index + len(parts)}]"]
+ header = str(path or "").strip()
+ if header:
+ block.append(f"[§ {header}]")
+ indent = len(texts) >= 2
+ for text in texts:
+ if indent:
+ block.append(
+ "\n".join(
+ (" " + ln if ln.strip() else ln) for ln in text.splitlines()
+ )
+ )
+ else:
+ block.append(text)
+ parts.append("\n".join(block).strip())
+ return "\n\n".join(parts)
diff --git a/packages/shared-python/shared/services/retrieval/hydration/legacy_evidence.py b/packages/shared-python/shared/services/retrieval/hydration/legacy_evidence.py
deleted file mode 100644
index 63b4ea358..000000000
--- a/packages/shared-python/shared/services/retrieval/hydration/legacy_evidence.py
+++ /dev/null
@@ -1,55 +0,0 @@
-from __future__ import annotations
-
-from collections import defaultdict
-from typing import Any
-
-
-def render_legacy_evidence_text(rows: list[dict[str, Any]]) -> str:
- """Render assembled retrieval rows into evidence-only context."""
- grouped_rows: dict[str, list[dict[str, Any]]] = defaultdict(list)
- for row in rows:
- doc_name = _source_value(row, "source_file_name") or "Unknown document"
- grouped_rows[doc_name].append(row)
-
- parts: list[str] = []
- for doc_name in sorted(grouped_rows):
- parts.append(f"[Document] {doc_name}")
- last_section = object()
- for row in sorted(grouped_rows[doc_name], key=_row_sort_key):
- section_path = _source_value(row, "section_path") or doc_name
- if section_path != last_section:
- parts.append(f"▸ {section_path}")
- last_section = section_path
- _append_content_lines(parts, row.get("content"))
-
- return "\n".join(parts)
-
-
-def _source_value(row: dict[str, Any], key: str) -> str:
- source = row.get("source")
- if isinstance(source, dict):
- value = source.get(key)
- if value:
- return str(value)
- value = row.get(key)
- return str(value) if value else ""
-
-
-def _row_sort_key(row: dict[str, Any]) -> tuple[str, int, str]:
- section_path = _source_value(row, "section_path")
- try:
- sort_order = int(row.get("sort_order") or 0)
- except (TypeError, ValueError):
- sort_order = 0
- chunk_id = str(row.get("chunk_id") or "")
- return section_path, sort_order, chunk_id
-
-
-def _append_content_lines(parts: list[str], content: object) -> None:
- text = str(content or "").strip()
- if not text:
- return
- for line in text.splitlines():
- stripped = line.strip()
- if stripped:
- parts.append(f" ┈ {stripped}")
diff --git a/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py b/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py
index 863bfadba..2f6aa54ad 100644
--- a/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py
+++ b/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py
@@ -5,9 +5,12 @@
from sqlalchemy.ext.asyncio import AsyncSession
+from shared.services.retrieval.hydration.asset_inline import (
+ inline_assets_at_placeholders,
+ strip_path_placeholders,
+)
from shared.services.retrieval.hydration.connected import hydrate_connected_target_rows
from shared.services.retrieval.hydration.row_utils import (
- clean_content,
extract_page_nums,
filter_excluded_rows,
iter_connected_target_ids,
@@ -70,18 +73,12 @@ async def assemble_retrieval_results(
assembled_row['content'] = _compose_table_content(row, rows_by_chunk_id)
assembled_row['content_source'] = 'summary'
elif chunk_type == 'text':
- related_parts = _connected_media_parts(row, rows_by_chunk_id)
- if base_content and related_parts:
- assembled_row['content'] = '\n\n'.join([base_content, *related_parts])
- elif related_parts:
- assembled_row['content'] = '\n\n'.join(related_parts)
- else:
- assembled_row['content'] = base_content
+ assembled_row['content'] = _compose_text_content(row, rows_by_chunk_id)
assembled_row['content_source'] = 'content'
else:
assembled_row['content'] = base_content
assembled_row['content_source'] = 'content'
- assembled_row['content'] = clean_content(assembled_row['content'])
+ assembled_row['content'] = strip_path_placeholders(assembled_row['content'])
assembled.append(assembled_row)
return assembled
@@ -93,6 +90,26 @@ def _page_summary(row: dict[str, Any]) -> str:
return str(metadata.get('summary') or '').strip()
+def _compose_text_content(
+ row: dict[str, Any],
+ rows_by_chunk_id: dict[str, dict[str, Any]],
+) -> str:
+ base_content = str(row.get('content') or '')
+ display_by_target = _connected_display_by_target(row, rows_by_chunk_id)
+ if not display_by_target:
+ return base_content
+ metadata = row.get('chunk_metadata') or row.get('metadata') or {}
+ connections = (
+ metadata.get('connect_to') if isinstance(metadata, dict) else None
+ ) or []
+ content, _embedded = inline_assets_at_placeholders(
+ base_content,
+ connections=connections if isinstance(connections, list) else [],
+ display_by_target=display_by_target,
+ )
+ return content
+
+
def _compose_table_content(
row: dict[str, Any],
rows_by_chunk_id: dict[str, dict[str, Any]],
@@ -102,11 +119,11 @@ def _compose_table_content(
return '\n\n'.join(part for part in parts if part)
-def _connected_media_parts(
+def _connected_display_by_target(
row: dict[str, Any],
rows_by_chunk_id: dict[str, dict[str, Any]],
-) -> list[str]:
- connected_targets: list[tuple[int, str]] = []
+) -> dict[str, str]:
+ display: dict[str, str] = {}
for target_id in iter_connected_target_ids(row):
target_row = rows_by_chunk_id.get(target_id)
if not target_row:
@@ -119,10 +136,8 @@ def _connected_media_parts(
else:
continue
if target_content:
- sort_key = int(target_row.get('sort_order', 0) or 0)
- connected_targets.append((sort_key, target_content))
- connected_targets.sort(key=lambda item: item[0])
- return [content for _, content in connected_targets]
+ display[target_id] = target_content
+ return display
def _connected_image_parts(
diff --git a/packages/shared-python/shared/services/retrieval/hydration/row_utils.py b/packages/shared-python/shared/services/retrieval/hydration/row_utils.py
index 32bcb2ba2..74fa382e9 100644
--- a/packages/shared-python/shared/services/retrieval/hydration/row_utils.py
+++ b/packages/shared-python/shared/services/retrieval/hydration/row_utils.py
@@ -1,6 +1,5 @@
from __future__ import annotations
-import re
from typing import Any
from shared.services.retrieval.search.section_filters import is_excluded_section
@@ -22,15 +21,6 @@
ReferenceLookupKey = tuple[str, str, str, str]
-_PATH_REF_RE = re.compile(r'\[(?:images|tables)/[^\]\n]+\]')
-_SAME_AS_RE = re.compile(r'\[SAME-AS [^\]]+\]')
-
-
-def clean_content(content: str) -> str:
- text = _PATH_REF_RE.sub('', content)
- text = _SAME_AS_RE.sub('', text)
- return text.strip()
-
def normalize_chunk_type(raw: object) -> str:
return str(raw or '').strip().split('\n', 1)[0].lower()
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_compose.py b/packages/shared-python/shared/services/retrieval/nav/nav_compose.py
index 3951a70c4..e23a5f825 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_compose.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_compose.py
@@ -12,6 +12,8 @@
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
+from shared.services.retrieval.hydration.evidence_text import render_evidence_blocks
+
from ._compat import Chunk
from ._compat import line_node_id
from ._compat import ToolSpace
@@ -100,17 +102,11 @@ def direct_parent_id(ts: ToolSpace, section_id: str, doc_id: str) -> Optional[st
return line_node_id(resolved, b.lines[p].line_id)
-def _section_title(ts: ToolSpace, section_id: str, doc_id: str, *, max_chars: int = 40) -> str:
+def _section_title(ts: ToolSpace, section_id: str, doc_id: str) -> str:
sid = str(section_id or "").strip()
if not sid:
return ""
- def _clip(text: str) -> str:
- t = (text or "").strip()
- if len(t) > max_chars:
- return t[:max_chars].rstrip()
- return t
-
# ``path_titles`` is a lightweight title lookup. Prefer it over
# ``get_structure`` because the latter may calculate the complete subtree
# chunk count, which is unnecessary while rendering evidence headers.
@@ -121,7 +117,7 @@ def _clip(text: str) -> str:
except TypeError:
path = str(path_titles(sid) or "").strip()
if path:
- return _clip(path.rsplit(" / ", 1)[-1])
+ return path
# Prefer structure title (Knowhere / ProviderToolSpace); never parse ids.
try:
@@ -131,7 +127,7 @@ def _clip(text: str) -> str:
if isinstance(st, dict):
raw = st.get("preview") or st.get("title") or ""
if isinstance(raw, str) and raw.strip():
- return _clip(raw.strip())
+ return raw.strip()
resolved = _section_doc_id(ts, sid, doc_id)
idx = getattr(ts, "_idx", None)
@@ -145,13 +141,13 @@ def _clip(text: str) -> str:
bb = getattr(idx, "_bundles", {}).get(resolved)
if bb and bb.lines:
title = (bb.lines[0].content or "").strip()
- return _clip(title) if title else sid
+ return title if title else sid
return sid
_, j = loc
if j < 0 or j >= len(b.lines):
return sid
title = (b.lines[j].content or "").strip()
- return _clip(title) if title else sid
+ return title if title else sid
def _chunk_body(chunk: Chunk) -> str:
@@ -302,7 +298,7 @@ def _build_groups(
if parent_id is None:
parent_id = owner
if parent_id not in groups:
- title = _section_title(ts, parent_id, owner_doc, max_chars=40)
+ title = _section_title(ts, parent_id, owner_doc)
groups[parent_id] = _ParentGroup(
parent_id=parent_id,
parent_title=title,
@@ -325,24 +321,13 @@ def _render_group(
selected: Sequence[_ChildItem],
*,
evidence_index: int,
- indent: bool,
) -> str:
- """Render one evidence block (full text only)."""
- parts: List[str] = [f"[E{evidence_index}]"]
- if group.parent_title:
- parts.append(f"[§ {group.parent_title}]")
- for child in selected:
- body = _chunk_body(child.chunk)
- if not body:
- continue
- if indent:
- indented = "\n".join(
- (" " + ln if ln.strip() else ln) for ln in body.splitlines()
- )
- parts.append(indented)
- else:
- parts.append(body)
- return "\n".join(parts).strip()
+ """Render one evidence block via the shared evidence renderer."""
+ bodies = [_chunk_body(child.chunk) for child in selected]
+ return render_evidence_blocks(
+ [(group.parent_title or "", bodies)],
+ start_index=evidence_index,
+ )
def _scored_flat(groups: Sequence[_ParentGroup]) -> List[Tuple[Chunk, float]]:
@@ -399,9 +384,8 @@ def _render_kept(
]
if not entries:
continue
- indent = len(entries) >= 2
block = _render_group(
- g, entries, evidence_index=len(parts) + 1, indent=indent
+ g, entries, evidence_index=len(parts) + 1
)
add = len(block) + (len(sep) if parts else 0)
if used + add <= budget_chars:
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py b/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py
index 80b9cde01..ead3dfc6e 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py
@@ -36,6 +36,10 @@
runtime_checkable,
)
+from shared.services.retrieval.hydration.asset_inline import (
+ inline_assets_at_placeholders,
+)
+
if TYPE_CHECKING:
from .knowhere_hybrid import PersistedScoreCorpus
@@ -179,17 +183,72 @@ def parent_id(self, section_id: str) -> Optional[str]:
return str(parent) if parent else None
def _node_unit_span(self, section_id: str) -> Tuple[str, int, int]:
- """(joined text, first sort_order, unit count) for one node's own units."""
+ """(joined text, first sort_order, unit count) for one node's own units.
+
+ Evidence display only: text units insert connected assets at placeholders.
+ Scoring still uses ``materialize_self_only_chunks`` / raw ``unit_text``.
+ """
self_units = getattr(self._provider, "self_units", None)
unit_text = getattr(self._provider, "unit_text", None)
if not callable(self_units) or not callable(unit_text):
return "", 0, 0
units = list(self_units(section_id) or ())
- texts = [t for t in (str(unit_text(u) or "").strip() for u in units) if t]
- if not texts:
- return "", 0, len(units)
+ if not units:
+ return "", 0, 0
first_order = int(getattr(units[0], "sort_order", 0) or 0)
- return "\n".join(texts), first_order, len(units)
+
+ asset_types = {"image", "table"}
+ text_units: List[Any] = []
+ asset_by_id: Dict[str, str] = {}
+ for unit in units:
+ chunk_type = str(getattr(unit, "chunk_type", "") or "").strip().lower()
+ chunk_id = str(getattr(unit, "chunk_id", "") or "").strip()
+ body = str(unit_text(unit) or "").strip()
+ if chunk_type in asset_types:
+ if chunk_id and body:
+ asset_by_id[chunk_id] = body
+ continue
+ text_units.append(unit)
+
+ if not text_units:
+ texts = [body for body in asset_by_id.values() if body]
+ return "\n".join(texts), first_order, len(units)
+
+ parts: List[str] = []
+ used_assets: Set[str] = set()
+ for unit in text_units:
+ content = str(unit_text(unit) or "").strip()
+ meta = getattr(unit, "metadata", None) or {}
+ connections = (
+ meta.get("connect_to") if isinstance(meta, dict) else None
+ ) or []
+ if not isinstance(connections, list):
+ connections = []
+ wanted = {
+ str(item.get("target") or "").strip()
+ for item in connections
+ if isinstance(item, dict)
+ }
+ display = {
+ target_id: asset_by_id[target_id]
+ for target_id in wanted
+ if target_id in asset_by_id
+ }
+ content, embedded = inline_assets_at_placeholders(
+ content,
+ connections=connections,
+ display_by_target=display,
+ )
+ used_assets.update(embedded)
+ if content:
+ parts.append(content)
+
+ for target_id, body in asset_by_id.items():
+ if target_id in used_assets or not body:
+ continue
+ parts.append(body)
+
+ return "\n".join(parts), first_order, len(units)
def _make_chunk(
self, node_id: str, doc_id: str, text: str, order: int, section_id: str
diff --git a/packages/shared-python/shared/testing/contract_runtime.py b/packages/shared-python/shared/testing/contract_runtime.py
index 27a3c7de1..f5fbbd5f4 100644
--- a/packages/shared-python/shared/testing/contract_runtime.py
+++ b/packages/shared-python/shared/testing/contract_runtime.py
@@ -350,6 +350,13 @@ def configure_contract_environment(
"QSTASH_CURRENT_SIGNING_KEY": "qstash-current-test-key",
"QSTASH_NEXT_SIGNING_KEY": "qstash-next-test-key",
"QSTASH_CALLBACK_BASE_URL": "http://localhost:5005/api/v1",
+ # Pin product defaults so a developer's local apps/api/.env cannot
+ # change contract expectations (credits seed, upload allow-list).
+ "FREE_PLAN_INITIAL_CREDITS": "5",
+ "SUPPORTED_EXTENSIONS": (
+ ".doc,.docx,.pdf,.txt,.xls,.xlsx,.csv,.pptx,"
+ ".jpg,.jpeg,.png,.md,.html,.htm"
+ ),
}
if "BILLING_ENABLED" not in os.environ:
diff --git a/packages/shared-python/shared/tests/test_asset_inline.py b/packages/shared-python/shared/tests/test_asset_inline.py
new file mode 100644
index 000000000..a59652f09
--- /dev/null
+++ b/packages/shared-python/shared/tests/test_asset_inline.py
@@ -0,0 +1,154 @@
+"""Unit tests for placeholder-based asset inlining."""
+
+from __future__ import annotations
+
+import pytest
+
+from shared.services.retrieval.hydration.asset_inline import (
+ inline_assets_at_placeholders,
+)
+from shared.services.retrieval.hydration.result_assembly import (
+ assemble_retrieval_results,
+)
+from shared.services.retrieval.nav.nav_hierarchy import ProviderToolSpace
+from shared.services.retrieval.nav.nav_knowhere import (
+ KnowhereProvider,
+ SectionRow,
+ UnitRow,
+)
+
+
+def test_inline_replaces_placeholder_with_newlines() -> None:
+ body, embedded = inline_assets_at_placeholders(
+ "see [images/a.png] here",
+ connections=[
+ {
+ "target": "img-1",
+ "relation": "embeds",
+ "ref": "[images/a.png]",
+ }
+ ],
+ display_by_target={"img-1": "[Image: images/a.png]\nsummary"},
+ )
+ assert body == "see \n[Image: images/a.png]\nsummary\n here"
+ assert embedded == {"img-1"}
+ assert "[images/" not in body
+
+
+def test_inline_appends_when_placeholder_missing() -> None:
+ body, embedded = inline_assets_at_placeholders(
+ "plain text",
+ connections=[{"target": "img-1", "ref": "[images/a.png]"}],
+ display_by_target={"img-1": "[Image: images/a.png]"},
+ )
+ assert body == "plain text\n\n[Image: images/a.png]"
+ assert embedded == {"img-1"}
+
+
+def test_inline_does_not_duplicate_target() -> None:
+ body, embedded = inline_assets_at_placeholders(
+ "x [images/a.png] y",
+ connections=[
+ {"target": "img-1", "ref": "[images/a.png]"},
+ {"target": "img-1", "ref": "[images/a.png]"},
+ ],
+ display_by_target={"img-1": "[Image: images/a.png]"},
+ )
+ assert body.count("[Image: images/a.png]") == 1
+ assert embedded == {"img-1"}
+
+
+@pytest.mark.asyncio
+async def test_assemble_inserts_table_at_placeholder() -> None:
+ rows = [
+ {
+ "chunk_id": "text-1",
+ "chunk_type": "text",
+ "content": "见表 [tables/table-1.html] 结束",
+ "chunk_metadata": {
+ "connect_to": [
+ {
+ "target": "table-1",
+ "relation": "embeds",
+ "ref": "[tables/table-1.html]",
+ }
+ ]
+ },
+ },
+ {
+ "chunk_id": "table-1",
+ "chunk_type": "table",
+ "content": "",
+ "file_path": "tables/table-1.html",
+ "asset_url": "https://assets.example.com/job-1/tables/table-1.html",
+ "chunk_metadata": {
+ "summary": "企业入驻信息登记模板",
+ "keywords": ["企业名称"],
+ },
+ },
+ ]
+ assembled = await assemble_retrieval_results(
+ rows=rows,
+ exclude_document_ids=[],
+ exclude_sections=[],
+ )
+ assert len(assembled) == 1
+ content = assembled[0]["content"]
+ assert "[tables/" not in content
+ assert content.index("见表") < content.index("[Table:")
+ assert content.index("[Table:") < content.index("结束")
+ assert "企业入驻信息登记模板" in content
+ assert "SHOULD NOT LEAK" not in content
+
+
+def test_node_unit_span_inlines_section_assets() -> None:
+ provider = KnowhereProvider(
+ doc_id="doc-1",
+ sections=[
+ SectionRow(
+ section_id="sec-1",
+ parent_section_id=None,
+ section_path="One",
+ section_title="One",
+ section_level=1,
+ summary="",
+ sort_order=0,
+ )
+ ],
+ units=[
+ UnitRow(
+ chunk_id="text-1",
+ section_id="sec-1",
+ chunk_type="text",
+ content="see [images/a.png] end",
+ sort_order=0,
+ metadata={
+ "connect_to": [
+ {
+ "target": "img-1",
+ "relation": "embeds",
+ "ref": "[images/a.png]",
+ }
+ ]
+ },
+ ),
+ UnitRow(
+ chunk_id="img-1",
+ section_id="sec-1",
+ chunk_type="image",
+ content="images/a.png",
+ sort_order=1,
+ file_path="images/a.png",
+ metadata={"summary": "chart summary"},
+ ),
+ ],
+ )
+ ts = ProviderToolSpace(provider)
+ text, _order, count = ts._node_unit_span("sec-1")
+ assert count == 2
+ assert "[images/" not in text
+ assert text.index("see") < text.index("[Image:")
+ assert text.index("[Image:") < text.index("end")
+ assert "chart summary" in text
+ # Asset must not also appear as a trailing standalone copy.
+ assert text.count("[Image:") == 1