Merge

dotagents._merge

Managed-block merge for init's AGENTS.md and the harnesses' own entry files.

init must never clobber a user-customized AGENTS.md. Content owned by dotagents is delimited by literal marker lines and treated as a block within the file: create if missing, prepend if present-without-markers, refresh-in-place if the markers are already there. Detection is by marker presence only, never by prose matching, so it survives user reformatting.

A marker counts only when it is a LINE of its own (surrounding whitespace allowed): the base AGENTS.md itself mentions the marker names in prose, and a substring match would treat that mention as a block.

BEGIN_MARKER = '<!-- dotagents:begin -->' module-attribute

CONTEXT_BEGIN_MARKER = '<!-- dotagents:context:begin -->' module-attribute

CONTEXT_END_MARKER = '<!-- dotagents:context:end -->' module-attribute

END_MARKER = '<!-- dotagents:end -->' module-attribute

find_block(text, begin_marker=BEGIN_MARKER, end_marker=END_MARKER)

(start, end) character span of the managed block in text -- from the first BEGIN marker line through the first END marker line after it, ignoring marker lines inside fenced code -- or None when there is no block. A BEGIN with no END after it is not a block (the caller decides whether that is an error).

Source code in src/dotagents/_merge.py

def find_block(
    text: str, begin_marker: str = BEGIN_MARKER, end_marker: str = END_MARKER
) -> "tuple[int, int] | None":
    """``(start, end)`` character span of the managed block in ``text`` -- from
    the first BEGIN marker line through the first END marker line after it,
    ignoring marker lines inside fenced code -- or ``None`` when there is no
    block. A BEGIN with no END after it is not a block (the caller decides
    whether that is an error)."""
    begins = _marker_lines(text, begin_marker)
    if not begins:
        return None
    ends = _marker_lines(text, end_marker, begins[0].end())
    if not ends:
        return None
    return begins[0].start(), ends[0].end()

merge_block(target, block_source_text, *, force=False, dry_run=False, backup_root=None, begin_marker=BEGIN_MARKER, end_marker=END_MARKER, append=False)

