tenseleyflow/documentlanguagemodel / dd38498

+### `dlm synth`

+

+Synthesize instruction or preference training data (Sprint 43).

+

+```

+dlm synth instructions <path> [--teacher T] [--per-section N]

+                             [--strategy {extraction,expansion,both}]

+                             [--filter {sway,none,dedup-only}]

+                             [--threshold F] [--max-pairs N]

+                             [--max-new-tokens N] [--temp F] [--top-p F]

+                             [--seed N] [--apply | --dry-run]

+dlm synth preferences <path> [--samples N] [--judge J] [--threshold F]

+                             [--max-pairs N] [--temp F] [--top-p F]

+                             [--backend {auto,pytorch,mlx}] [--adapter NAME]

+                             [--apply]

+dlm synth revert <path>

+dlm synth list <path>

+```

+

+| Option | Default | Notes |

+|---|---|---|

+| `--teacher T` | `self` | Teacher selector: `self`, `hf:<model>`, `openai:<model>`, `anthropic:<model>`, or `vllm-server:<url>`. |

+| `--per-section N` | `3` | Accepted instruction pairs to request per prose section before filtering. |

+| `--strategy {extraction,expansion,both}` | `extraction` | `extraction` asks for questions answered directly by the prose, `expansion` extrapolates beyond it, and `both` splits the per-section budget across both prompts. |

+| `--filter {sway,none,dedup-only}` | `sway` | Filter pipeline after generation. `sway` reuses Sprint 42's judge, `dedup-only` keeps near-duplicate suppression but skips judging, `none` accepts every deduped pair. |

+| `--threshold F` | judge default | Minimum sway-judge margin. Only valid with `--filter sway`. |

+| `--max-pairs N` | unlimited | Cap the number of accepted synth pairs from one invocation. |

+| `--max-new-tokens N` | `512` | Teacher-side completion cap per prompt. |

+| `--temp F` | `0.0` | Teacher sampling temperature. |

+| `--top-p F` | None | Optional top-p cutoff for teacher sampling. |

+| `--seed N` | None | Optional teacher sampling seed. |

+| `--apply` | false | Write accepted auto-synth `::instruction::` sections directly to the `.dlm`. |

+| `--dry-run` | false | Preview the synth plan without staging or writing anything. Default behavior stages the accepted plan under the store for inspection via `dlm synth list`. |

+

+`dlm synth instructions` prints the raw synth plan, then the filter

+summary (`generated`, `dedup`, `judge passed`, `threshold/accepted`).

+Without `--apply` or `--dry-run`, the accepted auto-synth

+`::instruction::` sections are staged under the store root so `dlm

+synth list` can show them before a later rerun. `dlm synth revert`

+strips every `auto_synth: true` instruction section from the document.

+

+`dlm synth preferences` is an alias over `dlm preference mine` for the

+same Sprint 42 preference-mining loop. Use it when you want the

+umbrella synth surface but the output should be `::preference::`

+sections instead of `::instruction::` sections.

+

+# `dlm synth instructions|preferences|revert|list` — synthetic data loop.

+_synth_app = typer.Typer(

+    help="Synthesize instruction or preference training data.",

+    no_args_is_help=True,

+)

+_synth_app.command("instructions")(commands.synth_instructions_cmd)

+_synth_app.command("preferences")(commands.preference_mine_cmd)

+_synth_app.command("revert")(commands.synth_revert_cmd)

+_synth_app.command("list")(commands.synth_list_cmd)

+app.add_typer(_synth_app, name="synth")

+

+from collections.abc import Sequence

+from typing import TYPE_CHECKING, Annotated, Any, Literal, cast

+# --- synth -----------------------------------------------------------------

+

+

+def synth_instructions_cmd(

+    path: Annotated[

+        Path, typer.Argument(help=".dlm file to synthesize instruction sections from.")

+    ],

+    teacher: Annotated[

+        str,

+        typer.Option(

+            "--teacher",

+            help=(

+                "Teacher selector: self, hf:<model>, openai:<model>, "

+                "anthropic:<model>, or vllm-server:<url>."

+            ),

+        ),

+    ] = "self",

+    per_section: Annotated[

+        int,

+        typer.Option(

+            "--per-section",

+            help="Instruction pairs to generate per prose section.",

+            min=1,

+        ),

+    ] = 3,

+    strategy: Annotated[

+        str,

+        typer.Option(

+            "--strategy",

+            help="Synthesis strategy: extraction, expansion, or both.",

+        ),

+    ] = "extraction",

+    filter_kind: Annotated[

+        str,

+        typer.Option(

+            "--filter",

+            help="Filter pipeline: sway, none, or dedup-only.",

+        ),

+    ] = "sway",

+    threshold: Annotated[

+        float | None,

+        typer.Option(

+            "--threshold",

+            help="Optional minimum sway-judge margin when --filter=sway.",

+            min=0.0,

+        ),

+    ] = None,

+    max_pairs: Annotated[

+        int | None,

+        typer.Option(

+            "--max-pairs",

+            help="Maximum accepted synth pairs to keep from this run.",

+            min=1,

+        ),

+    ] = None,

+    max_new_tokens: Annotated[

+        int,

+        typer.Option(

+            "--max-new-tokens",

+            help="Maximum new tokens the teacher may emit per prompt.",

+            min=1,

+        ),

+    ] = 512,

+    temp: Annotated[

+        float,

+        typer.Option("--temp", help="Teacher sampling temperature.", min=0.0),

+    ] = 0.0,

+    top_p: Annotated[

+        float | None,

+        typer.Option(

+            "--top-p",

+            help="Optional top-p cutoff for teacher sampling.",

+            min=0.0,

+            max=1.0,

+        ),

+    ] = None,

+    seed: Annotated[

+        int | None,

+        typer.Option("--seed", help="Optional teacher sampling seed."),

+    ] = None,

+    apply: Annotated[

+        bool,

+        typer.Option(

+            "--apply",

+            help="Write accepted auto-synth sections directly to the .dlm.",

+        ),

+    ] = False,

+    dry_run: Annotated[

+        bool,

+        typer.Option(

+            "--dry-run",

+            help="Preview the synth plan without staging or writing anything.",

+        ),

+    ] = False,

+) -> None:

+    """Generate, stage, or apply auto-synth instruction sections."""

+    from rich.console import Console

+

+    from dlm.doc.errors import DlmParseError

+    from dlm.doc.parser import parse_file

+    from dlm.preference import JudgeUnavailableError, build_judge

+    from dlm.store.paths import for_dlm

+    from dlm.synth import (

+        InvalidTeacherSpecError,

+        TeacherInvocationError,

+        TeacherUnavailableError,

+        build_synth_plan,

+        build_teacher,

+        clear_pending_plan,

+        filter_synth_plan,

+        render_filter_report,

+        render_synth_plan,

+        save_pending_plan,

+    )

+    from dlm.synth import (

+        apply_plan as apply_synth_plan,

+    )

+    from dlm.synth import (

+        build_apply_plan as build_synth_apply_plan,

+    )

+    from dlm.synth import (

+        render_apply_plan as render_synth_apply_plan,

+    )

+

+    console = Console(stderr=True)

+    out_console = Console()

+

+    if strategy not in ("extraction", "expansion", "both"):

+        console.print(

+            "[red]synth:[/red] --strategy must be one of extraction|expansion|both "

+            f"(got {strategy!r})."

+        )

+        raise typer.Exit(code=2)

+    if filter_kind not in ("sway", "none", "dedup-only"):

+        console.print(

+            f"[red]synth:[/red] --filter must be one of sway|none|dedup-only (got {filter_kind!r})."

+        )

+        raise typer.Exit(code=2)

+    if apply and dry_run:

+        console.print("[red]synth:[/red] --apply and --dry-run are mutually exclusive.")

+        raise typer.Exit(code=2)

+    if threshold is not None and filter_kind != "sway":

+        console.print("[red]synth:[/red] --threshold is only valid when --filter is `sway`.")

+        raise typer.Exit(code=2)

+

+    try:

+        parsed = parse_file(path)

+    except (DlmParseError, OSError) as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=1) from exc

+

+    store = for_dlm(parsed.frontmatter.dlm_id)

+

+    try:

+        strategy_value = cast(Literal["extraction", "expansion", "both"], strategy)

+        teacher_obj = build_teacher(teacher, dlm_path=path)

+        plan = build_synth_plan(

+            parsed,

+            teacher_obj,

+            per_section=per_section,

+            strategy=strategy_value,

+            max_pairs=max_pairs,

+            max_new_tokens=max_new_tokens,

+            temperature=temp,

+            top_p=top_p,

+            seed=seed,

+        )

+    except InvalidTeacherSpecError as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=2) from exc

+    except TeacherUnavailableError as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=1) from exc

+    except TeacherInvocationError as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=1) from exc

+    except ValueError as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=2) from exc

+

+    judge_obj = None

+    if filter_kind == "sway":

+        try:

+            judge_obj = build_judge("sway", dlm_path=path)

+        except JudgeUnavailableError as exc:

+            console.print(f"[red]synth:[/red] {exc}")

+            raise typer.Exit(code=1) from exc

+

+    try:

+        filter_value = cast(Literal["sway", "none", "dedup-only"], filter_kind)

+        filtered = filter_synth_plan(

+            plan,

+            filter_kind=filter_value,

+            judge=judge_obj,

+            threshold=threshold,

+        )

+    except ValueError as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=2) from exc

+

+    out_console.print(render_synth_plan(plan))

+    out_console.print("")

+    out_console.print(render_filter_report(filtered))

+

+    if not filtered.additions:

+        if not dry_run:

+            clear_pending_plan(store)

+        out_console.print(

+            "\n[yellow]no synth additions accepted[/yellow] — either generation "

+            "yielded no valid pairs, dedup removed them, or the filter rejected them."

+        )

+        raise typer.Exit(code=2)

+

+    sections = [addition.addition.section for addition in filtered.additions]

+

+    if apply:

+        apply_plan = build_synth_apply_plan(parsed, sections)

+        out_console.print("")

+        out_console.print(render_synth_apply_plan(apply_plan))

+        summary = apply_synth_plan(parsed, apply_plan, target=path)

+        clear_pending_plan(store)

+        out_console.print(

+            f"\n[green]synth:[/green] wrote {summary.added} section(s) to {path} "

+            f"({summary.skipped} skipped)"

+        )

+        return

+

+    if dry_run:

+        out_console.print("\n[green]synth:[/green] dry-run only — nothing staged.")

