Add phase lifecycle callbacks and DNS zone file validation tests

netengine/cli/main.py

@@ -348,8 +348,41 @@ async def _up(
 
     orchestrator = Orchestrator(spec, mock_mode=mock)
     click.echo(f"Booting world from {spec_file} (phases 0–{up_to})…")
+
+    import time
+
+    _phase_start_times: dict[int, float] = {}
+
+    def _on_start(phase_num: int, phase_name: str) -> None:
+        _phase_start_times[phase_num] = time.monotonic()
+        click.echo(f"  ⧗  Phase {phase_num}: {phase_name}…")
+
+    def _on_complete(phase_num: int, phase_name: str) -> None:
+        elapsed = time.monotonic() - _phase_start_times.get(phase_num, time.monotonic())
+        click.echo(
+            click.style(f"  ✓  Phase {phase_num}: {phase_name}", fg="green")
+            + click.style(f"  ({elapsed:.1f}s)", fg="bright_black")
+        )
+
+    def _on_skip(phase_num: int, phase_name: str) -> None:
+        click.echo(
+            click.style(f"  –  Phase {phase_num}: {phase_name} (already done)", fg="bright_black")
+        )
+
+    def _on_error(phase_num: int, phase_name: str, exc: Exception) -> None:
+        click.echo(
+            click.style(f"  ✗  Phase {phase_num}: {phase_name} — {exc}", fg="red", bold=True),
+            err=True,
+        )
+
     try:
-        await orchestrator.execute_phases(up_to_phase=up_to)
+        await orchestrator.execute_phases(
+            up_to_phase=up_to,
+            on_phase_start=_on_start,
+            on_phase_complete=_on_complete,
+            on_phase_skip=_on_skip,
+            on_phase_error=_on_error,
+        )
     except Exception as exc:
         click.echo(f"Bootstrap failed: {exc}", err=True)
         sys.exit(1)

netengine/core/orchestrator.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 from pydantic import ValidationError
 
@@ -111,15 +111,28 @@ def _check_prerequisites(self, phase_num: int) -> None:
                 "Run earlier phases first."
             )
 
-    async def execute_phases(self, up_to_phase: int = 9) -> None:
+    async def execute_phases(
+        self,
+        up_to_phase: int = 9,
+        on_phase_start: Optional[Callable[[int, str], None]] = None,
+        on_phase_complete: Optional[Callable[[int, str], None]] = None,
+        on_phase_skip: Optional[Callable[[int, str], None]] = None,
+        on_phase_error: Optional[Callable[[int, str, Exception], None]] = None,
+    ) -> None:
         """Execute phases 0 through up_to_phase.
 
         Args:
             up_to_phase: Highest phase number to execute (default 9, all phases)
+            on_phase_start: Optional callback(phase_num, phase_name) called before each phase
+            on_phase_complete: Optional callback(phase_num, phase_name) called on success
+            on_phase_skip: Optional callback(phase_num, phase_name) called when phase is skipped
+            on_phase_error: Optional callback(phase_num, phase_name, exc) called on failure
 
         Raises:
             RuntimeError: If any phase fails or dependency validation fails
         """
+        from netengine.phase_labels import PHASE_LABELS
+
         if not self.runtime_state.world_spec:
             self.runtime_state.world_spec = self.spec.model_dump()
             self.runtime_state.save()
@@ -128,18 +141,23 @@ async def execute_phases(self, up_to_phase: int = 9) -> None:
             if phase_num > up_to_phase:
                 break
 
+            phase_name = PHASE_LABELS.get(str(phase_num), handler_class.__name__)
             handler = handler_class()
 
             if await handler.should_skip(self.context):
                 logger.info(
                     f"Phase {phase_num}: {handler_class.__name__} already completed, skipping"
                 )
                 self._mark_phase_complete(phase_num, handler)
+                if on_phase_skip:
+                    on_phase_skip(phase_num, phase_name)
                 continue
 
             self._check_prerequisites(phase_num)
 
             logger.info(f"Phase {phase_num}: {handler_class.__name__} starting")
+            if on_phase_start:
+                on_phase_start(phase_num, phase_name)
             try:
                 with record_phase(phase_num):
                     await handler.execute(self.context)
@@ -152,11 +170,15 @@ async def execute_phases(self, up_to_phase: int = 9) -> None:
                 self.runtime_state.save()
                 self.runtime_state.sync_to_supabase()
                 logger.info(f"Phase {phase_num} completed successfully")
+                if on_phase_complete:
+                    on_phase_complete(phase_num, phase_name)
 
             except Exception as e:
                 logger.error(f"Phase {phase_num} failed: {e}")
                 self.runtime_state.last_error = str(e)
                 self.runtime_state.save()
+                if on_phase_error:
+                    on_phase_error(phase_num, phase_name, e)
                 raise
 
     async def start_consumers(self) -> None:

tests/integration/test_e2e_bootstrap.py