Merge block_source_text (a fully marker-wrapped skeleton file's contents) into target, returning the branch taken: "created" / "block-inserted" / "block-refreshed" / "skipped (present)" (the block is already current) and, with force, "unchanged" / "replaced (--force, backed up)" / "replaced (--force)".

Never overwrites content outside the markers unless force is True (then the whole file is replaced, after being backed up under backup_root if the target pre-exists -- at a path mirroring the target's, and never over an earlier backup of the same file; "backed up" is reported only when a copy was made).

begin_marker/end_marker default to the Markdown/HTML comment pair; pass a different pair for other comment syntaxes (e.g. #-prefixed markers for TOML).

append puts the block at the END of an existing file rather than the start. Required for TOML: a [table] header captures every key line that follows it, so prepending a table would silently swallow the user's existing top-level keys into our table.

Files are written LF-only (:func:dotagents._fs.write_text_lf). A target carrying a BEGIN marker line with no END after it is malformed and refused (SystemExit) rather than gaining a second block.

Source code in src/dotagents/_merge.py

def merge_block(
    target: Path,
    block_source_text: str,
    *,
    force: bool = False,
    dry_run: bool = False,
    backup_root: "Path | None" = None,
    begin_marker: str = BEGIN_MARKER,
    end_marker: str = END_MARKER,
    append: bool = False,
) -> str:
    """Merge `block_source_text` (a fully marker-wrapped skeleton file's
    contents) into `target`, returning the branch taken:
    "created" / "block-inserted" / "block-refreshed" / "skipped (present)"
    (the block is already current) and, with `force`, "unchanged" /
    "replaced (--force, backed up)" / "replaced (--force)".

    Never overwrites content outside the markers unless `force` is True (then
    the whole file is replaced, after being backed up under `backup_root` if the
    target pre-exists -- at a path mirroring the target's, and never over an
    earlier backup of the same file; "backed up" is reported only when a copy
    was made).

    `begin_marker`/`end_marker` default to the Markdown/HTML comment pair; pass a
    different pair for other comment syntaxes (e.g. `#`-prefixed markers for TOML).

    `append` puts the block at the END of an existing file rather than the start.
    Required for TOML: a `[table]` header captures every key line that follows it,
    so prepending a table would silently swallow the user's existing top-level keys
    into our table.

    Files are written LF-only (:func:`dotagents._fs.write_text_lf`). A target
    carrying a BEGIN marker line with no END after it is malformed and refused
    (``SystemExit``) rather than gaining a second block.
    """
    block = _extract_block(block_source_text, begin_marker, end_marker)

    if force:
        if not target.exists():
            if not dry_run:
                write_text_lf(target, block_source_text, atomic=True)
            return "created"
        if target.read_text(encoding="utf-8-sig") == block_source_text:
            return "unchanged"
        backed_up = False
        if backup_root is not None:
            backed_up = dry_run or _backup(target, backup_root)
        if not dry_run:
            write_text_lf(target, block_source_text, atomic=True)
        return "replaced (--force, backed up)" if backed_up else "replaced (--force)"

    if not target.exists():
        if not dry_run:
            write_text_lf(target, block_source_text, atomic=True)
        return "created"

    # utf-8-sig: a leading BOM (PowerShell 5's `-Encoding UTF8`) would otherwise
    # hide a block on line 1 from the `^` anchor; it is not written back.
    existing = target.read_text(encoding="utf-8-sig")

    span = find_block(existing, begin_marker, end_marker)
    if span is not None:
        start, end = span
        new_text = existing[:start] + block + existing[end:]
        if new_text == existing:
            return "skipped (present)"
        if not dry_run:
            write_text_lf(target, new_text, atomic=True)
        return "block-refreshed"
    if _marker_lines(existing, begin_marker):
        raise SystemExit(
            "error: %s has a %r line with no %r after it -- fix the file, then re-run"
            % (target, begin_marker, end_marker)
        )

    if append:
        sep = "" if existing.endswith("\n") else "\n"
        new_text = existing + sep + "\n" + block + "\n"
    else:
        new_text = block + "\n\n" + existing
    if not dry_run:
        write_text_lf(target, new_text, atomic=True)
    return "block-inserted"

merge_context_block(target, context_text, *, dry_run=False)

Write an assembled context into target as a managed dotagents:context block: created if the file is absent, refreshed in place if the block is there, appended after the user's own content otherwise. Never a raw overwrite: the target may be a context SOURCE (a harness's own AGENTS.md), and overwriting it would re-inline the previous run's output.

Marker lines inside context_text (a store's AGENTS.md carries its own dotagents:begin/end) are dropped: embedded, a later include merge into the same file took them for its block and overwrote the context between them.

Source code in src/dotagents/_merge.py

def merge_context_block(target: Path, context_text: str, *, dry_run: bool = False) -> str:
    """Write an assembled context into ``target`` as a managed
    ``dotagents:context`` block: created if the file is absent, refreshed in
    place if the block is there, appended after the user's own content
    otherwise. Never a raw overwrite: the target may be a context SOURCE (a
    harness's own ``AGENTS.md``), and overwriting it would re-inline the
    previous run's output.

    Marker lines inside ``context_text`` (a store's ``AGENTS.md`` carries its
    own ``dotagents:begin``/``end``) are dropped: embedded, a later include
    merge into the same file took them for its block and overwrote the
    context between them."""
    body = "\n".join(
        ln for ln in context_text.strip().splitlines()
        if not _ANY_MARKER_RE.match(ln)
    )
    block_text = "%s\n%s\n%s\n" % (CONTEXT_BEGIN_MARKER, body, CONTEXT_END_MARKER)
    return merge_block(
        target, block_text, dry_run=dry_run,
        begin_marker=CONTEXT_BEGIN_MARKER, end_marker=CONTEXT_END_MARKER, append=True,
    )

merge_include_line(target, include_line, *, force=False, dry_run=False, backup_root=None)

Managed-block merge of a one-line @<path> include into a harness's entry file (~/.claude/CLAUDE.md, <project>/.claude/CLAUDE.md, ~/.gemini/GEMINI.md), so the harness actually loads the store's AGENTS.md.

If the file already carries include_line ANYWHERE (a hand-written include, inside or outside a managed block), it is left untouched (branch "skipped (present)") -- the include is the point, not the markers. Otherwise the marker-wrapped line is merged like any block, APPENDED so an existing file's own content stays first.

Source code in src/dotagents/_merge.py

def merge_include_line(
    target: Path,
    include_line: str,
    *,
    force: bool = False,
    dry_run: bool = False,
    backup_root: "Path | None" = None,
) -> str:
    """Managed-block merge of a one-line ``@<path>`` include into a harness's
    entry file (``~/.claude/CLAUDE.md``, ``<project>/.claude/CLAUDE.md``,
    ``~/.gemini/GEMINI.md``), so the harness actually loads the store's
    ``AGENTS.md``.

    If the file already carries ``include_line`` ANYWHERE (a hand-written include,
    inside or outside a managed block), it is left untouched (branch
    "skipped (present)") -- the include is the point, not the markers. Otherwise
    the marker-wrapped line is merged like any block, APPENDED so an existing
    file's own content stays first."""
    line = include_line.strip()
    if not force and target.exists():
        existing = target.read_text(encoding="utf-8-sig")
        if any(ln.strip() == line for ln in existing.splitlines()):
            return "skipped (present)"
        # Any spelling of the same file (`@~/.agents/AGENTS.md`, an absolute
        # path) already includes it; a second include would load it twice.
        wanted = _import_target(target, line[1:] if line.startswith("@") else line)
        for ref in re.findall(r"(?<!\S)@(\S+)", existing):
            if wanted is not None and _import_target(target, ref) == wanted:
                return "skipped (present)"
    block_text = "%s\n%s\n%s\n" % (BEGIN_MARKER, line, END_MARKER)
    return merge_block(
        target, block_text, force=force, dry_run=dry_run, backup_root=backup_root,
        append=True,
    )

remove_block(target, *, dry_run=False, begin_marker=BEGIN_MARKER, end_marker=END_MARKER)

Delete the managed block (markers included) from target, with the one blank line merge_block(append=True) put before it. Everything else stays. Returns "removed" or "absent" (no file, or no block in it).

Source code in src/dotagents/_merge.py

def remove_block(
    target: Path,
    *,
    dry_run: bool = False,
    begin_marker: str = BEGIN_MARKER,
    end_marker: str = END_MARKER,
) -> str:
    """Delete the managed block (markers included) from `target`, with the one
    blank line `merge_block(append=True)` put before it. Everything else stays.
    Returns "removed" or "absent" (no file, or no block in it)."""
    if not target.exists():
        return "absent"
    existing = target.read_text(encoding="utf-8-sig")
    span = find_block(existing, begin_marker, end_marker)
    if span is None:
        return "absent"
    start, end = span
    head, tail = existing[:start], existing[end:]
    if tail.startswith("\n"):
        tail = tail[1:]
    if head.endswith("\n\n"):
        head = head[:-1]
    if not dry_run:
        write_text_lf(target, head + tail, atomic=True)
    return "removed"

timestamped_backup_root(dest)

<dest>/install_backup/<timestamp>, ONE per store per process: every adapter of an init --force run backs up into the same root, so :func:_backup's never-overwrite rule sees the earlier copies.

Source code in src/dotagents/_merge.py

def timestamped_backup_root(dest: Path) -> Path:
    """``<dest>/install_backup/<timestamp>``, ONE per store per process: every
    adapter of an ``init --force`` run backs up into the same root, so
    :func:`_backup`'s never-overwrite rule sees the earlier copies."""
    key = Path(dest).resolve()
    if key not in _backup_roots:
        _backup_roots[key] = Path(dest) / "install_backup" / time.strftime("%Y%m%d-%H%M%S")
    return _backup_roots[key]