P3-A4 minimal citation authority repair on live main

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.
@@ -1929,7 +1944,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 +1992,17 @@ 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_ownership": {"citation_evidence_ownership_mismatch"},
             "citation_support": {"citation_unsupported_claim", "citation_support_unverified"},
             "citation_compound": {"citation_claim_split_required"},
@@ -2110,6 +2140,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 +2221,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 +2260,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 +5657,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 +5684,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 +6151,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 +6353,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,196 @@ 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 +6548,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 +6602,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 +6642,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):