+        return

+

+    pending = save_pending_plan(store, source_path=path.resolve(), sections=sections)

+    out_console.print(

+        f"\n[green]synth:[/green] staged {len(pending.sections)} auto-synth instruction "

+        f"section(s). Run [bold]dlm synth list {path}[/bold] to inspect them."

+    )

+

+

+def synth_revert_cmd(

+    path: Annotated[Path, typer.Argument(help=".dlm file to strip auto-synth instructions from.")],

+) -> None:

+    """Remove every `auto_synth: true` instruction section from the `.dlm`."""

+    from rich.console import Console

+

+    from dlm.doc.errors import DlmParseError

+    from dlm.doc.parser import parse_file

+    from dlm.synth import revert_all_auto_synth

+

+    console = Console(stderr=True)

+    out_console = Console()

+

+    try:

+        parsed = parse_file(path)

+    except (DlmParseError, OSError) as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=1) from exc

+

+    summary = revert_all_auto_synth(parsed, target=path)

+    out_console.print(

+        f"[green]synth:[/green] stripped {len(summary.added_section_ids)} "

+        f"auto-synth instruction section(s) from {path}"

+    )

+

+

+def synth_list_cmd(

+    path: Annotated[Path, typer.Argument(help=".dlm file whose auto-synth instructions we list.")],

+) -> None:

+    """List applied + staged auto-synth instruction sections."""

+    from rich.console import Console

+

+    from dlm.doc.errors import DlmParseError

+    from dlm.doc.parser import parse_file

+    from dlm.doc.sections import SectionType

+    from dlm.store.paths import for_dlm

+    from dlm.synth import PendingSynthPlanError, load_pending_plan

+

+    console = Console(stderr=True)

+    out_console = Console()

+

+    try:

+        parsed = parse_file(path)

+    except (DlmParseError, OSError) as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=1) from exc

+

+    store = for_dlm(parsed.frontmatter.dlm_id)

+    try:

+        pending = load_pending_plan(store)

+    except PendingSynthPlanError as exc:

+        console.print(f"[red]synth:[/red] {exc}")

+        raise typer.Exit(code=1) from exc

+

+    applied = [

+        section

+        for section in parsed.sections

+        if section.type is SectionType.INSTRUCTION and section.auto_synth

+    ]

+

+    out_console.print(f"[bold]{path}[/bold]")

+    out_console.print(f"  applied auto-synth: {len(applied)}")

+    out_console.print(f"  staged pending:     {len(pending.sections) if pending else 0}")

+

+    if not applied and pending is None:

+        out_console.print("  [dim]no auto-synth instruction sections yet[/dim]")

+        return

+

+    if applied:

+        _render_synth_listing(out_console, "Applied", applied)

+    if pending is not None:

+        _render_synth_listing(out_console, "Pending", pending.sections)

+

+

+def _render_synth_listing(

+    out_console: object,

+    heading: str,

+    sections: Sequence[object],

+) -> None:

+    from collections import Counter

+

+    from rich.console import Console

+

+    from dlm.doc.sections import Section

+

+    assert isinstance(out_console, Console)

+    typed_sections = [section for section in sections if isinstance(section, Section)]

+

+    out_console.print(f"\n[bold]{heading}[/bold]")

+

+    teacher_counts = Counter(section.synth_teacher or "unknown" for section in typed_sections)

+    strategy_counts = Counter(section.synth_strategy or "unknown" for section in typed_sections)

+    source_counts = Counter(section.source_section_id or "unknown" for section in typed_sections)

+

+    out_console.print("  by teacher:")

+    for teacher_name in sorted(teacher_counts):

+        out_console.print(f"    - {teacher_name}: {teacher_counts[teacher_name]}")

+

+    out_console.print("  by strategy:")

+    for strategy_name in sorted(strategy_counts):

+        out_console.print(f"    - {strategy_name}: {strategy_counts[strategy_name]}")

+

+    out_console.print("  by source section:")

+    for source_id in sorted(source_counts):

+        out_console.print(f"    - {source_id}: {source_counts[source_id]}")

+

+    out_console.print("  sections:")

+    for section in typed_sections:

+        prompt = _synth_prompt_summary(section.content, section_id=section.section_id)

+        out_console.print(

+            "    - "

+            f"{section.section_id}  teacher={section.synth_teacher or 'unknown'}  "

+            f"strategy={section.synth_strategy or 'unknown'}  "

+            f"source={section.source_section_id or 'unknown'}  "

+            f"prompt={prompt}"

+        )

+

+

+def _synth_prompt_summary(content: str, *, section_id: str) -> str:

+    """Best-effort prompt summary for `synth list`."""

+    from dlm.data.errors import InstructionParseError

+    from dlm.data.instruction_parser import parse_instruction_body

+

+    try:

+        pairs = parse_instruction_body(content, section_id=section_id)

+    except InstructionParseError:

+        return "<unparseable>"

+    if not pairs:

+        return "<empty>"

+    prompt = pairs[0].question.splitlines()[0].strip()

+    return prompt or "<blank>"

+

+

+from dlm.synth.apply import (

+    PlannedSynthAddition,

+    SkippedSynthAddition,

+    SynthApplyPlan,

+    SynthApplySkipReason,

+    SynthApplySummary,

+    apply_plan,

+    build_apply_plan,

+    render_apply_plan,

+    revert_all_auto_synth,

+)

