P3-A4 resilient citation repair

core/generation_runner.py

@@ -1902,6 +1902,21 @@ class GenerationRunner:
             rows.append(' | '.join(parts))
         return '\n'.join(f'- {row}' for row in rows)
 
+    @staticmethod
+    def _section_forbids_external_citations(section_title: str = "", section_id: str = "") -> bool:
+        """One authority for manuscript units that must remain citation-free.
+
+        P3-A4 closes a historic split where Results/Findings were declared
+        citation-free in the Writer prompt but did not enter the same quarantine
+        validation path as Conclusion/Appendix-style units.
+        """
+        title = str(section_title or '').casefold()
+        sid = str(section_id or '').strip().casefold()
+        return (
+            any(word in title for word in ("abstract", "keyword", "conclusion", "references", "appendix"))
+            or sid in {"results", "findings"}
+        )
+
     @staticmethod
     def _section_violation_severity(row, *, section_name: str = "", mode: str = "", exhausted: bool = False) -> str:
         """Map one validation issue to REPAIRABLE/FATAL/ADVISORY.
@@ -1921,6 +1936,16 @@ class GenerationRunner:
                 return "ADVISORY"
             return "REPAIRABLE"
 
+        # Delivery-first resilience: uncertain noncanonical citation-like forms
+        # get a DeepSeek repair opportunity, but they are not allowed to terminate
+        # an otherwise complete manuscript after bounded repair.  Definite
+        # authority/scope violations remain fail-closed below.
+        if kind in {
+            "citation_noncanonical_unverified_marker",
+            "citation_suspected_in_quarantined_unit",
+        }:
+            return "ADVISORY"
+
         always_fatal = {
             "causal_language_violation",
             "sample_size_mismatch", "sample_count_mismatch", "scale_points_mismatch",
@@ -1929,7 +1954,12 @@ class GenerationRunner:
             "unsupported_study_procedure", "enumeration_contract_invalid",
             "hypothesis_contract_missing_id", "hypothesis_contract_ordinal_alias",
             "hypothesis_contract_pairing_mismatch", "hypothesis_contract_unknown_id",
-            "citation_outside_frozen_section_whitelist", "authorial_ai_meta_trace",
+            "citation_outside_frozen_section_whitelist",
+            "citation_resolved_marker_outside_frozen_section_whitelist",
+            "citation_unknown_tag", "citation_resolved_marker_without_authority",
+            "citation_resolved_marker_ambiguous", "citation_noncanonical_resolved_marker",
+            "citation_forbidden_in_quarantined_unit",
+            "authorial_ai_meta_trace",
             "study_design_scope_mismatch", "unsupported_focal_construct", "unsupported_relation_claim",
             "unsupported_moderation_claim", "unsupported_mediation_claim", "unsupported_temporal_scope",
             "unsupported_design_tradition", "unsupported_unit_of_analysis", "unsupported_data_source_claim",
@@ -1972,7 +2002,21 @@ class GenerationRunner:
         item = dict(row or {}) if isinstance(row, dict) else {"type": "unknown_validation_issue"}
         kind = str(item.get("type") or "unknown_validation_issue")
         groups = {
-            "citation_scope": {"citation_outside_frozen_section_whitelist"},
+            "citation_scope": {
+                "citation_outside_frozen_section_whitelist",
+                "citation_resolved_marker_outside_frozen_section_whitelist",
+            },
+            "citation_authority": {
+                "citation_unknown_tag",
+                "citation_resolved_marker_without_authority",
+                "citation_resolved_marker_ambiguous",
+                "citation_noncanonical_resolved_marker",
+                "citation_forbidden_in_quarantined_unit",
+            },
+            "citation_style": {
+                "citation_noncanonical_unverified_marker",
+                "citation_suspected_in_quarantined_unit",
+            },
             "citation_ownership": {"citation_evidence_ownership_mismatch"},
             "citation_support": {"citation_unsupported_claim", "citation_support_unverified"},
             "citation_compound": {"citation_claim_split_required"},
@@ -2110,6 +2154,20 @@ class GenerationRunner:
                 blocking.append(row)
         return blocking, advisory
 
+    @classmethod
+    def _persistent_citation_authority_rows(cls, rows) -> list[dict]:
+        """Return citation identity/scope defects that may never cross a Section seal.
+
+        P3-A4 intentionally keeps this narrower than general validation failure.
+        Citation support/style advisories retain their existing adaptive-completion
+        behavior; only source identity and unit-scope authority stay fail-closed.
+        """
+        return [
+            row for row in list(rows or [])
+            if isinstance(row, dict)
+            and cls._section_violation_lineage(row) in {"citation_scope", "citation_authority"}
+        ]
+
     @staticmethod
     def _record_section_advisories(quality_notes: list, rows, *, section_name: str = "") -> None:
         """Append reader-facing, nonfatal notes without exposing control vocabulary."""
@@ -2177,6 +2235,7 @@ class GenerationRunner:
         self, *, agent, quant_engine, text: str, violations, mode: str, section_id: str,
         section_name: str, data_metadata: dict, study_evidence: str, evidence_contract: str,
         hypothesis_ledger: dict, task: dict, quote_invocation_plan: dict, quality_notes: list,
+        allowed_citation_tags=None, quarantine=False,
     ):
         """A2 last-mile semantic repair after lineage retry exhaustion.
 
