diff --git a/apps/worker/app/core/tasks/kb_tasks.py b/apps/worker/app/core/tasks/kb_tasks.py
index 4cba0c087..59c0ed233 100644
--- a/apps/worker/app/core/tasks/kb_tasks.py
+++ b/apps/worker/app/core/tasks/kb_tasks.py
@@ -686,6 +686,29 @@ def _parse(job_id: str, user_id: str | None):
             metadata_service.update_metadata(job_id, processing_timing_updates)
             job_metadata.update(processing_timing_updates)
 
+            # 1.5. Garbage Collection: Remove redundant local media files
+            try:
+                from shared.core.database_sync import get_sync_db_context
+                from shared.services.retrieval.publication_service import RetrievalPublicationService
+                from shared.models.database.job import Job
+                from sqlalchemy import select
+                
+                with get_sync_db_context() as db:
+                    job_record = db.execute(select(Job).where(Job.job_id == job_id)).scalar_one_or_none()
+                    if job_record:
+                        gc_namespace = JobMetadataHelper.get_field(job_metadata, "namespace") or "default"
+                        chunks, dedup_stats = RetrievalPublicationService.garbage_collect_and_dedup_local_media(
+                            db,
+                            job_id=job_id,
+                            user_id=str(job_record.user_id),
+                            namespace=gc_namespace,
+                            add_dir=str(add_dir) if add_dir else "",
+                            chunks=chunks,
+                        )
+            except Exception as e:
+                logger.error(f"[{job_id}] GC failed (non-fatal): {e}")
+                dedup_stats = None
+
             # Generate ZIP package
             zip_service = ZipResultService()
             zip_file_path, checksum, statistics, zip_size = (
@@ -734,6 +757,7 @@ def _parse(job_id: str, user_id: str | None):
                 stored_count=stored_count,
                 delivery_mode="url",
                 section_summaries=section_summaries,
+                chunk_dedup_stats=dedup_stats,
             )
 
             logger.info(
diff --git a/apps/worker/app/services/document_parser/doc_parser.py b/apps/worker/app/services/document_parser/doc_parser.py
index a4838470e..4d456131b 100755
--- a/apps/worker/app/services/document_parser/doc_parser.py
+++ b/apps/worker/app/services/document_parser/doc_parser.py
@@ -3,6 +3,7 @@
 import io
 import json
 import os
+import shutil
 import zipfile
 
 import pandas as pd
@@ -15,7 +16,11 @@
     remove_spaces,
 )
 from app.services.document_parser.html_parser import table2html
-from app.services.document_parser.image_parser import _get_vision_client, ask_image
+from app.services.document_parser.image_parser import (
+    _get_vision_client,
+    ask_image,
+    perceptual_hash,
+)
 from app.services.document_parser.layout_parser import pred_titles
 from app.services.document_parser.table_parser import sanitize_table_name_from_header
 from app.services.document_parser.toc_parser import (
@@ -101,10 +106,35 @@ def handle_image(
     current_heading,
     img_count,
     smart_summary=False,
+    seen_images=None,
 ):
     time_stamp = get_str_time()
-    client = _get_vision_client()
 
+    # Document-level dedup: use perceptual hash to catch visually-identical
+    # images that differ only in compression/metadata
+    img_hash = perceptual_hash(img_file["data"])
+    if seen_images is not None and img_hash in seen_images:
+        cached = seen_images[img_hash]
+        headings_stack[-1]["content"].append(cached["image_ref"])
+        df_list.append(
+            [
+                cached["image_ref"],
+                cached["img_path"],
+                "image",
+                len(cached["image_ref"]),
+                "",
+                cached["img_summary_field"],
+                cached["temp_uid"],
+                "",
+                "",
+                time_stamp,
+                "",
+            ]
+        )
+        logger.debug(f"Skipped duplicate image (hash={img_hash[:12]}...)")
+        return headings_stack, df_list, False  # False = cache hit, don't increment
+
+    client = _get_vision_client()
     last_context = _find_img_context(headings_stack)
 
     # Image index (always present)
@@ -150,7 +180,7 @@ def handle_image(
     img_path = os.path.join(img_dir, f"{img_name}{img_ext}")
     os.rename(img_raw_path, img_path)  # if summary fails, renaming is not applied
 
-    temp_uid = gen_str_codes(hashlib.sha256(img_file["data"]).hexdigest())
+    temp_uid = gen_str_codes(img_hash)
 
     # Build img_summary_field for df_list: image-n + optional summary
     if img_summary:
@@ -183,7 +213,17 @@ def handle_image(
             "",
         ]
     )
-    return headings_stack, df_list
+
+    # Cache result for document-level dedup
+    if seen_images is not None:
+        seen_images[img_hash] = {
+            "img_path": img_path,
+            "image_ref": image_ref,
+            "img_summary_field": img_summary_field,
+            "temp_uid": temp_uid,
+        }
+
+    return headings_stack, df_list, True  # True = new image processed
 
 
 def _first_cols_rows(table_block, max_items=10, max_chars=20):
@@ -245,6 +285,7 @@ def handle_table(
     cell_images=None,
     img_dir=None,
     img_count=0,
+    seen_images=None,
 ):
     time_stamp = get_str_time()
 
@@ -256,6 +297,31 @@ def handle_table(
         for (row_idx, col_idx), images in cell_images.items():
             descriptions = []
             for img_data in images:
+                # Document-level dedup: perceptual hash for visual duplicates
+                cell_img_hash = perceptual_hash(img_data["data"])
+                if seen_images is not None and cell_img_hash in seen_images:
+                    cached = seen_images[cell_img_hash]
+                    descriptions.append(f"[{cached['img_summary_field']}]")
+                    table_img_entries.append(
+                        [
+                            cached["image_ref"],
+                            cached["img_path"],
+                            "image",
+                            len(cached["image_ref"]),
+                            "",
+                            cached["img_summary_field"],
+                            cached["temp_uid"],
+                            "",
+                            "",
+                            time_stamp,
+                            "",
+                        ]
+                    )
+                    logger.debug(
+                        f"Skipped duplicate table cell image (hash={cell_img_hash[:12]}...)"
+                    )
+                    continue
+
                 img_count += 1
                 img_ext = os.path.splitext(img_data["image_name"])[-1]
                 image_index = f"image-{img_count}"
@@ -286,7 +352,7 @@ def handle_table(
                 descriptions.append(f"[{effective_desc}]")
 
                 # Also add as IMAGE entry in df_list for indexing
-                temp_uid = gen_str_codes(hashlib.sha256(img_data["data"]).hexdigest())
+                temp_uid = gen_str_codes(cell_img_hash)
                 img_summary_field = (
                     f"{image_index}\n{img_summary}" if img_summary else image_index
                 )
@@ -312,12 +378,22 @@ def handle_table(
                     ]
                 )
 
+                # Cache result for document-level dedup
+                if seen_images is not None:
+                    seen_images[cell_img_hash] = {
+                        "img_path": relative_img_path,
+                        "image_ref": image_ref,
+                        "img_summary_field": img_summary_field,
+                        "temp_uid": temp_uid,
+                    }
+
             cell_image_map[(row_idx, col_idx)] = " ".join(descriptions)
 
         logger.info(
             f"Extracted {sum(len(v) for v in cell_images.values())} images from table-{table_count + 1} cells"
         )
 
+
     # Generate HTML with image descriptions embedded
     tb_html_str = table2html(
         block, cell_image_map=cell_image_map if cell_image_map else None
@@ -417,7 +493,10 @@ def iter_block_items(doc_data):
             "a": "http://schemas.openxmlformats.org/drawingml/2006/main",
             "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
             "w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main",
+            "v": "urn:schemas-microsoft-com:vml",
+            "o": "urn:schemas-microsoft-com:office:office",
         }
+        r_ns = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}"
 
         root = etree.fromstring(xml)
         body = root.find(".//w:body", namespaces=ns)
@@ -500,15 +579,17 @@ def iter_block_items(doc_data):
                     yield ele_num, p_obj or text, label, meta
                     ele_num += 1
 
-                # images
+                # images (DrawingML: <a:blip>)
+                seen_rids = set()
                 blips = elem.xpath(".//a:blip", namespaces=ns)
                 for b in blips:
-                    rid = b.get(
-                        "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed"
-                    )
+                    rid = b.get(f"{r_ns}embed")
+                    if not rid or rid in seen_rids:
+                        continue
                     target = rel_map.get(rid)
                     if not target or not target.startswith("media/"):
                         continue
+                    seen_rids.add(rid)
                     data = docx.read("word/" + target)
                     yield (
                         ele_num,
@@ -522,6 +603,50 @@ def iter_block_items(doc_data):
                         },
                     )
                     ele_num += 1
