R6A-R4-R1 evidence-first qualitative quote invocation

core/generation_runner.py

@@ -271,6 +271,237 @@ class GenerationRunner:
             raise RuntimeError(f"Qualitative quotations lack a Findings owner: {missing[:8]}")
         return owners
 
+    @staticmethod
+    def _qualitative_task_key(task):
+        task = task or {}
+        explicit = str(task.get("task_key") or "").strip()
+        if explicit:
+            return explicit
+        section_id = str(task.get("section_id") or task.get("section") or "section").strip()
+        stable_unit = str(task.get("theme_id") or task.get("subsection_id") or section_id).strip()
+        part_index = int(task.get("part_index") or 0) + 1
+        return f"{section_id}:{stable_unit}:part:{part_index:02d}"
+
+    @classmethod
+    def _build_qualitative_quote_invocation_plan(cls, tasks, data_metadata):
+        """Freeze one deterministic body invocation for every retained quote.
+
+        Logical ownership is a stable Theme object. Runtime Part chunks only
+        select where the already-owned quote is rendered; they never become the
+        quote owner and can never change cardinality.
+        """
+        metadata = data_metadata or {}
+        evidence_by_tag = {
+            str(item.get("tag") or "").strip(): dict(item)
+            for item in metadata.get("evidence", [])
+            if isinstance(item, dict) and str(item.get("tag") or "").strip()
+        }
+        theme_id_by_title = {}
+        for index, theme in enumerate(metadata.get("themes") or [], start=1):
+            if not isinstance(theme, dict):
+                continue
+            title = re.sub(r"\s+", " ", str(theme.get("title") or "")).strip().casefold()
+            if title:
+                theme_id_by_title[title] = str(theme.get("theme_id") or f"THEME-{index:02d}").strip()
+
+        plan = {}
+        task_slots = {}
+        for task in tasks or []:
+            if str(task.get("section_id") or "") != "findings":
+                continue
+            task_key = cls._qualitative_task_key(task)
+            task_theme_id = str(task.get("theme_id") or "").strip()
+            task_theme_title = str(task.get("theme_title") or task.get("subsection") or "").strip()
+            for item in task.get("theme_evidence") or []:
+                if not isinstance(item, dict):
+                    continue
+                tag = str(item.get("tag") or "").strip()
+                if not tag:
+                    continue
+                evidence = evidence_by_tag.get(tag)
+                if evidence is None:
+                    raise RuntimeError(f"Qualitative quote plan references unknown evidence: {tag}")
+                if tag in plan:
+                    raise RuntimeError(f"Qualitative quote plan has duplicate render assignment: {tag}")
+                owner_theme_id = str(
+                    task_theme_id
+                    or evidence.get("theme_id")
+                    or theme_id_by_title.get(
+                        re.sub(r"\s+", " ", str(evidence.get("theme") or task_theme_title)).strip().casefold()
+                    )
+                    or ""
+                ).strip()
+                if not owner_theme_id:
+                    raise RuntimeError(f"Qualitative quote lacks a stable theme owner: {tag}")
+                quote = str(evidence.get("quote") or "").strip()
+                if not quote:
+                    raise RuntimeError(f"Qualitative quote is empty: {tag}")
+                quote_hash = str(evidence.get("quote_hash") or "").strip() or hashlib.sha256(
+                    re.sub(r"\s+", " ", quote).strip().encode("utf-8")
+                ).hexdigest()
+                slot_ordinal = task_slots.get(task_key, 0) + 1
+                task_slots[task_key] = slot_ordinal
+                plan[tag] = {
+                    "quote_id": tag,
+                    "quote_hash": quote_hash,
+                    "owner_theme_id": owner_theme_id,
+                    "owner_theme_title": str(evidence.get("theme") or task_theme_title).strip(),
+                    "render_task_key": task_key,
+                    "render_part_index": int(task.get("part_index") or 0),
+                    "render_part_count": int(task.get("part_count") or 1),
+                    "slot_ordinal": slot_ordinal,
+                    "cardinality": 1,
+                }
+
+        missing = sorted(set(evidence_by_tag) - set(plan))
+        extra = sorted(set(plan) - set(evidence_by_tag))
+        if missing or extra:
+            raise RuntimeError(
+                f"Qualitative quote invocation plan mismatch: missing={missing[:8]} extra={extra[:8]}"
+            )
+        cls._validate_qualitative_quote_invocation_plan(plan, tasks, metadata)
+        return plan
+
+    @classmethod
+    def _validate_qualitative_quote_invocation_plan(cls, plan, tasks, data_metadata):
+        metadata = data_metadata or {}
+        evidence_by_tag = {
+            str(item.get("tag") or "").strip(): dict(item)
+            for item in metadata.get("evidence", [])
+            if isinstance(item, dict) and str(item.get("tag") or "").strip()
+        }
+        task_by_key = {cls._qualitative_task_key(task): task for task in tasks or []}
+        if set(plan or {}) != set(evidence_by_tag):
+            raise RuntimeError("Qualitative quote invocation plan does not match the frozen evidence ledger.")
+        for tag, record in (plan or {}).items():
+            if int(record.get("cardinality") or 0) != 1:
+                raise RuntimeError(f"Qualitative quote cardinality must equal one: {tag}")
+            owner_theme_id = str(record.get("owner_theme_id") or "").strip()
+            if not owner_theme_id or "part " in owner_theme_id.casefold():
+                raise RuntimeError(f"Qualitative quote owner is not a stable Theme object: {tag}")
+            task_key = str(record.get("render_task_key") or "").strip()
+            task = task_by_key.get(task_key)
+            if not task or str(task.get("section_id") or "") != "findings":
+                raise RuntimeError(f"Qualitative quote render task is invalid: {tag} -> {task_key!r}")
+            task_theme_id = str(task.get("theme_id") or "").strip()
+            if task_theme_id and task_theme_id != owner_theme_id:
+                raise RuntimeError(f"Qualitative quote render task crossed theme ownership: {tag}")
+            evidence = evidence_by_tag[tag]
+            quote = str(evidence.get("quote") or "").strip()
+            expected_hash = str(evidence.get("quote_hash") or "").strip() or hashlib.sha256(
+                re.sub(r"\s+", " ", quote).strip().encode("utf-8")
+            ).hexdigest()
+            if str(record.get("quote_hash") or "") != expected_hash:
+                raise RuntimeError(f"Qualitative quote hash drifted after planning: {tag}")
+        return True
+
+    @staticmethod
+    def _strip_model_controlled_qualitative_quotes(text, data_metadata):
+        """Remove model-emitted quote controls/literals before Executor rendering.
+
+        This is a containment boundary, not the quote control plane: quotation
+        presence and cardinality come exclusively from the frozen invocation
+        plan. The model's attempted tags/literals therefore have zero authority.
+        """
+        value = str(text or "")
+        value = re.sub(r"(?<!\w)\[?QUOTE-\d+\]?(?!\w)", "", value, flags=re.I)
+        quotes = sorted(
+            {
+                str(item.get("quote") or "").strip()
+                for item in (data_metadata or {}).get("evidence", [])
+                if isinstance(item, dict) and str(item.get("quote") or "").strip()
+            },
+            key=len,
+            reverse=True,
+        )
+        for quote in quotes:
+            for wrapped in (f'"{quote}"', f'“{quote}”', f'«{quote}»', quote):
+                value = value.replace(wrapped, "")
+        value = re.sub(r"[ \t]+([,.;:!?])", r"\1", value)
+        value = re.sub(r"([,;:])\s*([.!?])", r"\2", value)
+        value = re.sub(r"[ \t]{2,}", " ", value)
+        value = re.sub(r"\n[ \t]+", "\n", value)
+        return value.strip()
+
+    @classmethod
+    def _render_planned_qualitative_quotes(cls, text, *, task, data_metadata, quote_invocation_plan):
+        """Deterministically insert frozen quotations into one rendered task."""
+        value = cls._strip_model_controlled_qualitative_quotes(text, data_metadata)
+        task_key = cls._qualitative_task_key(task)
+        scheduled = [
+            dict(record)
+            for record in (quote_invocation_plan or {}).values()
+            if str(record.get("render_task_key") or "") == task_key
+        ]
+        scheduled.sort(key=lambda record: (int(record.get("slot_ordinal") or 0), str(record.get("quote_id") or "")))
+        if not scheduled:
+            return value
+
+        evidence_by_tag = {
+            str(item.get("tag") or "").strip(): dict(item)
+            for item in (data_metadata or {}).get("evidence", [])
+            if isinstance(item, dict) and str(item.get("tag") or "").strip()
+        }
+        blocks = [block.strip() for block in re.split(r"\n\s*\n", value) if block.strip()]
+        anchor_indices = [
+            index for index, block in enumerate(blocks)
+            if not re.match(r"^(?:#{1,6}\s|\||[-*+]\s|>)", block)
+        ]
+        if not anchor_indices:
+            anchor_indices = [len(blocks) - 1] if blocks else []
+
+        insertions = {}
+        for index, record in enumerate(scheduled):
+            tag = str(record.get("quote_id") or "").strip()
+            quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip()
+            if not quote:
+                raise RuntimeError(f"Planned qualitative quotation is unavailable at render time: {tag}")
+            if anchor_indices:
+                anchor_pos = min(
+                    len(anchor_indices) - 1,
+                    max(0, int(((index + 1) * len(anchor_indices)) / (len(scheduled) + 1))),
+                )
+                block_index = anchor_indices[anchor_pos]
+                insertions.setdefault(block_index, []).append(f'"{quote}"')
+            else:
+                blocks.append(f'"{quote}"')
+
+        rendered_blocks = []
+        for index, block in enumerate(blocks):
+            rendered_blocks.append(block)
+            rendered_blocks.extend(insertions.get(index, []))
+        rendered = "\n\n".join(rendered_blocks).strip()
+        for record in scheduled:
+            tag = str(record.get("quote_id") or "").strip()
+            quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip()
+            if rendered.count(quote) != 1:
+                raise RuntimeError(f"Qualitative deterministic quote rendering failed: {tag}")
+        return rendered
+
+    @staticmethod
+    def _validate_qualitative_quote_rendering(text, quote_invocation_plan, data_metadata):
+        body_text = re.split(
+            r"(?m)^#\s+Appendix(?:[:\s].*)?$", str(text or ""), maxsplit=1
+        )[0]
+        unresolved = sorted(set(re.findall(r"\bQUOTE-\d+\b", body_text, flags=re.I)))
+        if unresolved:
+            raise RuntimeError(f"Unresolved qualitative quote controls in body: {unresolved[:8]}")
+        normalized_body = re.sub(r"\s+", " ", body_text).casefold()
+        evidence_by_tag = {
+            str(item.get("tag") or "").strip(): dict(item)
+            for item in (data_metadata or {}).get("evidence", [])
+            if isinstance(item, dict) and str(item.get("tag") or "").strip()
+        }
+        for tag, record in (quote_invocation_plan or {}).items():
+            quote = re.sub(r"\s+", " ", str((evidence_by_tag.get(tag) or {}).get("quote") or "")).strip().casefold()
+            expected = int(record.get("cardinality") or 0)
+            actual = normalized_body.count(quote) if quote else 0
+            if actual != expected:
+                raise RuntimeError(
+                    f"Qualitative quote invocation invariant failed: {tag} actual={actual} planned={expected}"
+                )
+        return True
+
     @staticmethod
     def _redact_qualitative_quotes(text, data_metadata):
         """Hide prior verbatim quote text from cross-unit continuity context."""
