R3.2.8.6-R1.3C-KK1-R6A-R2 Qualitative Quote Ownership and Empirical-Literature Evidence Boundary Fix

core/generation_runner.py

@@ -239,6 +239,132 @@ class GenerationRunner:
         text = re.sub(r"[ \t]{2,}", " ", text)
         return text.strip()
 
+    @staticmethod
+    def _build_qualitative_quote_ownership(tasks, data_metadata):
+        """Freeze one and only one Findings owner for every qualitative quote."""
+        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()
+        }
+        owners = {}
+        for task in tasks or []:
+            if str(task.get("section_id") or "") != "findings":
+                continue
+            owner = str(task.get("display_name") 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
+                if tag not in evidence_by_tag:
+                    raise RuntimeError(f"Qualitative quotation ownership references unknown tag: {tag}")
+                if tag in owners:
+                    raise RuntimeError(
+                        f"Qualitative quotation has multiple Findings owners: {tag} -> {owners[tag]!r}, {owner!r}"
+                    )
+                owners[tag] = owner
+        missing = sorted(set(evidence_by_tag) - set(owners))
+        if missing:
+            raise RuntimeError(f"Qualitative quotations lack a Findings owner: {missing[:8]}")
+        return owners
+
+    @staticmethod
+    def _redact_qualitative_quotes(text, data_metadata):
+        """Hide prior verbatim quote text from cross-unit continuity context."""
+        value = str(text or "")
+        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:
+            value = value.replace(f'"{quote}"', "[prior qualitative quotation omitted]")
+            value = value.replace(quote, "[prior qualitative quotation omitted]")
+        return value
+
+    @classmethod
+    def _bind_owned_qualitative_quotes(
+        cls,
+        text,
+        *,
+        display_name,
+        data_metadata,
+        quote_ownership,
+        prior_manuscript="",
+    ):
+        """Bind only the current Findings unit's owned quotation tags.
+
+        Other manuscript units are not allowed to contain frozen quote text or
+        quote tags.  A direct verbatim reproduction of an owned quote is
+        normalized back to its tag before deterministic binding, preserving the
+        frozen wording without letting model prose create a second copy.
+        """
+        value = str(text or "")
+        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()
+        }
+        allowed = {
+            tag for tag, owner in (quote_ownership or {}).items()
+            if str(owner) == str(display_name)
+        }
+
+        raw_tags = set(re.findall(r"\bQUOTE-\d+\b", value))
+        unauthorized_tags = sorted(raw_tags - allowed)
+        if unauthorized_tags:
+            raise RuntimeError(
+                f"Qualitative quotation tag used outside owning Findings unit {display_name!r}: {unauthorized_tags[:8]}"
+            )
+
+        # Quote text itself is also fenced.  Owned verbatim text is converted
+        # back to a tag and then bound once; unowned verbatim text fails early.
+        for tag, item in evidence_by_tag.items():
+            quote = str(item.get("quote") or "").strip()
+            if not quote or quote not in value:
+                continue
+            if tag not in allowed:
+                raise RuntimeError(
+                    f"Qualitative quotation text used outside owning Findings unit: {tag} in {display_name!r}"
+                )
+            value = value.replace(f'"{quote}"', tag)
+            value = value.replace(quote, tag)
+
+        # A quote already present in frozen prior prose cannot be rebound later,
+        # even after resume. This makes ownership idempotent without new state.
+        normalized_prior = re.sub(r"\s+", " ", str(prior_manuscript or "")).casefold()
+        for tag in sorted(allowed):
+            quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip()
+            if quote and re.sub(r"\s+", " ", quote).casefold() in normalized_prior:
+                raise RuntimeError(f"Qualitative quotation already used before owning unit: {tag}")
+
+        for tag in sorted(allowed):
+            count = len(re.findall(rf"\b{re.escape(tag)}\b", value))
+            if count > 1:
+                raise RuntimeError(
+                    f"Qualitative quotation repeated within owning Findings unit: {tag} count={count}"
+                )
+            if count == 1:
+                quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip()
+                value = re.sub(
+                    rf"\b{re.escape(tag)}\b",
+                    lambda _match, frozen_quote=quote: f'"{frozen_quote}"',
+                    value,
+                    count=1,
+                )
+
+        # No unresolved quote tag may escape into citation QA or final output.
+        unresolved = sorted(set(re.findall(r"\bQUOTE-\d+\b", value)))
+        if unresolved:
+            raise RuntimeError(f"Unresolved qualitative quotation tags: {unresolved[:8]}")
+        return value
+
     def _write_preview_payload(self, *, force: bool = False) -> None:
         with self._preview_lock:
             if self.preview_path is None:
@@ -797,6 +923,16 @@ class GenerationRunner:
             )
             self._checkpoint("evidence_ready", state, 6)
 
+        # ASQ planning never needs access to verbatim quotations.  Keep the
+        # research-logic/front-matter stage on a quote-free internal-evidence
+        # summary; verbatim text is disclosed later only to its owning Findings
+        # unit (or the appendix).
+        planning_data_md = (
+            qual_engine.build_scoped_evidence_md(data_metadata, include_themes=True)
+            if mode == "asq"
+            else user_data_md
+        )
+
         refs_text = str(state.get("refs_text") or "")
         if not refs_text:
             self._progress(8, "S2 + OpenAlex")
@@ -838,7 +974,7 @@ class GenerationRunner:
             research_logic_response = self._stream_text(
                 agent.stream_generate_research_logic(
                     topic, keywords, synthesis_result, refs_text,
-                    data_evidence=user_data_md, mode=mode,
+                    data_evidence=planning_data_md, mode=mode,
                 )
             )
             research_logic_contract = re.sub(
@@ -851,7 +987,7 @@ class GenerationRunner:
             self._progress(14, "working_front_matter")
             working_front = agent.generate_working_front_matter(
                 topic, keywords, research_logic_contract, synthesis_result,
-                data_evidence=user_data_md, mode=mode,
+                data_evidence=planning_data_md, mode=mode,
             )
             working_abstract = str(working_front.get("abstract") or "").strip()
             working_keywords = list(working_front.get("keywords") or [])
@@ -964,7 +1100,7 @@ class GenerationRunner:
         if not working_abstract and not state.get("english_finalized"):
             working_front = agent.generate_working_front_matter(
                 topic, keywords, research_logic_contract, synthesis_result,
-                data_evidence=user_data_md, mode=mode,
+                data_evidence=planning_data_md, mode=mode,
             )
             working_abstract = str(working_front.get("abstract") or "").strip()
             working_keywords = list(working_front.get("keywords") or [])
@@ -1029,6 +1165,12 @@ class GenerationRunner:
             )
             state["citation_plan"] = citation_plan
 
+        quote_ownership = (
+            self._build_qualitative_quote_ownership(adaptive_structure, data_metadata)
+            if mode == "asq"
+            else {}
+        )
+
         total_steps = len(adaptive_structure)
         for sec in adaptive_structure:
             display_name = sec["display_name"]
@@ -1075,6 +1217,11 @@ class GenerationRunner:
                     "Never stack sources, substitute tags, or use literature as evidence for this study's own facts.\n"
                     f"<CLAIM_EVIDENCE_PLAN>\n{claim_plan_text}\n</CLAIM_EVIDENCE_PLAN>"
                 )
+            if mode == "asq" and section_id not in {"findings", "appendix"}:
+                citation_rule += (
+                    "\nASQ EVIDENCE-CHANNEL RULE: A sentence whose factual subject is this study's setting, participants, interviews, data sources, cases, actors, observations, events, quotations, analytic steps, or empirical findings must not carry an external [CITE-XX] tag. "
+                    "Place any prior-literature or methodological-rationale claim in a separate sentence and cite only that external claim."
+                )
             memory_context = agent._build_memory_context(sec_title, 2500)
             memory_block = f"\n<GLOBAL_MEMORY>\n{memory_context}\n</GLOBAL_MEMORY>\n" if memory_context else ""
             snippets = rag_engine.get_relevant_context(display_name) if rag_engine is not None else ""
@@ -1114,9 +1261,36 @@ class GenerationRunner:
                 "Preserve any frozen stage/case/deviant role and use deviant evidence to qualify rather than erase the pattern."
                 if concepts or evidence_labels else ""
             )