+
+                # TODO: Re-evaluate VML group extraction strategy. 
+                # Complex VML composite images (<v:group>) are currently skipped because extracting 
+                # <v:imagedata> piece-by-piece loses textual overlay and positioning. 
+                # Future plan: Use LibreOffice headless conversion to render the entire document 
+                # and map the perfectly rendered images back to the layout via text anchors.
+                """
+                # images (VML: <v:imagedata>) — convert to PNG
+                from PIL import Image as PILImage
+
+                vml_images = elem.xpath(".//v:imagedata", namespaces=ns)
+                for v in vml_images:
+                    rid = v.get(f"{r_ns}id")
+                    if not rid or rid in seen_rids:
+                        continue
+                    target = rel_map.get(rid)
+                    if not target or not target.startswith("media/"):
+                        continue
+                    seen_rids.add(rid)
+                    raw_data = docx.read("word/" + target)
+                    # Convert to PNG for uniform downstream handling
+                    try:
+                        pil_img = PILImage.open(io.BytesIO(raw_data))
+                        png_buf = io.BytesIO()
+                        pil_img.save(png_buf, format="PNG")
+                        png_data = png_buf.getvalue()
+                    except Exception as e:
+                        logger.warning(f"Failed to convert VML image to PNG: {e}")
+                        continue
+                    orig_name = target.split("/")[-1]
+                    png_name = os.path.splitext(orig_name)[0] + ".png"
+                    yield (
+                        ele_num,
+                        None,
+                        "IMAGE",
+                        {
+                            "image_name": png_name,
+                            "from": "paragraph_vml",
+                            "size": len(png_data),
+                            "data": png_data,
+                        },
+                    )
+                    ele_num += 1
+                """
                 map_index += 1
 
                 if toc_info["is_field_end"]:
@@ -538,15 +663,18 @@ def iter_block_items(doc_data):
                 cell_images = {}  # {(row_idx, col_idx): [{'image_name', 'data', 'size'}]}
                 for row_idx, tr in enumerate(elem.findall(".//w:tr", namespaces=ns)):
                     for col_idx, tc in enumerate(tr.findall(".//w:tc", namespaces=ns)):
-                        blips = tc.xpath(".//a:blip", namespaces=ns)
+                        cell_seen_rids = set()
                         imgs_in_cell = []
+                        # DrawingML images in cell
+                        blips = tc.xpath(".//a:blip", namespaces=ns)
                         for b in blips:
-                            rid = b.get(
-                                "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed"
-                            )
+                            rid = b.get(f"{r_ns}embed")
+                            if not rid or rid in cell_seen_rids:
+                                continue
                             target = rel_map.get(rid)
                             if not target or not target.startswith("media/"):
                                 continue
+                            cell_seen_rids.add(rid)
                             data = docx.read("word/" + target)
                             if (
                                 len(data) < 10 * 1024
@@ -559,6 +687,44 @@ def iter_block_items(doc_data):
                                     "size": len(data),
                                 }
                             )
+                        # TODO: VML in tables is temporarily skipped to avoid extracting 
+                        # fragmented textless background images. (Same as paragraph VML logic)
+                        """
+                        # VML images in cell — convert to PNG
+                        from PIL import Image as PILImage
+
+                        vml_in_cell = tc.xpath(".//v:imagedata", namespaces=ns)
+                        for v in vml_in_cell:
+                            rid = v.get(f"{r_ns}id")
+                            if not rid or rid in cell_seen_rids:
+                                continue
+                            target = rel_map.get(rid)
+                            if not target or not target.startswith("media/"):
+                                continue
+                            cell_seen_rids.add(rid)
+                            raw_data = docx.read("word/" + target)
+                            try:
+                                pil_img = PILImage.open(io.BytesIO(raw_data))
+                                png_buf = io.BytesIO()
+                                pil_img.save(png_buf, format="PNG")
+                                png_data = png_buf.getvalue()
+                            except Exception as e:
+                                logger.warning(
+                                    f"Failed to convert VML cell image to PNG: {e}"
+                                )
+                                continue
+                            if len(png_data) < 10 * 1024:
+                                continue
+                            orig_name = target.split("/")[-1]
+                            png_name = os.path.splitext(orig_name)[0] + ".png"
+                            imgs_in_cell.append(
+                                {
+                                    "image_name": png_name,
+                                    "data": png_data,
+                                    "size": len(png_data),
+                                }
+                            )
+                        """
                         if imgs_in_cell:
                             cell_images[(row_idx, col_idx)] = imgs_in_cell
 