@@ -452,3 +452,90 @@ def test_dry_run_exits_zero_even_with_nothing_to_remove(self, tmp_path, monkeypa
 
         assert result.exit_code == 0
         assert "would be removed" in result.output
+
+
+# ─────────────────────────────────────────────
+# Full MVP lifecycle
+# ─────────────────────────────────────────────
+
+
+class TestFullMVPLifecycle:
+    """Single end-to-end test covering the full dev-sandbox lifecycle:
+    bootstrap → API health → idempotent re-run → reload with new org.
+    """
+
+    @pytest.mark.asyncio
+    async def test_full_mvp_lifecycle(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("NETENGINE_STATE_FILE", str(tmp_path / "state.json"))
+        monkeypatch.setenv("NETENGINES_BOOTSTRAP_SECRET", "mvp-secret")
+
+        spec = load_spec(EXAMPLES_DIR / "minimal.yaml")
+
+        # ── Step 1: Full mock bootstrap, all 9 phases ──────────────────────
+        phase_events: list[tuple[str, int]] = []
+
+        def _on_start(n: int, name: str) -> None:
+            phase_events.append(("start", n))
+
+        def _on_complete(n: int, name: str) -> None:
+            phase_events.append(("complete", n))
+
+        def _on_skip(n: int, name: str) -> None:
+            phase_events.append(("skip", n))
+
+        with _all_phase_mocks():
+            orchestrator = Orchestrator(spec, mock_mode=True)
+            await orchestrator.execute_phases(
+                up_to_phase=9,
+                on_phase_start=_on_start,
+                on_phase_complete=_on_complete,
+                on_phase_skip=_on_skip,
+            )
+
+        state = orchestrator.runtime_state
+        for phase in range(10):
+            assert state.phase_completed.get(
+                str(phase)
+            ), f"Phase {phase} not complete after MVP bootstrap"
+
+        # DNSHandler covers phases 1+2 in one handler, so 9 handler invocations total
+        started = [n for event, n in phase_events if event == "start"]
+        completed = [n for event, n in phase_events if event == "complete"]
+        assert len(started) == 9, f"Expected 9 on_phase_start calls, got {len(started)}"
+        assert len(completed) == 9, f"Expected 9 on_phase_complete calls, got {len(completed)}"
+
+        # ── Step 2: World spec is recorded in state ────────────────────────
+        assert state.world_spec is not None
+        assert state.world_spec["metadata"]["name"] == spec.metadata.name
+
+        # ── Step 3: Reload adds a new org without full re-bootstrap ─────────
+        new_dict = spec.model_dump(mode="json")
+        new_dict["world_registry"]["organizations"].append(
+            {
+                "name": "mvp-corp",
+                "description": "Added in full lifecycle test",
+                "capabilities": ["host_services"],
+                "and_profile": "business",
+            }
+        )
+
+        from netengine.api.app import app
+
+        api_client = TestClient(app)
+        with patch(
+            "netengine.phases.phase_registries.RegistriesPhaseHandler.execute",
+            new_callable=AsyncMock,
+        ):
+            with patch(
+                "netengine.phases.phase_inworld_identity.InWorldIdentityPhaseHandler.execute",
+                new_callable=AsyncMock,
+            ):
+                resp = api_client.post(
+                    "/api/v1/reload",
+                    json={"spec_yaml": yaml.dump(new_dict)},
+                    headers={"X-Bootstrap-Secret": "mvp-secret"},
+                )
+
+        assert resp.status_code == 200
+        result = resp.json()
+        assert result["success"] is True, f"Reload failed: {result}"

tests/test_cli.py

@@ -49,7 +49,8 @@ def test_up_invokes_execute_phases_with_example_spec():
     mock_orchestrator_class.assert_called_once()
     spec_arg = mock_orchestrator_class.call_args.args[0]
     assert spec_arg.metadata.name == "minimal-example"
-    mock_orchestrator.execute_phases.assert_awaited_once_with(up_to_phase=9)
+    assert mock_orchestrator.execute_phases.await_count == 1
+    assert mock_orchestrator.execute_phases.call_args.kwargs["up_to_phase"] == 9
 
 
 def test_up_migration_failure_prevents_orchestrator_startup():
@@ -92,7 +93,8 @@ def test_up_allows_migration_failure_with_explicit_flag():
     assert result.exit_code == 0, result.output
     mock_migrations.assert_awaited_once_with("postgresql://example/db")
     mock_orchestrator_class.assert_called_once()
-    mock_orchestrator.execute_phases.assert_awaited_once_with(up_to_phase=9)
+    assert mock_orchestrator.execute_phases.await_count == 1
+    assert mock_orchestrator.execute_phases.call_args.kwargs["up_to_phase"] == 9
 
 
 def test_up_skip_migrations_bypasses_migration_execution():
@@ -113,7 +115,8 @@ def test_up_skip_migrations_bypasses_migration_execution():
     assert result.exit_code == 0, result.output
     mock_migrations.assert_not_awaited()
     mock_orchestrator_class.assert_called_once()
-    mock_orchestrator.execute_phases.assert_awaited_once_with(up_to_phase=9)
+    assert mock_orchestrator.execute_phases.await_count == 1
+    assert mock_orchestrator.execute_phases.call_args.kwargs["up_to_phase"] == 9
 
 
 def test_status_output_includes_phase_9():
@@ -148,7 +151,8 @@ def test_up_supports_environment_loader_option():
     mock_loader.assert_called_once_with(str(spec_file), environment="dev", overrides=None)
     spec_arg = mock_orchestrator_class.call_args.args[0]
     assert spec_arg.metadata.name == "minimal-example"
-    mock_orchestrator.execute_phases.assert_awaited_once_with(up_to_phase=9)
+    assert mock_orchestrator.execute_phases.await_count == 1
+    assert mock_orchestrator.execute_phases.call_args.kwargs["up_to_phase"] == 9
 
 
 def test_init_creates_spec_file(tmp_path: Path) -> None:
@@ -362,7 +366,8 @@ def test_up_supports_repeatable_set_overrides():
     spec_arg = mock_orchestrator_class.call_args.args[0]
     assert spec_arg.metadata.name == "my-world"
     assert spec_arg.world_services.mail.enabled is True
-    mock_orchestrator.execute_phases.assert_awaited_once_with(up_to_phase=9)
+    assert mock_orchestrator.execute_phases.await_count == 1
+    assert mock_orchestrator.execute_phases.call_args.kwargs["up_to_phase"] == 9
 
 
 def _write_cli_validate_spec(tmp_path: Path, **updates) -> Path:

tests/test_m1_handlers.py

@@ -204,6 +204,79 @@ async def test_should_skip_false_before_execute(self, phase_context: PhaseContex
 
         assert skip is False
 
+    async def test_root_zone_file_has_ns_delegation_per_tld(
+        self, phase_context: PhaseContext
+    ) -> None:
+        """Root zone file must include an NS delegation record for each TLD in the spec."""
+        handler = DNSHandler()
+        await handler.execute(phase_context)
+
+        root_zone_file = phase_context.runtime_state.dns_output["zone_files"]["root.internal"]
+        for tld in phase_context.spec.dns.tlds:
+            assert (
+                f"{tld.name}. NS" in root_zone_file
+            ), f"Root zone file missing NS delegation for TLD '{tld.name}'"
+
+    async def test_root_zone_file_has_glue_record_per_tld(
+        self, phase_context: PhaseContext
+    ) -> None:
+        """Root zone file must include a glue A record for each TLD nameserver."""
+        handler = DNSHandler()
+        await handler.execute(phase_context)
+
+        root_zone_file = phase_context.runtime_state.dns_output["zone_files"]["root.internal"]
+        for tld in phase_context.spec.dns.tlds:
+            tld_config = phase_context.runtime_state.dns_output["tlds"][tld.name]
+            assert tld_config["listen_ip"] in root_zone_file, (
+                f"Root zone file missing glue A record for TLD '{tld.name}' "
+                f"(expected IP {tld_config['listen_ip']})"
+            )
+
+    async def test_root_zone_file_has_platform_delegation(
+        self, phase_context: PhaseContext
+    ) -> None:
+        """Root zone file must delegate platform.internal to the platform nameserver."""
+        handler = DNSHandler()
+        await handler.execute(phase_context)
+
+        root_zone_file = phase_context.runtime_state.dns_output["zone_files"]["root.internal"]
+        assert (
+            "platform.internal. 3600 IN NS" in root_zone_file
+        ), "Root zone file missing NS delegation for platform.internal"
+        platform_ip = phase_context.runtime_state.dns_output["platform_zone"]["listen_ip"]
+        assert (
+            platform_ip in root_zone_file
+        ), "Root zone file missing platform.internal glue A record"
+
+    async def test_tld_output_has_listen_ip_per_tld(self, phase_context: PhaseContext) -> None:
+        """Each TLD in dns_output must have a listen_ip matching its spec config."""
+        handler = DNSHandler()
+        await handler.execute(phase_context)
+
+        tlds_output = phase_context.runtime_state.dns_output["tlds"]
+        for tld in phase_context.spec.dns.tlds:
+            assert tld.name in tlds_output, f"TLD '{tld.name}' missing from dns_output['tlds']"
+            assert tlds_output[tld.name]["listen_ip"] == tld.listen_ip, (
+                f"TLD '{tld.name}' listen_ip mismatch: "
+                f"expected {tld.listen_ip}, got {tlds_output[tld.name]['listen_ip']}"
+            )
+
+    async def test_tld_zone_files_have_correct_soa_and_ns(
+        self, phase_context: PhaseContext
+    ) -> None:
+        """Each TLD must have its own zone file with SOA and NS records."""
+        handler = DNSHandler()
+        await handler.execute(phase_context)
+
+        zone_files = phase_context.runtime_state.dns_output["zone_files"]
+        for tld in phase_context.spec.dns.tlds:
+            assert tld.name in zone_files, f"Zone file missing for TLD '{tld.name}'"
+            content = zone_files[tld.name]
+            assert "SOA" in content, f"TLD zone '{tld.name}' missing SOA record"
+            assert (
+                f"{tld.name}. 3600 IN NS" in content
+            ), f"TLD zone '{tld.name}' missing self NS record"
+
 
 class TestSubstrateAndDNSIntegration:
     """Integration tests for M1: Substrate → DNS execution order."""