@@ -723,10 +954,19 @@ class GenerationRunner:
         )
         updated_sections = dict(translation_sections or {})
         quote_ledger_record = dict(updated_sections.get("__quote_ledger__") or {})
+        current_quote_hashes = {
+            tag: hashlib.sha256(re.sub(r"\s+", " ", quote).strip().encode("utf-8")).hexdigest()
+            for tag, quote in getattr(self, "_qualitative_quote_sources", {}).items()
+        }
+        current_request_hash = str(getattr(self, "request_hash", "") or "")
+        stored_quote_hashes = quote_ledger_record.get("source_quote_hashes")
+        ledger_identity_ok = (
+            quote_ledger_record.get("target_lang") in {None, "", target_lang}
+            and quote_ledger_record.get("request_hash") in {None, "", current_request_hash}
+            and (stored_quote_hashes in {None, ""} or stored_quote_hashes == current_quote_hashes)
+        )
         quote_ledger = (
-            dict(quote_ledger_record.get("translations") or {})
-            if quote_ledger_record.get("target_lang") in {None, "", target_lang}
-            else {}
+            dict(quote_ledger_record.get("translations") or {}) if ledger_identity_ok else {}
         )
         translation_source, pending_quote_tags, reused_quote_tags = self._prepare_qualitative_translation_source(
             source, getattr(self, "_qualitative_quote_sources", {}), quote_ledger
@@ -768,7 +1008,12 @@ class GenerationRunner:
                 result = dict(result)
                 result['text'] = translated_text
                 updated_sections['__quote_ledger__'] = {
+                    'task_id': getattr(getattr(self, "task_dir", None), "name", ""),
+                    'request_hash': current_request_hash,
+                    'writer_execution_epoch': int(getattr(self, "execution_epoch", 0) or 0),
                     'target_lang': target_lang,
+                    'translation_contract_version': TRANSLATION_CONTRACT_VERSION,
+                    'source_quote_hashes': current_quote_hashes,
                     'translations': dict(quote_ledger),
                     'updated_at': datetime.now(timezone.utc).isoformat(),
                 }
@@ -1118,6 +1363,10 @@ class GenerationRunner:
             adaptive_structure = agent.build_generation_tasks(structure, body_target_words, safe_chunk_words)
             for task in adaptive_structure:
                 task["is_first"] = bool(task.get("section_first"))
+            quote_invocation_plan = (
+                self._build_qualitative_quote_invocation_plan(adaptive_structure, data_metadata)
+                if mode == "asq" else {}
+            )
             jit_texts = agent.prefetch_jit_literature(topic, keywords, adaptive_structure)
             if jit_texts:
                 rag_engine._initialize_chunks([jit_texts])
@@ -1139,6 +1388,7 @@ class GenerationRunner:
             state.update(
                 {
                     "adaptive_structure": adaptive_structure,
+                    "quote_invocation_plan": quote_invocation_plan,
                     "citation_plan": citation_plan,
                     "research_logic_contract": research_logic_contract,
                     "working_abstract": working_abstract,
@@ -1271,11 +1521,19 @@ class GenerationRunner:
             )
             state["citation_plan"] = citation_plan
 
-        quote_ownership = (
-            self._build_qualitative_quote_ownership(adaptive_structure, data_metadata)
-            if mode == "asq"
-            else {}
+        quote_invocation_plan = (
+            dict(state.get("quote_invocation_plan") or {}) if mode == "asq" else {}
         )
+        if mode == "asq":
+            if not quote_invocation_plan:
+                quote_invocation_plan = self._build_qualitative_quote_invocation_plan(
+                    adaptive_structure, data_metadata
+                )
+                state["quote_invocation_plan"] = quote_invocation_plan
+                self._checkpoint("quote_plan_ready", state, 15)
+            self._validate_qualitative_quote_invocation_plan(
+                quote_invocation_plan, adaptive_structure, data_metadata
+            )
 
         total_steps = len(adaptive_structure)
         for sec in adaptive_structure:
@@ -1352,20 +1610,22 @@ class GenerationRunner:
             )
             concepts = ", ".join(sec.get("concepts") or [])
             theme_evidence = [item for item in (sec.get("theme_evidence") or []) if isinstance(item, dict)]
-            evidence_labels = []
+            evidence_profiles = []
             for item in theme_evidence:
-                tag = str(item.get("tag") or "").strip()
                 stage = str(item.get("stage") or "Not specified").strip()
                 role = str(item.get("evidence_role") or "typical").strip()
                 case_label = str(item.get("case_label") or "Not specified").strip()
-                if tag:
-                    evidence_labels.append(f"{tag} [stage={stage}; case={case_label}; role={role}]")
+                evidence_profiles.append(f"stage={stage}; case={case_label}; role={role}")
+            planned_quote_count = sum(
+                1 for record in quote_invocation_plan.values()
+                if str(record.get("render_task_key") or "") == self._qualitative_task_key(sec)
+            )
             qualitative_rule = (
-                f"Assigned qualitative concepts: {concepts}. Assigned quotation tags: {', '.join(evidence_labels)}. "
-                "Use only these assigned quotation tags in this Findings unit and do not repeat a tag used elsewhere. "
-                "When invoking a quotation, output only its [QUOTE-XX] tag at the exact position; do not reproduce or paraphrase the quotation because the system inserts the frozen verbatim text. "
-                "Preserve any frozen stage/case/deviant role and use deviant evidence to qualify rather than erase the pattern."
-                if concepts or evidence_labels else ""
+                f"Assigned qualitative concepts: {concepts}. Assigned evidence profiles: {' | '.join(evidence_profiles)}. "
+                f"The Executor has frozen {planned_quote_count} quotation invocation(s) for this task. "
+                "Use the assigned evidence analytically, but output narrative prose only: do not reproduce any verbatim quotation and do not output any QUOTE-XX identifier. "
+                "The Executor inserts the frozen quotations deterministically after narrative/citation QA. Preserve stage/case/deviant meaning and use deviant evidence to qualify rather than erase the pattern."
+                if concepts or evidence_profiles else ""
             )
 
             scoped_data_md = user_data_md
@@ -1383,6 +1643,7 @@ class GenerationRunner:
                         theme_titles=[current_theme] if current_theme else None,
                         assigned_evidence=theme_evidence,
                         include_quotes=bool(theme_evidence),
+                        include_quote_ids=False,
                     )
                 elif section_id == "discussion":
                     scoped_data_md = qual_engine.build_scoped_evidence_md(
@@ -1452,36 +1713,34 @@ Output only final academic prose. Do not report the word count."""
                 prefix += f"# {sec_title}\n\n"
             if sec.get("subsection_first") and subsection:
                 prefix += f"## {subsection}\n\n"
-            en_raw = self._stream_text(
-                agent.stream_deepseek(
-                    sys_prompt, user_prompt, max_tokens=7000,
-                    model_role="writing", thinking=False, allow_fallback=False,
-                ),
-                preview_prefix=full_draft_en + prefix,
-                preview_stage=display_name,
-            )
-            en_raw = re.sub(r"<think>.*?</think>", "", en_raw, flags=re.S).strip()
-            en_raw = re.sub(rf"(?im)^#\s*{re.escape(sec_title)}\s*\n+", "", en_raw).strip()
-            if subsection:
-                en_raw = re.sub(rf"(?im)^##\s*{re.escape(subsection)}\s*\n+", "", en_raw).strip()
-            en_raw = re.sub(r"(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)", r"\1 \2", en_raw)
-            en_raw = re.sub(r"(?m)^#{4,}\s+(.*)", r"**\1**", en_raw)
-            en_raw = re.sub(r"(?m)^#\s+(.+)$", r"## \1", en_raw)
-            en_raw = agent.critique_and_repair_section(
-                en_raw, section_id=section_id, section_name=display_name, mode=mode,
-                evidence_contract=evidence_packet,
-            )
-            en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False, mode=mode)
-            # Bind only the current Findings unit's frozen quotation ownership.
-            # Other ASQ units cannot see or reproduce verbatim quote text.
-            if mode == "asq" and "appendix" not in sec_title.lower():
-                en_raw = self._bind_owned_qualitative_quotes(
-                    en_raw,
-                    display_name=display_name,
-                    data_metadata=data_metadata,
-                    quote_ownership=quote_ownership,
-                    prior_manuscript=full_draft_en,
+            if mode == "asq" and section_id == "appendix":
+                # R6A-R4: Appendix is a deterministic reader-facing projection
+                # of the Evidence Ledger. It is never regenerated by the LLM.
+                en_raw = qual_engine.build_appendix_evidence_md(data_metadata)
+            else:
+                en_raw = self._stream_text(
+                    agent.stream_deepseek(
+                        sys_prompt, user_prompt, max_tokens=7000,
+                        model_role="writing", thinking=False, allow_fallback=False,
+                    ),
+                    preview_prefix=full_draft_en + prefix,
+                    preview_stage=display_name,
+                )
+                en_raw = re.sub(r"<think>.*?</think>", "", en_raw, flags=re.S).strip()
+                en_raw = re.sub(rf"(?im)^#\s*{re.escape(sec_title)}\s*\n+", "", en_raw).strip()
+                if subsection:
+                    en_raw = re.sub(rf"(?im)^##\s*{re.escape(subsection)}\s*\n+", "", en_raw).strip()
+                en_raw = re.sub(r"(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)", r"\1 \2", en_raw)
+                en_raw = re.sub(r"(?m)^#{4,}\s+(.*)", r"**\1**", en_raw)
+                en_raw = re.sub(r"(?m)^#\s+(.+)$", r"## \1", en_raw)
+                en_raw = agent.critique_and_repair_section(
+                    en_raw, section_id=section_id, section_name=display_name, mode=mode,
+                    evidence_contract=evidence_packet,
                 )
+                en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False, mode=mode)
+                if mode == "asq":
+                    # Model prose has no authority over quote presence/count.
+                    en_raw = self._strip_model_controlled_qualitative_quotes(en_raw, data_metadata)
             allow_deterministic_citation_reinforcement = not (
                 mode == "asq" and section_id in {"methods", "discussion"}
             )
@@ -1531,6 +1790,13 @@ Output only final academic prose. Do not report the word count."""
                             en_resolved = en_resolved.rstrip() + "\n\n" + supplemental_resolved.strip()
                         used_after = set(agent.used_citations)
                         survived = used_after - used_before
+            if mode == "asq" and section_id == "findings":
+                en_resolved = self._render_planned_qualitative_quotes(
+                    en_resolved,
+                    task=sec,
+                    data_metadata=data_metadata,
+                    quote_invocation_plan=quote_invocation_plan,
+                )
             full_draft_en += en_resolved + "\n\n"
             if assigned_tag_list:
                 logging.info(
@@ -1586,6 +1852,9 @@ Output only final academic prose. Do not report the word count."""
             # sections, citation cleanup, and any Discussion rebalance are frozen.
             agent.validate_manuscript_structure(full_draft_en, adaptive_structure, translated=False)
             if mode == "asq":
+                self._validate_qualitative_quote_rendering(
+                    full_draft_en, quote_invocation_plan, data_metadata
+                )
                 repeated_quotes = agent.find_repeated_qualitative_quotes(full_draft_en, data_metadata)
                 if repeated_quotes:
                     raise RuntimeError(f"Repeated qualitative quotations: {repeated_quotes[:5]}")

core/pipeline_agent.py

@@ -1496,6 +1496,7 @@ USER KEYWORD SEEDS: {keywords}
                 seen_ids = set()
                 for index, theme in enumerate(themes, start=1):
                     title = str(theme.get("title") or f"Theme {index}").strip()
+                    theme_id = str(theme.get("theme_id") or f"THEME-{index:02d}").strip()
                     base_id = self._slugify_heading(title)
                     subsection_id = base_id
                     suffix = 2
@@ -1505,6 +1506,9 @@ USER KEYWORD SEEDS: {keywords}
                     seen_ids.add(subsection_id)
                     concepts = [str(item).strip() for item in theme.get("concepts", []) if str(item).strip()]
                     theme_evidence = [dict(item) for item in (theme.get("evidence") or []) if isinstance(item, dict)]
+                    for evidence_item in theme_evidence:
+                        evidence_item.setdefault("theme_id", theme_id)
+                        evidence_item.setdefault("theme", title)
                     stages = [str(item).strip() for item in (theme.get("stages") or []) if str(item).strip()]
                     has_deviant = bool(theme.get("has_deviant_evidence"))
                     dynamic_subsections.append(
@@ -1512,11 +1516,12 @@ USER KEYWORD SEEDS: {keywords}
                             "subsection_id": subsection_id,
                             "title": title,
                             "share": per_theme_share,
+                            "theme_id": theme_id,
                             "theme_title": title,
                             "concepts": concepts,
                             "theme_evidence": theme_evidence,
                             "required_content": (
-                                "Explain this theme using only its assigned concepts and tagged representative quotations. "
+                                "Explain this theme using only its assigned concepts and representative evidence. The Executor owns verbatim quotation rendering; do not reproduce quotation wording or internal quotation identifiers. "
                                 + (f"Preserve the frozen process order across these stages: {', '.join(stages)}. " if stages else "")
                                 + ("Use the assigned deviant/boundary evidence to qualify the pattern rather than hiding it. " if has_deviant else "")
                                 + "Do not reuse evidence assigned to another theme or invent a temporal sequence."
@@ -1549,6 +1554,7 @@ USER KEYWORD SEEDS: {keywords}
                     subsection_id = subsection.get("subsection_id") or self._slugify_heading(subsection.get("title"))
                     subsection_title = subsection.get("title")
                     required_content = subsection.get("required_content", "")
+                    theme_id = subsection.get("theme_id")
                     theme_title = subsection.get("theme_title")
                     concepts = subsection.get("concepts", [])
                     theme_evidence = subsection.get("theme_evidence", [])
@@ -1557,6 +1563,7 @@ USER KEYWORD SEEDS: {keywords}
                     subsection_id = None
                     subsection_title = None
                     required_content = ""
+                    theme_id = None
                     theme_title = None
                     concepts = []
                     theme_evidence = []
@@ -1574,18 +1581,25 @@ USER KEYWORD SEEDS: {keywords}
                         label += f" > {subsection_title}"
                     if part_count > 1:
                         label += f" (Part {part_index + 1}/{part_count})"
+                    section_id = section.get("section_id") or self._slugify_heading(section["section"])
+                    stable_unit_id = str(theme_id or subsection_id or section_id or "section")
+                    task_key = f"{section_id}:{stable_unit_id}:part:{part_index + 1:02d}"
                     tasks.append(
                         {
-                            "section_id": section.get("section_id") or self._slugify_heading(section["section"]),
+                            "section_id": section_id,
                             "section": section["section"],
                             "subsection_id": subsection_id,
                             "subsection": subsection_title,
                             "display_name": label,
+                            "task_key": task_key,
+                            "part_index": part_index,
+                            "part_count": part_count,
                             "target_words": part_target,
                             "section_first": unit_index == 0 and part_index == 0,
                             "subsection_first": bool(subsection_title) and part_index == 0,
                             "continuation": part_index > 0,
                             "required_content": required_content,
+                            "theme_id": theme_id,
                             "theme_title": theme_title,
                             "concepts": concepts,
                             "theme_evidence": part_theme_evidence,

core/qualitative_engine.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import logging
 import re
 
@@ -44,6 +45,7 @@ class QualitativeEngine:
         theme_titles=None,
         assigned_evidence=None,
         include_quotes=False,
+        include_quote_ids=True,
     ):
         """Render the minimum qualitative evidence needed by one drafting unit.
 
@@ -89,7 +91,7 @@ class QualitativeEngine:
             "[QUALITATIVE EVIDENCE CHANNELS]",
             "1. STUDY FACTS and THEME SUMMARIES below are this study's internal empirical evidence. Never attach an external [CITE-XX] tag to a sentence merely because it reports or interprets these frozen facts.",
             "2. External literature may support only prior-theory, prior-evidence, or methodological-rationale claims. Keep those claims in distinct sentences from this study's setting, sample, data, events, quotations, and findings.",
-            "3. ASSIGNED QUOTATIONS, when present, are verbatim evidence owned by the current Findings unit. Invoke them only by their [QUOTE-XX] tag; do not reuse a quotation owned by another unit.",
+            "3. ASSIGNED QUOTATIONS, when present, are analysis-only verbatim evidence owned by the current stable theme. Do not reproduce the verbatim wording or any internal quotation identifier in generated prose; the Executor renders frozen quotations deterministically after narrative QA.",
         ]
 
         allowed_theme_titles = None
@@ -151,17 +153,61 @@ class QualitativeEngine:
                     evidence_role = "typical"
                 quote_rows.append((tag, stage, case_label, evidence_role, quote))
             if quote_rows:
-                lines.extend([
-                    "",
-                    "### Assigned Qualitative Quotations",
-                    "| Tag | Process Stage | Case/Actor | Evidence Role | Verbatim Quote |",
-                    "| :--- | :--- | :--- | :--- | :--- |",
-                ])
-                lines.extend(
-                    f'| {tag} | {stage} | {case_label} | {role} | "{quote}" |'
-                    for tag, stage, case_label, role, quote in quote_rows
-                )
+                lines.extend(["", "### Assigned Qualitative Quotations"])
+                if include_quote_ids:
+                    lines.extend([
+                        "| Tag | Process Stage | Case/Actor | Evidence Role | Verbatim Quote |",
+                        "| :--- | :--- | :--- | :--- | :--- |",
+                    ])
+                    lines.extend(
+                        f'| {tag} | {stage} | {case_label} | {role} | "{quote}" |'
+                        for tag, stage, case_label, role, quote in quote_rows
+                    )
+                else:
+                    lines.extend([
+                        "| Process Stage | Case/Actor | Evidence Role | Verbatim Quote |",
+                        "| :--- | :--- | :--- | :--- |",
+                    ])
+                    lines.extend(
+                        f'| {stage} | {case_label} | {role} | "{quote}" |'
+                        for _tag, stage, case_label, role, quote in quote_rows
+                    )
+
+        return "\n".join(lines).strip()
+
+    @classmethod
+    def build_appendix_evidence_md(cls, metadata):
+        """Render the reader-facing qualitative evidence appendix deterministically.
+
+        Internal QUOTE identifiers are deliberately omitted. The same frozen
+        quotation literals are reused by the translation ledger, so Findings
+        and Appendix share one target-language rendering per quote.
+        """
+        metadata = metadata or {}
+
+        def cell(value, default="Not specified"):
+            return cls._clean_cell(value, default).replace("|", r"\|")
 
+        lines = [
+            "| Higher-order Theme | Empirical Concept | Process Stage | Case/Actor | Evidence Role | Representative Quote |",
+            "| :--- | :--- | :--- | :--- | :--- | :--- |",
+        ]
+        for item in metadata.get("evidence") or []:
+            if not isinstance(item, dict):
+                continue
+            quote = cls._clean_cell(item.get("quote"), "").replace('"', "'")
+            if not quote:
+                continue
+            lines.append(
+                "| {theme} | {concept} | {stage} | {case_label} | {role} | \"{quote}\" |".format(
+                    theme=cell(item.get("theme"), "Emergent pattern"),
+                    concept=cell(item.get("concept"), "Empirical observation"),
+                    stage=cell(item.get("stage")),
+                    case_label=cell(item.get("case_label")),
+                    role=cell(str(item.get("evidence_role") or "typical").strip().lower(), "typical"),
+                    quote=quote.replace("|", r"\|"),
+                )
+            )
         return "\n".join(lines).strip()
 
     def _build_request(self, topic, keywords, raw_interviews):
@@ -329,6 +375,7 @@ Rules:
                     "theme": theme,
                     "concept": concept,
                     "quote": quote,
+                    "quote_hash": hashlib.sha256(re.sub(r"\s+", " ", quote).strip().encode("utf-8")).hexdigest(),
                     "tag": tag,
                     "stage": stage,
                     "case_label": case_label,
@@ -337,10 +384,15 @@ Rules:
             )
 
         theme_map = {}
+        theme_ids = {}
         for item in clean_evidence:
+            theme_key = re.sub(r"\s+", " ", item["theme"]).strip().casefold()
+            theme_id = theme_ids.setdefault(theme_key, f"THEME-{len(theme_ids) + 1:02d}")
+            item["theme_id"] = theme_id
             theme_entry = theme_map.setdefault(
                 item["theme"],
                 {
+                    "theme_id": theme_id,
                     "title": item["theme"],
                     "concepts": [],
                     "evidence": [],
@@ -356,8 +408,10 @@ Rules:
                 theme_entry["has_deviant_evidence"] = True
             theme_entry["evidence"].append(
                 {
+                    "theme_id": item["theme_id"],
                     "tag": item["tag"],
                     "quote": item["quote"],
+                    "quote_hash": item["quote_hash"],
                     "stage": item["stage"],
                     "case_label": item["case_label"],
                     "evidence_role": item["evidence_role"],