+
+            scoped_data_md = user_data_md
+            if mode == "asq":
+                if section_id == "appendix":
+                    # The appendix is the only non-owning unit allowed to see
+                    # the full frozen evidence table. It is excluded from the
+                    # body duplicate-quote gate by design.
+                    scoped_data_md = user_data_md
+                elif section_id == "findings":
+                    current_theme = str(sec.get("theme_title") or "").strip()
+                    scoped_data_md = qual_engine.build_scoped_evidence_md(
+                        data_metadata,
+                        include_themes=True,
+                        theme_titles=[current_theme] if current_theme else None,
+                        assigned_evidence=theme_evidence,
+                        include_quotes=bool(theme_evidence),
+                    )
+                elif section_id == "discussion":
+                    scoped_data_md = qual_engine.build_scoped_evidence_md(
+                        data_metadata,
+                        include_themes=True,
+                    )
+                else:
+                    scoped_data_md = qual_engine.build_scoped_evidence_md(
+                        data_metadata,
+                        include_themes=False,
+                    )
             evidence_packet = (
-                f"{base_evidence_contract}\n<FROZEN_RESEARCH_EVIDENCE>\n{user_data_md}\n</FROZEN_RESEARCH_EVIDENCE>"
-                if user_data_md else base_evidence_contract
+                f"{base_evidence_contract}\n<FROZEN_RESEARCH_EVIDENCE>\n{scoped_data_md}\n</FROZEN_RESEARCH_EVIDENCE>"
+                if scoped_data_md else base_evidence_contract
             )
             completed_label = ", ".join(completed_sections)
             anti_repeat = (
@@ -1132,6 +1306,9 @@ class GenerationRunner:
             word_budget = APP_CONFIG.get("word_budget", {})
             min_ratio = float(word_budget.get("section_min_ratio", 0.92))
             max_ratio = float(word_budget.get("section_max_ratio", 1.08))
+            prompt_context_tail = context_tail[-2500:]
+            if mode == "asq":
+                prompt_context_tail = self._redact_qualitative_quotes(prompt_context_tail, data_metadata)
             user_prompt = f"""Write the assigned English manuscript unit.
 Section: {sec_title}
 Subsection: {subsection or 'None'}
@@ -1154,9 +1331,9 @@ ANCHOR PRECEDENCE: Frozen research evidence and the PROVISIONAL_RESEARCH_LOGIC o
 {memory_block}
 Literature synthesis: {synthesis_result}
 <FROZEN_RESEARCH_EVIDENCE>
-{user_data_md}
+{scoped_data_md}
 </FROZEN_RESEARCH_EVIDENCE>
-Previous tail: {context_tail[-2500:]}
+Previous tail: {prompt_context_tail}
 Output only final academic prose. Do not report the word count."""
             if agent._count_tokens(sys_prompt) + agent._count_tokens(user_prompt) > 60000:
                 user_prompt = agent._truncate_context(user_prompt, 60000 - agent._count_tokens(sys_prompt))
@@ -1189,14 +1366,20 @@ Output only final academic prose. Do not report the word count."""
                 evidence_contract=evidence_packet,
             )
             en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False, mode=mode)
-            # Keep QUOTE tags immutable through critic/style editing, then bind
-            # each tag to its frozen verbatim quotation exactly once. This
-            # prevents editorial passes from silently paraphrasing evidence.
-            if mode == "asq" and user_data_md and "appendix" not in sec_title.lower():
-                quote_dict = {tag: quote for quote, tag in re.findall(r'"([^"]+)"\s*\[(QUOTE-\d+)\]', user_data_md)}
-                for tag, quote in quote_dict.items():
-                    en_raw = en_raw.replace(tag, f'"{quote}"')
-            if not is_quarantined and assigned_tag_list:
+            # 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,
+                )
+            allow_deterministic_citation_reinforcement = not (
+                mode == "asq" and section_id in {"methods", "discussion"}
+            )
+            if not is_quarantined and assigned_tag_list and allow_deterministic_citation_reinforcement:
                 en_raw = agent.reinforce_claim_specific_citation_coverage(
                     en_raw, assigned_tag_list, section_name=display_name
                 )

core/qualitative_engine.py

@@ -35,6 +35,135 @@ class QualitativeEngine:
         lookup = {item.lower(): item for item in cls.ALLOWED_ANALYTIC_DESIGNS}
         return lookup.get(cleaned.lower(), "Iterative thematic/process analysis")
 
+    @classmethod
+    def build_scoped_evidence_md(
+        cls,
+        metadata,
+        *,
+        include_themes=True,
+        theme_titles=None,
+        assigned_evidence=None,
+        include_quotes=False,
+    ):
+        """Render the minimum qualitative evidence needed by one drafting unit.
+
+        Study facts/theme summaries are an internal empirical channel. Verbatim
+        quotations are a separate channel and are exposed only to their owning
+        Findings unit (or the final appendix when explicitly requested by the
+        caller). This prevents a quotation visible in one unit from being
+        opportunistically reused elsewhere and keeps external literature from
+        becoming evidence for this study's own facts.
+        """
+        metadata = metadata or {}
+        facts = metadata.get("study_facts") or {}
+        sources = facts.get("data_sources") or ["Not specified"]
+        if not isinstance(sources, list):
+            sources = [sources]
+        clean_sources = [cls._clean_cell(item) for item in sources]
+        participants = facts.get("participant_count")
+        interviews = facts.get("interview_count")
+        analytic_design = cls._normalize_analytic_design(
+            facts.get("analytic_design") or facts.get("analysis_method")
+        )
+        analytic_logic = cls._clean_cell(
+            facts.get("analytic_logic"),
+            "Iteratively compare empirical observations to refine a process/mechanism explanation.",
+        )
+        temporal_structure = cls._clean_cell(facts.get("temporal_structure"))
+        case_structure = cls._clean_cell(facts.get("case_structure"))
+        setting = cls._clean_cell(facts.get("research_setting"))
+
+        lines = [
+            "### Frozen Qualitative Study Facts",
+            "| Field | Frozen value |",
+            "| :--- | :--- |",
+            f"| Research setting | {setting} |",
+            f"| Data sources | {', '.join(clean_sources)} |",
+            f"| Participants | {participants if participants is not None else 'Not specified'} |",
+            f"| Interviews | {interviews if interviews is not None else 'Not specified'} |",
+            f"| Analytic design | {analytic_design} |",
+            f"| Analytic logic | {analytic_logic} |",
+            f"| Temporal structure | {temporal_structure} |",
+            f"| Case structure | {case_structure} |",
+            "",
+            "[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.",
+        ]
+
+        allowed_theme_titles = None
+        if theme_titles:
+            allowed_theme_titles = {
+                re.sub(r"\s+", " ", str(item or "")).strip().casefold()
+                for item in theme_titles
+                if str(item or "").strip()
+            }
+
+        if include_themes:
+            theme_rows = []
+            for theme in metadata.get("themes") or []:
+                if not isinstance(theme, dict):
+                    continue
+                title = cls._clean_cell(theme.get("title"), "Emergent pattern")
+                if allowed_theme_titles is not None and title.casefold() not in allowed_theme_titles:
+                    continue
+                concepts = ", ".join(
+                    cls._clean_cell(item, "")
+                    for item in (theme.get("concepts") or [])
+                    if cls._clean_cell(item, "")
+                ) or "Not specified"
+                stages = ", ".join(
+                    cls._clean_cell(item, "")
+                    for item in (theme.get("stages") or [])
+                    if cls._clean_cell(item, "") and cls._clean_cell(item, "") != "Not specified"
+                ) or "Not specified"
+                deviant = "yes" if theme.get("has_deviant_evidence") else "no"
+                theme_rows.append((title, concepts, stages, deviant))
+            if theme_rows:
+                lines.extend([
+                    "",
+                    "### Frozen Qualitative Theme Summary",
+                    "| Higher-order Theme | Empirical Concepts | Process Stages | Deviant/Boundary Evidence Present |",
+                    "| :--- | :--- | :--- | :--- |",
+                ])
+                lines.extend(
+                    f"| {title} | {concepts} | {stages} | {deviant} |"
+                    for title, concepts, stages, deviant in theme_rows
+                )
+
+        if include_quotes:
+            evidence_items = assigned_evidence
+            if evidence_items is None:
+                evidence_items = metadata.get("evidence") or []
+            quote_rows = []
+            for item in evidence_items or []:
+                if not isinstance(item, dict):
+                    continue
+                tag = cls._clean_cell(item.get("tag"), "")
+                quote = cls._clean_cell(item.get("quote"), "").replace('"', "'")
+                if not tag or not quote:
+                    continue
+                stage = cls._clean_cell(item.get("stage"))
+                case_label = cls._clean_cell(item.get("case_label"))
+                evidence_role = str(item.get("evidence_role") or "typical").strip().lower()
+                if evidence_role not in {"typical", "deviant"}:
+                    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
+                )
+
+        return "\n".join(lines).strip()
+
     def _build_request(self, topic, keywords, raw_interviews):
         schema = """Return ONLY one valid JSON object with this schema:
 {