+from dlm.synth.pending import (

+    PendingSynthPlan,

+    PendingSynthPlanError,

+    clear_pending_plan,

+    load_pending_plan,

+    save_pending_plan,

+)

+    "PendingSynthPlan",

+    "PendingSynthPlanError",

+    "PlannedSynthAddition",

+    "SkippedSynthAddition",

+    "SynthApplyPlan",

+    "SynthApplySkipReason",

+    "SynthApplySummary",

+    "apply_plan",

+    "build_apply_plan",

+    "clear_pending_plan",

+    "load_pending_plan",

+    "render_apply_plan",

+    "revert_all_auto_synth",

+    "save_pending_plan",

+"""Apply/revert staged auto-synth instruction sections."""

+

+from __future__ import annotations

+

+import dataclasses

+from dataclasses import dataclass

+from enum import StrEnum

+from pathlib import Path

+

+from dlm.doc.parser import ParsedDlm

+from dlm.doc.sections import Section, SectionType

+from dlm.doc.serializer import serialize

+from dlm.io.atomic import write_text as atomic_write_text

+

+

+class SynthApplySkipReason(StrEnum):

+    """Why a staged synth section did not make it into the plan."""

+

+    ALREADY_PRESENT = "already_present"

+    NOT_INSTRUCTION = "not_instruction"

+    NOT_AUTO_SYNTH = "not_auto_synth"

+

+

+@dataclass(frozen=True)

+class PlannedSynthAddition:

+    """One section that survived plan-time validation and dedupe."""

+

+    section: Section

+

+

+@dataclass(frozen=True)

+class SkippedSynthAddition:

+    """One input section that did not make it into the plan."""

+

+    section: Section

+    reason: SynthApplySkipReason

+    detail: str = ""

+

+

+@dataclass(frozen=True)

+class SynthApplyPlan:

+    """What the applier would do if executed."""

+

+    additions: tuple[PlannedSynthAddition, ...]

+    skipped: tuple[SkippedSynthAddition, ...]

+

+

+@dataclass(frozen=True)

+class SynthApplySummary:

+    """Outcome of applying or reverting auto-synth instruction sections."""

+

+    target: Path

+    added: int

+    skipped: int

+    added_section_ids: tuple[str, ...]

+

+

+def build_apply_plan(parsed: ParsedDlm, sections: list[Section]) -> SynthApplyPlan:

+    """Validate + dedupe staged synth sections against `parsed`."""

+    existing = {section.section_id for section in parsed.sections}

+    additions: list[PlannedSynthAddition] = []

+    skipped: list[SkippedSynthAddition] = []

+

+    for section in sections:

+        if section.type is not SectionType.INSTRUCTION:

+            skipped.append(

+                SkippedSynthAddition(

+                    section=section,

+                    reason=SynthApplySkipReason.NOT_INSTRUCTION,

+                    detail="only instruction sections can be applied",

+                )

+            )

+            continue

+        if not section.auto_synth:

+            skipped.append(

+                SkippedSynthAddition(

+                    section=section,

+                    reason=SynthApplySkipReason.NOT_AUTO_SYNTH,

+                    detail="section is not marked auto_synth=true",

+                )

+            )

+            continue

+        if section.section_id in existing:

+            skipped.append(

+                SkippedSynthAddition(

+                    section=section,

+                    reason=SynthApplySkipReason.ALREADY_PRESENT,

+                    detail=f"section_id {section.section_id} already in document",

+                )

+            )

+            continue

+        additions.append(PlannedSynthAddition(section=section))

+        existing.add(section.section_id)

+

+    return SynthApplyPlan(additions=tuple(additions), skipped=tuple(skipped))

+

+

+def render_apply_plan(plan: SynthApplyPlan) -> str:

+    """Plain-text form for dry-run output and tests."""

+    lines = [

+        f"synth apply plan: {len(plan.additions)} add, {len(plan.skipped)} skip",

+        "",

+    ]

+    if plan.additions:

+        lines.append("=== additions ===")

+        for add in plan.additions:

+            lines.append("")

+            lines.append(

+                "+ ::instruction:: "

+                "[section_id="

+                f"{add.section.section_id} teacher={add.section.synth_teacher} "

+                f"strategy={add.section.synth_strategy} source={add.section.source_section_id}]"

+            )

+    if plan.skipped:

+        lines.append("")

+        lines.append("=== skipped ===")

+        for skip in plan.skipped:

+            lines.append(f"- {skip.section.section_id}: {skip.reason.value} ({skip.detail})")

+    return "\n".join(lines)

+

+

+def apply_plan(

+    parsed: ParsedDlm,

+    plan: SynthApplyPlan,

+    *,

+    target: Path,

+) -> SynthApplySummary:

+    """Append plan additions to `parsed.sections` and atomically write them."""

+    new_sections = tuple(parsed.sections) + tuple(add.section for add in plan.additions)

+    updated = dataclasses.replace(parsed, sections=new_sections)

+    atomic_write_text(target, serialize(updated))

+    return SynthApplySummary(

+        target=target,

+        added=len(plan.additions),

+        skipped=len(plan.skipped),

+        added_section_ids=tuple(add.section.section_id for add in plan.additions),

+    )

+

+

