P3-A4 citation authority repair

core/pipeline_agent.py

@@ -1640,117 +1640,6 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         return cleaned
 
 
-    @staticmethod
-    def _english_reader_surface_has_foreign_script(value):
-        """Return True only for scripts that cannot remain in an English reader-facing surface.
-
-        User intent may be entered in Chinese, Russian, or Kazakh, but the English
-        manuscript projection must not leak Han or Cyrillic source text. Latin
-        acronyms, numerals, punctuation, and ordinary English remain untouched.
-        """
-        return bool(re.search(r'[\u3400-\u9fff\u3040-\u30ff\u0400-\u052f]', str(value or '')))
-
-    @classmethod
-    def validate_reader_facing_english_projection(cls, projection, *, expected_keyword_count=None):
-        """Validate a language projection without changing the user's semantic authority."""
-        record = dict(projection or {})
-        title = str(record.get('title') or record.get('english_title') or '').strip()
-        raw_keywords = record.get('keywords')
-        if raw_keywords is None:
-            raw_keywords = record.get('english_keywords')
-        keywords = [str(item).strip() for item in (raw_keywords or []) if str(item).strip()]
-        if not title:
-            raise DeepSeekRequestError('English reader-facing title is empty.')
-        if expected_keyword_count is not None and len(keywords) != int(expected_keyword_count):
-            raise DeepSeekRequestError(
-                f'English reader-facing keyword cardinality changed: expected={int(expected_keyword_count)} observed={len(keywords)}.'
-            )
-        combined = ' '.join([title] + keywords)
-        if cls._english_reader_surface_has_foreign_script(combined):
-            raise DeepSeekRequestError('English reader-facing title/keywords contain untranslated Han or Cyrillic source text.')
-        return {'title': title, 'keywords': keywords}
-
-    def normalize_reader_facing_front_matter_to_english(self, topic, keywords):
-        """Create a faithful English language projection of user Title/Keywords.
-
-        The original user strings remain semantic authority in the study contract.
-        This method is presentation-only: English input is passed through without a
-        model call; non-English/mixed-script input is translated, never optimized.
-        Keyword cardinality and order are invariant.
-        """
-        source_title = str(topic or '').strip()
-        source_keywords = parse_keyword_terms(keywords, explicit_user=True)
-        if not source_title:
-            raise DeepSeekRequestError('User research title is empty before English language projection.')
-
-        source_surface = ' '.join([source_title] + source_keywords)
-        if not self._english_reader_surface_has_foreign_script(source_surface):
-            return {
-                'title': source_title,
-                'keywords': list(source_keywords),
-                'translation_applied': False,
-            }
-
-        indexed_keywords = [
-            {'index': index, 'text': value}
-            for index, value in enumerate(source_keywords, start=1)
-        ]
-        payload = {'title': source_title, 'keywords': indexed_keywords}
-        base_prompt = f"""Translate ONLY the supplied user research Title and Keywords into academic English.
-Return JSON only: {{"title":"...","keywords":["..."]}}.
-
-LANGUAGE-AUTHORITY CONTRACT:
-1. This is translation, not title optimization, rewriting, topic expansion, or keyword generation.
-2. Preserve the complete research meaning: focal constructs, direction/relationship wording, mediator/moderator roles, population/context, method qualifiers, acronyms, numbers, and scope.
-3. Do not add, delete, merge, split, reorder, generalize, specialize, or reinterpret any keyword. Return exactly {len(source_keywords)} keywords in the supplied order.
-4. Preserve existing English technical identities and acronyms. Translate/transliterate Han or Cyrillic reader-facing text so the returned Title/Keywords contain no Chinese, Russian, or Kazakh script.
-5. Do not mention the translation process and do not wrap JSON in Markdown.
-
-<SOURCE_USER_FRONT_MATTER>
-{json.dumps(payload, ensure_ascii=False)}
-</SOURCE_USER_FRONT_MATTER>"""
-        current_prompt = base_prompt
-        last_error = None
-        for local_attempt in range(2):
-            response = self.call_deepseek(
-                'You are a precise academic translator. Preserve research semantics exactly and change only language surface.',
-                current_prompt,
-                max_tokens=max(700, min(1600, 500 + len(source_keywords) * 90)),
-                temp=0.0,
-                model_role='writing',
-                thinking=False,
-                allow_fallback=False,
-                response_format={'type': 'json_object'},
-                logical_node='front_matter_writing',
-            )
-            try:
-                parsed = self._extract_json_from_text(response)
-                if not isinstance(parsed, dict):
-                    raise DeepSeekRequestError('English front-matter language projection was not valid JSON.')
-                candidate = self.validate_reader_facing_english_projection(
-                    parsed, expected_keyword_count=len(source_keywords)
-                )
-                return {
-                    'title': candidate['title'],
-                    'keywords': candidate['keywords'],
-                    'translation_applied': True,
-                }
-            except DeepSeekRequestError as exc:
-                last_error = exc
-                if local_attempt >= 1:
-                    raise
-                logging.warning(
-                    'English front-matter language projection retry | attempt=%s/2 | reason=%s',
-                    local_attempt + 1, self._safe_source_error(exc),
-                )
-                current_prompt = base_prompt + (
-                    '\n\nLOCAL VALIDATION RETRY: The prior response violated the language/cardinality contract. '
-                    'Return a fresh complete JSON object only. Translate the same Title and the same indexed Keywords '
-                    'faithfully into English, preserve keyword order/count exactly, and remove all Han/Cyrillic source text. '
-                    'Do not optimize or rewrite the research intent.'
-                )
-        raise DeepSeekRequestError(f'English front-matter language projection failed: {last_error}')
-
     @staticmethod
     def _front_matter_has_citation(value):
         text = str(value or '')
