test(disclosure): cover venue-track contract (#615) (#622)

This commit is contained in:
Edward Cheng-I Wu
2026-08-02 00:39:41 +08:00
committed by GitHub
parent eba1d3842d
commit c804e945f6
4 changed files with 1597 additions and 0 deletions
+1
View File
@@ -6,6 +6,7 @@ All notable changes to this project will be documented in this file.
### Added
- **Deterministic venue-track contract coverage (#615).** A test-only, offline contract oracle and 43-test pytest suite now exercise the accepted 15-target disclosure behaviour without adding a runtime schema, live policy lookup, or general submission engine. Coverage includes the existing seven-field/order checker plus deletion and ordering mutations; selector/user-surface drift; complete intake with citation checking and preserved `OTHER / UNCLASSIFIED` use; unknown, incompatible, prohibited, and uncurated-policy halts; all four disclosure outcomes; anchor-track isolation; distinct purpose-specific multi-placement blocks; conditional `NOT_APPLICABLE` children; and strict `tool × task/run/artifact` fact binding with a cross-record borrowing mutation. Field fixtures pin the final Chinese Nursing, JAMA, Nature containment, International Eye Science, Frontiers, Lancet/Elsevier, Western hard-prohibition, and ICMJE-alongside contracts, including conceptual-versus-data figures, non-LLM and LLM study rights/prompt branches, generated proportion and clinical de-identification, graphical abstracts, primary-versus-research-method images, protected subjects, cover-art permissions, and cover-art-only versus mixed outcomes. The suite is registered in the unified pytest manifest and makes no policy-content change.
- **Frontiers disclosure action-carrier closeout + evidence/audit provenance (#619).** Frontiers factual-accuracy, plagiarism-free, and conditionally applicable figure accuracy-to-data checks move out of the Phase-2 render ledger into a labelled Phase-5 pre-submission checklist: created versus edited written/visual use and data-representing figures select the applicable actions, while false or unknown action state remains visibly outstanding without halting an otherwise complete disclosure or producing a false confirmation. AI authorship and editor/reviewer external upload remain separate hard prohibitions. The venue evidence row now reflects that distinction. JAMA retains its live current A Piece of My Mind and Poetry drafting prohibitions while explicitly recording that both clauses were verified live on 2026-08-01 and are absent from the cited 2026-07-01 exact-URL snapshot. The reusable external-contribution audit prompt, previously present only on the maintainer bakeoff branch, is brought onto main-line history with the exact 2026-08-01 PR #599 heads, model/effort, finding counts, first-party recheck, and closure outcome. A focused fail-closed checker and mutation suite cover these three closeout surfaces without adding a disclosure schema or a general submission-policy engine.
- **E4 promotion integrity and resume-from-bundle recovery (#616).** The E4 dispatch harness now freezes a prompt-invisible `recovery-state.json` before record installation: invocation context plus the dispatch/retry/abort event ledger and a path/type/SHA-256 manifest, with no operator-supplied closed status fields. `resume_e4_record.py` re-emits through the original atomic record builder after a post-dispatch emission failure without constructing a model transport, retrying a call, or re-running a checker; it supports only the documented rolled-back `<work-dir>/bundle` and canonical installed-raw states and refuses changed, inserted, missing, redirected, ambiguous, inconsistent, or already-consumed evidence. `check_e4_promotion.py` independently verifies a manually promoted record/raw pair against its work-directory original — canonical scored/blocked layout, complete relative path/type set, SHA-256 identity for every file, and safe resolution of `raw_bundle` plus all `*_location` fields — without reinterpreting output or verdicts. Acceptance and mutation tests are wired into the unified pytest manifest; reviewer prompts, contracts, fixtures, checker verdicts, dispatch ordering, and the frozen 2026-07-27 `NOT COMPUTABLE` cohort remain unchanged.
- **Reviewer protocol text single-source, public role naming, and lightweight calibration tier (#611).** The five sprint-reviewer Phase 1/2 prompt pairs and the synthesizer protocol now have one marked canonical source (`reviewer_sprint_prompt_source.md`) while every dispatched section remains fully inline for `--bare --tools ""`; a byte-exact render check plus explicit SHA-256 re-pin lock intentional edits without changing prompt semantics or dispatch behavior. The former public EIC seat is displayed consistently as **Journal-Fit Reviewer**, with `eic_agent`, `contract_role: eic`, serialized `EIC`/`EIC-W<n>` source IDs, frozen evidence, and real-journal Editor-in-Chief references preserved as compatibility boundaries; those tokens do not select Stage 3' agent files—the synthesizer emits first-round decisions, while contract-governed re-review uses three dedicated calls and a checker-derived outcome. Calibration keeps the existing panel engine and default 5-20-paper full tier (5 runs, 3-run budget override) while adding an explicit opt-in directional tier of exactly three gold papers (Minor, Major, and one Accept/Reject extreme), one fresh panel each, gold-label isolation, exact/raw Minor-Major boundary reporting, raw #215 severity-risk counts, and a hard prohibition on error-rate/profile claims. Its cross-model branch is a canonical non-sprint single-call Reviewer 2 transport with attempt-atomic fallback, so only a homogeneous substrate plan can feed metrics or disclosures; per-dimension score error remains `NOT COMPUTABLE` without adjudicated dimension-level gold scores, and every partially annotated dimension reports its own `annotated_n/N` plus missingness instead of implying gold-set-wide coverage. Three fail-closed lints and their mutation suites are wired into spec consistency and the unified pytest manifest.
+4
View File
@@ -371,3 +371,7 @@ path = "scripts/test_check_calibration_tiers.py"
[[pytest]]
id = "619-disclosure-closeout"
path = "scripts/test_check_619_disclosure_closeout.py"
[[pytest]]
id = "615-venue-disclosure-contract"
path = "scripts/test_venue_disclosure_contract.py"
+755
View File
@@ -0,0 +1,755 @@
#!/usr/bin/env python3
"""Deterministic venue-track contract coverage for issue #615."""
from __future__ import annotations
import shutil
import subprocess
import sys
from pathlib import Path
import pytest
from venue_disclosure_contract_harness import (
ALL_CATEGORIES,
ALL_OPERATIONS,
ALL_OUTCOMES,
ALL_TARGETS,
Fact,
UseRecord,
evaluate,
known,
surface_sync_errors,
unknown,
)
REPO_ROOT = Path(__file__).resolve().parents[1]
POLICY_CHECKER = REPO_ROOT / "scripts" / "check_venue_disclosure_policies.py"
POLICIES = "academic-paper/references/venue_disclosure_policies.md"
PROTOCOL = "academic-paper/references/disclosure_mode_protocol.md"
def _categories(**updates: str) -> dict[str, str]:
values = {name: "NOT_USED" for name in ALL_CATEGORIES}
values.update(updates)
return values
def _record(
record_id: str = "use-1",
*,
tool: str = "ExampleAI",
task: str = "draft-1",
artifact: str = "manuscript.md",
category: str = "DRAFTING_ASSISTANCE",
operations: tuple[str, ...] = ("SUBSTANTIVELY_DRAFTED",),
targets: tuple[str, ...] = ("METHODS",),
research_use: bool = False,
facts: dict[str, Fact] | None = None,
) -> UseRecord:
bound = {
name: value if value.owner is not None else Fact(value.state, value.value, record_id)
for name, value in (facts or {}).items()
}
return UseRecord(
record_id=record_id,
tool=tool,
task=task,
artifact=artifact,
category=category,
operations=operations,
targets=targets,
research_use=research_use,
facts=bound,
)
def _base_case(
venue: str,
records: tuple[UseRecord, ...] = (),
*,
categories: dict[str, str] | None = None,
global_facts: dict[str, Fact] | None = None,
) -> dict[str, object]:
facts = {
"external_use_inventory_confirmed": known("list supplied" if records else "none"),
"ai_listed_or_proposed_as_author": known(False),
}
facts.update(global_facts or {})
return {
"venue": venue,
"categories": categories or _categories(
**({records[0].category: "USED"} if records else {})
),
"records": records,
"global_facts": facts,
}
def _iclrfacts() -> dict[str, Fact]:
return {
"specific_assisted_tasks": known("checked source-to-claim alignment"),
"author_accepts_full_responsibility": known(True),
"affected_content": known("reference list and linked claims"),
}
def _jama_facts(*, llm: bool, included: bool = False, protected_input: bool = False) -> dict[str, Fact]:
facts = {
"author_review_accuracy": known(True),
"author_accepts_content_integrity_responsibility": known(True),
"model_or_tool_version": known("2.1"),
"extension_numbers_applicable": known(False),
"manufacturer": known("Example Labs"),
"dates_of_use": known("2026-07-30"),
"use_description": known("assisted the registered analysis"),
"affected_portions": known("Methods"),
"specific_research_use": known("classified preregistered records"),
"study_uses_llm": known(llm),
"copyright_protected_content_entered": known(protected_input),
"ai_generated_content_included_in_submission": known(included),
}
if llm:
facts.update({
"llm_prompts": known(("classify record", "explain exclusion")),
"llm_prompt_sequence": known((1, 2)),
"llm_prompt_revisions": known("second prompt narrowed the date range"),
})
if protected_input:
facts.update({
"copyright_permission_copy": known("permissions/input-license.pdf"),
"copyright_permission_methods_description": known("licensed corpus under agreement 7"),
})
if included:
facts.update({
"included_content_type": known("supplementary classification table"),
"publication_rights_basis": known("service terms grant publication rights"),
})
return facts
def _ies_facts(*, data: bool, clinical: bool = False) -> dict[str, Fact]:
facts = {
"policy_tool_scope": known("GENAI_OR_AIGC"),
"use_scope": known("OTHER_CONFIRMED_NON_CORE_RESEARCH_STEP"),
"use_scope_basis": known("terminology harmonization after analysis"),
"tool_is_overseas": known(False),
"generated_core_main_text_conclusion_analysis_viewpoint_or_innovation_claim": known(False),
"fabricated_experimental_plan_technical_route_or_citation": known(False),
"replaced_author_in_experimental_design_or_data_validation": known(False),
"fabricated_data_invented_results_or_tampered_conclusions": known(False),
"rewrote_plagiarized_work_to_evade_detection": known(False),
"generated_peer_review_response_grant_contribution_or_integrity_statement": known(False),
"uploaded_secret_research_data_or_unpublished_results_to_public_ai_platform": known(False),
"use_involves_data": known(data),
"aigc_generated_or_tampered_data": known(False),
"aigc_replaced_core_analysis": known(False),
"uploaded_undeidentified_data_to_aigc": known(False),
"uploaded_data_lacking_required_ethics_review_to_aigc": known(False),
"fabricated_data_or_ethics_proof": known(False),
"version": known("2026.7"),
"purpose": known("terminology harmonization"),
"generated_proportion": known("8%"),
}
if data:
facts.update({
"data_types": known(("coded outcome labels",)),
"data_verification_status": known("dual human verification complete"),
"data_involves_clinical_or_case_data": known(clinical),
})
if clinical:
facts["de_identification_measures"] = known("direct identifiers removed before use")
return facts
def _frontiers_facts(*, represents_data: bool | None, created: bool = True) -> dict[str, Fact]:
return {
"policy_tool_scope": known("GENAI_OR_AIGC"),
"version": known("4.2"),
"model": known("Example Vision"),
"source_provider": known("Example Labs"),
"content_operation": known("CREATED" if created else "EDITED"),
"affected_content_kind": known("VISUAL"),
"affected_content": known("Figure 2 conceptual workflow"),
"figure_represents_data": unknown() if represents_data is None else known(represents_data),
}
def _lancet_common() -> dict[str, Fact]:
return {
"ai_replaced_authors_intellectual_contribution": known(False),
"generated_media_duplicates_or_refers_to_protected_subject": known(False),
"visual_accuracy_confirmed": known(True),
"visual_originality_confirmed": known(True),
"based_on_existing_artwork_or_graphics": known(False),
}
def _copy_contract_tree(tmp_path: Path) -> Path:
rels = (
POLICIES,
PROTOCOL,
"academic-paper/SKILL.md",
"commands/ars-disclosure.md",
"academic-paper/references/mode_selection_guide.md",
"README.md",
"README.ja-JP.md",
"README.ko-KR.md",
"README.zh-CN.md",
"README.zh-TW.md",
)
for rel in rels:
dst = tmp_path / rel
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy(REPO_ROOT / rel, dst)
return tmp_path
def _mutate(root: Path, rel: str, old: str, new: str) -> None:
path = root / rel
text = path.read_text(encoding="utf-8")
assert old in text, f"missing mutation anchor: {old!r}"
path.write_text(text.replace(old, new, 1), encoding="utf-8")
# Static database and vocabulary regression coverage.
def test_existing_15_venue_structural_checker_passes() -> None:
result = subprocess.run(
[sys.executable, str(POLICY_CHECKER)],
cwd=REPO_ROOT,
capture_output=True,
text=True,
)
assert result.returncode == 0, result.stderr
@pytest.mark.parametrize("mutation", ("field-deletion", "ordering-drift"))
def test_existing_structural_checker_rejects_mutations(tmp_path: Path, mutation: str) -> None:
policies = tmp_path / "policies.md"
shutil.copy(REPO_ROOT / POLICIES, policies)
text = policies.read_text(encoding="utf-8")
if mutation == "field-deletion":
text = text.replace("| Authorship rule |", "| Removed field |", 1)
else:
acl = text.index("## Venue: ACL")
bmj = text.index("## Venue: BMJ")
chinese = text.index("## Venue: Chinese Nursing", bmj)
first = text[acl:bmj]
second = text[bmj:chinese]
text = text[:acl] + second + first + text[chinese:]
policies.write_text(text, encoding="utf-8")
result = subprocess.run(
[sys.executable, str(POLICY_CHECKER), str(policies)],
capture_output=True,
text=True,
)
assert result.returncode == 1
def test_closed_vocabulary_matches_protocol() -> None:
protocol = (REPO_ROOT / PROTOCOL).read_text(encoding="utf-8")
for token in ALL_OUTCOMES | ALL_OPERATIONS | ALL_TARGETS:
assert f"`{token}`" in protocol
for label in (
"Citation checking",
"Other / unclassified AI use (logged or external)",
):
assert label in protocol
def test_selector_and_user_surfaces_are_in_sync() -> None:
assert surface_sync_errors(REPO_ROOT) == []
def test_selector_surface_drift_is_detected(tmp_path: Path) -> None:
tree = _copy_contract_tree(tmp_path)
_mutate(
tree,
"commands/ars-disclosure.md",
" / journal-level 国际眼科杂志",
"",
)
assert any("ars-disclosure.md" in error for error in surface_sync_errors(tree))
# Intake, dispatch, halt, and four-outcome fixtures.
def test_citation_checking_is_a_first_class_used_category() -> None:
record = _record(
category="CITATION_CHECKING",
operations=("CHECKED_CITATIONS",),
targets=("REFERENCE_OR_CITATION",),
facts=_iclrfacts(),
)
result = evaluate(_base_case("ICLR", (record,), categories=_categories(CITATION_CHECKING="USED")))
assert (result.outcome, result.execution_status) == ("REQUIRED", "READY")
assert result.ledger[record.record_id]["specific_assisted_tasks"].value
def test_unknown_external_inventory_halts_before_categorization() -> None:
case = _base_case("ICLR")
case["global_facts"]["external_use_inventory_confirmed"] = unknown() # type: ignore[index]
result = evaluate(case)
assert (result.outcome, result.execution_status, result.halt_reason) == (
"UNKNOWN", "HALTED", "UNRESOLVED_INPUT"
)
assert "Phase 2a" not in result.phases
def test_used_other_unclassified_is_preserved_and_halted() -> None:
record = _record(
category="OTHER_UNCLASSIFIED",
operations=("OTHER_CONFIRMED",),
targets=("OTHER_CONFIRMED",),
facts={"verbatim_description": known("AI sorted an uncategorized submission object")},
)
result = evaluate(
_base_case("ICLR", (record,), categories=_categories(OTHER_UNCLASSIFIED="USED"))
)
assert result.outcome == "UNKNOWN"
assert result.halt_reason == "UNRESOLVED_INPUT"
assert "AI sorted an uncategorized submission object" in result.diagnostics[0]
def test_unknown_venue_never_falls_back_or_emits_placeholders() -> None:
result = evaluate(_base_case("Imaginary Journal"))
assert (result.outcome, result.execution_status, result.halt_reason) == (
"UNKNOWN", "HALTED", "UNCURATED_POLICY"
)
rendered = " ".join(block.text for block in result.blocks)
assert result.blocks == ()
assert "generic" not in rendered.casefold()
assert "[" not in rendered
def test_all_four_outcomes_are_exercised() -> None:
required = evaluate(
_base_case(
"ICLR",
(_record(facts=_iclrfacts()),),
categories=_categories(DRAFTING_ASSISTANCE="USED"),
)
)
not_required = evaluate(_base_case("ICLR"))
unknown_result = evaluate(_base_case("Unknown Venue"))
cover_facts = _lancet_common() | {
"artifact_class": known("COVER_ART"),
"editor_permission": known(True),
"publisher_permission": known(True),
"cover_art_contains_third_party_material": known(False),
"content_attribution": known("none_applicable"),
}
action_only = evaluate(
_base_case(
"The Lancet",
(_record(category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=cover_facts),),
categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED"),
)
)
assert {required.outcome, not_required.outcome, unknown_result.outcome, action_only.outcome} == ALL_OUTCOMES
assert action_only.blocks == ()
assert "editor permission" in " ".join(action_only.actions).casefold()
def test_unknown_required_fact_and_incompatible_confirmed_fact_halt_before_render() -> None:
missing = _record(facts=_iclrfacts() | {"author_accepts_full_responsibility": unknown()})
incompatible = _record(facts=_iclrfacts() | {"author_accepts_full_responsibility": known(False)})
missing_result = evaluate(_base_case("ICLR", (missing,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
incompatible_result = evaluate(_base_case("ICLR", (incompatible,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert (missing_result.outcome, missing_result.halt_reason, missing_result.blocks) == (
"REQUIRED", "UNRESOLVED_INPUT", ()
)
assert (incompatible_result.outcome, incompatible_result.halt_reason, incompatible_result.blocks) == (
"REQUIRED", "INCOMPATIBLE_FACT", ()
)
def test_policy_anchor_never_enters_venue_phases() -> None:
result = evaluate({"policy_anchor": "icmje", "categories": _categories(), "records": (), "global_facts": {"external_use_inventory_confirmed": known("none")}})
assert result.track == "anchor"
assert all(not phase.startswith("Venue Phase") for phase in result.phases)
assert result.outcome is None
def test_nature_consistent_dual_selector_routes_to_anchor() -> None:
result = evaluate({"venue": "Nature Medicine", "policy_anchor": "nature", "categories": _categories(), "records": (), "global_facts": {"external_use_inventory_confirmed": known("none")}})
assert result.track == "anchor"
assert "Venue Phase 2a" not in result.phases
def test_multi_placement_blocks_are_distinct_and_purpose_specific() -> None:
facts = {
"ai_generated_material_used_as_primary_source": known(False),
"human_review_editing_performed": known(True),
"no_plagiarism_confirmed": known(True),
"technology_description": known("ExampleAI 2.1"),
"produced_content": known("citation verification notes"),
}
record = _record(category="CITATION_CHECKING", operations=("CHECKED_CITATIONS",), targets=("REFERENCE_OR_CITATION",), facts=facts)
result = evaluate(_base_case("NEJM", (record,), categories=_categories(CITATION_CHECKING="USED")))
assert [block.placement for block in result.blocks] == ["COVER_LETTER", "SUBMITTED_WORK"]
assert len({block.text for block in result.blocks}) == 2
assert len({block.purpose for block in result.blocks}) == 2
# Record binding and venue-specific fixtures from the accepted #599 vocabulary.
def test_per_record_facts_cannot_be_borrowed_across_tools_or_figures() -> None:
first = _record("figure-1", artifact="figure-1.png", facts=_frontiers_facts(represents_data=False))
borrowed = _frontiers_facts(represents_data=True)
borrowed["model"] = Fact("KNOWN", "Example Vision", "figure-1")
second = _record("figure-2", artifact="figure-2.png", facts=borrowed)
result = evaluate(
_base_case(
"Frontiers",
(first, second),
categories=_categories(DRAFTING_ASSISTANCE="USED"),
)
)
assert result.execution_status == "HALTED"
assert result.halt_reason == "UNRESOLVED_INPUT"
assert "cross-record" in " ".join(result.diagnostics)
@pytest.mark.parametrize(
("represents_data", "expected", "absent"),
(
(False, {"factual accuracy", "plagiarism-free"}, "accuracy to data"),
(True, {"factual accuracy", "plagiarism-free", "accuracy to data"}, ""),
),
)
def test_frontiers_conceptual_vs_data_figure_actions(
represents_data: bool, expected: set[str], absent: str
) -> None:
record = _record(
category="VISUAL_ARTWORK_MEDIA_ASSISTANCE",
operations=("GENERATED",),
targets=("RESEARCH_FIGURE_OR_MEDIA",),
facts=_frontiers_facts(represents_data=represents_data),
)
result = evaluate(_base_case("Frontiers", (record,), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
actions = " ".join(result.actions).casefold()
assert result.execution_status == "READY"
assert all(item in actions for item in expected)
if absent:
assert absent not in actions
def test_frontiers_unknown_data_routing_is_outstanding_not_halted() -> None:
record = _record(
category="VISUAL_ARTWORK_MEDIA_ASSISTANCE",
operations=("EDITED",),
targets=("RESEARCH_FIGURE_OR_MEDIA",),
facts=_frontiers_facts(represents_data=None, created=False),
)
result = evaluate(_base_case("Frontiers", (record,), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert result.execution_status == "READY"
assert "resolve figure_represents_data" in result.actions
assert "accuracy to data" not in " ".join(result.actions).casefold()
def test_frontiers_missing_model_or_source_halts() -> None:
facts = _frontiers_facts(represents_data=False)
facts["source_provider"] = unknown()
record = _record(facts=facts)
result = evaluate(_base_case("Frontiers", (record,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert (result.execution_status, result.halt_reason) == ("HALTED", "UNRESOLVED_INPUT")
def test_frontiers_mixed_genai_and_other_ai_scope_halts_without_partial_bundle() -> None:
genai = _record("genai", facts=_frontiers_facts(represents_data=False))
other_facts = _frontiers_facts(represents_data=False)
other_facts["policy_tool_scope"] = known("OTHER_AI")
other = _record("other-ai", tool="Classifier", task="screen", facts=other_facts)
result = evaluate(_base_case("Frontiers", (genai, other), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert (result.outcome, result.halt_reason, result.blocks) == (
"UNKNOWN", "POLICY_SCOPE_GAP", ()
)
def test_jama_non_llm_study_requires_date_and_rights_but_not_prompt_history() -> None:
facts = _jama_facts(llm=False, included=True, protected_input=True) | {
"jama_submission_type": known("ORIGINAL_INVESTIGATION"),
"jama_submission_type_is_prohibited": known(False),
}
record = _record(research_use=True, facts=facts)
result = evaluate(_base_case("JAMA", (record,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert result.execution_status == "READY"
ledger = result.ledger[record.record_id]
assert ledger["dates_of_use"].value == "2026-07-30"
assert ledger["llm_prompts"].state == "NOT_APPLICABLE"
assert ledger["copyright_permission_copy"].value.endswith(".pdf")
assert ledger["publication_rights_basis"].value
def test_jama_llm_study_requires_prompt_sequence_and_revisions() -> None:
facts = _jama_facts(llm=True) | {
"jama_submission_type": known("ORIGINAL_INVESTIGATION"),
"jama_submission_type_is_prohibited": known(False),
}
record = _record(research_use=True, facts=facts)
result = evaluate(_base_case("JAMA", (record,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert result.execution_status == "READY"
assert result.ledger[record.record_id]["llm_prompt_sequence"].value == (1, 2)
def test_conditional_children_are_not_applicable_only_after_explicit_false_parent() -> None:
facts = _jama_facts(llm=False)
facts.update({
"jama_submission_type": known("ORIGINAL_INVESTIGATION"),
"jama_submission_type_is_prohibited": known(False),
})
ready = evaluate(_base_case("JAMA", (_record(research_use=True, facts=facts),), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert ready.ledger["use-1"]["llm_prompt_sequence"].state == "NOT_APPLICABLE"
facts["study_uses_llm"] = unknown()
halted = evaluate(_base_case("JAMA", (_record(research_use=True, facts=facts),), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert halted.execution_status == "HALTED"
assert "llm_prompt_sequence" not in halted.ledger.get("use-1", {}) or halted.ledger["use-1"]["llm_prompt_sequence"].state != "NOT_APPLICABLE"
def test_jama_prohibited_manuscript_class_halts_as_known_prohibited_use() -> None:
facts = _jama_facts(llm=False) | {
"jama_submission_type": known("POETRY"),
"jama_submission_type_is_prohibited": known(True),
}
record = _record(research_use=False, facts=facts)
result = evaluate(_base_case("JAMA", (record,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert (result.outcome, result.halt_reason) == ("REQUIRED", "PROHIBITED_USE")
def test_international_eye_science_non_core_data_and_deidentification_fields() -> None:
record = _record(
category="ANALYSIS_ASSISTANCE",
operations=("ANALYSED",),
targets=("ORIGINAL_RESEARCH_DATA",),
research_use=True,
facts=_ies_facts(data=True, clinical=True),
)
result = evaluate(_base_case("International Eye Science", (record,), categories=_categories(ANALYSIS_ASSISTANCE="USED")))
assert result.execution_status == "READY"
ledger = result.ledger[record.record_id]
assert ledger["generated_proportion"].value == "8%"
assert ledger["data_types"].value == ("coded outcome labels",)
assert ledger["data_verification_status"].value
assert ledger["de_identification_measures"].value
def test_international_eye_science_other_ai_only_and_mixed_halt_as_scope_gap() -> None:
genai = _record("genai", facts=_ies_facts(data=False))
other_facts = _ies_facts(data=False)
other_facts["policy_tool_scope"] = known("OTHER_AI")
other = _record("other", facts=other_facts)
for records in ((other,), (genai, other)):
result = evaluate(_base_case("International Eye Science", records, categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert (result.outcome, result.halt_reason, result.blocks) == (
"UNKNOWN", "POLICY_SCOPE_GAP", ()
)
def test_chinese_nursing_scientific_contribution_hard_prohibition() -> None:
facts = {
"policy_tool_scope": known("GENAI_OR_AIGC"),
"ai_performed_scientific_or_intellectual_contribution": known(True),
"generated_research_figure_or_media": known(False),
"altered_original_research_data_process_or_results": known(False),
"used_unverified_genai_reference": known(False),
}
record = _record(facts=facts)
result = evaluate(_base_case("Chinese Nursing Journals Publishing House", (record,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert result.halt_reason == "PROHIBITED_USE"
def test_chinese_nursing_confirmed_non_core_use_reaches_render() -> None:
facts = {
"policy_tool_scope": known("GENAI_OR_AIGC"),
"ai_performed_scientific_or_intellectual_contribution": known(False),
"generated_research_figure_or_media": known(False),
"altered_original_research_data_process_or_results": known(False),
"used_unverified_genai_reference": known(False),
"purpose": known("surface-level terminology consistency"),
"affected_content": known("non-scientific submission metadata"),
"human_review_editing_performed": known(True),
"author_accepts_full_responsibility": known(True),
}
record = _record(
category="EDITING_ASSISTANCE",
operations=("EDITED",),
targets=("OTHER_SUBMISSION_TEXT",),
facts=facts,
)
result = evaluate(
_base_case(
"Chinese Nursing Journals Publishing House",
(record,),
categories=_categories(EDITING_ASSISTANCE="USED"),
)
)
assert (result.outcome, result.execution_status) == ("REQUIRED", "READY")
def test_nature_venue_image_is_contained_before_phase2b() -> None:
record = _record(category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",))
result = evaluate(_base_case("Nature", (record,), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert (result.outcome, result.halt_reason) == ("UNKNOWN", "CONTRACT_GAP")
assert "Venue Phase 2b" not in result.phases
assert "NATURE_VENUE_IMAGE_CONTAINMENT" in result.diagnostics
@pytest.mark.parametrize(
("venue", "facts", "target"),
(
("ICMJE", {"ai_generated_material_used_as_primary_source": known(True), "ai_cited_as_author": known(False)}, "REFERENCE_OR_CITATION"),
("NEJM", {"ai_generated_material_used_as_primary_source": known(True)}, "REFERENCE_OR_CITATION"),
("PLOS", {"ai_fabricated_or_misrepresented_primary_research_data": known(True)}, "ORIGINAL_RESEARCH_DATA"),
),
)
def test_western_hard_prohibitions_halt(venue: str, facts: dict[str, Fact], target: str) -> None:
record = _record(category="CITATION_CHECKING", operations=("CHECKED_CITATIONS",), targets=(target,), facts=facts)
result = evaluate(_base_case(venue, (record,), categories=_categories(CITATION_CHECKING="USED")))
assert result.halt_reason == "PROHIBITED_USE"
def test_icmje_member_venue_gets_separate_advisory_without_channel_merge() -> None:
facts = {
"technology_description": known("ExampleAI"),
"why_used": known("draft organization"),
"how_used": known("outlined the discussion"),
}
record = _record(facts=facts)
result = evaluate(_base_case("BMJ", (record,), categories=_categories(DRAFTING_ASSISTANCE="USED")))
assert result.execution_status == "READY"
assert "ICMJE-alongside advisory" in result.advisories
assert all(block.purpose != "ICMJE-alongside advisory" for block in result.blocks)
def test_lancet_graphical_abstract_uses_dedicated_caption_path() -> None:
facts = _lancet_common() | {
"artifact_class": known("GRAPHICAL_ABSTRACT"),
"graphical_abstract_used_ai_or_ai_assisted_illustration": known(True),
"graphical_abstract_tool_class": known("DEDICATED_SCIENTIFIC_OR_PROFESSIONAL_ILLUSTRATION_TOOL"),
"publication_rights_basis": known("tool terms grant publication rights"),
}
record = _record(category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=facts)
result = evaluate(_base_case("The Lancet", (record,), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert result.execution_status == "READY"
assert [block.placement for block in result.blocks] == ["GRAPHICAL_ABSTRACT_CAPTION"]
def test_lancet_general_purpose_graphical_abstract_tool_is_prohibited() -> None:
facts = _lancet_common() | {
"artifact_class": known("GRAPHICAL_ABSTRACT"),
"graphical_abstract_used_ai_or_ai_assisted_illustration": known(True),
"graphical_abstract_tool_class": known("GENERAL_PURPOSE_GENERATIVE_AI_IMAGE_TOOL"),
}
record = _record(category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=facts)
result = evaluate(_base_case("The Lancet", (record,), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert result.halt_reason == "PROHIBITED_USE"
def test_lancet_primary_image_vs_research_method_paths() -> None:
common = _lancet_common() | {"artifact_class": known("PRIMARY_RESEARCH_IMAGE")}
primary = common | {
"ai_is_formal_research_design_or_method": known(False),
"image_output_directly_obtained_in_research_through_that_method": known(False),
}
prohibited = evaluate(_base_case("The Lancet", (_record(category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=primary),), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert prohibited.halt_reason == "PROHIBITED_USE"
method = _lancet_common() | {
"artifact_class": known("RESEARCH_METHOD_IMAGE"),
"ai_is_formal_research_design_or_method": known(True),
"image_output_directly_obtained_in_research_through_that_method": known(True),
"reproducible_method_details": known("registered segmentation pipeline and seed"),
"model_or_tool_version": known("3.0"),
"developer_or_manufacturer_applicable": known(True),
"developer_or_manufacturer": known("Example Labs"),
}
ready = evaluate(_base_case("The Lancet", (_record(category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=method),), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert ready.execution_status == "READY"
assert [block.placement for block in ready.blocks] == ["METHODS"]
def test_lancet_protected_subject_halts_before_visual_classification() -> None:
facts = _lancet_common() | {
"generated_media_duplicates_or_refers_to_protected_subject": known(True),
}
record = _record(category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=facts)
result = evaluate(_base_case("The Lancet", (record,), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert result.halt_reason == "PROHIBITED_USE"
assert "artifact_class" not in result.ledger.get(record.record_id, {})
def test_lancet_cover_art_permission_missing_halts_and_mixed_use_is_required() -> None:
cover = _lancet_common() | {
"artifact_class": known("COVER_ART"),
"editor_permission": known(True),
"publisher_permission": known(False),
"cover_art_contains_third_party_material": known(False),
"content_attribution": known("none_applicable"),
}
cover_record = _record("cover", category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=cover)
halted = evaluate(_base_case("The Lancet", (cover_record,), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED")))
assert halted.halt_reason == "INCOMPATIBLE_FACT"
cover["publisher_permission"] = known(True)
cover_record = _record("cover", category="VISUAL_ARTWORK_MEDIA_ASSISTANCE", operations=("GENERATED",), targets=("RESEARCH_FIGURE_OR_MEDIA",), facts=cover)
prep = _lancet_common() | {
"tool_service_name": known("ExampleAI"),
"purpose": known("language revision"),
"extent_of_human_oversight": known("sentence-level review"),
"author_reviewed_and_edited": known(True),
"author_accepts_full_responsibility": known(True),
}
prep_record = _record("prep", facts=prep)
mixed = evaluate(_base_case("The Lancet", (cover_record, prep_record), categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED", DRAFTING_ASSISTANCE="USED")))
assert mixed.outcome == "REQUIRED"
assert mixed.blocks
assert "editor permission" in " ".join(mixed.actions).casefold()
@pytest.mark.parametrize(
("mutated_fact", "replacement", "expected_reason"),
(
("editor_permission", known(False), "INCOMPATIBLE_FACT"),
("publisher_permission", known(False), "INCOMPATIBLE_FACT"),
("third_party_material_permission", unknown(), "UNRESOLVED_INPUT"),
("rights_holder_permission", unknown(), "UNRESOLVED_INPUT"),
),
)
def test_lancet_permission_failures_always_halt(
mutated_fact: str, replacement: Fact, expected_reason: str
) -> None:
facts = _lancet_common() | {
"artifact_class": known("COVER_ART"),
"editor_permission": known(True),
"publisher_permission": known(True),
"cover_art_contains_third_party_material": known(True),
"third_party_material_permission": known("license/third-party.pdf"),
"content_attribution": known("artist and source credited"),
"based_on_existing_artwork_or_graphics": known(True),
"rights_holder_permission": known("license/source-art.pdf"),
"existing_artwork_attribution": known("source artist credited"),
}
facts[mutated_fact] = replacement
record = _record(
category="VISUAL_ARTWORK_MEDIA_ASSISTANCE",
operations=("GENERATED",),
targets=("RESEARCH_FIGURE_OR_MEDIA",),
facts=facts,
)
result = evaluate(
_base_case(
"The Lancet",
(record,),
categories=_categories(VISUAL_ARTWORK_MEDIA_ASSISTANCE="USED"),
)
)
assert (result.execution_status, result.halt_reason, result.blocks) == (
"HALTED",
expected_reason,
(),
)
@@ -0,0 +1,837 @@
#!/usr/bin/env python3
"""Deterministic test oracle for the venue disclosure contract.
This module exists only to exercise the documentation-owned contract in CI. It
is deliberately not imported by the disclosure runtime and is not a general
submission-policy engine.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping
import re
ALL_OUTCOMES = frozenset({"REQUIRED", "ACTION_ONLY", "NOT_REQUIRED", "UNKNOWN"})
ALL_CATEGORIES = frozenset(
{
"RESEARCH_ASSISTANCE",
"CITATION_CHECKING",
"DRAFTING_ASSISTANCE",
"REVISION_ASSISTANCE",
"EDITING_ASSISTANCE",
"ANALYSIS_ASSISTANCE",
"VISUAL_ARTWORK_MEDIA_ASSISTANCE",
"PEER_REVIEW_SIMULATION",
"OTHER_UNCLASSIFIED",
}
)
ALL_OPERATIONS = frozenset(
{
"GENERATED",
"SUBSTANTIVELY_DRAFTED",
"EDITED",
"ANALYSED",
"SEARCHED",
"CHECKED_CITATIONS",
"FORMATTED_CITATIONS",
"OTHER_CONFIRMED",
}
)
ALL_TARGETS = frozenset(
{
"WHOLE_PAPER",
"TITLE",
"ABSTRACT",
"INTRODUCTION_OR_BACKGROUND",
"METHODS",
"RESULTS",
"DISCUSSION",
"CONCLUSION",
"RESULT_INTERPRETATION",
"CORE_ARGUMENT",
"INNOVATION_CLAIM",
"RESEARCH_FIGURE_OR_MEDIA",
"ORIGINAL_RESEARCH_DATA",
"RESEARCH_PROCESS",
"SUPPORTING_DATA_FILE",
"REFERENCE_OR_CITATION",
"CODE",
"PEER_REVIEW_MATERIAL",
"OTHER_SUBMISSION_TEXT",
"OTHER_CONFIRMED",
}
)
CANONICAL_VENUES = (
"ACL",
"BMJ",
"Chinese Nursing Journals Publishing House",
"EMNLP",
"Frontiers",
"ICLR",
"ICMJE",
"International Eye Science",
"JAMA",
"Nature",
"NEJM",
"NeurIPS",
"PLOS",
"Science",
"The Lancet",
)
ALIASES = {
"the bmj": "BMJ",
"中华护理杂志社": "Chinese Nursing Journals Publishing House",
"frontiers journals": "Frontiers",
"international committee of medical journal editors": "ICMJE",
"国际眼科杂志": "International Eye Science",
"journal of the american medical association": "JAMA",
"the new england journal of medicine": "NEJM",
"new england journal of medicine": "NEJM",
"plos journals": "PLOS",
"plos one": "PLOS",
"lancet": "The Lancet",
}
for _venue in CANONICAL_VENUES:
ALIASES[_venue.casefold()] = _venue
ICMJE_MEMBER_TARGETS = frozenset({"BMJ", "JAMA", "NEJM", "The Lancet"})
AUTHORSHIP_TARGETS = frozenset(set(CANONICAL_VENUES) - {"PLOS"})
SCOPE_TARGETS = frozenset(
{
"Chinese Nursing Journals Publishing House",
"Frontiers",
"International Eye Science",
}
)
TEXT_TARGETS = frozenset(
{
"WHOLE_PAPER",
"TITLE",
"ABSTRACT",
"INTRODUCTION_OR_BACKGROUND",
"METHODS",
"RESULTS",
"DISCUSSION",
"CONCLUSION",
"RESULT_INTERPRETATION",
"CORE_ARGUMENT",
"INNOVATION_CLAIM",
"OTHER_SUBMISSION_TEXT",
}
)
@dataclass(frozen=True)
class Fact:
state: str
value: Any = None
owner: str | None = None
def known(value: Any, owner: str | None = None) -> Fact:
return Fact("KNOWN", value, owner)
def unknown(owner: str | None = None) -> Fact:
return Fact("UNKNOWN", None, owner)
def not_applicable(owner: str | None = None) -> Fact:
return Fact("NOT_APPLICABLE", None, owner)
@dataclass(frozen=True)
class UseRecord:
record_id: str
tool: str
task: str
artifact: str
category: str
operations: tuple[str, ...]
targets: tuple[str, ...]
research_use: bool = False
facts: Mapping[str, Fact] = field(default_factory=dict)
@dataclass(frozen=True)
class Block:
placement: str
purpose: str
text: str
@dataclass(frozen=True)
class ContractResult:
track: str
outcome: str | None
execution_status: str
halt_reason: str | None = None
phases: tuple[str, ...] = ()
blocks: tuple[Block, ...] = ()
actions: tuple[str, ...] = ()
advisories: tuple[str, ...] = ()
ledger: Mapping[str, Mapping[str, Fact]] = field(default_factory=dict)
diagnostics: tuple[str, ...] = ()
class _ContractStop(RuntimeError):
pass
class _Evaluator:
def __init__(self, case: Mapping[str, object]) -> None:
self.case = case
self.track = "venue"
self.outcome: str | None = None
self.status = "READY"
self.halt_reason: str | None = None
self.phases: list[str] = []
self.blocks: list[Block] = []
self.actions: list[str] = []
self.advisories: list[str] = []
self.ledger: dict[str, dict[str, Fact]] = {}
self.diagnostics: list[str] = []
self.records: tuple[UseRecord, ...] = tuple(case.get("records", ())) # type: ignore[arg-type]
self.global_facts: Mapping[str, Fact] = case.get("global_facts", {}) # type: ignore[assignment]
self.venue: str | None = None
def result(self) -> ContractResult:
return ContractResult(
track=self.track,
outcome=self.outcome,
execution_status=self.status,
halt_reason=self.halt_reason,
phases=tuple(self.phases),
blocks=tuple(self.blocks),
actions=tuple(self.actions),
advisories=tuple(self.advisories),
ledger={key: dict(value) for key, value in self.ledger.items()},
diagnostics=tuple(self.diagnostics),
)
def halt(self, outcome: str, reason: str, diagnostic: str) -> None:
self.outcome = outcome
self.status = "HALTED"
self.halt_reason = reason
self.diagnostics.append(diagnostic)
self.blocks.clear()
raise _ContractStop
def require_global(self, name: str, required_value: object | None = None) -> Fact:
fact = self.global_facts.get(name, unknown())
if fact.state != "KNOWN":
self.halt(self.outcome or "UNKNOWN", "UNRESOLVED_INPUT", f"global fact {name} is UNKNOWN")
if required_value is not None and fact.value != required_value:
self.halt(self.outcome or "UNKNOWN", "INCOMPATIBLE_FACT", f"global fact {name} is incompatible")
return fact
def require_record(
self,
record: UseRecord,
name: str,
*,
required_value: object | None = None,
) -> Fact:
fact = record.facts.get(name, unknown(record.record_id))
self.ledger.setdefault(record.record_id, {})[name] = fact
if fact.owner not in {None, record.record_id}:
self.halt(
self.outcome or "UNKNOWN",
"UNRESOLVED_INPUT",
f"cross-record fact borrowing: {name} for {record.record_id} belongs to {fact.owner}",
)
if fact.state != "KNOWN":
self.halt(self.outcome or "UNKNOWN", "UNRESOLVED_INPUT", f"{record.record_id}.{name} is UNKNOWN")
if required_value is not None and fact.value != required_value:
self.halt(
self.outcome or "UNKNOWN",
"INCOMPATIBLE_FACT",
f"{record.record_id}.{name} must be {required_value!r}",
)
return fact
def prohibit_true(self, record: UseRecord, name: str) -> Fact:
fact = self.require_record(record, name)
if fact.value is True:
self.halt(self.outcome or "REQUIRED", "PROHIBITED_USE", f"prohibited predicate true: {record.record_id}.{name}")
if fact.value is not False:
self.halt(self.outcome or "REQUIRED", "INCOMPATIBLE_FACT", f"{record.record_id}.{name} is not boolean")
return fact
def conditional(
self,
record: UseRecord,
parent: str,
children: tuple[str, ...],
) -> bool:
fact = self.require_record(record, parent)
if fact.value is False:
for child in children:
supplied = record.facts.get(child)
if supplied is not None and supplied.state == "KNOWN":
self.halt(
self.outcome or "REQUIRED",
"INCOMPATIBLE_FACT",
f"{child} cannot be KNOWN when {parent} is false",
)
self.ledger.setdefault(record.record_id, {})[child] = not_applicable(record.record_id)
return False
if fact.value is not True:
self.halt(self.outcome or "REQUIRED", "INCOMPATIBLE_FACT", f"{parent} is not boolean")
for child in children:
self.require_record(record, child)
return True
def run(self) -> ContractResult:
try:
self._dispatch()
if self.track == "anchor":
return self.result()
self._intake()
self._phase2a()
if self.outcome == "ACTION_ONLY":
self._cover_actions()
self._member_advisory()
return self.result()
if self.outcome == "NOT_REQUIRED":
self._member_advisory()
return self.result()
self._phase2b()
self._render()
self._phase5_actions()
self._member_advisory()
except _ContractStop:
pass
return self.result()
def _dispatch(self) -> None:
self.phases.append("Selector dispatch")
raw_venue = self.case.get("venue")
raw_anchor = self.case.get("policy_anchor")
if raw_anchor is not None:
anchor = str(raw_anchor).casefold()
venue_text = str(raw_venue).strip() if raw_venue is not None else None
nature_pair = anchor == "nature" and venue_text is not None and (
venue_text.casefold() in {
"nature",
"nature portfolio",
"nature (nature publishing group)",
"nature publishing group",
}
or venue_text.startswith("Nature ")
)
if raw_venue is not None and not nature_pair:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "selector conflict")
if anchor not in {"prisma-traice", "icmje", "nature", "ieee"}:
self.halt("UNKNOWN", "UNRESOLVED_INPUT", "unknown policy anchor")
self.track = "anchor"
self.phases.append("Anchor intake")
return
if raw_venue is None:
self.halt("UNKNOWN", "UNRESOLVED_INPUT", "selector required")
key = str(raw_venue).strip().casefold()
self.venue = ALIASES.get(key)
if self.venue is None:
self.phases.append("Venue lookup")
self.halt(
"UNKNOWN",
"UNCURATED_POLICY",
f"I do not have a curated executable policy for {raw_venue}",
)
def _intake(self) -> None:
self.phases.append("Intake")
self.outcome = "UNKNOWN"
self.require_global("external_use_inventory_confirmed")
categories = self.case.get("categories")
if not isinstance(categories, Mapping) or set(categories) != ALL_CATEGORIES:
self.halt("UNKNOWN", "UNRESOLVED_INPUT", "complete category inventory required")
invalid = {value for value in categories.values() if value not in {"USED", "NOT_USED", "UNCERTAIN"}}
if invalid or "UNCERTAIN" in categories.values():
self.halt("UNKNOWN", "UNRESOLVED_INPUT", "category inventory unresolved")
used_categories = {name for name, state in categories.items() if state == "USED"}
if used_categories and not self.records:
self.halt("UNKNOWN", "UNRESOLVED_INPUT", "USED category has no use record")
for record in self.records:
if record.category not in ALL_CATEGORIES or record.category not in used_categories:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", f"record/category mismatch: {record.record_id}")
if not record.record_id or not record.tool or not record.task or not record.artifact:
self.halt("UNKNOWN", "UNRESOLVED_INPUT", "tool x task/run/artifact identity incomplete")
if not record.operations or not set(record.operations) <= ALL_OPERATIONS:
self.halt("UNKNOWN", "UNRESOLVED_INPUT", f"operation unresolved: {record.record_id}")
if not record.targets or not set(record.targets) <= ALL_TARGETS:
self.halt("UNKNOWN", "UNRESOLVED_INPUT", f"target unresolved: {record.record_id}")
other = [record for record in self.records if record.category == "OTHER_UNCLASSIFIED"]
if other:
description = other[0].facts.get("verbatim_description", unknown()).value
self.halt(
"UNKNOWN",
"UNRESOLVED_INPUT",
f"unclassified use preserved: {description}",
)
def _phase2a(self) -> None:
assert self.venue is not None
self.phases.extend(("Venue Phase 2", "Venue Phase 2a"))
if self.venue in AUTHORSHIP_TARGETS:
author_fact = self.require_global("ai_listed_or_proposed_as_author")
if author_fact.value is True:
self.halt("REQUIRED" if self.records else "NOT_REQUIRED", "INCOMPATIBLE_FACT", "AI cannot be listed as author")
if author_fact.value is not False:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "authorship fact is not boolean")
if self.venue in SCOPE_TARGETS and self.records:
scopes = [self.require_record(record, "policy_tool_scope").value for record in self.records]
if any(scope == "OTHER_AI" for scope in scopes):
self.halt("UNKNOWN", "POLICY_SCOPE_GAP", "OTHER_AI inventory requires current non-generative policy")
if any(scope != "GENAI_OR_AIGC" for scope in scopes):
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "invalid policy_tool_scope")
self.outcome = "REQUIRED" if self.records else "NOT_REQUIRED"
if not self.records:
return
handler = {
"Chinese Nursing Journals Publishing House": self._phase2a_chinese_nursing,
"ICMJE": self._phase2a_icmje,
"International Eye Science": self._phase2a_ies,
"JAMA": self._phase2a_jama,
"Nature": self._phase2a_nature,
"NEJM": self._phase2a_nejm,
"PLOS": self._phase2a_plos,
"The Lancet": self._phase2a_lancet,
}.get(self.venue)
if handler is not None:
handler()
def _phase2a_chinese_nursing(self) -> None:
names = (
"ai_performed_scientific_or_intellectual_contribution",
"generated_research_figure_or_media",
"altered_original_research_data_process_or_results",
"used_unverified_genai_reference",
)
for record in self.records:
for name in names:
self.prohibit_true(record, name)
def _phase2a_icmje(self) -> None:
for record in self.records:
if "REFERENCE_OR_CITATION" in record.targets:
self.prohibit_true(record, "ai_generated_material_used_as_primary_source")
self.prohibit_true(record, "ai_cited_as_author")
def _phase2a_ies(self) -> None:
allowed_scopes = {
"LANGUAGE_POLISHING",
"LITERATURE_RETRIEVAL",
"DATA_ORGANIZATION",
"CHART_ANNOTATION",
"OTHER_CONFIRMED_NON_CORE_RESEARCH_STEP",
"CORE_RESEARCH_STEP",
}
prohibitions = (
"generated_core_main_text_conclusion_analysis_viewpoint_or_innovation_claim",
"fabricated_experimental_plan_technical_route_or_citation",
"replaced_author_in_experimental_design_or_data_validation",
"fabricated_data_invented_results_or_tampered_conclusions",
"rewrote_plagiarized_work_to_evade_detection",
"generated_peer_review_response_grant_contribution_or_integrity_statement",
"uploaded_secret_research_data_or_unpublished_results_to_public_ai_platform",
"aigc_generated_or_tampered_data",
"aigc_replaced_core_analysis",
"uploaded_undeidentified_data_to_aigc",
"uploaded_data_lacking_required_ethics_review_to_aigc",
"fabricated_data_or_ethics_proof",
)
for record in self.records:
scope = self.require_record(record, "use_scope").value
if scope not in allowed_scopes:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "invalid International Eye Science use_scope")
if scope == "CORE_RESEARCH_STEP":
self.halt("REQUIRED", "PROHIBITED_USE", "CORE_RESEARCH_STEP is prohibited")
if scope == "OTHER_CONFIRMED_NON_CORE_RESEARCH_STEP":
self.require_record(record, "use_scope_basis")
overseas = self.require_record(record, "tool_is_overseas")
if overseas.value is True:
self.require_record(record, "lawful_compliance_qualification", required_value=True)
elif overseas.value is not False:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "tool_is_overseas is not boolean")
for name in prohibitions:
self.prohibit_true(record, name)
data = self.require_record(record, "use_involves_data")
if scope in {"DATA_ORGANIZATION", "CHART_ANNOTATION"} and data.value is not True:
self.halt("REQUIRED", "INCOMPATIBLE_FACT", "data scope requires use_involves_data=true")
def _phase2a_jama(self) -> None:
for record in self.records:
generated_text = bool(
set(record.operations) & {"GENERATED", "SUBSTANTIVELY_DRAFTED"}
and set(record.targets) & TEXT_TARGETS
)
if generated_text or "OTHER_CONFIRMED" in record.operations or "OTHER_CONFIRMED" in record.targets:
submission_type = self.require_record(record, "jama_submission_type")
prohibited = self.require_record(record, "jama_submission_type_is_prohibited")
known_prohibited = submission_type.value in {
"OPINION_MANUSCRIPT",
"LETTER_TO_THE_EDITOR",
"ONLINE_COMMENT",
"A_PIECE_OF_MY_MIND",
"POETRY",
}
if prohibited.value is not known_prohibited:
self.halt("REQUIRED", "INCOMPATIBLE_FACT", "JAMA submission-type predicate contradicts exact type")
if known_prohibited:
self.halt("REQUIRED", "PROHIBITED_USE", f"JAMA prohibits AI drafting for {submission_type.value}")
if "RESEARCH_FIGURE_OR_MEDIA" in record.targets:
created = self.require_record(record, "clinical_image_or_illustration_created_or_manipulated")
if created.value is True:
self.require_record(record, "part_of_formal_research_design_or_methods", required_value=True)
elif created.value is not False:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "clinical image predicate is not boolean")
def _phase2a_nature(self) -> None:
if any("RESEARCH_FIGURE_OR_MEDIA" in record.targets for record in self.records):
self.halt("UNKNOWN", "CONTRACT_GAP", "NATURE_VENUE_IMAGE_CONTAINMENT")
def _phase2a_nejm(self) -> None:
for record in self.records:
if "REFERENCE_OR_CITATION" in record.targets:
self.prohibit_true(record, "ai_generated_material_used_as_primary_source")
def _phase2a_plos(self) -> None:
data_targets = {"ORIGINAL_RESEARCH_DATA", "RESULTS", "SUPPORTING_DATA_FILE"}
for record in self.records:
if set(record.targets) & data_targets:
self.prohibit_true(record, "ai_fabricated_or_misrepresented_primary_research_data")
def _phase2a_lancet(self) -> None:
cover_count = 0
for record in self.records:
self.prohibit_true(record, "ai_replaced_authors_intellectual_contribution")
if "RESEARCH_FIGURE_OR_MEDIA" not in record.targets:
continue
self.prohibit_true(record, "generated_media_duplicates_or_refers_to_protected_subject")
artifact_class = self.require_record(record, "artifact_class").value
if artifact_class == "PRIMARY_RESEARCH_IMAGE":
formal = self.require_record(record, "ai_is_formal_research_design_or_method")
direct = self.require_record(record, "image_output_directly_obtained_in_research_through_that_method")
if formal.value is not True or direct.value is not True:
self.halt("REQUIRED", "PROHIBITED_USE", "AI-created primary research image is prohibited")
elif artifact_class == "RESEARCH_METHOD_IMAGE":
self.require_record(record, "ai_is_formal_research_design_or_method", required_value=True)
self.require_record(record, "image_output_directly_obtained_in_research_through_that_method", required_value=True)
self.require_record(record, "reproducible_method_details")
elif artifact_class == "GRAPHICAL_ABSTRACT":
self.require_record(record, "graphical_abstract_used_ai_or_ai_assisted_illustration", required_value=True)
tool_class = self.require_record(record, "graphical_abstract_tool_class").value
if tool_class == "GENERAL_PURPOSE_GENERATIVE_AI_IMAGE_TOOL":
self.halt("REQUIRED", "PROHIBITED_USE", "general-purpose GenAI graphical abstracts are prohibited")
if tool_class != "DEDICATED_SCIENTIFIC_OR_PROFESSIONAL_ILLUSTRATION_TOOL":
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "unknown graphical-abstract tool class")
elif artifact_class == "COVER_ART":
cover_count += 1
self.require_record(record, "editor_permission", required_value=True)
self.require_record(record, "publisher_permission", required_value=True)
has_third_party = self.require_record(record, "cover_art_contains_third_party_material")
if has_third_party.value is True:
self.require_record(record, "third_party_material_permission")
elif has_third_party.value is False:
self.ledger[record.record_id]["third_party_material_permission"] = not_applicable(record.record_id)
else:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "third-party predicate is not boolean")
self.require_record(record, "content_attribution")
elif artifact_class not in {"EXPLANATORY_IMAGE", "DATA_VISUALIZATION"}:
self.halt("UNKNOWN", "CONTRACT_GAP", "unmodelled visual/media class")
self.require_record(record, "visual_accuracy_confirmed", required_value=True)
self.require_record(record, "visual_originality_confirmed", required_value=True)
based = self.require_record(record, "based_on_existing_artwork_or_graphics")
if based.value is True:
self.require_record(record, "rights_holder_permission")
self.require_record(record, "existing_artwork_attribution")
elif based.value is False:
self.ledger[record.record_id]["rights_holder_permission"] = not_applicable(record.record_id)
self.ledger[record.record_id]["existing_artwork_attribution"] = not_applicable(record.record_id)
else:
self.halt("UNKNOWN", "INCOMPATIBLE_FACT", "existing-artwork predicate is not boolean")
if artifact_class == "DATA_VISUALIZATION":
self.require_record(record, "directly_derived_from_data_by_reproducible_method", required_value=True)
if cover_count == len(self.records):
self.outcome = "ACTION_ONLY"
def _phase2b(self) -> None:
assert self.venue is not None
self.phases.append("Venue Phase 2b")
for record in self.records:
if self.venue == "ICLR":
self.require_record(record, "specific_assisted_tasks")
self.require_record(record, "author_accepts_full_responsibility", required_value=True)
self.require_record(record, "affected_content")
elif self.venue == "BMJ":
for name in ("technology_description", "why_used", "how_used"):
self.require_record(record, name)
elif self.venue == "Chinese Nursing Journals Publishing House":
for name in ("purpose", "affected_content", "human_review_editing_performed", "author_accepts_full_responsibility"):
required = True if name in {"human_review_editing_performed", "author_accepts_full_responsibility"} else None
self.require_record(record, name, required_value=required)
elif self.venue == "Frontiers":
for name in ("version", "model", "source_provider", "content_operation", "affected_content_kind", "affected_content"):
self.require_record(record, name)
elif self.venue == "ICMJE":
for name in ("technology_description", "how_used"):
self.require_record(record, name)
elif self.venue == "International Eye Science":
self._phase2b_ies(record)
elif self.venue == "JAMA":
self._phase2b_jama(record)
elif self.venue == "Nature":
for name in ("how_used", "affected_content", "author_accepts_accountability"):
required = True if name == "author_accepts_accountability" else None
self.require_record(record, name, required_value=required)
elif self.venue == "NEJM":
for name in ("technology_description", "produced_content", "human_review_editing_performed", "no_plagiarism_confirmed"):
required = True if name in {"human_review_editing_performed", "no_plagiarism_confirmed"} else None
self.require_record(record, name, required_value=required)
elif self.venue == "PLOS":
for name in ("how_used", "outputs_validated", "affected_content"):
self.require_record(record, name)
elif self.venue == "The Lancet":
self._phase2b_lancet(record)
def _phase2b_ies(self, record: UseRecord) -> None:
for name in ("version", "purpose", "use_scope", "generated_proportion"):
self.require_record(record, name)
data = self.require_record(record, "use_involves_data")
data_children = ("data_types", "data_verification_status", "data_involves_clinical_or_case_data")
if data.value is False:
for child in data_children:
self.ledger[record.record_id][child] = not_applicable(record.record_id)
self.ledger[record.record_id]["de_identification_measures"] = not_applicable(record.record_id)
return
if data.value is not True:
self.halt("REQUIRED", "INCOMPATIBLE_FACT", "use_involves_data is not boolean")
for child in data_children:
self.require_record(record, child)
clinical = self.ledger[record.record_id]["data_involves_clinical_or_case_data"]
if clinical.value is True:
self.require_record(record, "de_identification_measures")
elif clinical.value is False:
self.ledger[record.record_id]["de_identification_measures"] = not_applicable(record.record_id)
else:
self.halt("REQUIRED", "INCOMPATIBLE_FACT", "clinical-data predicate is not boolean")
def _phase2b_jama(self, record: UseRecord) -> None:
self.require_record(record, "author_review_accuracy", required_value=True)
self.require_record(record, "author_accepts_content_integrity_responsibility", required_value=True)
for name in ("model_or_tool_version", "manufacturer", "dates_of_use", "use_description", "affected_portions"):
self.require_record(record, name)
self.conditional(record, "extension_numbers_applicable", ("extension_numbers",))
self.ledger[record.record_id]["ai_used_in_scientific_study"] = known(
record.research_use, record.record_id
)
if not record.research_use:
return
self.require_record(record, "specific_research_use")
self.conditional(
record,
"study_uses_llm",
("llm_prompts", "llm_prompt_sequence", "llm_prompt_revisions"),
)
self.conditional(
record,
"copyright_protected_content_entered",
("copyright_permission_copy", "copyright_permission_methods_description"),
)
self.conditional(
record,
"ai_generated_content_included_in_submission",
("included_content_type", "publication_rights_basis"),
)
def _phase2b_lancet(self, record: UseRecord) -> None:
artifact = self.ledger.get(record.record_id, {}).get("artifact_class")
if artifact is not None and artifact.value == "COVER_ART":
return
if artifact is not None and artifact.value == "GRAPHICAL_ABSTRACT":
self.require_record(record, "publication_rights_basis")
return
if artifact is not None and artifact.value == "RESEARCH_METHOD_IMAGE":
self.require_record(record, "model_or_tool_version")
self.conditional(
record,
"developer_or_manufacturer_applicable",
("developer_or_manufacturer",),
)
return
if artifact is not None and artifact.value in {"EXPLANATORY_IMAGE", "DATA_VISUALIZATION"}:
self.require_record(record, "model_or_tool_version")
return
for name in (
"tool_service_name",
"purpose",
"extent_of_human_oversight",
"author_reviewed_and_edited",
"author_accepts_full_responsibility",
):
required = True if name in {"author_reviewed_and_edited", "author_accepts_full_responsibility"} else None
self.require_record(record, name, required_value=required)
def _render(self) -> None:
assert self.venue is not None
self.phases.extend(("Venue Phase 3", "Venue Phase 4", "Venue Phase 5"))
if self.venue == "NEJM":
self.blocks.extend(
(
Block("COVER_LETTER", "submission disclosure", "Tell the editor which technology was used and what it produced."),
Block("SUBMITTED_WORK", "reader-facing disclosure", "Describe the reviewed AI-produced material and originality confirmation in the submitted work."),
)
)
elif self.venue == "ICMJE":
self.blocks.extend(
(
Block("COVER_LETTER", "editor disclosure", "Describe the AI-assisted technology and use for the editor."),
Block("SUBMITTED_WORK", "article disclosure", "Describe the AI-assisted technology and use in the appropriate article section."),
)
)
elif self.venue == "JAMA":
if any(record.research_use for record in self.records):
self.blocks.append(Block("METHODS", "AI-use disclosure portion", "Describe the specific research AI use and confirmed conditional rights facts."))
if any(not record.research_use for record in self.records):
self.blocks.append(Block("ACKNOWLEDGEMENTS", "manuscript-preparation disclosure", "Identify the tool, dates, affected portions, review, and responsibility."))
if any(
self.ledger.get(record.record_id, {}).get("ai_generated_content_included_in_submission", unknown()).value is True
for record in self.records
):
self.blocks.append(Block("RELEVANT_LEGEND", "item-specific publication-rights disclosure", "State the confirmed publication-rights basis for the affected item."))
elif self.venue == "The Lancet":
for record in self.records:
artifact = self.ledger.get(record.record_id, {}).get("artifact_class")
if artifact is None:
self.blocks.append(Block("DECLARATION_BEFORE_REFERENCES", f"manuscript-preparation declaration for {record.record_id}", f"Declare {record.tool}'s purpose, oversight, review, and author responsibility."))
elif artifact.value == "GRAPHICAL_ABSTRACT":
self.blocks.append(Block("GRAPHICAL_ABSTRACT_CAPTION", "illustration-tool disclosure", f"Name {record.tool} and its publication-rights basis in this caption."))
elif artifact.value in {"RESEARCH_METHOD_IMAGE", "DATA_VISUALIZATION"}:
self.blocks.append(Block("METHODS", f"research visual method for {record.record_id}", f"Describe reproducible use of {record.tool} for {record.artifact}."))
elif artifact.value == "EXPLANATORY_IMAGE":
self.blocks.append(Block("IMAGE_CAPTION", f"explanatory-image disclosure for {record.record_id}", f"Identify {record.tool} for {record.artifact}."))
else:
placements = {
"ACL": "ACKNOWLEDGEMENTS",
"BMJ": "ACKNOWLEDGEMENTS_OR_METHODS",
"Chinese Nursing Journals Publishing House": "END_OF_MAIN_TEXT",
"EMNLP": "ACKNOWLEDGEMENTS",
"Frontiers": "ACKNOWLEDGEMENTS_OR_METHODS",
"ICLR": "PAPER_BODY",
"International Eye Science": "END_OF_MAIN_TEXT",
"Nature": "METHODS_OR_ACKNOWLEDGEMENTS",
"PLOS": "METHODS",
"Science": "ACKNOWLEDGEMENTS_OR_METHODS",
}
placement = placements.get(self.venue, "POLICY_SPECIFIED_LOCATION")
self.blocks.append(Block(placement, "venue AI-use disclosure", f"Render confirmed per-record facts for {self.venue}."))
def _phase5_actions(self) -> None:
if self.venue == "Frontiers":
for record in self.records:
operation = self.ledger[record.record_id]["content_operation"].value
kind = self.ledger[record.record_id]["affected_content_kind"].value
if operation == "CREATED":
self.actions.append(f"{record.record_id}: factual accuracy")
self.actions.append(f"{record.record_id}: plagiarism-free")
if kind == "VISUAL":
represents = record.facts.get("figure_represents_data", unknown(record.record_id))
if represents.state != "KNOWN":
self.actions.append("resolve figure_represents_data")
elif represents.value is True:
self.actions.append(f"{record.record_id}: accuracy to data")
self._cover_actions()
def _cover_actions(self) -> None:
if self.venue != "The Lancet":
return
for record in self.records:
artifact = self.ledger.get(record.record_id, {}).get("artifact_class")
if artifact is not None and artifact.value == "COVER_ART":
self.actions.extend(
(
f"{record.record_id}: editor permission confirmed",
f"{record.record_id}: publisher permission confirmed",
f"{record.record_id}: third-party material and attribution checked",
)
)
def _member_advisory(self) -> None:
if self.venue in ICMJE_MEMBER_TARGETS and self.records:
self.advisories.append("ICMJE-alongside advisory")
def evaluate(case: Mapping[str, object]) -> ContractResult:
"""Evaluate one synthetic fixture against the documentation-owned contract."""
return _Evaluator(case).run()
SURFACE_FILES = (
"academic-paper/SKILL.md",
"commands/ars-disclosure.md",
"academic-paper/references/mode_selection_guide.md",
"README.md",
"README.ja-JP.md",
"README.ko-KR.md",
"README.zh-CN.md",
"README.zh-TW.md",
)
SURFACE_TOKENS = {
"ACL": ("ACL",),
"BMJ": ("BMJ",),
"Chinese Nursing Journals Publishing House": (
"Chinese Nursing Journals Publishing House",
"中华护理杂志社",
),
"EMNLP": ("EMNLP",),
"Frontiers": ("Frontiers",),
"ICLR": ("ICLR",),
"ICMJE": ("ICMJE",),
"International Eye Science": ("International Eye Science", "国际眼科杂志"),
"JAMA": ("JAMA",),
"Nature": ("Nature",),
"NEJM": ("NEJM",),
"NeurIPS": ("NeurIPS",),
"PLOS": ("PLOS",),
"Science": ("Science",),
"The Lancet": ("The Lancet",),
}
def surface_sync_errors(root: Path) -> list[str]:
"""Return deterministic selector/database/user-surface drift diagnostics."""
errors: list[str] = []
policies_path = root / "academic-paper/references/venue_disclosure_policies.md"
protocol_path = root / "academic-paper/references/disclosure_mode_protocol.md"
try:
policies = policies_path.read_text(encoding="utf-8")
protocol = protocol_path.read_text(encoding="utf-8")
except FileNotFoundError as exc:
return [f"required contract surface missing: {exc.filename}"]
headings = re.findall(r"^## Venue: (.+)$", policies, re.MULTILINE)
labels = tuple(heading.split(" (", 1)[0].strip() for heading in headings)
if labels != CANONICAL_VENUES:
errors.append("runnable database canonical inventory/order drift")
for venue, tokens in SURFACE_TOKENS.items():
if not any(token in protocol for token in tokens):
errors.append(f"disclosure_mode_protocol.md missing selector for {venue}")
for rel in SURFACE_FILES:
path = root / rel
try:
text = path.read_text(encoding="utf-8")
except FileNotFoundError:
errors.append(f"{rel}: surface missing")
continue
for venue, tokens in SURFACE_TOKENS.items():
if not any(token in text for token in tokens):
errors.append(f"{rel}: selector drift for {venue}")
return errors