+def revert_all_auto_synth(

+    parsed: ParsedDlm,

+    *,

+    target: Path,

+) -> SynthApplySummary:

+    """Strip every auto-synth instruction section and atomically rewrite `target`."""

+    survivors = tuple(

+        section

+        for section in parsed.sections

+        if not (section.type is SectionType.INSTRUCTION and section.auto_synth)

+    )

+    removed_ids = tuple(

+        section.section_id

+        for section in parsed.sections

+        if section.type is SectionType.INSTRUCTION and section.auto_synth

+    )

+    updated = dataclasses.replace(parsed, sections=survivors)

+    atomic_write_text(target, serialize(updated))

+    return SynthApplySummary(

+        target=target,

+        added=0,

+        skipped=0,

+        added_section_ids=removed_ids,

+    )

+"""Persist staged auto-synth instruction sections between CLI steps."""

+

+from __future__ import annotations

+

+import json

+from dataclasses import dataclass

+from datetime import UTC, datetime

+from pathlib import Path

+from typing import TYPE_CHECKING, Any

+

+from dlm.doc.sections import Section, SectionType

+from dlm.io.atomic import write_text as atomic_write_text

+from dlm.synth.errors import SynthError

+

+if TYPE_CHECKING:

+    from collections.abc import Sequence

+

+    from dlm.store.paths import StorePath

+

+

+class PendingSynthPlanError(SynthError):

+    """Raised when the staged synth plan cannot be read or validated."""

+

+

+@dataclass(frozen=True)

+class PendingSynthPlan:

+    """One staged synth plan for a store."""

+

+    source_path: Path

+    created_at: str

+    sections: tuple[Section, ...]

+

+

+def pending_plan_path(store: StorePath) -> Path:

+    """Path to the staged synth payload for `store`."""

+    return store.root / "synth" / "pending.json"

+

+

+def save_pending_plan(

+    store: StorePath,

+    *,

+    source_path: Path,

+    sections: Sequence[Section],

+) -> PendingSynthPlan:

+    """Persist `sections` as the staged synth plan for `store`."""

+    plan = PendingSynthPlan(

+        source_path=source_path.resolve(),

+        created_at=_utcnow(),

+        sections=tuple(sections),

+    )

+    path = pending_plan_path(store)

+    path.parent.mkdir(parents=True, exist_ok=True)

+    payload = {

+        "schema_version": 1,

+        "source_path": str(plan.source_path),

+        "created_at": plan.created_at,

+        "sections": [_section_to_payload(section) for section in plan.sections],

+    }

+    atomic_write_text(path, json.dumps(payload, indent=2, sort_keys=True) + "\n")

+    return plan

+

+

+def load_pending_plan(store: StorePath) -> PendingSynthPlan | None:

+    """Return the staged synth plan for `store`, or None when absent."""

+    path = pending_plan_path(store)

+    if not path.exists():

+        return None

+    try:

+        raw = json.loads(path.read_text(encoding="utf-8"))

+    except OSError as exc:

+        raise PendingSynthPlanError(f"could not read staged synth plan: {exc}") from exc

+    except json.JSONDecodeError as exc:

+        raise PendingSynthPlanError(f"staged synth plan is not valid JSON: {exc}") from exc

+

+    if not isinstance(raw, dict):

+        raise PendingSynthPlanError("staged synth plan must be a JSON object")

+    if raw.get("schema_version") != 1:

+        raise PendingSynthPlanError(

+            f"unsupported staged synth plan schema_version={raw.get('schema_version')!r}"

+        )

+

+    source_path = raw.get("source_path")

+    created_at = raw.get("created_at")

+    sections_raw = raw.get("sections")

+    if not isinstance(source_path, str) or not source_path:

+        raise PendingSynthPlanError("staged synth plan is missing source_path")

+    if not isinstance(created_at, str) or not created_at:

+        raise PendingSynthPlanError("staged synth plan is missing created_at")

+    if not isinstance(sections_raw, list):

+        raise PendingSynthPlanError("staged synth plan is missing sections")

+

+    sections: list[Section] = []

+    for idx, entry in enumerate(sections_raw):

+        try:

+            sections.append(_section_from_payload(entry))

+        except (TypeError, ValueError, KeyError) as exc:

+            raise PendingSynthPlanError(f"invalid section payload at index {idx}: {exc}") from exc

+

+    return PendingSynthPlan(

+        source_path=Path(source_path),

+        created_at=created_at,

+        sections=tuple(sections),

+    )

+

+

+def clear_pending_plan(store: StorePath) -> bool:

+    """Delete the staged synth plan for `store`. Returns True iff it existed."""

+    path = pending_plan_path(store)

+    if not path.exists():

+        return False

+    path.unlink()

+    return True

+

+

+def _utcnow() -> str:

+    return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")

+

+

+def _section_to_payload(section: Section) -> dict[str, Any]:

+    return {

+        "type": section.type.value,

+        "content": section.content,

+        "start_line": section.start_line,

+        "adapter": section.adapter,

+        "tags": dict(section.tags),

+        "auto_harvest": section.auto_harvest,

+        "harvest_source": section.harvest_source,

+        "auto_mined": section.auto_mined,

+        "judge_name": section.judge_name,

+        "judge_score_chosen": section.judge_score_chosen,

+        "judge_score_rejected": section.judge_score_rejected,

+        "mined_at": section.mined_at,

+        "mined_run_id": section.mined_run_id,

+        "auto_synth": section.auto_synth,

+        "synth_teacher": section.synth_teacher,

+        "synth_strategy": section.synth_strategy,

+        "synth_at": section.synth_at,

+        "source_section_id": section.source_section_id,

+        "media_path": section.media_path,

+        "media_alt": section.media_alt,

+        "media_blob_sha": section.media_blob_sha,

+        "media_transcript": section.media_transcript,

+    }

