diff --git a/cognitive_evolve_runtime/nexus/final_projection.py b/cognitive_evolve_runtime/nexus/final_projection.py index bd41624..e370d25 100644 --- a/cognitive_evolve_runtime/nexus/final_projection.py +++ b/cognitive_evolve_runtime/nexus/final_projection.py @@ -100,9 +100,10 @@ def build_final_projection(*, population: CandidatePopulation, synthesis: Any, g certificate.setdefault("blocking_reasons", []) if isinstance(certificate["blocking_reasons"], list) and selection.blocked_reason: certificate["blocking_reasons"].append(selection.blocked_reason) - best = _best_answer_candidate(population.candidates, contract=contract) or select_best_current_direction(population.candidates, contract=contract) - if best is not None: - return _projection_for_candidate(best, status="completed", objective_solved=False, certificate=certificate, answer_text="", contract=contract, graded_output=graded_output) + if not (answer_text and candidate is None): + best = _best_answer_candidate(population.candidates, contract=contract) or select_best_current_direction(population.candidates, contract=contract) + if best is not None: + return _projection_for_candidate(best, status="completed", objective_solved=False, certificate=certificate, answer_text="", contract=contract, graded_output=graded_output) if answer_text and candidate is None: return _projection_for_candidate(None, status="completed", objective_solved=False, certificate=certificate, answer_text=answer_text, contract=contract, graded_output=graded_output) return FinalProjection( diff --git a/cognitive_evolve_runtime/nexus/reproduction.py b/cognitive_evolve_runtime/nexus/reproduction.py index 9294f61..417f98f 100644 --- a/cognitive_evolve_runtime/nexus/reproduction.py +++ b/cognitive_evolve_runtime/nexus/reproduction.py @@ -80,8 +80,7 @@ def dedupe_offspring_against_population(offspring: list[CandidateGenome], popula unique.append(candidate) else: candidate.metadata["duplicate_offspring_reason"] = "duplicate_semantic_signature" - record_candidate_budget_decision(candidate, source="dedupe_offspring_against_population", reason="duplicate_semantic_signature", action="soft_retain") - unique.append(candidate) + record_candidate_budget_decision(candidate, source="dedupe_offspring_against_population", reason="duplicate_semantic_signature", action="hard_exclude") return unique diff --git a/cognitive_evolve_runtime/nexus/runtime.py b/cognitive_evolve_runtime/nexus/runtime.py index dc3c9b1..41650d1 100644 --- a/cognitive_evolve_runtime/nexus/runtime.py +++ b/cognitive_evolve_runtime/nexus/runtime.py @@ -35,6 +35,7 @@ from cognitive_evolve_runtime.nexus.runtime_options import option_bool, resolve_runtime_options, restore_runtime_options from cognitive_evolve_runtime.nexus.runtime_services import NexusPersistenceService, NexusProjectVerificationService from cognitive_evolve_runtime.nexus._shared import MODEL_BOUNDARY_ERRORS, positive_int +from cognitive_evolve_runtime.nexus.stop_reasons import normalize_external_review_stop_reason from cognitive_evolve_runtime.persistence.checkpoint import CheckpointStore, contract_payload_for_persistence @@ -152,9 +153,11 @@ def run_text( min_population_size = min_population_size if min_population_size is not None else budget.initial_candidate_count population = seed_population(contract=contract, world=world, policy=policy, model=self.model_routes.model_for(NexusModelRole.SEED), min_population_size=min_population_size) archives = ArchiveManager(policy.archive_schema) - verification_plan = VerificationSynthesizer(model=self.model).synthesize({"goal": goal, "contract": contract.to_dict()}) world_payload = _world_to_dict_with_latent_metadata(world, contract) observer = self._live_observer(mode="text", contract=contract, world=world_payload, max_rounds=budget.max_rounds, budget=budget.to_dict()) + if observer is not None: + observer({"phase": "post_seeding", "round": 0, "population": population, "archives": archives, "policy": policy, "progress_event": {"type": "nexus_post_seeding_checkpoint", "round": 0, "phase": "post_seeding", "max_rounds": budget.max_rounds}, "runtime_options": runtime_options}) + verification_plan = VerificationSynthesizer(model=self.model).synthesize({"goal": goal, "contract": contract.to_dict()}) result = evolve_once( population=population, archives=archives, @@ -248,6 +251,10 @@ def run_project( min_population_size=min_population_size, provided_context=initial_context_result.to_source_context(), ) + project_world_payload = _world_to_dict_with_latent_metadata({"snapshot": snapshot.to_dict(), "project_world_model": world.to_dict()}, contract) + observer = self._live_observer(mode="project", contract=contract, world=project_world_payload, max_rounds=budget.max_rounds, budget=budget.to_dict()) + if observer is not None: + observer({"phase": "post_seeding", "round": 0, "population": population, "archives": archives, "policy": policy, "progress_event": {"type": "nexus_post_seeding_checkpoint", "round": 0, "phase": "post_seeding", "max_rounds": budget.max_rounds}, "runtime_options": runtime_options}) verification_plan = VerificationSynthesizer(model=self.model).synthesize({"goal": user_goal, "contract": contract.to_dict(), "mode": "project"}) context_result = self.context_orchestrator.build_for_parents( contract=contract, @@ -278,8 +285,6 @@ def verify_offspring(candidates: list[Any]) -> list[ProjectVerificationSummary]: verification_summaries.extend(summaries) return summaries - project_world_payload = _world_to_dict_with_latent_metadata({"snapshot": snapshot.to_dict(), "project_world_model": world.to_dict()}, contract) - observer = self._live_observer(mode="project", contract=contract, world=project_world_payload, max_rounds=budget.max_rounds, budget=budget.to_dict()) result = evolve_once( population=population, archives=archives, @@ -377,6 +382,14 @@ def verify_offspring(candidates: list[Any]) -> list[ProjectVerificationSummary]: else: provided_context = dict(restored.get("provided_context") or {}) budget_data = dict(getattr(checkpoint, "budget", {}) or {}) + terminal_stop = normalize_external_review_stop_reason(budget_data.get("stop_reason")) + resume_does_not_extend = max_rounds is None or int(max_rounds) <= int(checkpoint.max_rounds or 0) + if terminal_stop and resume_does_not_extend: + run_result_path = self.output_dir / "run-result.json" + if not run_result_path.exists(): + raise FileNotFoundError(f"terminal checkpoint resume requires persisted run-result.json: {run_result_path}") + payload = json.loads(run_result_path.read_text(encoding="utf-8")) + return NexusRunResult(**payload) adaptive_resume = bool(budget_data.get("adaptive")) if max_rounds is not None: target_rounds = max(int(max_rounds), int(checkpoint.round or 0) + 1) diff --git a/cognitive_evolve_runtime/nexus/runtime_services.py b/cognitive_evolve_runtime/nexus/runtime_services.py index 845024c..9d37a43 100644 --- a/cognitive_evolve_runtime/nexus/runtime_services.py +++ b/cognitive_evolve_runtime/nexus/runtime_services.py @@ -15,6 +15,7 @@ from cognitive_evolve_runtime.nexus.consistency import assert_runtime_consistency from cognitive_evolve_runtime.nexus.loop import EvolutionBudget, EvolutionLoopResult from cognitive_evolve_runtime.nexus.final_projection import build_final_projection +from cognitive_evolve_runtime.nexus.nextgen import user_facing_verification_status from cognitive_evolve_runtime.nexus.project_verification import ProjectCandidateVerifier, ProjectVerificationSummary from cognitive_evolve_runtime.persistence.checkpoint import build_checkpoint_state from cognitive_evolve_runtime.nexus.seed_coverage import SEED_RESERVOIR_SIDECAR_PAYLOAD_KEY, persist_seed_reservoir_sidecar @@ -258,23 +259,40 @@ def _candidate_table_section(result: EvolutionLoopResult) -> str: rows = _candidate_rows(result, limit=12) if not rows: return "" - lines = ["", "", "## Candidate summary", "", "| rank | direction | rank score | target files | patch status |", "|---:|---|---:|---|---|"] - lines.extend(f"| {row['rank']} | {row['direction']} | {row['score']:.3f} | {row['target_files']} | {row['patch_status']} |" for row in rows) - return "\n".join(lines) + show_uncertainty = any(row["uncertainty"] for row in rows) + lines = ["", "", "## Candidate portfolio summary", ""] + groups: dict[str, list[dict[str, Any]]] = {} + for row in rows: + groups.setdefault(row["family_id"], []).append(row) + for family_id, family_rows in groups.items(): + lines.extend([f"### Mechanism family: {_clip_md(family_id, 120)}", ""]) + if show_uncertainty: + lines.extend(["| rank | direction | verification status | uncertainty |", "|---:|---|---|---|"]) + lines.extend(f"| {row['rank']} | {row['direction']} | {row['verification_status']} | {row['uncertainty'] or 'not recorded'} |" for row in family_rows) + else: + lines.extend(["| rank | direction | verification status |", "|---:|---|---|"]) + lines.extend(f"| {row['rank']} | {row['direction']} | {row['verification_status']} |" for row in family_rows) + lines.append("") + return "\n".join(lines[:-1]) def _candidate_rows(result: EvolutionLoopResult, *, limit: int) -> list[dict[str, Any]]: population = getattr(result, "population", None) candidates = list(getattr(population, "candidates", []) or []) ranked = sorted(candidates, key=lambda candidate: (_rank_score(candidate), str(getattr(candidate, "id", ""))), reverse=True)[:limit] + graded_output = _graded_output_from_result(result) + final_certificate = dict((getattr(getattr(result, "synthesis", None), "closure_certificate", {}) or {}).get("final_certificate") or {}) return [ { "rank": idx, "id": str(getattr(candidate, "id", "")), + "family_id": _candidate_family_id(candidate), "direction": _clip_md(str(getattr(candidate, "concise_claim", "") or getattr(candidate, "core_mechanism", "") or getattr(candidate, "artifact_type", "candidate")), 80), "score": _rank_score(candidate), "target_files": _clip_md(", ".join(_candidate_target_files(candidate)[:5]), 120), "patch_status": _patch_status(candidate), + "verification_status": _clip_md(user_facing_verification_status(candidate, final_certificate=final_certificate, graded_output=graded_output), 60), + "uncertainty": _candidate_uncertainty(candidate), } for idx, candidate in enumerate(ranked, start=1) ] @@ -284,6 +302,23 @@ def _candidate_row_markdown(row: dict[str, Any]) -> str: return f"| {row['rank']} | `{row['id']}` | {row['direction']} | {row['score']:.3f} | {row['target_files']} | {row['patch_status']} |" +def _candidate_family_id(candidate: Any) -> str: + metadata = getattr(candidate, "metadata", {}) + nextgen = metadata.get("nextgen") if isinstance(metadata, dict) and isinstance(metadata.get("nextgen"), dict) else {} + return str(nextgen.get("canonical_mechanism_family_id") or "unrecorded") + + +def _candidate_uncertainty(candidate: Any) -> str: + parts: list[str] = [] + missing = [str(value) for value in (getattr(candidate, "missing_parts", []) or [])[:1] if value] + notes = [str(value) for value in (getattr(candidate, "uncertainty_notes", []) or [])[:1] if value] + if missing: + parts.append(f"missing: {missing}") + if notes: + parts.append(f"notes: {notes}") + return _clip_md("; ".join(parts), 120) + + def _rank_score(candidate: Any) -> float: scores = getattr(candidate, "multihead_scores", {}) if isinstance(getattr(candidate, "multihead_scores", {}), dict) else {} for key in ("rank_score", "frontier_score", "objective_alignment", "answer_likelihood"): diff --git a/scripts/launch-core-self-evolve-openai-screen.sh b/scripts/launch-core-self-evolve-openai-screen.sh index fae982a..af401e2 100755 --- a/scripts/launch-core-self-evolve-openai-screen.sh +++ b/scripts/launch-core-self-evolve-openai-screen.sh @@ -38,8 +38,9 @@ Options: --branch-factor N default: 4 --initial-candidates N default: 16 --min-rounds-before-stop N default: 8 - --require-upstream-health require --upstream-health-url before model calls - --upstream-health-url URL operator-supplied upstream health endpoint + --require-upstream-health force upstream health gating even without a URL (fails fast) + --upstream-health-url URL operator-supplied upstream health endpoint; setting it + (or COGEV_UPSTREAM_HEALTH_URL) auto-enables health gating --dry-run print the launch plan without starting services/runner -h, --help show this help HELP @@ -110,6 +111,12 @@ fi if [[ "$INCLUDE_TESTS" == "true" ]]; then runner_args+=(--include-tests) fi +# A configured upstream health URL (flag or COGEV_UPSTREAM_HEALTH_URL) auto-enables +# gating; an empty URL never blocks the run. Explicit --require-upstream-health +# without a URL still fails fast below. +if [[ -n "$UPSTREAM_HEALTH_URL" ]]; then + REQUIRE_UPSTREAM_HEALTH="true" +fi if [[ "$REQUIRE_UPSTREAM_HEALTH" == "true" ]]; then runner_args+=(--require-upstream-health --upstream-health-url "$UPSTREAM_HEALTH_URL") fi diff --git a/tests/test_nextgen_cbt_pcbg_landing.py b/tests/test_nextgen_cbt_pcbg_landing.py index 0409966..2200d52 100644 --- a/tests/test_nextgen_cbt_pcbg_landing.py +++ b/tests/test_nextgen_cbt_pcbg_landing.py @@ -310,9 +310,38 @@ def test_final_projection_binds_synthesis_answer_to_best_current_candidate_id() assert projection.candidate_id == "chosen" assert projection.best_current_direction["candidate_id"] == "chosen" + assert projection.best_current_direction["blocked_from_verified_claim_reason"] != "answer_unbound_to_candidate_artifact" assert "answer_unbound_to_candidate_artifact" not in projection.advisory_issues +def test_final_projection_keeps_unbound_model_answer_unbound() -> None: + candidates = [ + CandidateGenome(id="candidate-a", artifact={"mechanism": "archive fallback"}, concise_claim="Archive fallback"), + CandidateGenome(id="candidate-b", artifact={"mechanism": "rank fallback"}, concise_claim="Rank fallback"), + ] + synthesis = SynthesizedResult( + status="model_synthesized", + final_answer="A free-text model answer that intentionally has no candidate binding.", + warnings=["model_final_answer_unbound_to_candidate_artifact"], + best_current_direction={"candidate_id": "", "route": "best_current"}, + ) + + projection = build_final_projection( + population=CandidatePopulation(candidates), + synthesis=synthesis, + graded_output=_graded_portfolio(), + ) + + assert projection.status == "completed" + assert projection.candidate_id == "" + assert projection.artifact == synthesis.final_answer + assert projection.best_current_direction["candidate_id"] == "" + assert projection.best_current_direction["candidate_id"] not in {candidate.id for candidate in candidates} + assert projection.best_current_direction["verification_status"] == "unverified" + assert projection.best_current_direction["blocked_from_verified_claim_reason"] == "answer_unbound_to_candidate_artifact" + assert "answer_unbound_to_candidate_artifact" in projection.advisory_issues + + def test_final_projection_unwraps_best_current_direction_carrier_to_real_direction() -> None: direction = CandidateGenome( id="direction", @@ -446,13 +475,14 @@ def test_seed_reservoir_soft_retains_low_relevance_and_duplicates() -> None: assert duplicate.metadata["candidate_budget_decision"]["action"] == "soft_reservoir" -def test_duplicate_offspring_is_soft_retained_with_budget_trace() -> None: +def test_duplicate_offspring_is_hard_excluded_with_budget_trace() -> None: parent = CandidateGenome(id="p", artifact="same", concise_claim="same", core_mechanism="same") child = CandidateGenome(id="c", parent_ids=["p"], artifact="same", concise_claim="same", core_mechanism="same") kept = dedupe_offspring_against_population([child], CandidatePopulation([parent])) - assert kept == [child] + assert kept == [] + assert child.metadata["candidate_budget_decision"]["action"] == "hard_exclude" assert child.metadata["candidate_budget_decision"]["reason"] == "duplicate_semantic_signature" assert "productive_child_observation" in child.metadata["nextgen"] diff --git a/tests/test_nexus_exploration_persistence.py b/tests/test_nexus_exploration_persistence.py index 00ed5fd..82243e5 100644 --- a/tests/test_nexus_exploration_persistence.py +++ b/tests/test_nexus_exploration_persistence.py @@ -4,10 +4,16 @@ from pathlib import Path from typing import Any -from cognitive_evolve_runtime.candidates.genome import CandidateGenome +import pytest + +from cognitive_evolve_runtime.archives.manager import ArchiveManager +from cognitive_evolve_runtime.candidates.genome import CandidateGenome, CandidatePopulation +from cognitive_evolve_runtime.contracts.objective_contract import NexusObjectiveContract from cognitive_evolve_runtime.engine.orchestrator import EngineOrchestrator from cognitive_evolve_runtime.llm.env import LLMResponseError -from cognitive_evolve_runtime.nexus.runtime import NexusRuntime +from cognitive_evolve_runtime.nexus.policy import EvolutionPolicy +from cognitive_evolve_runtime.nexus.runtime import NexusRunResult, NexusRuntime +from cognitive_evolve_runtime.persistence.checkpoint import CheckpointStore class NarrowSeedModel: @@ -74,6 +80,156 @@ def diagnose_search_state(self, **_: Any) -> dict[str, Any]: return super().diagnose_search_state(**_) +def _write_resume_fixture( + out: Path, + *, + stop_reason: str, + checkpoint_round: int = 1, + checkpoint_max_rounds: int = 2, + phase: str = "terminal", + write_run_result: bool = True, +) -> dict[str, Any]: + contract = NexusObjectiveContract(original_user_goal="resume goal", normalized_goal="resume goal") + policy = EvolutionPolicy() + world = {"kind": "text", "goal_summary": "resume goal"} + population = CandidatePopulation( + [ + CandidateGenome( + id="C0", + artifact="persisted answer", + concise_claim="persisted answer", + core_mechanism="fixture", + multihead_scores={"objective_alignment": 0.9, "answer_likelihood": 0.9}, + ) + ] + ) + CheckpointStore(out / "checkpoint.json").save_state( + round=checkpoint_round, + max_rounds=checkpoint_max_rounds, + population=population, + archives=ArchiveManager(), + policy=policy, + contract=contract, + world=world, + mode="text", + progress_event={"type": "evolution_progress", "round": checkpoint_round, "max_rounds": checkpoint_max_rounds, "phase": phase}, + budget={"current_round": checkpoint_round, "max_rounds": checkpoint_max_rounds, "stop_reason": stop_reason}, + verification_plan={"verifier_id": "noop", "strength": "NONE", "modality": "none", "verifier_fingerprint": "fixture"}, + ) + payload = NexusRunResult( + mode="text", + contract=contract.to_dict(), + policy=policy.to_dict(), + world=world, + evolution={"synthesis": {"final_answer": "persisted answer"}, "stop_reason": stop_reason, "persisted": True}, + artifacts={"run_result": str(out / "run-result.json")}, + ).to_dict() + if write_run_result: + (out / "run-result.json").write_text(json.dumps(payload), encoding="utf-8") + return payload + + +def test_post_seeding_checkpoint_survives_verification_synthesizer_failure(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + def fail_synthesize(self: object, *_args: Any, **_kwargs: Any) -> object: + raise LLMResponseError("verification synthesize failed after seed") + + monkeypatch.setattr("cognitive_evolve_runtime.nexus.runtime.VerificationSynthesizer.synthesize", fail_synthesize) + repo = tmp_path / "repo" + repo.mkdir() + (repo / "mod.py").write_text("def value():\n return 1\n", encoding="utf-8") + + cases = [ + ("text", tmp_path / "text", lambda out: NexusRuntime(model=NarrowSeedModel(), output_dir=out).run_text("seed then fail", max_rounds=4, min_population_size=3)), + ("project", tmp_path / "project", lambda out: NexusRuntime(output_dir=out).run_project(repo, user_goal="seed then fail", max_rounds=4, min_population_size=3)), + ] + for mode, out, run in cases: + with pytest.raises(LLMResponseError, match="verification synthesize failed after seed"): + run(out) + + checkpoint_path = out / "checkpoint.json" + checkpoint = json.loads(checkpoint_path.read_text(encoding="utf-8")) + restored = CheckpointStore(checkpoint_path).restore_state() + assert restored is not None + assert checkpoint["mode"] == mode + assert checkpoint["round"] == 0 + assert checkpoint["progress_event"]["phase"] == "post_seeding" + assert checkpoint["budget"]["current_round"] == 0 + assert checkpoint["policy"]["policy_id"] == "nexus-evolution-policy" + assert checkpoint["archives"]["archive_schema"] + assert len(checkpoint["population"]["candidates"]) >= 3 + assert len(restored["population"].candidates) >= 3 + + +def test_terminal_resume_reuses_persisted_run_result_without_evolving(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + calls: list[bool] = [] + payload = _write_resume_fixture(tmp_path, stop_reason="candidate_ready_for_external_review") + + def fail_evolve_once(**_: Any) -> None: + calls.append(True) + pytest.fail("terminal resume should not call evolve_once") + + monkeypatch.setattr("cognitive_evolve_runtime.nexus.runtime.evolve_once", fail_evolve_once) + + resumed = NexusRuntime(output_dir=tmp_path).resume_from_checkpoint(max_rounds=2) + + assert calls == [] + assert resumed.to_dict() == payload + + +def test_terminal_resume_extending_rounds_enters_evolve_once(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + calls: list[int] = [] + _write_resume_fixture(tmp_path, stop_reason="candidate_ready_for_external_review") + + class ReachedEvolveOnce(Exception): + pass + + def stop_at_evolve_once(**kwargs: Any) -> None: + calls.append(kwargs["budget"].max_rounds) + raise ReachedEvolveOnce + + monkeypatch.setattr("cognitive_evolve_runtime.nexus.runtime.evolve_once", stop_at_evolve_once) + + with pytest.raises(ReachedEvolveOnce): + NexusRuntime(output_dir=tmp_path).resume_from_checkpoint(max_rounds=3) + + assert calls == [3] + + +def test_post_seeding_resume_does_not_use_terminal_short_circuit(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + calls: list[int] = [] + _write_resume_fixture(tmp_path, stop_reason="", checkpoint_round=0, phase="post_seeding") + + class ReachedEvolveOnce(Exception): + pass + + def stop_at_evolve_once(**kwargs: Any) -> None: + calls.append(kwargs["budget"].current_round) + raise ReachedEvolveOnce + + monkeypatch.setattr("cognitive_evolve_runtime.nexus.runtime.evolve_once", stop_at_evolve_once) + + with pytest.raises(ReachedEvolveOnce): + NexusRuntime(output_dir=tmp_path).resume_from_checkpoint(max_rounds=2) + + assert calls == [0] + + +def test_terminal_resume_requires_persisted_run_result(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + calls: list[bool] = [] + _write_resume_fixture(tmp_path, stop_reason="candidate_ready_for_external_review", write_run_result=False) + + def fail_evolve_once(**_: Any) -> None: + calls.append(True) + pytest.fail("terminal resume should fail closed before evolve_once") + + monkeypatch.setattr("cognitive_evolve_runtime.nexus.runtime.evolve_once", fail_evolve_once) + + with pytest.raises(FileNotFoundError, match="terminal checkpoint resume requires persisted run-result.json"): + NexusRuntime(output_dir=tmp_path).resume_from_checkpoint() + + assert calls == [] + + def test_nexus_amplifies_narrow_model_seed_pool_and_marks_search_seeds(tmp_path: Path) -> None: result = NexusRuntime(model=NarrowSeedModel(), output_dir=tmp_path).run_text( "Solve a hard math problem.", diff --git a/tests/test_nexus_single_runtime_purity.py b/tests/test_nexus_single_runtime_purity.py index cb2e9b9..eb46c8c 100644 --- a/tests/test_nexus_single_runtime_purity.py +++ b/tests/test_nexus_single_runtime_purity.py @@ -232,9 +232,99 @@ def test_final_answer_artifact_includes_human_candidate_table() -> None: text = final_answer_artifact_text(result) candidates = candidates_markdown(result) - assert "## Candidate summary" in text - assert "pkg/parser.py" in text + assert "## Candidate portfolio summary" in text + assert "update parser" in text + assert "| rank | candidate | direction | rank score | target files | patch status |" in candidates + assert "pkg/parser.py" in candidates assert "| 1 | `C1` | update parser" in candidates + assert "Mechanism family" not in candidates + + +def test_final_answer_candidate_portfolio_groups_by_canonical_family() -> None: + from cognitive_evolve_runtime.candidates.genome import CandidateGenome, CandidatePopulation + from cognitive_evolve_runtime.nexus.runtime_services import final_answer_artifact_text + + alpha = CandidateGenome( + id="A1", + concise_claim="alpha direction", + missing_parts=["alpha gap", "second gap"], + uncertainty_notes=["alpha uncertainty", "second note"], + metadata={"nextgen": {"canonical_mechanism_family_id": "alpha#m1"}}, + multihead_scores={"rank_score": 0.9}, + ) + beta = CandidateGenome( + id="B1", + concise_claim="beta direction", + metadata={"nextgen": {"canonical_mechanism_family_id": "beta#m2"}}, + multihead_scores={"rank_score": 0.8}, + ) + result = SimpleNamespace( + completion_status="completed", + stop_reason="max_rounds", + synthesis=SimpleNamespace(best_candidate_id="A1", final_answer="\nAnswer body", closure_certificate={}), + population=CandidatePopulation([alpha, beta]), + graded_output={}, + ) + + text = final_answer_artifact_text(result) + + assert "### Mechanism family: alpha#m1" in text + assert "### Mechanism family: beta#m2" in text + assert "| rank | direction | verification status | uncertainty |" in text + assert "missing: ['alpha gap']; notes: ['alpha uncertainty']" in text + assert "second gap" not in text + + +def test_final_answer_candidate_portfolio_missing_canonical_uses_unrecorded() -> None: + from cognitive_evolve_runtime.candidates.genome import CandidateGenome, CandidatePopulation + from cognitive_evolve_runtime.nexus.runtime_services import final_answer_artifact_text + + result = SimpleNamespace( + completion_status="completed", + stop_reason="max_rounds", + synthesis=SimpleNamespace(best_candidate_id="C1", final_answer="\nAnswer body", closure_certificate={}), + population=CandidatePopulation( + [ + CandidateGenome(id="C1", concise_claim="first", multihead_scores={"rank_score": 0.9}), + CandidateGenome(id="C2", concise_claim="second", multihead_scores={"rank_score": 0.8}), + ] + ), + graded_output={}, + ) + + text = final_answer_artifact_text(result) + + assert text.count("### Mechanism family: unrecorded") == 1 + assert "### Mechanism family: C1" not in text + assert "### Mechanism family: C2" not in text + + +def test_final_answer_candidate_status_is_answer_first_not_local_verified() -> None: + from cognitive_evolve_runtime.candidates.genome import CandidateGenome, CandidatePopulation + from cognitive_evolve_runtime.nexus.runtime_services import final_answer_artifact_text + + result = SimpleNamespace( + completion_status="completed", + stop_reason="max_rounds", + synthesis=SimpleNamespace(best_candidate_id="C1", final_answer="\nAnswer body", closure_certificate={}), + population=CandidatePopulation( + [ + CandidateGenome( + id="C1", + concise_claim="local verified only", + verification_result={"passed": True}, + multihead_scores={"rank_score": 0.9}, + ) + ] + ), + graded_output={}, + ) + + text = final_answer_artifact_text(result) + + assert "correctness_verdict: external_validation_required" in text + assert "project_correctness_claim: not_claimed" in text + assert "| 1 | local verified only | advisory |" in text def test_final_answer_artifact_defers_correctness_to_external_review() -> None: