L1+L2 staged landing: 5 narrow slices (P0-P4)

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(

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
 
 

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)

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"):

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

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"]