+

+

+def _section_from_payload(raw: object) -> Section:

+    if not isinstance(raw, dict):

+        raise TypeError(f"expected object, got {type(raw).__name__}")

+    section_type = SectionType(str(raw["type"]))

+    tags = raw.get("tags", {})

+    if not isinstance(tags, dict):

+        raise TypeError("tags must be an object")

+    if not all(isinstance(k, str) and isinstance(v, str) for k, v in tags.items()):

+        raise TypeError("tags keys and values must be strings")

+    return Section(

+        type=section_type,

+        content=str(raw["content"]),

+        start_line=int(raw.get("start_line", 0)),

+        adapter=_optional_str(raw.get("adapter")),

+        tags=dict(tags),

+        auto_harvest=bool(raw.get("auto_harvest", False)),

+        harvest_source=_optional_str(raw.get("harvest_source")),

+        auto_mined=bool(raw.get("auto_mined", False)),

+        judge_name=_optional_str(raw.get("judge_name")),

+        judge_score_chosen=_optional_float(raw.get("judge_score_chosen")),

+        judge_score_rejected=_optional_float(raw.get("judge_score_rejected")),

+        mined_at=_optional_str(raw.get("mined_at")),

+        mined_run_id=_optional_int(raw.get("mined_run_id")),

+        auto_synth=bool(raw.get("auto_synth", False)),

+        synth_teacher=_optional_str(raw.get("synth_teacher")),

+        synth_strategy=_optional_str(raw.get("synth_strategy")),

+        synth_at=_optional_str(raw.get("synth_at")),

+        source_section_id=_optional_str(raw.get("source_section_id")),

+        media_path=_optional_str(raw.get("media_path")),

+        media_alt=_optional_str(raw.get("media_alt")),

+        media_blob_sha=_optional_str(raw.get("media_blob_sha")),

+        media_transcript=_optional_str(raw.get("media_transcript")),

+    )

+

+

+def _optional_str(value: object) -> str | None:

+    if value is None:

+        return None

+    if not isinstance(value, str):

+        raise TypeError(f"expected string or null, got {type(value).__name__}")

+    return value

+

+

+def _optional_float(value: object) -> float | None:

+    if value is None:

+        return None

+    if isinstance(value, bool) or not isinstance(value, int | float):

+        raise TypeError(f"expected float or null, got {type(value).__name__}")

+    return float(value)

+

+

+def _optional_int(value: object) -> int | None:

+    if value is None:

+        return None

+    if isinstance(value, bool) or not isinstance(value, int):

+        raise TypeError(f"expected int or null, got {type(value).__name__}")

+    return value

+

+

+def test_reference_doc_covers_synth_surface() -> None:

+    section = _section("synth")

+    help_text = _normalized_help("synth", "instructions")

+

+    for flag in (

+        "--teacher",

+        "--per-section",

+        "--strategy",

+        "--filter",

+        "--threshold",

+        "--apply",

+        "--dry-run",

+    ):

+        assert flag in help_text

+        assert flag in section

+

+    assert "dlm synth preferences <path>" in section

+    assert "dlm synth revert <path>" in section

+    assert "dlm synth list <path>" in section

+"""CLI tests for `dlm synth` (Sprint 43)."""

+

+from __future__ import annotations

+

+import re

+from collections import deque

+from datetime import datetime

+from pathlib import Path

+

+import pytest

+from typer.testing import CliRunner

+

+from dlm.base_models import BaseModelSpec

+from dlm.cli.app import app

+from dlm.doc.parser import parse_file

+from dlm.doc.sections import SectionType

+from dlm.preference.judge import PairScore

+from dlm.preference.pending import load_pending_plan as load_pending_preference_plan

+from dlm.store.manifest import Manifest, TrainingRunSummary, save_manifest

+from dlm.store.paths import for_dlm

+from dlm.synth.pending import load_pending_plan

+

+_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")

+_DLM_ID = "01KPQ9X1000000000000000000"

+_REV = "0123456789abcdef0123456789abcdef01234567"

+

+

+def _normalized_output(result: object) -> str:

+    text = getattr(result, "output", "") + getattr(result, "stderr", "")

+    return " ".join(_ANSI_RE.sub("", text).split())

+

+

+def _write_synth_doc(path: Path) -> None:

+    path.write_text(

+        "---\n"

+        f"dlm_id: {_DLM_ID}\n"

+        "dlm_version: 15\n"

+        "base_model: smollm2-135m\n"

+        "---\n"

+        "DGEMM multiplies two dense matrices and optionally accumulates the result.\n",

+        encoding="utf-8",

+    )

+

+

+def _write_preference_doc(path: Path) -> None:

