Skip to content

jiti.cli.merge.orchestrator

Internal API

Documented for transparency — the supported public surface is the top-level jiti package. Internals can change in any release.

jiti.cli.merge.orchestrator

Drive jiti merge: resolve targets, gate, dispatch source rewrites and test folding.

Gating runs in full before any write because introspecting one stub reads source by line number — rewriting an earlier section would corrupt the next one's spec check. So we resolve every chosen section against the intact source first, then apply the (pure-text) rewrites. After impls land in source, test files are folded in (see tests.py), and ruff formats every touched file in a single batch.

Classes

Functions

source_files 

source_files(root: Path) -> dict[str, Path]

Map each importable module in the project to its source file (excludes the mirror).

Source code in src/jiti/cli/merge/orchestrator.py 
31
32
33
34
35
36
37
def source_files(root: Path) -> dict[str, Path]:
    """Map each importable module in the project to its source file (excludes the mirror)."""
    mapping: dict[str, Path] = {}
    for path in walk_py_files(root):
        name, _ = module_name(path)
        mapping.setdefault(name, path)
    return mapping

select 

select(targets: Sequence[str], refs: Sequence[SectionRef]) -> list[SectionRef]

Resolve user targets (file path, dotted module, or qualname) to the sections they name.

Source code in src/jiti/cli/merge/orchestrator.py 
40
41
42
43
44
45
46
47
48
def select(targets: Sequence[str], refs: Sequence[SectionRef]) -> list[SectionRef]:
    """Resolve user targets (file path, dotted module, or qualname) to the sections they name."""
    chosen: dict[str, SectionRef] = {}
    for target in targets:
        matches = _match(target, refs)
        if not matches:
            raise MergeError(f"no generated section matched '{target}' — try `jiti status`.")
        chosen.update((ref.key, ref) for ref in matches)
    return list(chosen.values())

run_merge 

run_merge(
    root: Path, targets: Sequence[str], merge_all: bool, dry_run: bool, prune_scratch: bool = False
) -> int

Fold the selected generated sections back into their source files. Returns an exit code.

--prune drops the agent's scratch tests instead of ejecting them to a real test file.

Source code in src/jiti/cli/merge/orchestrator.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def run_merge(
    root: Path,
    targets: Sequence[str],
    merge_all: bool,
    dry_run: bool,
    prune_scratch: bool = False,
) -> int:
    """Fold the selected generated sections back into their source files. Returns an exit code.

    `--prune` drops the agent's scratch tests instead of ejecting them to a real test file.
    """
    mirror = root / ".jiti"
    refs = inventory(mirror)
    if not refs:
        print("nothing to merge (.jiti/ is empty or absent).")
        return 0

    chosen = list(refs) if merge_all else select(targets, refs)
    sources = source_files(root)
    store = JitiStore(mirror)
    # required_for gates contribute to each declaration's spec hash, so test modules must be
    # imported before the gate check below — otherwise the check sees a stale spec.
    import_test_modules([str(root)])

    plans: list[tuple[SectionRef, Path, Action]] = []
    blocked = False
    for ref in chosen:
        try:
            plans.append(_gate(ref, sources, store))
        except MergeError as error:
            print(f"skip {ref.key}: {error}")
            blocked = blocked or not merge_all

    # `apply_section` strips @jiti from source, leaving the wrapper unreachable via import.
    # Snapshot each ref's gate locations now while the wrapper is still bound to the user's stub.
    gate_index: dict[str, list[GateLocation]] = {
        ref.key: gate_locations_for(ref) for ref, _, _ in plans
    }

    written: list[Path] = []
    for ref, source_path, action in plans:
        if not dry_run:
            written.append(apply_section(ref, source_path))
        print(f"{'would merge' if dry_run else 'merged'} {ref.key}  ({action.value})")

    if not dry_run and plans:
        written.extend(merge_test_files(root, mirror, plans, gate_index, prune_scratch))
        _ruff_batch(written)
        remove_empty_dirs(mirror)
        if not _any_jiti_left(sources):
            print("no @jiti remains — you can drop jiti from your dependencies.")
    print(f"\n{'would merge' if dry_run else 'merged'} {len(plans)} section(s).")
    return 1 if blocked else 0