diff --git a/.github/workflows/paper-contracts.yml b/.github/workflows/paper-contracts.yml new file mode 100644 index 0000000..705b7e7 --- /dev/null +++ b/.github/workflows/paper-contracts.yml @@ -0,0 +1,34 @@ +name: Paper contracts + +on: + pull_request: + paths: + - "templates/default/ui/v2_*.py" + - "templates/default/schemas/**" + - "templates/default/research_trajectory/**/*.example.json" + - "tests/paper_contracts/**" + - ".github/workflows/paper-contracts.yml" + push: + branches: [main] + paths: + - "templates/default/ui/v2_*.py" + - "templates/default/schemas/**" + - "templates/default/research_trajectory/**/*.example.json" + - "tests/paper_contracts/**" + - ".github/workflows/paper-contracts.yml" + workflow_dispatch: + +permissions: + contents: read + +jobs: + contracts: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Run paper-facing contract tests + run: python -m unittest discover -s tests/paper_contracts -p 'test_*.py' -v diff --git a/templates/default/ui/v2_merge.py b/templates/default/ui/v2_merge.py index a2823f0..4f441b4 100644 --- a/templates/default/ui/v2_merge.py +++ b/templates/default/ui/v2_merge.py @@ -287,6 +287,31 @@ def _decoded_projection(projection: Mapping[str, Any]) -> dict[str, Any]: } +def _canonical_card_status_sets(projection: Mapping[str, Any]) -> dict[str, set[str]]: + """Return service-canonical card identities by effective findings status. + + Merely mentioning a card ID elsewhere in canonical state (for example in a + material exclusion) does not make that card accepted evidence. + """ + + decoded = _decoded_projection(projection) + findings = next( + ( + value + for value in _projection_json_values(decoded) + if value.get("artifact_type") == "current_findings" + ), + None, + ) + if not findings: + return {"accepted": set(), "qualified": set(), "superseded": set()} + return { + "accepted": set(findings.get("accepted_card_ids", ())), + "qualified": set(findings.get("qualified_card_ids", ())), + "superseded": set(findings.get("superseded_card_ids", ())), + } + + def _projection_errors( before: Mapping[str, Any], after: Mapping[str, Any], @@ -332,7 +357,9 @@ def _projection_errors( if removed := old_evidence - new_evidence: errors.append(f"negative/conflicting evidence was removed: {sorted(removed)}") - old_known = _all_card_ids(before) + old_status = _canonical_card_status_sets(before) + old_effective = old_status["accepted"] | old_status["qualified"] + old_history = old_effective | old_status["superseded"] accepted_now = { str(item["card_id"]) for item in card_decisions @@ -341,19 +368,19 @@ def _projection_errors( cards_by_id = { str(card.get("id")): card for card in _result_card_list(result_cards) } - known = old_known | accepted_now + known_active = old_effective | accepted_now referenced = _field_card_ids( after, _CARD_REFERENCE_FIELDS - {"superseded_card_ids"} ) - if unknown := referenced - known: + if unknown := referenced - known_active: errors.append( f"effective projection references unknown/unaccepted cards: {sorted(unknown)}" ) superseded = _field_card_ids(after, {"superseded_card_ids"}) - # Recorded historical proposals may be corrected without promoting them - # to active evidence. Only the service supplies these verified identities. - if unknown := superseded - old_known - set(cards_by_id) - set(prior_card_ids): + # Receipt-verified deferred cards may be named only as supersession targets; + # an arbitrary historical mention does not establish a valid identity. + if unknown := superseded - old_history - set(cards_by_id) - set(prior_card_ids): errors.append( f"effective projection references unknown superseded cards: {sorted(unknown)}" ) @@ -362,7 +389,8 @@ def _projection_errors( for card_id in accepted_now for prior_id in cards_by_id.get(card_id, {}).get("supersedes", ()) } - if unjustified := superseded - old_known - declared_superseded: + newly_superseded = superseded - old_status["superseded"] + if unjustified := newly_superseded - declared_superseded: errors.append( "effective projection marks cards superseded without an accepted " f"superseding card: {sorted(unjustified)}" @@ -381,6 +409,19 @@ def _projection_errors( None, ) if old_findings and findings: + old_effective_findings = ( + set(old_findings.get("accepted_card_ids", ())) + | set(old_findings.get("qualified_card_ids", ())) + ) + new_effective_findings = ( + set(findings.get("accepted_card_ids", ())) + | set(findings.get("qualified_card_ids", ())) + ) + if unauthorized := (new_effective_findings - old_effective_findings) - accepted_now: + errors.append( + "canonical findings promote cards without an accepted current decision: " + f"{sorted(unauthorized)}" + ) old_canonical = ( set(old_findings.get("accepted_card_ids", ())) | set(old_findings.get("qualified_card_ids", ())) @@ -544,7 +585,13 @@ def prepare_card_decisions(*, merge_request, result_cards, operations, candidate f"Merge Request repeats card decisions: {sorted(duplicate_requests)}" ) try: - old_card_ids = _all_card_ids(_decoded_projection(base_projection)) | set(prior_card_ids) + old_status = _canonical_card_status_sets(base_projection) + old_card_ids = ( + old_status["accepted"] + | old_status["qualified"] + | old_status["superseded"] + | set(prior_card_ids) + ) except (TypeError, ValueError) as exc: old_card_ids = set() errors.append(str(exc)) diff --git a/templates/default/ui/v2_runtime.py b/templates/default/ui/v2_runtime.py index c90307e..547d87d 100644 --- a/templates/default/ui/v2_runtime.py +++ b/templates/default/ui/v2_runtime.py @@ -2202,6 +2202,41 @@ def _preexecution_output_paths(self, trial_id: str, stage_id: str) -> list[str]: ) return sorted(set(outputs)) + def _validate_plan_critical_path_target( + self, plan: Mapping[str, Any] + ) -> None: + """Bind the Plan to one current service-derived Critical Path item. + + The JSON schema can only require a non-empty identifier. This runtime + check supplies the relational invariant: the target must name an item + in the currently published project state for the current canonical + revision. Scientific move selection remains agent reasoning; this + check validates only the Plan-to-control-state reference. + """ + + state = self._load_artifact( + "research_trajectory/STATE.json", "project_state" + ) + if state.get("project_id") != self.project_id: + raise V2RuntimeError("published project state belongs to another project") + current_revision = self.transactions.current_revision() + if state.get("canonical_revision") != current_revision: + raise V2RuntimeError( + "published project state revision differs from the current canonical revision" + ) + target = str(plan.get("critical_path_target") or "").strip() + current = { + str(item.get("id")): item + for item in state.get("critical_path", ()) + if isinstance(item, Mapping) and item.get("status") != "complete" + } + if target not in current: + available = sorted(current) + raise V2RuntimeError( + "PLAN critical_path_target does not reference a current actionable " + f"Critical Path item: {target or ''}; available={available}" + ) + def _plan_context( self, trial_id: str, @@ -2216,6 +2251,7 @@ def _plan_context( route_relative = f"{trial_root}/EXPERT_ROUTE.json" review_relative = f"{trial_root}/reviews/PLAN_REVIEW.json" plan = self._load_artifact(plan_relative, "plan", paired=True) + self._validate_plan_critical_path_target(plan) route = self._load_artifact(route_relative, "expert_route", paired=True) self._validate_expert_route(route, plan) review = self._load_artifact( diff --git a/tests/paper_contracts/README.md b/tests/paper_contracts/README.md new file mode 100644 index 0000000..77a1fd6 --- /dev/null +++ b/tests/paper_contracts/README.md @@ -0,0 +1,20 @@ +# Paper contract tests + +These tests expose a small, public set of software invariants that correspond to +claims in the CoAutoResearch system paper. They use synthetic artifacts and +synthetic passing reviewer outputs. They do **not** evaluate scientific quality, +autonomous-agent performance, or real reviewer judgments. + +Run from the repository root: + +```bash +python -m unittest discover -s tests/paper_contracts -p 'test_*.py' -v +``` + +The suite currently checks: + +- historical mentions do not become accepted evidence merely by appearing in canonical state; +- superseded evidence cannot silently return as active support; +- one accepted Result Card may be composed into multiple Research Lines; +- an execution-approved Plan must target a current actionable Critical Path item; +- a publication smoke path binds Publish Receipt, transaction, and canonical revision. diff --git a/tests/paper_contracts/runtime_smoke.py b/tests/paper_contracts/runtime_smoke.py new file mode 100644 index 0000000..9b5dddb --- /dev/null +++ b/tests/paper_contracts/runtime_smoke.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""Synthetic V2Runtime integration smoke checks. + +This exercises the real runtime stage/review/merge/transaction/publication path +using synthetic artifacts and synthetic passing reviewer outputs. It is not a +scientific evaluation and does not invoke a coding agent or UI. +""" +from __future__ import annotations +import argparse, hashlib, json, os, shutil, sys, tempfile +from copy import deepcopy +from pathlib import Path + +STAMP='2026-07-14T12:00:00Z' +PROJECT='coar-demo' +TRIAL='000001_test-main-signal' +STAGE='STAGE-000001-aaaaaaaa' + +def sha(b: bytes) -> str: + return hashlib.sha256(b).hexdigest() + +def run(repo: Path, *, invalid_cp: bool=False) -> dict: + template=repo/'templates/default' + sys.path.insert(0,str(template/'ui')) + os.environ['COAUTO_SCHEMA_DIR']=str(template/'schemas') + from v2_runtime import V2Runtime + from v2_artifacts import result_card_payload_hash + from v2_contracts import canonical_json_bytes + + def ex(name: str): + paths=list(template.rglob(name+'.example.json')) + if len(paths)!=1: + raise AssertionError(f'expected one {name}.example.json, got {paths}') + return json.loads(paths[0].read_text(encoding='utf-8')) + + with tempfile.TemporaryDirectory(prefix='coar-runtime-smoke-') as td, \ + tempfile.TemporaryDirectory(prefix='coar-plan-guard-') as initial_guard, \ + tempfile.TemporaryDirectory(prefix='coar-exec-guard-') as execution_guard, \ + tempfile.TemporaryDirectory(prefix='coar-review-guard-') as review_guard: + root=Path(td)/'project' + shutil.copytree(template,root) + rt=V2Runtime(root,PROJECT) + + state=ex('PROJECT_STATE') + state.update(canonical_revision=0,active_line_id='L0001', + critical_path=[{'id':'CP1','description':'Establish the main signal.','status':'open','source':'L0001/current_bottleneck'}], + goal_gate_path=None,next_step='Test the main signal.') + findings=ex('CURRENT_FINDINGS') + findings.update(canonical_revision=0,accepted_card_ids=[],qualified_card_ids=[],superseded_card_ids=[],synthesis=[]) + line=ex('LINE') + line.update(last_published_revision=0,supporting_cards=[],limiting_cards=[],conflicting_cards=[],material_exclusions=[],current_bottleneck='Establish the main signal.') + campaign=ex('CAMPAIGN') + campaign.update(last_published_revision=0,aggregate_status='in_progress',current_blockers=['Main signal not yet established.']) + campaign['components'][0].update(component_id='main_signal',title='Establish the main signal',status='not_started',satisfying_card_ids=[],rationale='Synthetic fixture.') + rt._write_pair('research_trajectory/STATE.json',state) + rt._write_pair('research_trajectory/CURRENT_FINDINGS.json',findings) + rt._write_pair('research_trajectory/lines/L0001.json',line) + rt._write_pair('research_trajectory/campaigns/C0001.json',campaign) + + trial=ex('TRIAL') + trial.update(base_revision=0,stage_id=None,lifecycle_state='planned',publish_revision=None,trial_outcome=None,closed_at=None) + rt._write_artifact(f'research_trajectory/trials/{TRIAL}/TRIAL.json',trial) + init=rt.initialize_trial(TRIAL,agent_guard_dir=initial_guard,stage_id=STAGE) + assert init['status']=='plan_ready',init + + plan=ex('PLAN') + plan['critical_path_target']='CP404' if invalid_cp else 'CP1' + plan['active_line_ids']=['L0001']; plan['campaign_ids']=['C0001'] + route=ex('EXPERT_ROUTE'); route['review_triggers']=[] + rt._write_pair(f'research_trajectory/trials/{TRIAL}/PLAN.json',plan) + rt._write_pair(f'research_trajectory/trials/{TRIAL}/EXPERT_ROUTE.json',route) + plan_bytes=(root/f'research_trajectory/trials/{TRIAL}/PLAN.json').read_bytes() + plan_review=ex('REVIEWER_OUTPUT') + plan_review.update(review_id='RV-000001-plan-01',reviewer='plan',scope='plan',phase='pre_execution',decision='pass', + summary='Synthetic plan-review smoke fixture.',stage_id=None,stage_manifest_hash=None,gate_effect='none', + instruction_file='instructions/reviewers/PLAN_REVIEWER.md', + reviewed_inputs=[f'research_trajectory/trials/{TRIAL}/PLAN.json',f'research_trajectory/trials/{TRIAL}/EXPERT_ROUTE.json'], + evidence_checked=[],strengths=[],blockers=[],required_actions=[],qualified_partial_passes=[],unassessed_areas=[],human_task_candidates=[],response_to_human=None) + plan_review['extensions']={'plan_binding':{'plan_revision':1,'plan_sha256':sha(plan_bytes)}} + rt._write_pair(f'research_trajectory/trials/{TRIAL}/reviews/PLAN_REVIEW.json',plan_review) + approval=rt.approve_plan(TRIAL,STAGE,plan_guard_dir=initial_guard,agent_guard_dir=execution_guard,created_at=STAMP) + if invalid_cp: + assert approval['status']=='plan_rejected',approval + assert any('critical_path_target' in e for e in approval.get('errors',[])),approval + assert not (root/f'research_trajectory/.staging/{TRIAL}/{STAGE}/PLAN_APPROVAL.json').exists() + return {'mode':'invalid_cp','status':'passed','approval_status':approval['status'],'errors':approval.get('errors',[])} + assert approval['status']=='execution_ready',approval + + evidence_rel=f'research_trajectory/trials/{TRIAL}/artifacts/results.json' + evidence_bytes=canonical_json_bytes({'synthetic_runtime_smoke':True,'scientific_result':None}) + evidence_path=root/evidence_rel; evidence_path.parent.mkdir(parents=True,exist_ok=True); evidence_path.write_bytes(evidence_bytes) + evidence_hash=sha(evidence_bytes) + report=ex('REPORT') + report.update(summary='Synthetic runtime publication smoke only.',work_performed=['Exercised V2Runtime publication path.'],procedures=[], + findings=['Synthetic software-path finding.'],negative_results=[],limitations=['Not scientific evidence.'],interpretation='Software path only.', + line_effects=[],campaign_effects=[{'campaign_id':'C0001','component_id':'main_signal','proposed_status':'tentative','rationale':'Synthetic fixture.'}], + venue_impact='No venue change.') + report['artifacts']=[{'path':evidence_rel,'role':'primary_result','sha256':evidence_hash,'extensions':{}}] + cards=ex('RESULT_CARDS'); card=cards['cards'][0] + card.update(type='process_lesson',claim_ids=[],effect='informs',summary='Synthetic software-path result.') + card['evidence'][0]['path']=evidence_rel; card['evidence'][0]['sha256']=evidence_hash + card['content_hash']=result_card_payload_hash(card) + request=ex('MERGE_REQUEST') + request.update(stage_id=STAGE,staged_update_manifest_path=f'research_trajectory/.staging/{TRIAL}/{STAGE}/STAGED_UPDATE_MANIFEST.json', + requested_line_updates=[],requested_campaign_updates=['C0001/main_signal -> tentative'],summary='Synthetic runtime smoke merge.') + rt._write_pair(f'research_trajectory/trials/{TRIAL}/REPORT.json',report) + rt._write_pair(f'research_trajectory/trials/{TRIAL}/RESULT_CARDS.json',cards) + rt._write_pair(f'research_trajectory/trials/{TRIAL}/MERGE_REQUEST.json',request) + + stage_root=f'research_trajectory/.staging/{TRIAL}/{STAGE}' + brief=ex('HUMAN_BRIEF') + brief['evidence'][0]['path']=evidence_rel; brief['evidence'][0]['sha256']=evidence_hash + brief.update(line_impact='No Research Line claim was changed by this synthetic smoke run.',campaign_impact='C0001 main_signal proposed tentative.',one_line_outcome='Synthetic runtime smoke produced one proposed result.') + gate=ex('GATE_EVIDENCE'); gate.update(final_ready=False,recommended_status='continue',reasons=['Synthetic smoke leaves real research incomplete.']) + rt._write_pair(f'{stage_root}/HUMAN_BRIEF.json',brief) + rt._write_pair(f'{stage_root}/GATE_EVIDENCE.json',gate) + + candidate_findings=deepcopy(findings) + candidate_findings.update(canonical_revision=1,qualified_card_ids=['RC-000001-01'], + synthesis=[{'finding_id':'F0001','statement':'Synthetic runtime smoke finding.','status':'accepted_with_qualification','card_ids':['RC-000001-01']}]) + candidate_campaign=deepcopy(campaign) + candidate_campaign['last_published_revision']=1 + candidate_campaign['components'][0].update(status='tentative',satisfying_card_ids=['RC-000001-01'],rationale='Synthetic smoke result.') + candidate_campaign.update(aggregate_status='tentative',current_blockers=['Replication remains missing.']) + candidate_root=f'{stage_root}/candidate' + rt._write_pair(f'{candidate_root}/research_trajectory/CURRENT_FINDINGS.json',candidate_findings) + rt._write_pair(f'{candidate_root}/research_trajectory/campaigns/C0001.json',candidate_campaign) + + route_inputs={ + 'target':'evaluation','line_effect':'no_change','new_or_changed_claim':False, + 'campaign_component_promoted_beyond_in_progress':True, + 'external_sources_or_new_citations':False,'target_venue_configured_or_venue_impact':False, + 'manuscript_or_claim_hierarchy_changed':False,'active_figure_or_table_changed':False, + 'central_line_effect_not_no_change':False,'campaign_component_marked_passed_or_waived':False, + 'target_venue_lock_change_requested':False,'gate_candidate_pass':False,'final_candidate':False, + 'high_risk_ethics_legal_human_subjects':False,'venue_impact':False, + 'agent_requested_level':'standard','enabled_registries':[]} + staged=rt.stage_trial(TRIAL,STAGE,agent_guard_dir=execution_guard,review_guard_dir=review_guard,route_inputs=route_inputs,created_at=STAMP) + assert staged['status']=='needs_review',staged + manifest=staged['review_manifest'] + for reviewer,path in staged['review_output_paths'].items(): + if reviewer=='plan': continue + review=ex('REVIEWER_OUTPUT') + review.update(reviewer=reviewer,scope=reviewer,review_id=f'RV-000001-{reviewer}-01',phase='post_stage',decision='pass', + summary=f'Synthetic {reviewer} runtime smoke review.',stage_id=STAGE,stage_manifest_hash=manifest['stage_manifest_hash'],gate_effect='none', + reviewed_inputs=[f'research_trajectory/trials/{TRIAL}/REPORT.json',f'research_trajectory/trials/{TRIAL}/RESULT_CARDS.json',f'{stage_root}/STAGED_UPDATE_MANIFEST.json'], + strengths=[],blockers=[],required_actions=[],qualified_partial_passes=[],unassessed_areas=[],human_task_candidates=[],response_to_human=None) + review['extensions']={'card_eligibility':[{'card_id':'RC-000001-01','requested_decision':'accept_with_qualification','eligible':True,'reason':'Synthetic exact-stage runtime smoke eligibility.'}]} + review['evidence_checked']=[{'path':evidence_rel,'sha256':evidence_hash,'source_kind':'artifact','scope':'synthetic software path','description':'Synthetic smoke artifact','extensions':{}}] + review['instruction_file']=f"instructions/reviewers/{'PROCESS_REVIEWER.md' if reviewer=='process' else 'EVIDENCE_REVIEWER.md'}" + rt._write_pair(path,review) + + completed=rt.complete_trial(TRIAL,STAGE,review_guard_dir=review_guard,gate_projection={},created_at=STAMP,publish=True) + assert completed['status']=='published' and completed.get('published') is True,completed + receipt=json.loads((root/f'research_trajectory/trials/{TRIAL}/PUBLISH_RECEIPT.json').read_text()) + revision=json.loads((root/'research_trajectory/CANONICAL_REVISION.json').read_text()) + final_findings=json.loads((root/'research_trajectory/CURRENT_FINDINGS.json').read_text()) + assert receipt['published_revision']==revision['revision']==1 + assert receipt['transaction_id']==revision['transaction_id'] + assert 'RC-000001-01' in final_findings['qualified_card_ids'] + return { + 'mode':'publication','status':'passed','approval_status':approval['status'], + 'stage_status':staged['status'],'completion_status':completed['status'], + 'published_revision':revision['revision'],'transaction_id':revision['transaction_id'], + 'gate_status':completed['goal_gate']['status'], + 'qualified_card_ids':final_findings['qualified_card_ids'], + 'published_paths':[x['path'] for x in receipt['published_files']], + } + +def main(): + ap=argparse.ArgumentParser() + ap.add_argument('--repo',type=Path,required=True) + ap.add_argument('--invalid-cp',action='store_true') + ap.add_argument('--output',type=Path) + args=ap.parse_args() + result=run(args.repo.resolve(),invalid_cp=args.invalid_cp) + text=json.dumps(result,indent=2,sort_keys=True)+'\n' + if args.output: + args.output.parent.mkdir(parents=True,exist_ok=True); args.output.write_text(text,encoding='utf-8') + print(text,end='') + +if __name__=='__main__': main() diff --git a/tests/paper_contracts/support.py b/tests/paper_contracts/support.py new file mode 100644 index 0000000..4dfc077 --- /dev/null +++ b/tests/paper_contracts/support.py @@ -0,0 +1,236 @@ +"""Synthetic fixtures for paper-facing contract tests. + +These fixtures exercise software invariants only. They are not scientific +results and do not represent real reviewer judgments. +""" +from __future__ import annotations + +from copy import deepcopy +import hashlib +import json +import os +from pathlib import Path +import sys + +REPO = Path(__file__).resolve().parents[2] +TEMPLATE = REPO / "templates/default" +os.environ["COAUTO_SCHEMA_DIR"] = str(TEMPLATE / "schemas") +sys.path.insert(0, str(TEMPLATE / "ui")) + +from v2_artifacts import result_card_payload_hash, validate_artifact +from v2_contracts import canonical_json_bytes +from v2_merge import MergeEvaluator +from v2_stage import build_stage_manifest + +FINDINGS = "research_trajectory/CURRENT_FINDINGS.json" +LINE1 = "research_trajectory/lines/L0001.json" +LINE2 = "research_trajectory/lines/L0002.json" +CARD = "RC-000001-01" +PRIOR = "RC-000000-01" +EXCLUDED = "RC-000000-02" +TRIAL = "000001_test-main-signal" +STAGE = "STAGE-000001-aaaaaaaa" +STAMP = "2026-07-14T12:00:00Z" +PROJECT = "coar-demo" + + +def example(name: str): + paths = list(TEMPLATE.rglob(name + ".example.json")) + if len(paths) != 1: + raise AssertionError(f"Expected one {name} fixture; got {paths}") + return json.loads(paths[0].read_text(encoding="utf-8")) + + +def digest(value) -> str: + raw = value if isinstance(value, bytes) else canonical_json_bytes(value) + return hashlib.sha256(raw).hexdigest() + + +def line(line_id="L0001", status="active", support=None, limit=None): + value = example("LINE") + value.update( + line_id=line_id, + status=status, + supporting_cards=list(support or []), + limiting_cards=list(limit or []), + conflicting_cards=[], + current_bottleneck="Synthetic fixture bottleneck; not an observed result.", + ) + return value + + +def findings(accepted=None, qualified=None, superseded=None, revision=None): + if revision is None: + revision = 1 if accepted or qualified or superseded else 0 + value = example("CURRENT_FINDINGS") + value.update( + canonical_revision=revision, + accepted_card_ids=list(accepted or []), + qualified_card_ids=list(qualified or []), + superseded_card_ids=list(superseded or []), + synthesis=[], + ) + return value + + +def deferred_cards(): + value = example("RESULT_CARDS") + value["cards"][0].update( + type="process_lesson", + summary="Synthetic current proposal; deferred, not accepted.", + ) + return value + + +def defer_request(): + return [{"card_id": CARD, "decision": "defer", "qualification": ""}] + + +def case(before=None, after=None, requests=None, cards=None): + before = deepcopy(before if before is not None else {FINDINGS: findings()}) + after = deepcopy( + after + if after is not None + else {FINDINGS: findings(qualified=[CARD], revision=1), LINE1: line(support=[CARD])} + ) + cards = deepcopy(cards if cards is not None else example("RESULT_CARDS")) + request = example("MERGE_REQUEST") + if requests is not None: + request["requested_card_decisions"] = deepcopy(requests) + base_revision = before.get(FINDINGS, {}).get("canonical_revision", 0) + request["base_revision"] = base_revision + if FINDINGS in after: + after[FINDINGS]["canonical_revision"] = base_revision + 1 + for value in after.values(): + if isinstance(value, dict) and value.get("artifact_type") == "line": + value["last_published_revision"] = base_revision + 1 + + candidate_files = {} + material_files = {} + prefix = f"research_trajectory/.staging/{TRIAL}/{STAGE}/candidate/" + for path, value in sorted(after.items()): + if path in before and before[path] == value: + continue + cp = prefix + path + candidate_files[cp] = canonical_json_bytes(value) + material_files[cp] = candidate_files[cp] + + trial_root = f"research_trajectory/trials/{TRIAL}" + stage_root = f"research_trajectory/.staging/{TRIAL}/{STAGE}" + trial = example("TRIAL") + trial.update(base_revision=base_revision, stage_id=STAGE) + plan = example("PLAN") + plan["primary_local_question"] = "Does this synthetic record satisfy the tested software contract?" + plan["active_line_ids"] = sorted( + { + value["line_id"] + for value in after.values() + if isinstance(value, dict) and value.get("artifact_type") == "line" + } + ) + report = example("REPORT") + report.update( + summary="Synthetic test input; no research was executed.", + work_performed=["Constructed software-test fixtures only."], + procedures=[], + findings=[], + negative_results=[], + limitations=["Not scientific evidence."], + line_effects=[], + campaign_effects=[], + ) + brief = example("HUMAN_BRIEF") + brief["one_line_outcome"] = "Synthetic fixture only; no real research outcome." + evidence_path = f"{trial_root}/artifacts/results.json" + evidence_bytes = canonical_json_bytes({"synthetic_fixture": True, "scientific_result": None}) + material_files[evidence_path] = evidence_bytes + + for artifact in (report, cards, brief): + def bind(value): + if isinstance(value, dict): + if value.get("path") == evidence_path and "sha256" in value: + value["sha256"] = digest(evidence_bytes) + for child in value.values(): + bind(child) + elif isinstance(value, list): + for child in value: + bind(child) + bind(artifact) + for card in cards.get("cards", []): + card["content_hash"] = result_card_payload_hash(card) + + for name, artifact in { + "TRIAL": trial, + "PLAN": plan, + "EXPERT_ROUTE": example("EXPERT_ROUTE"), + "REPORT": report, + "RESULT_CARDS": cards, + "MERGE_REQUEST": request, + }.items(): + material_files[f"{trial_root}/{name}.json"] = canonical_json_bytes(artifact) + for name, artifact in { + "HUMAN_BRIEF": brief, + "GATE_EVIDENCE": example("GATE_EVIDENCE"), + }.items(): + material_files[f"{stage_root}/{name}.json"] = canonical_json_bytes(artifact) + + stage = build_stage_manifest( + project_id=PROJECT, + trial_id=TRIAL, + stage_id=STAGE, + base_revision=base_revision, + files=material_files, + base_files=before, + created_at=STAMP, + updated_at=STAMP, + ) + manifest = example("REVIEW_MANIFEST") + manifest["stage_manifest_hash"] = stage["stage_content_hash"] + outputs = [] + for reviewer in manifest["required_reviewers"]: + value = example("REVIEWER_OUTPUT") + value.update( + reviewer=reviewer, + scope=reviewer, + review_id=f"RV-000001-{reviewer}-01", + summary="Synthetic passing review fixture. No scientific review occurred.", + ) + if reviewer == "plan": + value.update(phase="pre_execution", stage_id=None, stage_manifest_hash=None) + value["extensions"] = {} + value["reviewed_inputs"] = [f"{trial_root}/PLAN.json", f"{trial_root}/EXPERT_ROUTE.json"] + value["instruction_file"] = "instructions/reviewers/PLAN_REVIEWER.md" + else: + value["stage_manifest_hash"] = stage["stage_content_hash"] + value["extensions"]["card_eligibility"] = [ + { + "card_id": item["card_id"], + "requested_decision": item["decision"], + "eligible": True, + "reason": "Synthetic contract fixture only.", + } + for item in request["requested_card_decisions"] + ] + for item in value["evidence_checked"]: + if item.get("path") == evidence_path: + item["sha256"] = digest(evidence_bytes) + errors = validate_artifact(value, expected_type="reviewer_output") + assert not errors, errors + outputs.append(value) + + return dict( + stage_manifest=stage, + merge_request=request, + review_manifest=manifest, + reviewer_outputs=outputs, + result_cards=cards, + candidate_files=candidate_files, + base_projection=before, + created_at=STAMP, + ) + + +def evaluate(bundle, revision=None): + if revision is None: + revision = bundle["stage_manifest"]["base_revision"] + return MergeEvaluator(current_revision=revision).evaluate(**bundle) diff --git a/tests/paper_contracts/test_paper_contracts.py b/tests/paper_contracts/test_paper_contracts.py new file mode 100644 index 0000000..d87e6db --- /dev/null +++ b/tests/paper_contracts/test_paper_contracts.py @@ -0,0 +1,102 @@ +"""Public regression tests for claims made by the CoAutoResearch system paper. + +The tests are deliberately narrow and synthetic. They establish software +contracts, not scientific effectiveness. +""" +from __future__ import annotations + +from copy import deepcopy +from pathlib import Path +import unittest + +from support import ( + CARD, + EXCLUDED, + FINDINGS, + LINE1, + LINE2, + PRIOR, + case, + deferred_cards, + defer_request, + evaluate, + findings, + line, +) +from runtime_smoke import run as runtime_smoke + +REPO = Path(__file__).resolve().parents[2] + + +class EvidenceAdmissionContracts(unittest.TestCase): + def assert_rejected(self, bundle, fragment: str): + result = evaluate(bundle) + self.assertFalse(result["publishable"], result) + self.assertIn(fragment, "; ".join(result["errors"])) + + def test_exclusion_only_card_cannot_become_line_support(self): + before = {FINDINGS: findings(accepted=[PRIOR]), LINE1: line(support=[PRIOR])} + before[LINE1]["material_exclusions"] = [ + {"id": EXCLUDED, "reason": "Synthetic excluded history; never accepted."} + ] + after = deepcopy(before) + after[FINDINGS]["canonical_revision"] = 1 + after[LINE1]["supporting_cards"].append(EXCLUDED) + self.assert_rejected( + case(before, after, requests=defer_request(), cards=deferred_cards()), + "unknown/unaccepted", + ) + + def test_exclusion_only_card_cannot_become_accepted_finding(self): + before = {FINDINGS: findings(accepted=[PRIOR]), LINE1: line(support=[PRIOR])} + before[LINE1]["material_exclusions"] = [ + {"id": EXCLUDED, "reason": "Synthetic excluded history; never accepted."} + ] + after = deepcopy(before) + after[FINDINGS]["canonical_revision"] = 1 + after[FINDINGS]["accepted_card_ids"].append(EXCLUDED) + self.assert_rejected( + case(before, after, requests=defer_request(), cards=deferred_cards()), + "promote cards without an accepted current decision", + ) + + def test_superseded_history_cannot_return_as_active_support(self): + before = { + FINDINGS: findings(accepted=[PRIOR], superseded=[EXCLUDED]), + LINE1: line(support=[PRIOR]), + } + after = deepcopy(before) + after[FINDINGS]["canonical_revision"] = 1 + after[LINE1]["supporting_cards"].append(EXCLUDED) + self.assert_rejected( + case(before, after, requests=defer_request(), cards=deferred_cards()), + "unknown/unaccepted", + ) + + def test_old_accepted_card_can_support_multiple_research_lines(self): + before = {FINDINGS: findings(accepted=[PRIOR]), LINE1: line(support=[PRIOR])} + after = deepcopy(before) + after[FINDINGS]["canonical_revision"] = 1 + after[LINE2] = line("L0002", "candidate", support=[PRIOR]) + result = evaluate(case(before, after, requests=defer_request(), cards=deferred_cards())) + self.assertTrue(result["publishable"], result["errors"]) + self.assertIn(PRIOR, result["effective_projection"][LINE1]["supporting_cards"]) + self.assertIn(PRIOR, result["effective_projection"][LINE2]["supporting_cards"]) + + +class RuntimeContracts(unittest.TestCase): + def test_plan_must_target_current_critical_path_item(self): + result = runtime_smoke(REPO, invalid_cp=True) + self.assertEqual(result["status"], "passed") + self.assertEqual(result["approval_status"], "plan_rejected") + + def test_publication_receipt_and_revision_are_bound(self): + result = runtime_smoke(REPO, invalid_cp=False) + self.assertEqual(result["status"], "passed") + self.assertEqual(result["completion_status"], "published") + self.assertEqual(result["published_revision"], 1) + self.assertIn(CARD, result["qualified_card_ids"]) + + +if __name__ == "__main__": + unittest.main()