+    path.write_text(

+        "---\n"

+        f"dlm_id: {_DLM_ID}\n"

+        "dlm_version: 15\n"

+        "base_model: smollm2-135m\n"

+        "---\n"

+        "::instruction::\n"

+        "### Q\n"

+        "What is DGEMM?\n"

+        "### A\n"

+        "A matrix multiply.\n",

+        encoding="utf-8",

+    )

+

+

+def _write_manifest(home: Path, doc: Path, *, run_id: int = 7) -> None:

+    store = for_dlm(_DLM_ID, home=home)

+    store.ensure_layout()

+    save_manifest(

+        store.manifest,

+        Manifest(

+            dlm_id=_DLM_ID,

+            base_model="smollm2-135m",

+            base_model_revision=_REV,

+            source_path=doc.resolve(),

+            training_runs=[

+                TrainingRunSummary(

+                    run_id=run_id,

+                    started_at=datetime(2026, 4, 24, 12, 0, 0),

+                    ended_at=datetime(2026, 4, 24, 12, 1, 0),

+                    adapter_version=1,

+                    seed=123,

+                    steps=12,

+                )

+            ],

+        ),

+    )

+

+

+def _spec() -> BaseModelSpec:

+    return BaseModelSpec.model_validate(

+        {

+            "key": "smollm2-135m",

+            "hf_id": "HuggingFaceTB/SmolLM2-135M-Instruct",

+            "revision": _REV,

+            "architecture": "LlamaForCausalLM",

+            "params": 135_000_000,

+            "target_modules": ["q_proj", "v_proj"],

+            "template": "chatml",

+            "gguf_arch": "llama",

+            "tokenizer_pre": "default",

+            "license_spdx": "Apache-2.0",

+            "license_url": None,

+            "requires_acceptance": False,

+            "redistributable": True,

+            "size_gb_fp16": 0.3,

+            "context_length": 4096,

+            "recommended_seq_len": 2048,

+        }

+    )

+

+

+class _FakeTeacher:

+    def __init__(self, name: str, payload: str) -> None:

+        self.name = name

+        self._payload = payload

+

+    def generate(self, *_args: object, **_kwargs: object) -> str:

+        return self._payload

+

+

+class _FakeJudge:

+    name = "sway:preference_judge"

+    suggested_threshold = 0.1

+

+    def score_pair(self, prompt: str, candidate_a: str, candidate_b: str) -> PairScore:

+        _ = prompt, candidate_a, candidate_b

+        return PairScore(score_a=0.8, score_b=0.2)

+

+

+class _FakeBackend:

+    def __init__(self, responses: dict[str, list[str]]) -> None:

+        self._responses = {prompt: deque(items) for prompt, items in responses.items()}

+

+    def load(self, spec: object, store: object, *, adapter_name: str | None = None) -> None:

+        _ = spec, store, adapter_name

+

+    def generate(self, prompt: str, **_kwargs: object) -> str:

+        return self._responses[prompt].popleft()

+

+    def unload(self) -> None:

+        return None

+

+

+def _patch_synth_runtime(monkeypatch: pytest.MonkeyPatch) -> None:

+    payloads = {

+        "self": ('[{"question":"What does DGEMM do?","answer":"It multiplies dense matrices."}]'),

+        "hf:stub/model": (

+            '[{"question":"When would you call DGEMM?","answer":"When you need a BLAS matrix multiplication."}]'

+        ),

+    }

+

+    def _build_teacher(raw: str, **_kwargs: object) -> _FakeTeacher:

+        payload = payloads.get(raw, payloads["self"])

+        return _FakeTeacher(raw, payload)

+

+    monkeypatch.setattr("dlm.synth.build_teacher", _build_teacher)

+    monkeypatch.setattr("dlm.preference.build_judge", lambda *args, **kwargs: _FakeJudge())

+

+

+def _patch_preference_alias_runtime(monkeypatch: pytest.MonkeyPatch) -> None:

+    monkeypatch.setattr("dlm.base_models.resolve", lambda *args, **kwargs: _spec())

+    monkeypatch.setattr(

+        "dlm.hardware.doctor",

+        lambda: type("R", (), {"capabilities": object()})(),

+    )

+    monkeypatch.setattr(

+        "dlm.inference.backends.select_backend",

+        lambda *args, **kwargs: "pytorch",

+    )

+    monkeypatch.setattr(

+        "dlm.inference.backends.build_backend",

+        lambda *args, **kwargs: _FakeBackend({"What is DGEMM?": ["bad answer", "good answer"]}),

+    )

+    monkeypatch.setattr("dlm.preference.build_judge", lambda *args, **kwargs: _FakeJudge())

+

+

+class TestSynthCmd:

+    def test_instructions_stage_pending_plan_by_default(

+        self,

+        tmp_path: Path,

+        monkeypatch: pytest.MonkeyPatch,

+    ) -> None:

+        home = tmp_path / "home"

+        doc = tmp_path / "doc.dlm"

+        _write_synth_doc(doc)

+        _patch_synth_runtime(monkeypatch)

+

+        runner = CliRunner()

+        result = runner.invoke(

+            app,

+            ["--home", str(home), "synth", "instructions", str(doc), "--per-section", "1"],

+        )

+

+        assert result.exit_code == 0, result.output

+        normalized = _normalized_output(result)

+        assert "synth plan: 1 add, 0 skip" in normalized

+        assert "synth filter: generated 1, dedup 1, judge passed 1, threshold 1" in normalized

+        assert "staged 1 auto-synth instruction section" in normalized

+

+        pending = load_pending_plan(for_dlm(_DLM_ID, home=home))

+        assert pending is not None

+        assert len(pending.sections) == 1

+        assert pending.sections[0].auto_synth is True

+        assert pending.sections[0].synth_teacher == "self"

+        assert pending.sections[0].synth_strategy == "extraction"

+

+    def test_apply_writes_auto_synth_sections_and_clears_pending(

+        self,

+        tmp_path: Path,

+        monkeypatch: pytest.MonkeyPatch,

+    ) -> None:

+        home = tmp_path / "home"

+        doc = tmp_path / "doc.dlm"

+        _write_synth_doc(doc)

+        _patch_synth_runtime(monkeypatch)

+

+        runner = CliRunner()

+        result = runner.invoke(

+            app,

+            [

+                "--home",

+                str(home),

+                "synth",

+                "instructions",

+                str(doc),

+                "--per-section",

+                "1",

+                "--apply",

+            ],

+        )

+

+        assert result.exit_code == 0, result.output

+        normalized = _normalized_output(result)

+        assert "synth apply plan: 1 add, 0 skip" in normalized

+        assert "wrote 1 section(s)" in normalized

+        assert load_pending_plan(for_dlm(_DLM_ID, home=home)) is None

+

+        parsed = parse_file(doc)

+        assert any(

+            section.type is SectionType.INSTRUCTION and section.auto_synth

+            for section in parsed.sections

+        )

+

+    def test_revert_strips_auto_synth_sections(

+        self,

+        tmp_path: Path,

+        monkeypatch: pytest.MonkeyPatch,

+    ) -> None:

+        home = tmp_path / "home"

+        doc = tmp_path / "doc.dlm"

+        _write_synth_doc(doc)

+        _patch_synth_runtime(monkeypatch)

+

+        runner = CliRunner()

+        apply_result = runner.invoke(

+            app,

+            [

+                "--home",

+                str(home),

+                "synth",

+                "instructions",

+                str(doc),

+                "--per-section",

+                "1",

+                "--apply",

+            ],

+        )

+        assert apply_result.exit_code == 0, apply_result.output

+

+        revert_result = runner.invoke(

+            app,

+            ["--home", str(home), "synth", "revert", str(doc)],

+        )

+

+        assert revert_result.exit_code == 0, revert_result.output

+        assert "stripped 1 auto-synth instruction section" in _normalized_output(revert_result)

+

+        parsed = parse_file(doc)

+        assert not any(

+            section.type is SectionType.INSTRUCTION and section.auto_synth

+            for section in parsed.sections

+        )

+

+    def test_list_shows_counts_for_applied_and_pending_sections(

+        self,

+        tmp_path: Path,

+        monkeypatch: pytest.MonkeyPatch,

+    ) -> None:

+        home = tmp_path / "home"

+        doc = tmp_path / "doc.dlm"

+        _write_synth_doc(doc)

+        _patch_synth_runtime(monkeypatch)

+

+        runner = CliRunner()

+        apply_result = runner.invoke(

+            app,

+            [

+                "--home",

+                str(home),

+                "synth",

+                "instructions",

+                str(doc),

+                "--per-section",

+                "1",

+                "--strategy",

+                "extraction",

+                "--apply",

+            ],

+        )

+        assert apply_result.exit_code == 0, apply_result.output

+

+        stage_result = runner.invoke(

+            app,

+            [

+                "--home",

+                str(home),

+                "synth",

+                "instructions",

+                str(doc),

+                "--teacher",

+                "hf:stub/model",

+                "--per-section",

+                "1",

+                "--strategy",

+                "expansion",

+            ],

+        )

+        assert stage_result.exit_code == 0, stage_result.output

+

+        list_result = runner.invoke(

+            app,

+            ["--home", str(home), "synth", "list", str(doc)],

+        )

+

+        assert list_result.exit_code == 0, list_result.output

+        normalized = _normalized_output(list_result)

+        source_id = parse_file(doc).sections[0].section_id

+        assert "applied auto-synth: 1" in normalized

+        assert "staged pending: 1" in normalized

+        assert "self: 1" in normalized

+        assert "hf:stub/model: 1" in normalized

+        assert "extraction: 1" in normalized

+        assert "expansion: 1" in normalized

+        assert f"{source_id}: 1" in normalized

+

+    def test_preferences_alias_routes_through_preference_mine(

+        self,

+        tmp_path: Path,

+        monkeypatch: pytest.MonkeyPatch,

+    ) -> None:

+        home = tmp_path / "home"

+        doc = tmp_path / "doc.dlm"

+        _write_preference_doc(doc)

+        _write_manifest(home, doc)

+        _patch_preference_alias_runtime(monkeypatch)

+

+        runner = CliRunner()

+        result = runner.invoke(

+            app,

+            [

+                "--home",

+                str(home),

+                "synth",

+                "preferences",

+                str(doc),

+                "--samples",

+                "2",

+            ],

+        )

+

+        assert result.exit_code == 0, result.output

+        assert "preference mine plan: 1 add, 0 skip" in _normalized_output(result)

+        pending = load_pending_preference_plan(for_dlm(_DLM_ID, home=home))

+        assert pending is not None

+        assert len(pending.sections) == 1

+        assert pending.sections[0].auto_mined is True