@@ -1790,8 +1679,6 @@ LANGUAGE-AUTHORITY CONTRACT:
         word_count = cls._working_abstract_word_count(text)
         if cls._front_matter_has_citation(text):
             raise DeepSeekRequestError('Working Abstract must not contain citations.')
-        if cls._english_reader_surface_has_foreign_script(text):
-            raise DeepSeekRequestError('Working Abstract contains untranslated Han or Cyrillic source text.')
         forbidden = (
             ('RESULT_ASSERTION', r'\bwe\s+(?:find|found|show|demonstrate|observe|establish)\b'),
             ('RESULT_ASSERTION', r'\b(?:results?|findings?)\s+(?:show|indicate|reveal|demonstrate|confirm|support)\b'),
@@ -1988,28 +1875,24 @@ LANGUAGE-AUTHORITY CONTRACT:
         return issues
 
 
-    def reconcile_quantitative_reader_front_matter(
-        self, *, topic, keywords, contract, manuscript="", reader_title=None, reader_keywords=None
-    ):
-        """Preserve source semantic authority while allowing a faithful English surface.
+    def reconcile_quantitative_reader_front_matter(self, *, topic, keywords, contract, manuscript=""):
+        """Preserve R5 user intent while retaining legacy contract reconciliation.
 
-        ``user_research_intent`` remains the immutable semantic authority.  When
-        ``reader_title``/``reader_keywords`` are supplied they are a language-only
-        English projection of that same authority; they may change script/language
-        but not constructs, relations, scope, order, or keyword cardinality.
+        Contracts that carry ``user_research_intent`` are authoritative: the
+        finalizer must not repair an upstream design mismatch by silently
+        renaming the user study.  Older contracts without that explicit R5
+        authority retain the pre-R5 bounded reconciliation path so historical
+        checkpoints and compatibility behavior remain stable.
         """
         record = dict(contract or {})
         intent = dict(record.get("user_research_intent") or {})
         policy = keyword_policy_from_contract(record, fallback_keywords=keywords)
         source_topic = str(intent.get("source_topic") or record.get("source_topic") or topic or "").strip()
         source_keywords = list(intent.get("source_keywords") or [])
-        reader_source_topic = str(reader_title or source_topic or topic or "").strip()
-        reader_source_keywords = parse_keyword_terms(reader_keywords, explicit_user=True) if reader_keywords is not None else list(source_keywords)
         if str(policy.get("authority_source") or "").upper() == "USER_EXPLICIT" and source_keywords:
-            authoritative_reader = reader_source_keywords or source_keywords
             return {
-                "title": reader_source_topic,
-                "keywords": apply_keyword_policy(authoritative_reader, policy, authoritative=authoritative_reader),
+                "title": source_topic,
+                "keywords": apply_keyword_policy(source_keywords, policy, authoritative=source_keywords),
             }
 
         # No explicit user keyword authority: retain user-authoritative title when
@@ -2027,7 +1910,7 @@ LANGUAGE-AUTHORITY CONTRACT:
             logging.warning("Front-matter semantic contract missing frozen constructs; using degraded fallback.")
             fallback = self._quantitative_front_matter_fallback(record)
             if intent and source_topic:
-                fallback["title"] = reader_source_topic
+                fallback["title"] = source_topic
             return fallback
 
         source_keywords = list(record.get("source_keywords") or [])
@@ -2116,7 +1999,7 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
             extra = [str(item).strip() for item in ((critic or {}).get("unmeasured_focal_constructs") or []) if str(item).strip()]
             missing = [str(item).strip() for item in ((critic or {}).get("missing_frozen_constructs") or []) if str(item).strip()]
             if not invalid and not extra and not missing:
-                resolved_title = reader_source_topic if intent else re.sub(r"\s+", " ", str(candidate.get("title") or "")).strip()
+                resolved_title = source_topic if intent else re.sub(r"\s+", " ", str(candidate.get("title") or "")).strip()
                 return {
                     "title": resolved_title,
                     "keywords": apply_keyword_policy(candidate.get("keywords") or [], policy),
@@ -2130,11 +2013,7 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
 
         fallback = self._quantitative_front_matter_fallback(record)
         if intent and source_topic:
-            fallback["title"] = reader_source_topic
-        if str(policy.get("authority_source") or "").upper() == "USER_EXPLICIT" and reader_source_keywords:
-            fallback["keywords"] = apply_keyword_policy(
-                reader_source_keywords, policy, authoritative=reader_source_keywords
-            )
+            fallback["title"] = source_topic
         logging.warning("Adaptive front-matter reconciliation exhausted; using deterministic degraded fallback instead of failing manuscript generation.")
         return fallback
 
@@ -2149,7 +2028,7 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
         prompt = f"""Create WORKING front matter for an empirical {str(mode).upper()}-oriented manuscript before body drafting.
 Return JSON only: {{\"abstract\":\"...\",\"keywords\":[\"...\"]}}.
 Aim for approximately 120-180 English words, have no citations, and function only as a continuity anchor. Word count is guidance, not a failure condition; prioritize Conciseness, Accuracy, and Rigor over mechanical compliance. It may state the research problem, theoretical tension, focal concepts, question/hypothesis architecture, design or method class, and intended contribution domain. It MUST NOT state coefficients, p-values, fit statistics, sample statistics unless already indispensable to define the design, supported/unsupported hypotheses, result direction, substantive findings, or a conclusion that has not yet been frozen. Do not predict findings.
-Preserve the supplied English keyword identities and research scope. They are a faithful language projection of the user input. Do not silently replace, delete, reorder, or reinterpret an explicit construct/relation because direct literature support is sparse. Do not translate them back into the source input language.
+Preserve the user keyword identities and research scope. Do not silently replace, delete, or reinterpret an explicit construct/relation because direct literature support is sparse. The user-supplied keyword list is authoritative for reader-facing keyword identity; wording changes belong only to target-language translation.
 TOPIC: {topic}
 USER KEYWORDS: {keywords}
 <PROVISIONAL_RESEARCH_LOGIC>
@@ -2186,7 +2065,7 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
         prompt = f"""Write FINAL English front matter for this completed {str(mode).upper()} manuscript.
 Return JSON only: {{\"abstract\":\"...\",\"keywords\":[\"...\"]}}.
 Aim for a concise Abstract of roughly 120-200 words, use no citations, and accurately state the research problem, validated design, main frozen finding(s), and contribution. Word count is guidance, not a failure condition; prioritize Conciseness, Accuracy, and Rigor. Include only the minimum numerical evidence required; do not list coefficients, p-values, and fit statistics unless one value is indispensable. Do not infer beyond the frozen evidence or completed manuscript.
-Preserve the supplied English keyword identities and research scope. They are a faithful language projection of the user input. Do not silently replace, delete, reorder, or reinterpret an explicit construct/relation because direct literature support is sparse. Do not translate them back into the source input language.
+Preserve the user keyword identities and research scope. Do not silently replace, delete, or reinterpret an explicit construct/relation because direct literature support is sparse. The user-supplied keyword list is authoritative for reader-facing keyword identity; wording changes belong only to target-language translation.
 TOPIC: {topic}
 USER KEYWORDS: {keywords}
 <FROZEN_RESEARCH_EVIDENCE>
@@ -2206,8 +2085,6 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
             abstract = re.sub(r'(?im)^#+\s*abstract\s*', '', str(parsed.get('abstract') or '')).strip()
             if self._front_matter_has_citation(abstract):
                 raise DeepSeekRequestError('Final Abstract must not contain citations.')
-            if self._english_reader_surface_has_foreign_script(abstract):
-                raise DeepSeekRequestError('Final Abstract contains untranslated Han or Cyrillic source text.')
             meta_findings = self.inspect_authorial_ai_meta(abstract)
             if meta_findings:
                 raise DeepSeekRequestError('Final Abstract contains high-confidence authorial AI-writing meta-commentary.')
@@ -3054,17 +2931,8 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
                 future_lane[executor.submit(self._s2_worker, query, 0)] = 'method_authority'
                 future_lane[executor.submit(self._oa_worker, query, 1, "language:en,type:article,is_paratext:false")] = 'method_authority'
             for future in concurrent.futures.as_completed(future_lane):
+                results = future.result()
                 lane = future_lane[future]
-                try:
-                    results = future.result()
-                except PipelineCancelledError:
-                    raise
-                except Exception as exc:
-                    logging.warning(
-                        "Delivery-first METHOD_AUTHORITY source skipped | lane=%s error=%s",
-                        lane, type(exc).__name__,
-                    )
-                    continue
                 with self.lock:
                     for res in results:
                         result_keys = self._reference_keys(res)
@@ -3098,44 +2966,28 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
         sys_prompt = """You are an expert academic librarian. Generate exactly 3 concise English search queries.
 [CRITICAL SEARCH QUERY RULES]
 1. DO NOT use complex, deeply nested boolean logic (e.g., Avoid `(A OR B) AND (C OR D)`).
-2. DO NOT use excessive quotation marks for long phrases.
-3. Output simple, space-separated keyword combinations or simple AND logic.
+2. DO NOT use excessive quotation marks for long phrases. 
+3. Output simple, space-separated keyword combinations or simple AND logic. 
    Good: "educational chatbot" "recommendation system" context-aware
    Bad: ("educational chatbot" OR "pedagogical agent") AND ("recommendation system")
 4. Keep the query strictly UNDER 5 terms to avoid triggering server API timeouts.
 Return ONLY a valid JSON array of strings."""
         user_prompt = f"Topic: {topic}\nKeywords: {keywords}"
         for attempt in range(3):
-            try:
-                queries = self._extract_json_from_text(self.call_deepseek(
-                    sys_prompt,
-                    user_prompt,
-                    temp=0.1 + attempt * 0.1,
-                    model_role="fast",
-                    logical_node='search_jit',
-                ))
-            except PipelineCancelledError:
-                raise
-            except Exception as exc:
-                logging.warning(
-                    "Delivery-first search-query generation fallback | attempt=%s/3 error=%s",
-                    attempt + 1, type(exc).__name__,
-                )
-                queries = []
+            queries = self._extract_json_from_text(self.call_deepseek(
+                sys_prompt,
+                user_prompt,
+                temp=0.1 + attempt * 0.1,
+                model_role="fast",
+                logical_node='search_jit',
+            ))
             if queries and isinstance(queries, list) and len(queries) > 0:
                 final_queries = [str(q).strip() for q in queries if self._is_english_dominant(str(q).strip())]
-                if final_queries:
+                if final_queries: 
                     return final_queries[:3]
             time.sleep(1)
-
-        clean_chunks = [
-            c.strip() for c in re.findall(r'[a-zA-Z0-9\s\-]{4,}', f"{topic} {keywords}")
-            if self._is_english_dominant(c.strip())
-        ]
-        if clean_chunks:
-            logging.warning(
-                "Delivery-first search-query provider exhausted; using deterministic English user-text query fallback."
-            )
+            
+        clean_chunks = [c.strip() for c in re.findall(r'[a-zA-Z0-9\s\-]{4,}', f"{topic} {keywords}") if self._is_english_dominant(c.strip())]
         return clean_chunks[:3] if clean_chunks else []
 
     @staticmethod
@@ -3566,18 +3418,7 @@ Return ONLY a valid JSON array of strings."""
                 for future in concurrent.futures.as_completed(future_sources):
                     self._check_cancelled()
                     source = future_sources[future]
-                    try:
-                        rows = future.result()
-                    except PipelineCancelledError:
-                        raise
-                    except Exception as exc:
-                        self._record_source_error(source, 'RETRIEVAL_FUTURE_ERROR', detail=exc)
-                        logging.warning(
-                            "Delivery-first literature source future skipped | source=%s error=%s",
-                            source, type(exc).__name__,
-                        )
-                        rows = []
-                    collected[source].extend(rows)
+                    collected[source].extend(future.result())
                     if progress_callback:
                         progress_callback()
 
@@ -3586,24 +3427,15 @@ Return ONLY a valid JSON array of strings."""
             if phase == 0:
                 terms = self.generate_search_queries(topic, keywords)
             else:
-                try:
-                    response = self.call_deepseek(
-                        'You are an academic librarian. Return only a JSON array with three new concise English search queries, each under five terms.',
-                        f"Topic: {topic}\nKeywords: {keywords}\nUsed: {sorted(all_used_terms)}",
-                        temp=0.4,
-                        model_role='fast',
-                        logical_node='search_jit',
-                    )
-                    parsed = self._extract_json_from_text(response)
-                    terms = parsed if isinstance(parsed, list) else []
-                except PipelineCancelledError:
-                    raise
-                except Exception as exc:
-                    logging.warning(
-                        'Delivery-first auxiliary search-query phase skipped | phase=%s error=%s',
-                        phase, type(exc).__name__,
-                    )
-                    terms = []
+                response = self.call_deepseek(
+                    'You are an academic librarian. Return only a JSON array with three new concise English search queries, each under five terms.',
+                    f"Topic: {topic}\nKeywords: {keywords}\nUsed: {sorted(all_used_terms)}",
+                    temp=0.4,
+                    model_role='fast',
+                    logical_node='search_jit',
+                )
+                parsed = self._extract_json_from_text(response)
+                terms = parsed if isinstance(parsed, list) else []
 
             normalized_terms = []
             for value in terms:
@@ -5256,84 +5088,6 @@ INPUT:
             return True
         return False
 
-    @staticmethod
-    def _has_explicit_current_qualitative_study_anchor(sentence):
-        """Identify an explicit current-study owner even in a cited sentence.
-
-        A method citation may justify a procedure, but it must not make this
-        manuscript's own empirical/procedural clause externally owned.  Keep
-        this narrower than the general qualitative ownership heuristic so a
-        cited phrase such as ``the study conducted interviews`` is not
-        accidentally reclassified as the current manuscript.
-        """
-        value = str(sentence or "")
-        return bool(re.search(
-            r"(?i)\b(?:we|our\s+study|this\s+study|the\s+present\s+study|"
-            r"research\s+team|our\s+data|our\s+analysis|our\s+interviews?|"
-            r"our\s+focus\s+groups?|our\s+observations?|in\s+this\s+study|"
-            r"in\s+our\s+study)\b",
-            value,
-        ))
-
-    @staticmethod
-    def _normalize_qualitative_analytic_design_identity(value):
-        folded = re.sub(r"\s+", " ", str(value or "")).strip().casefold()
-        if not folded:
-            return ""
-        if "gioia" in folded:
-            return "gioia"
-        if "grounded theory" in folded:
-            return "grounded_theory"
-        if "comparative case" in folded:
-            return "comparative_case"
-        if "longitudinal" in folded and "qualitative" in folded:
-            return "longitudinal_qualitative"
-        if "thematic" in folded:
-            return "thematic_process"
-        if "process analysis" in folded:
-            return "process_analysis"
-        return folded
-
-    @classmethod
-    def _performed_qualitative_analytic_design_identity(cls, sentence):
-        """Return a performed current-study qualitative design, if explicit.
-
-        This is a prose-level evidence guard, not a new method router.  It
-        prevents a cited method name from silently replacing the analytic design
-        frozen in the qualitative evidence packet.
-        """
-        value = re.sub(r"\s+", " ", str(sentence or "")).strip()
-        if not value or not cls._has_explicit_current_qualitative_study_anchor(value):
-            return ""
-        candidates = (
-            ("gioia", r"\bGioia(?:-style)?(?:\s+inductive\s+coding|\s+methodology|\s+method|\s+approach)?\b"),
-            ("grounded_theory", r"\bgrounded\s+theory\b"),
-            ("comparative_case", r"\bcomparative\s+case\s+analysis\b"),
-            ("longitudinal_qualitative", r"\blongitudinal\s+qualitative\s+analysis\b"),
-            ("process_analysis", r"\bprocess\s+analysis\b"),
-            ("thematic_process", r"\b(?:iterative\s+)?thematic(?:/process)?\s+analysis\b|\bthematic\s+analysis\b"),
-        )
-        for identity, pattern in candidates:
-            match = re.search(pattern, value, flags=re.I)
-            if not match:
-                continue
-            before = value[max(0, match.start() - 120):match.start()]
-            if re.search(
-                r"(?i)\b(?:did\s+not|does\s+not|do\s+not|was\s+not|were\s+not|"
-                r"not|without|never)\b[^.;:]{0,45}$",
-                before,
-            ):
-                continue
-            full_before = value[:match.start()]
-            if re.search(
-                r"(?i)\b(?:use|uses|used|employ|employs|employed|apply|applies|applied|"
-                r"adopt|adopts|adopted|follow|follows|followed|conduct|conducts|conducted|"
-                r"analy[sz]e|analy[sz]es|analy[sz]ed|code|codes|coded)\b[^.;:]{0,180}$",
-                full_before,
-            ):
-                return identity
-        return ""
-
     @classmethod
     def _is_study_owned_qualitative_claim(cls, sentence, section_name=""):
         """Conservatively identify only this manuscript's empirical/procedural claims.
@@ -5348,11 +5102,6 @@ INPUT:
             return False
         if re.match(r'^["“‘\']', stripped) or re.search(r"\bQUOTE-\d+\b", value, flags=re.I):
             return False
-        # Explicit current-study ownership dominates an appended method or
-        # theory citation.  The cited source may justify the procedure, but it
-        # cannot own the statement that *this* study performed it.
-        if cls._has_explicit_current_qualitative_study_anchor(value):
-            return True
         if cls._has_external_literature_marker(value):
             return False
         if re.search(r"(?i)\b(?:prior|previous|earlier|existing)\s+(?:research|stud(?:y|ies)|literature)\b", value):
@@ -5406,9 +5155,6 @@ INPUT:
             "ethics": [str(x or "").strip().casefold() for x in (facts.get("ethics_procedures") or []) if str(x or "").strip()],
             "reflexivity": [str(x or "").strip().casefold() for x in (facts.get("reflexivity_procedures") or []) if str(x or "").strip()],
         }
-        authorized_design = cls._normalize_qualitative_analytic_design_identity(
-            facts.get("analytic_design") or facts.get("analysis_method")
-        )
         procedure_patterns = {
             "trustworthiness": r"\bmember\s+check(?:ing|s)?\b|\bpeer\s+debrief(?:ing|ed)?\b|\baudit\s+trail\b|\bexternal\s+audit\b",
             "ethics": r"\binformed\s+consent\b|\bethics?\s+approval\b|\binstitutional\s+review\s+board\b|\bIRB\b|\bapproval\s+(?:number|code)\b",
@@ -5445,18 +5191,6 @@ INPUT:
             if not core.strip() or not cls._is_study_owned_qualitative_claim(core, section_name):
                 continue
 
-            observed_design = cls._performed_qualitative_analytic_design_identity(core)
-            if observed_design and authorized_design and observed_design != authorized_design:
-                sig = ("analytic_design", observed_design)
-                if sig not in seen:
-                    seen.add(sig)
-                    violations.append({
-                        "type":"unsupported_qualitative_analytic_design",
-                        "observed":observed_design,
-                        "expected":authorized_design,
-                        "preview":core.strip()[:180],
-                    })
-
             # Check every explicit procedure mention; do not stop after the first.
             for category, pattern in procedure_patterns.items():
                 for match in re.finditer(pattern, core, flags=re.I):
@@ -5519,9 +5253,6 @@ INPUT:
             "ethics": [str(x or "").strip() for x in (facts.get("ethics_procedures") or []) if str(x or "").strip()],
             "reflexivity": [str(x or "").strip() for x in (facts.get("reflexivity_procedures") or []) if str(x or "").strip()],
         }
-        authorized_design = cls._normalize_qualitative_analytic_design_identity(
-            facts.get("analytic_design") or facts.get("analysis_method")
-        )
         allowed_by_actor = {}
         canonical_relations = []
         canonical_by_actor = {}
@@ -5556,16 +5287,6 @@ INPUT:
             lower = core.casefold()
             if not core.strip() or not cls._is_study_owned_qualitative_claim(core, section_name):
                 continue
-            observed_design = cls._performed_qualitative_analytic_design_identity(core)
-            if observed_design and authorized_design and observed_design != authorized_design:
-                replacements.append((start, end, ""))
-                diagnostics.append({
-                    "type":"unsupported_qualitative_analytic_design_removed",
-                    "observed":observed_design,
-                    "expected":authorized_design,
-                    "preview":core.strip()[:180],
-                })
-                continue
             removed = False
             for category, pattern in procedure_patterns.items():
                 procedure_match = re.search(pattern, core, flags=re.I)
@@ -6349,6 +6070,199 @@ Return exactly:
         )
         return revised
 
+    @staticmethod
+    def _high_confidence_noncanonical_citation_spans(text):
+        """Return high-confidence direct author-year citation forms outside canonical parentheses.
+
+        These forms are never rebound heuristically.  They are surfaced as typed
+        section violations so the owning writer can retry with its frozen CITE
+        capability or narrow the unsupported claim.  Patterns intentionally avoid
+        bare years and tokens containing digits to reduce false positives for
+        model/version/statistical expressions.
+        """
+        value = str(text or '')
+        surname = r"[A-ZÀ-ÖØ-Þ][A-Za-zÀ-ÖØ-öø-ÿ'’.-]{1,39}"
+        conjunction = r"(?:&|and|AND)"
+        et_al = r"(?:et|Et|ET)\s+(?:al|Al|AL)\."
+        author_group = rf"(?:{surname}(?:\s+{conjunction}\s+{surname})?|{surname}\s+{et_al})"
+        year = r"(?:19|20)\d{2}[a-z]?"
+        patterns = (
+            re.compile(rf"\[(?P<cite>{author_group}\s*,\s*{year})\]"),
+            re.compile(rf"(?<![\w\[])\b(?P<cite>{author_group})\s*\(\s*(?P<year>{year})\s*\)"),
+        )
+        spans = []
+        for pattern in patterns:
+            for match in pattern.finditer(value):
+                candidate = match.group(0).strip()
+                # Canonical SmartPaper markers are handled separately and always
+                # contain the author and year inside the same parentheses.
+                if candidate.startswith('(') and candidate.endswith(')'):
+                    continue
+                spans.append((match.start(), match.end(), candidate))
+        # Stable de-duplication for overlapping patterns.
+        unique = []
+        seen = set()
+        for row in sorted(spans, key=lambda item: (item[0], item[1], item[2])):
+            key = (row[0], row[1], row[2])
+            if key not in seen:
+                seen.add(key)
+                unique.append(row)
+        return unique
+
+    def _normalize_pre_resolved_citation_authority(self, text, *, allowed_tags=None, section_name=''):
+        """Normalize only exact, authoritative pre-resolved markers back to CITE tags.
+
+        The Section Writer is supposed to emit frozen ``[CITE-XX]`` capabilities.
+        A model-emitted reader-facing author-year marker therefore has no provenance
+        until it is matched exactly against the *compiled* citation authority.
+
+        Safe normalization requires all of the following:
+        - exact marker equality after ``compile_citation_identities()``;
+        - authoritative reference metadata;
+        - one distinct source identity for the visible marker;
+        - at least one matching tag inside this manuscript unit's whitelist.
+
+        Anything else becomes a typed violation.  No fuzzy author/year matching,
+        metadata synthesis, or silent citation deletion is permitted here.
+        """
+        value = str(text or '')
+        allowed = None if allowed_tags is None else {
+            str(tag or '').strip() for tag in allowed_tags if str(tag or '').strip()
+        }
+        self.compile_citation_identities()
+
+        authoritative = {}
+        visible_any = {}
+        for tag, metadata in (self.citation_map or {}).items():
+            meta = metadata or {}
+            marker = str(meta.get('in_text') or '').strip()
+            if not marker:
+                continue
+            visible_any.setdefault(marker, []).append(str(tag))
+            if str(meta.get('ref') or '').strip():
+                authoritative.setdefault(marker, []).append(str(tag))
+
+        violations = []
+        replacements = []
+        for marker in list(dict.fromkeys(self._resolved_parenthetical_citations(value))):
+            tags = list(authoritative.get(marker) or [])
+            if not tags:
+                visible_tags = sorted(visible_any.get(marker) or [], key=self._citation_tag_number)
+                violations.append({
+                    'type': 'citation_resolved_marker_without_authority',
+                    'citation': marker,
+                    'tag': visible_tags[0] if len(visible_tags) == 1 else '',
+                    'sentence': marker,
+                    'reason': 'A reader-facing author-year citation was emitted without authoritative reference metadata.',
+                    'repair_instruction': (
+                        'Do not emit author-year citations directly. Use only an authorized CITE tag from this manuscript unit '
+                        'when its evidence directly supports the claim, or narrow/rewrite the claim so no unsupported citation is needed.'
+                    ),
+                })
+                continue
+
+            distinct_sources = {
+                self._citation_source_identity(self.citation_map.get(tag) or {})
+                for tag in tags
+            }
+            if len(distinct_sources) != 1:
+                violations.append({
+                    'type': 'citation_resolved_marker_ambiguous',
+                    'citation': marker,
+                    'sentence': marker,
+                    'reason': 'The reader-facing author-year marker maps to multiple distinct authoritative sources.',
+                    'repair_instruction': (
+                        'Do not guess which source was intended. Rewrite this claim using exactly one authorized CITE tag '
+                        'from the frozen evidence plan, or remove/narrow the unsupported citation-dependent claim.'
+                    ),
+                })
+                continue
+
+            eligible = tags if allowed is None else [tag for tag in tags if tag in allowed]
+            if not eligible:
+                violations.append({
+                    'type': 'citation_resolved_marker_outside_frozen_section_whitelist',
+                    'citation': marker,
+                    'tag': sorted(tags, key=self._citation_tag_number)[0] if tags else '',
+                    'sentence': marker,
+                    'reason': 'The citation is authoritative globally but is not authorized for this manuscript unit.',
+                    'repair_instruction': (
+                        'Use only citation tags frozen for this manuscript unit. If no eligible source directly supports the claim, '
+                        'narrow or rewrite the claim without external citation support.'
+                    ),
+                })
+                continue
+
+            selected = sorted(eligible, key=self._citation_tag_number)[0]
+            replacements.append((marker, selected))
+            self._append_citation_audit(
+                section_name, 'normalize_exact_pre_resolved_marker_to_authoritative_tag',
+                marker, selected, 'Exact compiled marker + authoritative reference + current-unit whitelist.'
+            )
+
+        # Reject other high-confidence author-year forms rather than attempting a
+        # lossy conversion to canonical identity.  The bounded Section retry owns
+        # prose repair and claim narrowing.
+        for _start, _end, citation in self._high_confidence_noncanonical_citation_spans(value):
+            violations.append({
+                'type': 'citation_noncanonical_resolved_marker',
+                'citation': citation,
+                'sentence': citation,
+                'reason': 'A direct author-year citation form bypasses the frozen CITE-tag authority path.',
+                'repair_instruction': (
+                    'Do not emit narrative or bracketed author-year citations directly. Use one authorized CITE tag '
+                    'from this manuscript unit when directly supported, or rewrite the claim without that citation.'
+                ),
+            })
+
+        for marker, tag in replacements:
+            value = value.replace(marker, tag)
+        return value, violations
+
+    def _quarantined_unit_citation_violations(self, text):
+        """Return typed citations found in a manuscript unit that forbids literature.
+
+        Conclusion/Appendix-style units are explicitly writer-quarantined from
+        external literature.  P3-A4 therefore treats any high-confidence raw or
+        reader-facing citation form as a retry-owned scope violation instead of
+        silently deleting the marker and leaving the surrounding claim intact.
+        Bare years and ordinary numeric/model-version expressions are excluded by
+        the same conservative scanners used by the normal citation path.
+        """
+        value = str(text or '')
+        found = []
+        seen = set()
+
+        for raw in re.findall(r'\[?CITE-\d+\]?', value, flags=re.I):
+            token = raw.upper()
+            key = ('raw', token)
+            if key not in seen:
+                seen.add(key)
+                found.append(token)
+
+        for marker in self._resolved_parenthetical_citations(value):
+            key = ('resolved', marker)
+            if key not in seen:
+                seen.add(key)
+                found.append(marker)
+
+        for _start, _end, citation in self._high_confidence_noncanonical_citation_spans(value):
+            key = ('noncanonical', citation)
+            if key not in seen:
+                seen.add(key)
+                found.append(citation)
+
+        return [{
+            'type': 'citation_forbidden_in_quarantined_unit',
+            'citation': citation,
+            'sentence': citation,
+            'reason': 'This manuscript unit forbids external literature citations, but a citation form was emitted.',
+            'repair_instruction': (
+                'Rewrite this unit without external literature citations. Preserve only documented study evidence, '
+                'study-owned interpretation, and the section role; do not replace the citation with another source.'
+            ),
+        } for citation in found]
+
     def audit_and_resolve_citations(
         self, text, section_name='', section_id='', quarantine=False, allowed_tags=None
     ):
@@ -6358,12 +6272,41 @@ Return exactly:
         )
         value = self.strip_ineligible_current_study_citations(str(text or ''), section_name)
         if quarantine:
-            return self.resolve_citations(value, quarantine=True)
+            quarantine_rows = self._quarantined_unit_citation_violations(value)
+            if ownership_rows or quarantine_rows:
+                # Do not silently delete a citation-dependent claim. The owning
+                # Section retry/adaptive-repair path receives a typed violation
+                # and must rewrite the unit without external literature.
+                self._last_citation_semantic_violations = (
+                    list(ownership_rows) + list(quarantine_rows)
+                )
+                for row in quarantine_rows:
+                    self._append_citation_audit(
+                        section_name, 'reject_citation_in_quarantined_unit',
+                        str(row.get('citation') or ''), '', str(row.get('reason') or '')
+                    )
+                return value
+            return value
+
+        # P3-A4: collect deterministic citation-authority violations separately.
+        # semantic_audit_claim_specific_citations() owns its own last-violation
+        # buffer, so these rows are merged back only after that audit completes.
+        authority_rows = []
         unknown = sorted({f"[{tag.upper()}]" for tag in re.findall(r'CITE-\d+', value, re.I)} - set(self.citation_map))
         if unknown:
             for tag in unknown:
+                authority_rows.append({
+                    'type': 'citation_unknown_tag',
+                    'tag': tag,
+                    'sentence': tag,
+                    'reason': 'The manuscript unit emitted a citation capability that does not exist in authoritative citation_map.',
+                    'repair_instruction': (
+                        'Use only citation tags frozen for this manuscript unit, or narrow/rewrite the claim so no unsupported citation is needed.'
+                    ),
+                })
                 value = re.sub(rf'\s*\[?{re.escape(tag.strip("[]"))}\]?\s*', ' ', value, flags=re.I)