@@ -2215,10 +2274,22 @@ class GenerationRunner:
             repaired, _enum_rows = agent.repair_and_validate_enumeration_contract(
                 repaired, section_name=section_name
             )
+            # P3-A4: every model-owned last-mile rewrite must re-enter the same
+            # citation authority + claim-specific semantic Gate before it can be
+            # considered sealable.  Otherwise an unrelated adaptive repair could
+            # reintroduce a pre-resolved/unknown/wrong-scope citation after the
+            # earlier Section citation audit had already passed.
+            repaired = agent.audit_and_resolve_citations(
+                repaired, section_name=section_name, section_id=section_id,
+                quarantine=quarantine, allowed_tags=allowed_citation_tags,
+            )
+            adaptive_citation_rows = copy.deepcopy(
+                list(getattr(agent, "_last_citation_semantic_violations", []) or [])
+            )
             rows = self._section_validation_bundle(
                 mode, agent, quant_engine, repaired, data_metadata, section_id=section_id,
                 section_name=section_name, evidence_contract=evidence_contract, include_critic=False,
-                hypothesis_ledger=hypothesis_ledger,
+                citation_violations=adaptive_citation_rows, hypothesis_ledger=hypothesis_ledger,
             )
             blocking, advisory = self._partition_section_violations(
                 rows, section_name=section_name, mode=mode, exhausted=True
@@ -5600,11 +5671,8 @@ class GenerationRunner:
             )
             sec_title = sec["section"]
             subsection = sec.get("subsection")
-            is_quarantined = any(
-                word in sec_title.lower()
-                for word in ["abstract", "keyword", "conclusion", "references", "appendix"]
-            )
             section_id = str(sec.get("section_id") or "")
+            is_quarantined = self._section_forbids_external_citations(sec_title, section_id)
             section_logical_node = self._section_logical_node(
                 mode, section_id, sec.get("subsection_id")
             )
@@ -5630,7 +5698,7 @@ class GenerationRunner:
             elif section_id == "conclusion":
                 mode_directive = str(APP_CONFIG.get("conclusion_anchor_directive") or "")
 
-            if is_quarantined or section_id in {"results", "findings"}:
+            if is_quarantined:
                 citation_rule = (
                     "NO EXTERNAL LITERATURE CITATIONS ALLOWED IN THIS UNIT. "
                     "Report only documented empirical/qualitative evidence and the interpretation permitted by the section role."
@@ -6097,6 +6165,29 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                         if adaptive_advisory:
                             self._record_section_advisories(quality_notes, adaptive_advisory, section_name=display_name)
                         if adaptive_blocking:
+                            # P3-A4: citation identity/scope authority is a hard
+                            # pre-seal invariant.  Adaptive completion may retain
+                            # reader-facing advisories for unrelated quality
+                            # defects, but it may never seal a Section that still
+                            # contains an unauthorized/ambiguous citation identity
+                            # or a citation outside the frozen unit whitelist.
+                            persistent_citation_authority = self._persistent_citation_authority_rows(
+                                adaptive_blocking
+                            )
+                            if persistent_citation_authority:
+                                self._checkpoint_section_retry_exhausted(
+                                    state, section_id=section_id, section_name=display_name,
+                                    violations=persistent_citation_authority, pct=pct, attempts=section_attempt,
+                                )
+                                raise RuntimeError(
+                                    "citation_authority_repair_exhausted:"
+                                    + str(display_name)
+                                    + ":"
+                                    + ",".join(sorted({
+                                        str((row or {}).get("type") or "citation_authority")
+                                        for row in persistent_citation_authority if isinstance(row, dict)
+                                    }))
+                                )
                             self._record_adaptive_completion_residual(
                                 quality_notes, adaptive_blocking, section_name=display_name, stage="phase1_semantic_repair"
                             )
@@ -6276,7 +6367,26 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                             study_evidence=writer_scoped_data_md, evidence_contract=evidence_packet,
                             hypothesis_ledger=hypothesis_ledger, task=sec,
                             quote_invocation_plan=quote_invocation_plan, quality_notes=quality_notes,
+                            allowed_citation_tags=allowed_section_citation_tags,
+                            quarantine=is_quarantined,
                         )
+                        persistent_citation_authority = self._persistent_citation_authority_rows(
+                            residual_rows
+                        )
+                        if persistent_citation_authority:
+                            self._checkpoint_section_retry_exhausted(
+                                state, section_id=section_id, section_name=display_name,
+                                violations=persistent_citation_authority, pct=pct, attempts=section_attempt,
+                            )
+                            raise RuntimeError(
+                                "citation_authority_repair_exhausted:"
+                                + str(display_name)
+                                + ":"
+                                + ",".join(sorted({
+                                    str((row or {}).get("type") or "citation_authority")
+                                    for row in persistent_citation_authority if isinstance(row, dict)
+                                }))
+                            )
                         logging.warning(
                             "Adaptive completion accepted section | unit=%s residual=%s",
                             display_name, len(residual_rows),

core/pipeline_agent.py

@@ -6349,6 +6349,314 @@ Return exactly:
         )
         return revised
 
+    @staticmethod
+    def _high_confidence_noncanonical_citation_spans(text):
+        """Return only *suspected* noncanonical citation forms worth one soft repair pass.
+
+        P3-A4-R4 delivery-first rule:
+        - exact aliases of authoritative citation_map entries are handled separately
+          and deterministically by ``_normalize_pre_resolved_citation_authority``;
+        - unmatched narrative ``Name (year)`` forms are surfaced only when nearby
+          attribution language makes citation intent high-confidence;
+        - bracketed ``[Name, year]`` forms remain suspicious, but are treated as
+          repairable/advisory rather than a terminal authority failure when they do
+          not map to citation_map.
+
+        This deliberately avoids treating ordinary product/method/version prose such
+        as ``Stata (2024)``, ``Mplus (2024)``, ``SmartPLS (2024)``,
+        ``DeepSeek (2024)``, or ``SEM (2024)`` as citations merely because the
+        surface form resembles an author-year citation.
+        """
+        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]?"
+        bracketed = re.compile(rf"\[(?P<cite>{author_group}\s*,\s*{year})\]")
+        narrative = re.compile(rf"(?<![\w\[])(?P<cite>{author_group})\s*\(\s*(?P<year>{year})\s*\)")
+        attribution_after = re.compile(
+            r"^\s*(?:,\s*)?(?:argues?|reports?|finds?|found|suggests?|shows?|showed|"
+            r"demonstrates?|demonstrated|proposes?|notes?|observes?|states?|concludes?|"
+            r"documents?|identifies?|reported|suggested|proposed|noted|observed|stated|"
+            r"concluded|documented|identified)\b",
+            re.I,
+        )
+        attribution_before = re.compile(
+            r"(?:according\s+to|as\s+(?:reported|noted|shown|argued|observed|stated)\s+by)\s*$",
+            re.I,
+        )
+
+        spans = []
+        for match in bracketed.finditer(value):
+            spans.append((match.start(), match.end(), match.group(0).strip()))
+        for match in narrative.finditer(value):
+            before = value[max(0, match.start() - 48):match.start()]
+            after = value[match.end():min(len(value), match.end() + 64)]
+            if attribution_after.search(after) or attribution_before.search(before):
+                spans.append((match.start(), match.end(), match.group(0).strip()))
+
+        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 _authoritative_noncanonical_citation_aliases(self):
+        """Map exact narrative/bracketed aliases to authoritative CITE tags.
+
+        The alias set is derived only from compiled citation_map markers carrying
+        authoritative reference metadata.  No free-form author/year guessing is
+        performed.
+        """
+        self.compile_citation_identities()
+        aliases = {}
+        for tag, metadata in (self.citation_map or {}).items():
+            meta = metadata or {}
+            marker = str(meta.get('in_text') or '').strip()
+            if not marker or not str(meta.get('ref') or '').strip():
+                continue
+            match = re.fullmatch(
+                r"\((?P<authors>[^()]+?)\s*,\s*(?P<year>(?:19|20)\d{2}[a-z]?)\)",
+                marker,
+                flags=re.I,
+            )
+            if not match:
+                continue
+            authors = match.group('authors').strip()
+            year = match.group('year').strip()
+            for alias in (f"{authors} ({year})", f"[{authors}, {year}]"):
+                aliases.setdefault(alias, []).append(str(tag))
+        return aliases
+    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.'
+            )
+
+        # Exact narrative/bracketed aliases of authoritative citation_map entries
+        # are a formatting defect, not a reason to kill generation.  Normalize
+        # them deterministically back to the owning CITE tag and let the existing
+        # claim-specific semantic audit validate the claim/source relationship.
+        alias_authority = self._authoritative_noncanonical_citation_aliases()
+        authoritative_aliases_seen = set()
+        for alias, alias_tags in alias_authority.items():
+            if alias not in value:
+                continue
+            tags = list(alias_tags or [])
+            authoritative_aliases_seen.add(alias)
+            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': alias,
+                    'sentence': alias,
+                    'reason': 'The noncanonical 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': alias,
+                    'tag': sorted(tags, key=self._citation_tag_number)[0] if tags else '',
+                    'sentence': alias,
+                    '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((alias, selected))
+            self._append_citation_audit(
+                section_name, 'normalize_exact_noncanonical_marker_to_authoritative_tag',
+                alias, selected, 'Exact authoritative alias + current-unit whitelist.'
+            )
+
+        # Unmatched suspicious author-year forms get one DeepSeek-owned repair
+        # opportunity but are explicitly *soft* after bounded repair.  This keeps
+        # the system resilient when a product/method/model label happens to look
+        # like a surname + year while still asking the model to clean genuinely
+        # citation-like prose when attribution context is strong.
+        for _start, _end, citation in self._high_confidence_noncanonical_citation_spans(value):
+            if citation in authoritative_aliases_seen:
+                continue
+            violations.append({
+                'type': 'citation_noncanonical_unverified_marker',
+                'citation': citation,
+                'sentence': citation,
+                'reason': 'A citation-like author-year form could not be bound exactly to frozen citation authority.',
+                'repair_instruction': (
+                    'If this is intended as a literature citation, replace it with one authorized CITE tag whose evidence directly '
+                    'supports the claim. If it is a product, method, model, software, or ordinary year label, keep the factual prose '
+                    'but rewrite it so it cannot be mistaken for an external author-year citation.'
+                ),
+            })
+
+        for marker, tag in replacements:
+            value = value.replace(marker, tag)
+        return value, violations
+    def _quarantined_unit_citation_violations(self, text):
+        """Return definite and suspected citation forms in citation-free units.
+
+        Delivery-first classification:
+        - raw CITE tags, canonical resolved parenthetical markers, and exact
+          authoritative narrative/bracketed aliases are definite literature
+          citations and remain fail-closed after bounded repair;
+        - unmatched citation-like narrative/bracketed forms are only suspected.
+          They receive a DeepSeek repair opportunity but may degrade to an
+          advisory instead of terminating an otherwise complete manuscript.
+        """
+        value = str(text or '')
+        rows = []
+        seen = set()
+
+        def add(kind, citation, reason, instruction):
+            key = (kind, str(citation or ''))
+            if key in seen:
+                return
+            seen.add(key)
+            rows.append({
+                'type': kind,
+                'citation': str(citation or ''),
+                'sentence': str(citation or ''),
+                'reason': reason,
+                'repair_instruction': instruction,
+            })
+
+        definite_reason = 'This manuscript unit forbids external literature citations, but a definite citation form was emitted.'
+        definite_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 raw in re.findall(r'\[?CITE-\d+\]?', value, flags=re.I):
+            add('citation_forbidden_in_quarantined_unit', raw.upper(), definite_reason, definite_instruction)
+
+        for marker in self._resolved_parenthetical_citations(value):
+            add('citation_forbidden_in_quarantined_unit', marker, definite_reason, definite_instruction)
+
+        authoritative_aliases = self._authoritative_noncanonical_citation_aliases()
+        authoritative_seen = set()
+        for alias in authoritative_aliases:
+            if alias in value:
+                authoritative_seen.add(alias)
+                add('citation_forbidden_in_quarantined_unit', alias, definite_reason, definite_instruction)
+
+        for _start, _end, citation in self._high_confidence_noncanonical_citation_spans(value):
+            if citation in authoritative_seen:
+                continue
+            add(
+                'citation_suspected_in_quarantined_unit',
+                citation,
+                'A citation-like author-year form appeared in a citation-free manuscript unit but was not bound to frozen authority.',
+                (
+                    'Use DeepSeek to rewrite the passage without external literature if this is a citation. If it is instead a '
+                    'product, method, model, software, or ordinary year label, preserve the factual content and clarify the wording. '
+                    'Do not terminate manuscript generation solely because this soft ambiguity remains after bounded repair.'
+                ),
+            )
+
+        return rows
     def audit_and_resolve_citations(
         self, text, section_name='', section_id='', quarantine=False, allowed_tags=None
     ):
@@ -6358,12 +6666,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 +6720,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 +6760,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):