refactor(render): scheme-agnostic build_manifest + schema v6 [LTV-Pn.1]

.agent-plan.md

@@ -68,10 +68,14 @@ lead-scoring byte-identical) merged (#119). `LTV-Pm` (early-pLTV
 tenure-anchored snapshot — `build_early_pltv_snapshot()` with a per-customer
 relative cutoff at `customer_start + early_tenure_weeks`; calendar + early
 builders unified on one per-customer-cutoff core; 19 tests) opened as
-**#120** — **LTV-M5 complete** (both observation regimes). Next: `LTV-M6`
-(`LTV-Pn` — register LifecycleScheme + recipe + manifest/schema-v6, fold in
-the deferred task-split writer for both regimes + the carried layering
-cleanups).
+**#120** — **LTV-M5 complete** (both observation regimes). **LTV-M6** (split
+`LTV-Pn` into four sub-PRs): `LTV-Pn.1` (scheme-agnostic `build_manifest` —
+drops the lead-scoring `world_graph` param for `generation_scheme` /
+`motif_family` / `extra_fields`; every manifest records `generation_scheme`;
+`BUNDLE_SCHEMA_VERSION` 5 → 6; lead-scoring data files byte-identical) opened
+as **#121**. Next: `LTV-Pn.2` (scheme-agnostic `WorldBundle` + exposure hook +
+shared bundle orchestrator), then `Pn.3` (lifecycle config + regression task
+model), `Pn.4` (complete `LifecycleScheme` + e2e bundle), `LTV-Po` (recipe).
 
 ---
 

docs/ltv/roadmap.md

@@ -46,7 +46,7 @@ protocol + registry, with the package physically reorganized into
 | `LTV-M3` | Customer population + lifecycle world | `LTV-Ph`, `LTV-Pi` | #113 (Ph) |
 | `LTV-M4` | Lifecycle simulation engine | `LTV-Pj`, `LTV-Pk` | #117 (Pj), #118 (Pk) |
 | `LTV-M5` | Customer snapshots + pLTV targets (both regimes) | `LTV-Pl`, `LTV-Pm` | #119 (Pl), #120 (Pm) |
-| `LTV-M6` | Register LifecycleScheme + recipe + manifest/version | `LTV-Pn`, `LTV-Po` | |
+| `LTV-M6` | Register LifecycleScheme + recipe + manifest/version | `LTV-Pn.1…4`, `LTV-Po` | #121 (Pn.1) |
 | `LTV-M7` | Validation + regression-metric calibration | `LTV-Pp` | |
 | `LTV-M8` | CLI, notebooks, publish | `LTV-Pq`, `LTV-Pr`, `LTV-Ps` | |
 
@@ -268,23 +268,43 @@ Total: ~19 PRs across 9 milestones.
 
 ## `LTV-M6` — Register LifecycleScheme + recipe + manifest/version
 
-- [ ] **`LTV-Pn`** — `feat(lifecycle): complete LifecycleScheme + manifest/version`.
-  Fill in the `LifecycleScheme` pipeline methods (population→sim→render→tasks);
-  add `n_customers` + lifecycle config (windows, early-tenure, observation
-  anchor) to `GenerationConfig`; record `generation_scheme` + `observation_date`
-  + windows in the manifest; bump `BUNDLE_SCHEMA_VERSION` 5 → 6 (D5); teach the
-  task-split writer the continuous-target path. Extend `CLAUDE.md` hard
-  constraints with the lifecycle snapshot-safety clause + the schemes/ layout.
-  - **Layering cleanup (carried debt, see `Known deferred cleanups` below):**
-    generalise `build_manifest` (drop the lead-scoring `world_graph` param) and
-    `apply_exposure` (stop hard-coding the lead-scoring hidden graph + latent
-    registry) so they are scheme-agnostic; with that done, remove the
-    `core.models` / `render.relational` **TYPE_CHECKING** back-references to
-    `leadforge.schemes.lead_scoring.*` introduced in `LTV-Pf.1` (a core→scheme
-    layering inversion), and lift the shared render orchestration out of each
-    scheme's `write_bundle` (the decomposition deferred in `LTV-Pe`).
-  - Tests: dispatch, lead-scoring path unaffected, manifest fields, regression
-    split writer, exposure filtering for new tables.
+`LTV-Pn` was too large for one reviewable, byte-identity-guarded PR (envelope
+generalization + 3 carried cleanups + config + task model + the lifecycle
+pipeline + schema bump).  Split into four sub-PRs in dependency order:
+
+- [x] **`LTV-Pn.1`** — `refactor(render): scheme-agnostic build_manifest +
+  schema v6` (**PR #121**).  `build_manifest` no longer takes the lead-scoring
+  `world_graph`: it takes `generation_scheme: str`, `motif_family: str | None`,
+  and an `extra_fields` mapping for scheme-specific keys.  Every manifest now
+  records `generation_scheme`; `BUNDLE_SCHEMA_VERSION` bumped 5 → 6.  Removes
+  the `manifests.py` → `lead_scoring.structure.graph` TYPE_CHECKING back-ref
+  (part of cleanup #3).  **Lead-scoring data files byte-identical** (tables/,
+  tasks/); only `manifest.json` changes (new field + version).  Schema
+  contract test renamed v5 → v6.
+  - Labels: `type: refactor`, `layer: render`
+- [ ] **`LTV-Pn.2`** — `refactor: scheme-agnostic WorldBundle + exposure hook +
+  shared bundle orchestrator`.  Generalise `WorldBundle` to hold scheme-owned
+  artifacts (finishing cleanup #3: drop the `core.models` / `render` →
+  `lead_scoring.*` back-refs); make `apply_exposure` / `write_metadata_dir`
+  scheme-agnostic via a hidden-truth hook (cleanup #2); lift a shared bundle
+  orchestrator with scheme render hooks out of `write_bundle` (cleanup #1).
+  Lead-scoring bundle byte-identical (full SHA-256 harness).
+  - Labels: `type: refactor`, `layer: api`, `layer: core`, `layer: render`
+- [ ] **`LTV-Pn.3`** — `feat: lifecycle config + regression task model`.  Add
+  `n_customers` + lifecycle config (forward windows, early-tenure, observation
+  anchor) to `GenerationConfig` (validated); add a regression `task_type`
+  (`regression` | `classification`) to `TaskManifest` + a continuous-target
+  split writer (the `LTV-Pc` / `LTV-Pl` / `LTV-Pm` deferral).  No e2e yet.
+  - Labels: `type: feature`, `layer: api`, `layer: schema`, `layer: render`
+- [ ] **`LTV-Pn.4`** — `feat(lifecycle): complete LifecycleScheme + e2e bundle`.
+  Implement `LifecycleScheme.build_world` (population → sim) and `write_bundle`
+  (lifecycle relational tables; both regime snapshots → two task families ×
+  3 windows + secondary churn; dataset card; manifest `observation_date` +
+  windows via `extra_fields`).  First end-to-end lifecycle bundle (programmatic;
+  recipe wiring is `LTV-Po`).  Extend `CLAUDE.md` hard constraints with the
+  lifecycle snapshot-safety clause + the `schemes/` layout.  Carries the
+  LTV-Pp validation flags: early-regime degenerate-column exemptions; the
+  dtype-preserving missingness opt-in.
   - Labels: `type: feature`, `layer: api`, `layer: render`
 - [ ] **`LTV-Po`** — `feat(recipes): b2b_saas_ltv_v1 recipe assets`. The three
   recipe YAMLs (`scheme: lifecycle`); register in the recipe registry;
@@ -329,7 +349,7 @@ Total: ~19 PRs across 9 milestones.
 
 The peer-schemes reorg deliberately defers a few cleanups to keep each M2 PR
 byte-identical and reviewable. They are tracked here and discharged in
-**`LTV-Pn`** (M6), where the manifest/exposure generalization makes them clean:
+**`LTV-Pn.1`/`LTV-Pn.2`** (M6), where the manifest/exposure generalization makes them clean:
 
 1. **Shared render orchestration** — `LTV-Pe` left each scheme owning its full
    `write_bundle`; only `write_relational_tables` is shared. A shared bundle
@@ -341,9 +361,11 @@ byte-identical and reviewable. They are tracked here and discharged in
 3. **core→scheme layering inversion** — `LTV-Pf.1` introduced
    `TYPE_CHECKING`-only imports of `leadforge.schemes.lead_scoring.*` in
    `core.models` (`WorldBundle.world_graph: WorldGraph | None`) and
-   `render.relational`. Harmless at runtime (no eager import), but `core`/shared
-   `render` should not reference a scheme. Remove once (2) makes
-   `WorldBundle` hold scheme-agnostic artifacts.
+   `render.*`. Harmless at runtime (no eager import), but `core`/shared
+   `render` should not reference a scheme. **Partly discharged in `LTV-Pn.1`**
+   (removed the `render.manifests` → `lead_scoring.structure.graph` back-ref);
+   the `core.models.WorldBundle` back-refs follow in `LTV-Pn.2` once
+   `WorldBundle` holds scheme-agnostic artifacts.
 
 ---
 

leadforge/cli/commands/inspect.py

@@ -52,6 +52,10 @@ def inspect(
 
     typer.echo(f"Bundle: {root}")
     typer.echo(f"  Recipe:        {manifest.get('recipe_id', '?')}")
+    # v6+ field: which peer generation scheme produced the bundle.  Conditional
+    # so pre-v6 bundles render without a "?" placeholder row.
+    if "generation_scheme" in manifest:
+        typer.echo(f"  Scheme:        {manifest['generation_scheme']}")
     typer.echo(f"  Seed:          {manifest.get('seed', '?')}")
     typer.echo(f"  Mode:          {manifest.get('exposure_mode', '?')}")
     typer.echo(f"  Difficulty:    {manifest.get('difficulty', '?')}")
@@ -82,7 +86,7 @@ def inspect(
                 names = ", ".join(cols[:3]) + ", ..."
             typer.echo(f"  Redactions:    {len(cols)} {noun} [{names}]")
 
-    typer.echo(f"  Motif family:  {manifest.get('motif_family', '?')}")
+    typer.echo(f"  Motif family:  {manifest.get('motif_family') or '?'}")
 
     typer.echo("")
     typer.echo("Tables:")

leadforge/render/manifests.py

@@ -22,7 +22,6 @@
 
 if TYPE_CHECKING:
     from leadforge.core.models import GenerationConfig
-    from leadforge.schemes.lead_scoring.structure.graph import WorldGraph
 
 # Bump this whenever the bundle layout or manifest schema changes.
 # History:
@@ -53,7 +52,16 @@
 #         consumers / validators can tell from the bundle alone whether
 #         the tables are snapshot-safe.  ``research_instructor`` bundles
 #         keep the full-horizon export (``relational_snapshot_safe = false``).
-BUNDLE_SCHEMA_VERSION = "5"
+#   "6" — LTV-Pn.1: every manifest now records ``generation_scheme`` (which
+#         peer generation scheme produced the bundle — ``lead_scoring`` or
+#         ``lifecycle``).  ``build_manifest`` is scheme-agnostic: it takes a
+#         ``motif_family`` string (or ``None``) instead of the lead-scoring
+#         ``world_graph``, and an ``extra_fields`` mapping for scheme-specific
+#         keys (the lifecycle scheme adds ``observation_date`` / forward
+#         windows in a later PR).  Data files (tables/, tasks/) are unchanged
+#         from v5 for the lead-scoring path; only ``manifest.json`` gains the
+#         ``generation_scheme`` field.
+BUNDLE_SCHEMA_VERSION = "6"
 
 # Manifest fields whose value is non-deterministic by design (wall-clock,
 # host metadata, etc.).  Determinism checks must ignore these fields when
@@ -63,13 +71,15 @@
 
 def build_manifest(
     config: GenerationConfig,
-    world_graph: WorldGraph,
+    generation_scheme: str,
     table_row_counts: dict[str, int],
     task_row_counts: dict[str, dict[str, int]],
     bundle_root: Path,
     generation_timestamp: str | None = None,
     redacted_columns: list[str] | None = None,
     relational_snapshot_safe: bool = False,
+    motif_family: str | None = None,
+    extra_fields: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Build the bundle manifest dict.
 
@@ -79,7 +89,10 @@ def build_manifest(
 
     Args:
         config: The resolved generation configuration.
-        world_graph: The sampled hidden world graph (provides motif_family).
+        generation_scheme: Name of the peer generation scheme that produced
+            the bundle (``lead_scoring`` / ``lifecycle``).  Recorded so a
+            consumer can tell which pipeline shape a bundle came from without
+            inspecting its tables.
         table_row_counts: Mapping of table name → row count.
         task_row_counts: Mapping of task_id → {split_name → row count}.
         bundle_root: Root directory of the written bundle.
@@ -96,6 +109,13 @@ def build_manifest(
             reading a v5+ bundle can tell from the manifest alone whether
             ``tables/`` is the snapshot-safe (public) shape or the
             full-horizon (instructor) shape.  Defaults to ``False``.
+        motif_family: The hidden-world motif family, when the scheme has one
+            (lead-scoring passes ``world_graph.motif_family``).  ``None`` for
+            schemes without a single named motif.  Recorded as
+            ``manifest.motif_family``.
+        extra_fields: Optional scheme-specific top-level manifest keys merged
+            into the result (e.g. the lifecycle scheme's ``observation_date``
+            and forward windows).  Must not collide with a core manifest key.
 
     Returns:
         A JSON-serialisable dict ready to be written as ``manifest.json``.
@@ -125,9 +145,10 @@ def build_manifest(
             entry[f"{split_name}_sha256"] = sha
         tasks[task_id] = entry
 
-    return {
+    manifest: dict[str, Any] = {
         "bundle_schema_version": BUNDLE_SCHEMA_VERSION,
         "package_version": config.package_version,
+        "generation_scheme": generation_scheme,
         "recipe_id": config.recipe_id,
         "seed": config.seed,
         "generation_timestamp": generation_timestamp,
@@ -140,13 +161,21 @@ def build_manifest(
         "primary_task": config.primary_task,
         "label_window_days": config.label_window_days,
         "snapshot_day": config.snapshot_day,
-        "motif_family": world_graph.motif_family,
+        "motif_family": motif_family,
         "redacted_columns": redacted_columns_list,
         "relational_snapshot_safe": bool(relational_snapshot_safe),
         "structural_redactions": _build_structural_redactions(bool(relational_snapshot_safe)),
         "tables": tables,
         "tasks": tasks,
     }
+    if extra_fields:
+        collisions = set(extra_fields) & set(manifest)
+        if collisions:
+            raise ValueError(
+                f"extra_fields would overwrite core manifest keys: {sorted(collisions)}"
+            )
+        manifest.update(extra_fields)
+    return manifest
 
 
 def _build_structural_redactions(relational_snapshot_safe: bool) -> dict[str, Any]:

leadforge/schemes/lead_scoring/__init__.py

@@ -265,7 +265,8 @@ def write_bundle(
         # ------------------------------------------------------------------
         manifest = build_manifest(
             config=config,
-            world_graph=world_graph,
+            generation_scheme=self.name,
+            motif_family=world_graph.motif_family,
             table_row_counts=table_row_counts,
             task_row_counts={task.task_id: task_row_counts},
             bundle_root=root,

tests/integration/test_snapshot_safe_bundle.py

@@ -1,12 +1,12 @@
-"""Integration tests for the snapshot-safe bundle write path (bundle schema v5).
+"""Integration tests for the snapshot-safe bundle write path (bundle schema v6).
 
 Covers the contract turned on in PR 2.2: ``student_public`` bundles
 route ``tables/`` through
 :func:`leadforge.schemes.lead_scoring.render.relational_snapshot_safe.to_dataframes_snapshot_safe`
 (the structural fix against the alpha-bundle reconstruction paths
 A-E), ``research_instructor`` bundles keep the full-horizon export,
 and the manifest is self-describing via ``relational_snapshot_safe``,
-``structural_redactions``, and ``bundle_schema_version == "5"``.
+``structural_redactions``, and ``bundle_schema_version == "6"``.
 
 Tests fall into three groups:
 

tests/render/test_bundle_schema_v5_contract.py → tests/render/test_bundle_schema_v6_contract.py

@@ -1,4 +1,4 @@
-"""Schema contract test for ``bundle_schema_version == "5"``.
+"""Schema contract test for ``bundle_schema_version == "6"``.
 
 The constants below are an *intentional* duplication of the column /
 table sets the bundle writer produces.  The duplication is the point:
@@ -20,6 +20,12 @@
 bundle is self-describing.  ``research_instructor`` bundles keep the
 full-horizon export.
 
+v6 vs v5: the lead-scoring published *shape* (columns, tables,
+snapshot-safe contract) is **unchanged**.  v6 adds a top-level
+``manifest.generation_scheme`` field (``lead_scoring`` here) recording
+which peer generation scheme produced the bundle.  The pinned column /
+table sets below therefore carry over from v5 verbatim.
+
 Task split column SET is unchanged from v4 — the structural fix lives
 in ``tables/``, not the snapshot.
 
@@ -173,12 +179,23 @@ def _leads_cols(bundle: Path) -> frozenset[str]:
     return _table_cols(bundle, "leads")
 
 
-def test_manifest_declares_v5(student_bundle: Path, instructor_bundle: Path) -> None:
+def test_manifest_declares_v6(student_bundle: Path, instructor_bundle: Path) -> None:
     for b in (student_bundle, instructor_bundle):
         manifest = json.loads((b / "manifest.json").read_text())
-        assert manifest["bundle_schema_version"] == "5", (
+        assert manifest["bundle_schema_version"] == "6", (
             f"{b.name}: bundle_schema_version is {manifest['bundle_schema_version']!r}, "
-            "expected '5'"
+            "expected '6'"
+        )
+
+
+def test_manifest_records_generation_scheme(student_bundle: Path, instructor_bundle: Path) -> None:
+    """v6 contract: every manifest records which peer generation scheme produced
+    the bundle.  Both fixtures come from the lead-scoring recipe."""
+    for b in (student_bundle, instructor_bundle):
+        manifest = json.loads((b / "manifest.json").read_text())
+        assert manifest["generation_scheme"] == "lead_scoring", (
+            f"{b.name}: generation_scheme is {manifest.get('generation_scheme')!r}, "
+            "expected 'lead_scoring'"
         )
 
 

tests/render/test_manifest_scheme_agnostic.py

@@ -0,0 +1,72 @@
+"""Unit tests for the scheme-agnostic build_manifest contract (LTV-Pn.1)."""
+
+from __future__ import annotations
+
+import pytest
+
+from leadforge.core.models import GenerationConfig
+from leadforge.render.manifests import BUNDLE_SCHEMA_VERSION, build_manifest
+
+
+def _config() -> GenerationConfig:
+    return GenerationConfig(seed=1, n_accounts=2, n_contacts=3, n_leads=4)
+
+
+def _manifest(tmp_path, **kwargs):
+    # Empty row-count dicts → no parquet files to hash; isolates the
+    # scheme-agnostic header fields under test.
+    return build_manifest(
+        config=_config(),
+        table_row_counts={},
+        task_row_counts={},
+        bundle_root=tmp_path,
+        generation_timestamp="2026-01-01T00:00:00+00:00",
+        **kwargs,
+    )
+
+
+def test_records_generation_scheme(tmp_path) -> None:
+    m = _manifest(tmp_path, generation_scheme="lifecycle")
+    assert m["generation_scheme"] == "lifecycle"
+    assert m["bundle_schema_version"] == BUNDLE_SCHEMA_VERSION
+
+
+def test_motif_family_defaults_to_none(tmp_path) -> None:
+    # A scheme without a single named motif (e.g. lifecycle) omits it.
+    m = _manifest(tmp_path, generation_scheme="lifecycle")
+    assert m["motif_family"] is None
+
+
+def test_motif_family_passthrough(tmp_path) -> None:
+    m = _manifest(tmp_path, generation_scheme="lead_scoring", motif_family="fit_dominant")
+    assert m["motif_family"] == "fit_dominant"
+
+
+def test_extra_fields_merged(tmp_path) -> None:
+    m = _manifest(
+        tmp_path,
+        generation_scheme="lifecycle",
+        extra_fields={"observation_date": "2026-06-01", "forward_windows_days": [90, 365, 730]},
+    )
+    assert m["observation_date"] == "2026-06-01"
+    assert m["forward_windows_days"] == [90, 365, 730]
+
+
+def test_extra_fields_cannot_overwrite_core_keys(tmp_path) -> None:
+    with pytest.raises(ValueError, match="overwrite core manifest keys"):
+        _manifest(
+            tmp_path,
+            generation_scheme="lifecycle",
+            extra_fields={"seed": 999, "generation_scheme": "evil"},
+        )
+
+
+def test_generation_scheme_is_required(tmp_path) -> None:
+    # Positional/keyword required argument — omitting it is a TypeError.
+    with pytest.raises(TypeError):
+        build_manifest(  # type: ignore[call-arg]
+            config=_config(),
+            table_row_counts={},
+            task_row_counts={},
+            bundle_root=tmp_path,
+        )

tests/render/test_render.py

@@ -360,7 +360,8 @@ def _make_manifest(self, sim_outputs, tmp_path):
 
         manifest = build_manifest(
             config=config,
-            world_graph=world_graph,
+            generation_scheme="lead_scoring",
+            motif_family=world_graph.motif_family,
             table_row_counts=table_row_counts,
             task_row_counts={"converted_within_90_days": task_counts},
             bundle_root=tmp_path,

tests/scripts/test_run_llm_critique.py

@@ -94,7 +94,7 @@ def _write_minimal_release(tmp_path: Path, *, tier: str = "intermediate") -> Pat
         "# Method\n", encoding="utf-8"
     )
     (bundle_dir / "manifest.json").write_text(
-        json.dumps({"bundle_schema_version": "5", "exposure_mode": "student_public"}),
+        json.dumps({"bundle_schema_version": "6", "exposure_mode": "student_public"}),
         encoding="utf-8",
     )
     (bundle_dir / "feature_dictionary.csv").write_text(