-            self._append_citation_audit(section_name, 'remove_unknown_tags', text, value, ','.join(unknown))
+            self._append_citation_audit(section_name, 'reject_unknown_tags_for_section_retry', text, value, ','.join(unknown))
+
         allowed = None if allowed_tags is None else {
             str(tag or '').strip() for tag in allowed_tags if str(tag or '').strip()
         }
@@ -6383,6 +6326,16 @@ Return exactly:
                 self._append_citation_audit(
                     section_name, 'reject_outside_frozen_section_whitelist', text, value, tag
                 )
+
+        # P3-A4 closes the resolved-marker bypass before section sealing.  Exact,
+        # unambiguous, current-unit-authorized markers are normalized back to raw
+        # capabilities so the existing claim-specific semantic audit still owns
+        # claim/evidence adjudication.  All other forms become typed violations.
+        value, resolved_authority_rows = self._normalize_pre_resolved_citation_authority(
+            value, allowed_tags=allowed, section_name=section_name
+        )
+        authority_rows.extend(resolved_authority_rows)
+
         has_citations = bool(re.search(r'CITE-\d+', value, re.I))
         violations = self.find_multi_citation_violations(value)
         if has_citations and self.citation_semantic_audit.get('enabled', True):
@@ -6413,12 +6366,21 @@ Return exactly:
                 value = self._local_citation_recovery(value, section_name, retain_lexical=False)
         elif violations:
             value = self._local_citation_recovery(value, section_name)
-        if ownership_rows or outside_rows:
+
+        pre_semantic_rows = list(ownership_rows) + list(outside_rows) + list(authority_rows)
+        if pre_semantic_rows:
             self._last_citation_semantic_violations = (
-                list(ownership_rows)
-                + list(self._last_citation_semantic_violations or [])
-                + list(outside_rows)
+                pre_semantic_rows + list(self._last_citation_semantic_violations or [])
             )
+
+        # P3-A4 reliability rule: once deterministic identity/scope authority has
+        # already rejected this attempt, return it to the owning bounded Section
+        # retry instead of letting the legacy multi-source hard exception preempt
+        # typed recovery.  The rejected bytes can never be sealed because runner
+        # treats these rows as a hard pre-seal invariant.
+        if outside_rows or authority_rows:
+            return value
+
         if self.find_multi_citation_violations(value):
             value = self._local_citation_recovery(value, section_name)
         if self.find_multi_citation_violations(value):