@@ -607,9 +773,15 @@ def parse_docx(
     headings_stack = [{"level": -1, "content": doc_structure}]
     current_heading = ""
 
+    # Clean old artifacts to prevent accumulation across debug runs.
+    # In production each job uses a fresh workspace so rmtree never triggers.
     tb_dir = os.path.join(output_dir, "tables")
+    if os.path.isdir(tb_dir):
+        shutil.rmtree(tb_dir)
     os.makedirs(tb_dir, exist_ok=True)
     img_dir = os.path.join(output_dir, "images")
+    if os.path.isdir(img_dir):
+        shutil.rmtree(img_dir)
     os.makedirs(img_dir, exist_ok=True)
 
     block_tuples = list(iter_block_items(doc_data))
@@ -671,6 +843,7 @@ def parse_docx(
     df_list = []
     table_count = 0
     image_count = 0
+    _seen_images: dict[str, dict] = {}  # sha256_hex -> cached image info for dedup
 
     logger.debug("Parsing docx file... total_blocks={}", len(block_tuples))
     for block_tuple in block_tuples:
@@ -708,7 +881,7 @@ def parse_docx(
             if meta and meta.get("size", 0) < 10 * 1024:
                 continue
 
-            headings_stack, df_list = handle_image(
+            headings_stack, df_list, is_new = handle_image(
                 df_list,
                 meta,
                 img_dir,
@@ -716,8 +889,10 @@ def parse_docx(
                 current_heading,
                 image_count,
                 llm_paras["summary_image"],
+                seen_images=_seen_images,
             )
-            image_count += 1
+            if is_new:
+                image_count += 1
             current_heading = last_heading_before_block
 
         elif label == "TABLE":
@@ -734,6 +909,7 @@ def parse_docx(
                 cell_images=meta,
                 img_dir=img_dir,
                 img_count=image_count,
+                seen_images=_seen_images,
             )
             table_count += 1
             current_heading = last_heading_before_block
diff --git a/apps/worker/app/services/document_parser/image_compressor.py b/apps/worker/app/services/document_parser/image_compressor.py
index 9893c3de0..892f85ac8 100644
--- a/apps/worker/app/services/document_parser/image_compressor.py
+++ b/apps/worker/app/services/document_parser/image_compressor.py
@@ -39,16 +39,6 @@ class CompressionStats(NamedTuple):
     rename_map: dict  # {"old_relative_path": "new_relative_path"}, e.g. {"images/foo.png": "images/foo.jpg"}
 
 
-def _has_transparency(img) -> bool:
-    """Check whether a PIL Image has meaningful alpha (transparency) data."""
-    if img.mode != "RGBA":
-        return False
-    # Sample alpha channel — if any pixel has alpha < 250, treat as transparent
-    alpha = img.getchannel("A")
-    extrema = alpha.getextrema()
-    return extrema[0] < 250
-
-
 def compress_output_images(
     output_dir: str,
     *,
@@ -104,38 +94,9 @@ def compress_output_images(
             w, h = img.size
             needs_resize = max(w, h) > max_side
             is_png = ext == ".png"
-            has_alpha = is_png and _has_transparency(img)
-
-            if is_png and not has_alpha:
-                # Opaque PNG → convert to JPEG
-                if needs_resize:
-                    ratio = max_side / max(w, h)
-                    new_w, new_h = int(w * ratio), int(h * ratio)
-                    img = img.resize((new_w, new_h), Image.LANCZOS)
-                    resized_count += 1
-
-                # Convert RGBA → RGB for JPEG
-                if img.mode in ("RGBA", "P", "LA"):
-                    img = img.convert("RGB")
-
-                jpg_path = os.path.splitext(file_path)[0] + ".jpg"
-                img.save(jpg_path, "JPEG", quality=jpeg_quality, optimize=True)
-                img.close()
-
-                # Remove original PNG
-                if jpg_path != file_path:
-                    os.remove(file_path)
-                    # Record rename for downstream reference updates
-                    old_rel = f"images/{filename}"
-                    new_rel = f"images/{os.path.basename(jpg_path)}"
-                    rename_map[old_rel] = new_rel
-
-                converted += 1
-                processed += 1
-                total_after += os.path.getsize(jpg_path)
 
-            elif is_png and has_alpha:
-                # Transparent PNG → keep as PNG but resize if needed
+            if is_png:
+                # Keep as PNG, only resize if needed
                 if needs_resize:
                     ratio = max_side / max(w, h)
                     new_w, new_h = int(w * ratio), int(h * ratio)
@@ -188,7 +149,7 @@ def compress_output_images(
         ratio = total_before / total_after if total_after > 0 else 0
         logger.info(
             f"[image_compressor] Compressed {processed} images "
-            f"({converted} PNG→JPG, {resized_count} resized), "
+            f"({resized_count} resized), "
             f"skipped {skipped}. "
             f"Size: {total_before / 1024 / 1024:.1f}MB → {total_after / 1024 / 1024:.1f}MB "
             f"({ratio:.1f}x reduction)"
diff --git a/apps/worker/app/services/document_parser/image_parser.py b/apps/worker/app/services/document_parser/image_parser.py
index f5a91c7f8..dc8ffb541 100755
--- a/apps/worker/app/services/document_parser/image_parser.py
+++ b/apps/worker/app/services/document_parser/image_parser.py
@@ -36,6 +36,25 @@
 g_img_lock = threading.Lock()
 
 
+def perceptual_hash(data: bytes) -> str:
+    """Compute a normalized pixel-data hash for image dedup.
+
+    Word/PDF may embed the same visual image with different compression
+    or metadata, making raw-byte SHA256 differ.  This function decodes
+    the image, converts to RGBA, and hashes the raw pixel buffer so
+    that visually-identical images always produce the same digest.
+
+    Falls back to raw-bytes hash when PIL cannot decode the data.
+    """
+    try:
+        img = Image.open(io.BytesIO(data))
+        pixels = img.convert("RGBA").tobytes()
+        return hashlib.sha256(pixels).hexdigest()
+    except Exception:
+        return hashlib.sha256(data).hexdigest()
+
+
+
 def _get_vision_client() -> OpenAICompatibleClientSync:
     """Create OpenAI-compatible client for vision models, auto-routing by IMAGE_MODEL name."""
     image_model = settings.IMAGE_MODEL or "qwen-vl-plus"
diff --git a/apps/worker/app/services/document_parser/layout_parser.py b/apps/worker/app/services/document_parser/layout_parser.py
index 12d467d7a..f308e7f5a 100755
--- a/apps/worker/app/services/document_parser/layout_parser.py
+++ b/apps/worker/app/services/document_parser/layout_parser.py
@@ -1197,7 +1197,7 @@ def hiearchy_llm(
     model_name=None,
     max_depth=6,
     toc_context=None,
-    max_len=2048,
+    max_len=8192,
     task="eval-headings",
 ):
     """Apply LLM to analyze the hierarchy of headings
diff --git a/apps/worker/app/services/document_parser/md_parser.py b/apps/worker/app/services/document_parser/md_parser.py
index c39d42f58..404d7b0cb 100755
--- a/apps/worker/app/services/document_parser/md_parser.py
+++ b/apps/worker/app/services/document_parser/md_parser.py
@@ -3,6 +3,7 @@
 import json
 import os
 import re
+import shutil
 from pathlib import Path
 
 import gevent
@@ -22,6 +23,7 @@
     _get_vision_client,
     ask_image,
     detect_summary_img_md,
+    perceptual_hash,
 )
 from app.services.document_parser.layout_parser import md_heading_match, pred_titles
 from app.services.document_parser.stage_profiler import stage_timer
@@ -309,10 +311,15 @@ def parse_md(
             json.dump(toc_hierarchies, f, ensure_ascii=False, indent=2)
         logger.info(f"Saved TOC hierarchies to {toc_json_path}")
 
-    # create local storage
+    # Clean old artifacts to prevent accumulation across debug runs.
+    # In production each job uses a fresh workspace so rmtree never triggers.
     tb_dir = os.path.join(output_dir, "tables")
+    if os.path.isdir(tb_dir):
+        shutil.rmtree(tb_dir)
     os.makedirs(tb_dir, exist_ok=True)
     img_dir = os.path.join(output_dir, "images")
+    if os.path.isdir(img_dir):
+        shutil.rmtree(img_dir)
     os.makedirs(img_dir, exist_ok=True)
 
     # initialize vars
@@ -332,6 +339,7 @@ def parse_md(
     img_count = 1
     path_counter = {}  # Track path occurrences for deduplication
     deferred_llm_tasks = []  # Collected during loop, executed in parallel after
+    _seen_images: dict[str, dict] = {}  # sha256_hex -> cached image info for dedup
 
     # Find layout.json path
     layout_json_path = os.path.join(output_dir, "layout.json")
@@ -474,6 +482,38 @@ def parse_md(
                     img_count += 1
                     continue
 
+                # Document-level dedup: perceptual hash for visual duplicates
+                with open(source_path, "rb") as f:
+                    img_binary_hash = perceptual_hash(f.read())
+
+                if img_binary_hash in _seen_images:
+                    cached = _seen_images[img_binary_hash]
+                    content_items.append(cached["img_content"])
+                    df_list.append(
+                        [
+                            cached["img_content"],
+                            cached["relative_img_path"],
+                            "image",
+                            len(cached["img_content"]),
+                            "",
+                            cached["img_summary_field"],
+                            cached["temp_uid"],
+                            "",
+                            "",
+                            time_stamp,
+                            str(current_pg_num) if current_pg_num > 0 else "",
+                        ]
+                    )
+                    logger.debug(
+                        f"Skipped duplicate image (hash={img_binary_hash[:12]}...)"
+                    )
+                    # Remove unused source file since we reuse the cached image
+                    try:
+                        source_path.unlink()
+                    except OSError:
+                        pass
+                    continue
+
                 os.rename(source_path, update_img_path)
 
                 # Image index (always present)
@@ -483,8 +523,6 @@ def parse_md(
                 effective_summary = img_summary or last_context or None
 
                 # Deterministic know_id: use image binary hash
-                with open(update_img_path, "rb") as img_f:
-                    img_binary_hash = hashlib.sha256(img_f.read()).hexdigest()
                 temp_uid = gen_str_codes(img_binary_hash)
                 relative_img_path = f"images/{img_name}{img_suffix}"
                 img_ref = build_chunk_ref(relative_img_path)
@@ -518,6 +556,15 @@ def parse_md(
                         str(current_pg_num) if current_pg_num > 0 else "",
                     ]
                 )
+
+                # Cache result for document-level dedup
+                _seen_images[img_binary_hash] = {
+                    "relative_img_path": relative_img_path,
+                    "img_content": img_content,
+                    "img_summary_field": img_summary_field,
+                    "temp_uid": temp_uid,
+                }
+
                 if base_llm_paras["summary_image"]:
                     # Store img_dir, img_name, img_suffix for post-loop rename (mirrors table deferred task)
                     deferred_llm_tasks.append(
diff --git a/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py b/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py
index 9332c044f..c57301e72 100644
--- a/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py
+++ b/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py
@@ -59,7 +59,6 @@ class ChunkPayload(TypedDict):
     content: str
     path: str
     metadata: ChunkMetadata
-    text: str
     order: int
     know_id: str
     keywords: list[str]
@@ -289,7 +288,6 @@ def dataframe_to_chunks(df: _ParserDataFrame | None) -> list[Dict[str, JsonValue
                 "content": content,
                 "path": path,
                 "metadata": metadata,
-                "text": content,
                 "order": index,
                 "know_id": str(know_id),
                 "keywords": metadata["keywords"],
diff --git a/packages/shared-python/shared/services/job_lifecycle_sync.py b/packages/shared-python/shared/services/job_lifecycle_sync.py
index f599d71ac..b0b7b197d 100644
--- a/packages/shared-python/shared/services/job_lifecycle_sync.py
+++ b/packages/shared-python/shared/services/job_lifecycle_sync.py
@@ -60,6 +60,7 @@ def finalize_job_success(
         stored_count: int = 0,
         delivery_mode: str = "url",
         section_summaries: Optional[Dict[str, str]] = None,
+        chunk_dedup_stats: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
         """Finalize a successful job in a single atomic transaction.
 
@@ -82,6 +83,7 @@ def finalize_job_success(
                     inline_payload=inline_payload,
                     result_s3_key=result_s3_key,
                     result_size=zip_size,
+                    chunk_dedup_stats=chunk_dedup_stats,
                 )
 
                 normalized_chunks = chunks or []
@@ -100,7 +102,7 @@ def finalize_job_success(
                         chunks=normalized_chunks,
                     )
                 )
-                if published_document_state is not None:
+                if published_document_state is not None and not published_document_state.get("skipped_all_duplicate"):
                     # Backfill DocumentSection.summary from enriched doc_nav data
                     if section_summaries:
                         self._backfill_section_summaries(
@@ -304,6 +306,7 @@ def _upsert_job_result(
         inline_payload: Optional[Dict[str, Any]] = None,
         result_s3_key: Optional[str] = None,
         result_size: Optional[int] = None,
+        chunk_dedup_stats: Optional[Dict[str, Any]] = None,
     ) -> JobResult:
         """Create or update JobResult row."""
         result = db.execute(select(JobResult).where(JobResult.job_id == job_id))
@@ -311,17 +314,24 @@ def _upsert_job_result(
 
         if existing:
             existing.delivery_mode = delivery_mode
-            existing.document_metadata = {}
+            doc_meta = existing.document_metadata or {}
+            if chunk_dedup_stats:
+                doc_meta["chunk_dedup"] = chunk_dedup_stats
+            existing.document_metadata = doc_meta
             existing.inline_payload = inline_payload
             existing.result_s3_key = result_s3_key
             existing.result_size = result_size
             db.flush()
             return existing
 
+        doc_meta = {}
+        if chunk_dedup_stats:
+            doc_meta["chunk_dedup"] = chunk_dedup_stats
+
         job_result = JobResult(
             job_id=job_id,
             delivery_mode=delivery_mode,
-            document_metadata={},
+            document_metadata=doc_meta,
             inline_payload=inline_payload,
             result_s3_key=result_s3_key,
             result_size=result_size,
diff --git a/packages/shared-python/shared/services/retrieval/publication_service.py b/packages/shared-python/shared/services/retrieval/publication_service.py
index 0bc95b163..aa2d3f675 100644
--- a/packages/shared-python/shared/services/retrieval/publication_service.py
+++ b/packages/shared-python/shared/services/retrieval/publication_service.py
@@ -8,12 +8,13 @@
 
 from __future__ import annotations
 
+from collections import defaultdict
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 from uuid import uuid4
 
 from loguru import logger
-from sqlalchemy import delete, select
+from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
 from shared.models.database.document import Document, DocumentChunk, DocumentSection
@@ -35,6 +36,141 @@ def utc_now_naive() -> datetime:
 
 
 class RetrievalPublicationService:
+
+    # ── Chunk-level content-hash dedup ──────────────────────────────────
+    # Mirrors graph_builder._dedup_chunks_by_content but operates on
+    # the DB document_chunks table instead of local knowledge_graph.json.
+
+    @staticmethod
+    def _collect_existing_chunk_id_map(
+        db: Session,
+        *,
+        user_id: str,
+        namespace: str,
+    ) -> Dict[str, str]:
+        """Return {chunk_id -> document_id} for all active document chunks
+        in the given (user_id, namespace) scope.
+
+        Only considers chunks belonging to the *current* revision of each
+        active document (Document.current_job_result_id == DocumentChunk.job_result_id).
+        """
+        rows = db.execute(
+            select(DocumentChunk.chunk_id, DocumentChunk.document_id)
+            .join(
+                Document,
+                (Document.document_id == DocumentChunk.document_id)
+                & (Document.current_job_result_id == DocumentChunk.job_result_id),
+            )
+            .where(
+                Document.user_id == user_id,
+                Document.namespace == namespace,
+                Document.status == "active",
+            )
+        ).all()
+        return {row[0]: row[1] for row in rows}
+
+    @staticmethod
+    def _dedup_chunks_by_content(
+        new_chunks: List[Dict[str, Any]],
+        existing_chunk_map: Dict[str, str],
+    ) -> tuple[List[Dict[str, Any]], Dict[str, int]]:
+        """Filter new_chunks: discard any whose chunk_id already exists.
+
+        Uses the same deterministic know_id (content-hash) comparison as
+        graph_builder._dedup_chunks_by_content.
+
+        Returns:
+            (deduped_chunks, overlap_by_document)
+            - deduped_chunks: chunks whose chunk_id is NOT in existing_chunk_map
+            - overlap_by_document: {document_id: count} of skipped chunks per
+              existing document (for observability logging)
+        """
+        overlap_by_document: Dict[str, int] = defaultdict(int)
+        deduped: List[Dict[str, Any]] = []
+        skipped = 0
+
+        for chunk in new_chunks:
+            cid = str(chunk.get("chunk_id") or chunk.get("know_id", ""))
+            if cid and cid in existing_chunk_map:
+                skipped += 1
+                overlap_by_document[existing_chunk_map[cid]] += 1
+            else:
+                deduped.append(chunk)
+
+        if skipped > 0:
+            logger.warning(
+                f"📊 DB chunk dedup: {skipped}/{len(new_chunks)} duplicate chunks "
+                f"skipped (by chunk_id), {len(deduped)} new chunks to insert. "
+                f"Overlap by document: {dict(overlap_by_document)}"
+            )
+        return deduped, dict(overlap_by_document)
+
+    @classmethod
+    def garbage_collect_and_dedup_local_media(
+        cls,
+        db: Session,
+        *,
+        job_id: str,
+        user_id: str,
+        namespace: str,
+        add_dir: str,
+        chunks: List[Dict[str, Any]],
+    ) -> tuple[List[Dict[str, Any]], Dict[str, Any]]:
+        """
+        Deduplicates chunks against the DB and physically deletes associated redundant 
+        media files (images/tables) from the local add_dir before ZIP packaging.
+        Returns the deduplicated chunks.
+        """
+        import os
+        
+        logger.info(f"[{job_id}] Starting local GC for redundant media files in namespace: {namespace}...")
+        try:
+            existing_map = cls._collect_existing_chunk_id_map(
+                db, user_id=user_id, namespace=namespace
+            )
+            deduped_chunks, overlap = cls._dedup_chunks_by_content(chunks, existing_map)
+            
+            stats = {
+                "total_incoming": len(chunks),
+                "duplicates_skipped": len(chunks) - len(deduped_chunks),
+                "new_chunks_inserted": len(deduped_chunks),
+                "overlap_by_document": overlap,
+            }
+            
+            if len(deduped_chunks) < len(chunks):
+                active_paths = set()
+                for c in deduped_chunks:
+                    fp = c.get("metadata", {}).get("file_path") or c.get("file_path")
+                    if fp:
+                        active_paths.add(fp)
+                        
+                deleted_count = 0
+                if add_dir and os.path.exists(add_dir):
+                    for c in chunks:
+                        fp = c.get("metadata", {}).get("file_path") or c.get("file_path")
+                        if fp and fp not in active_paths:
+                            abs_path = os.path.join(add_dir, fp)
+                            if os.path.exists(abs_path):
+                                os.remove(abs_path)
+                                deleted_count += 1
+                                
+                logger.info(f"[{job_id}] GC complete: permanently removed {deleted_count} redundant local media files.")
+                return deduped_chunks, stats
+            else:
+                logger.info(f"[{job_id}] GC complete: no redundant chunks found.")
+                return chunks, stats
+        except Exception as e:
+            logger.error(f"[{job_id}] GC failed (non-fatal): {e}")
+            stats = {
+                "total_incoming": len(chunks),
+                "duplicates_skipped": 0,
+                "new_chunks_inserted": len(chunks),
+                "overlap_by_document": {},
+            }
+            return chunks, stats
+
+    # ── Public API ──────────────────────────────────────────────────────
+
     def get_existing_document_scope(
         self,
         db: Session,
@@ -87,11 +223,27 @@ def _publish_document_state_for_job(
         chunks: List[Dict[str, Any]],
     ) -> Optional[Dict[str, str]]:
 
-        metadata = job.job_metadata or {}
-        namespace = metadata.get("namespace")
-        document_id = metadata.get("document_id")
-        source_file_name = metadata.get("source_file_name") or metadata.get("file_name")
+        job_metadata = job.job_metadata or {}
+        namespace = job_metadata.get("namespace") or "default"
+        document_id = job_metadata.get("document_id")
+        source_file_name = job_metadata.get("source_file_name") or job_metadata.get("file_name")
 
+        deduped_chunks = chunks
+
+        # If ALL chunks are duplicates → skip document creation entirely
+        if not deduped_chunks:
+            logger.warning(
+                f"⏭️  All chunks are duplicates of existing documents. "
+                f"Skipping document creation for job_id={job.job_id}."
+            )
+            return {
+                "user_id": str(job.user_id),
+                "namespace": namespace,
+                "document_id": None,
+                "skipped_all_duplicate": True,
+            }
+
+        # ── Document upsert (original logic, but only for deduped chunks) ──
         document = None
         if document_id:
             document = db.execute(
@@ -107,7 +259,7 @@ def _publish_document_state_for_job(
             document = Document(
                 document_id=document_id or f"doc_{uuid4().hex[:12]}",
                 user_id=str(job.user_id),
-                namespace=namespace or "default",
+                namespace=namespace,
                 status="active",
                 current_job_result_id=job_result_id,
                 source_file_name=source_file_name,
@@ -149,10 +301,11 @@ def _publish_document_state_for_job(
             .where(DocumentSection.job_result_id == job_result_id)
         )
 
+        # ── Insert only deduped (non-duplicate) chunks ──────────────────
         sections_by_path: Dict[str, DocumentSection] = {}
-        for index, chunk in enumerate(chunks):
-            metadata = chunk.get("metadata") or {}
-            source_path = metadata.get("path") or chunk.get("path")
+        for index, chunk in enumerate(deduped_chunks):
+            chunk_metadata = chunk.get("metadata") or {}
+            source_path = chunk_metadata.get("path") or chunk.get("path")
             section_path = section_path_from_chunk_path(source_path)
             section = sections_by_path.get(section_path)
             if section is None:
@@ -209,8 +362,8 @@ def _publish_document_state_for_job(
                     ),
                     term_search_text=build_term_search_text(chunk, path_text=path_text),
                     source_chunk_path=source_path,
-                    file_path=metadata.get("file_path") or chunk.get("file_path"),
-                    chunk_metadata=metadata,
+                    file_path=chunk_metadata.get("file_path") or chunk.get("file_path"),
+                    chunk_metadata=chunk_metadata,
                     sort_order=chunk.get("order", index),
                 )
             )
diff --git a/packages/shared-python/shared/utils/text_utils.py b/packages/shared-python/shared/utils/text_utils.py
index df77471d1..adbd292fc 100644
--- a/packages/shared-python/shared/utils/text_utils.py
+++ b/packages/shared-python/shared/utils/text_utils.py
@@ -199,7 +199,10 @@ def _tokenize_english_segment(text: str) -> list[str]:
 def _tokenize_cjk_segment(text: str) -> list[str]:
     if not text.strip():
         return []
-    return list(_jieba.lcut(text))
+    try:
+        return list(_jieba.lcut(text))
+    except AttributeError:
+        return list(_jieba.cut(text))
 
 
 def _resolve_retrieval_stopwords(
@@ -298,7 +301,10 @@ def tokenize2stw_remove(contents: List[str], stopwords: Optional[List[str]] = No
     for content in contents:
         # Pre-clean: remove IMAGE_/TABLE_ markers and reference labels
         content = _CHUNK_MARKER_RE.sub('', content)
-        raw_tokens = _jieba.lcut(content)
+        try:
+            raw_tokens = _jieba.lcut(content)
+        except AttributeError:
+            raw_tokens = list(_jieba.cut(content))
         # Filter: keep only tokens with meaningful characters (Chinese/English/numbers)
         tokens = [t for t in raw_tokens if _is_meaningful_token(t)]
         # Remove stopwords