our-claude-skills/docs/superpowers/plans/2026-06-27-notion-writer-cli-enhancements.md

Notion Writer CLI Enhancements Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add local file/image uploads (via the ntn CLI) and an opt-in native-markdown write engine to the notion-writer skill, without regressing the existing block-converter path.

Architecture: Keep the token-based notion-client core. Add two focused, pure-where-possible modules — ntn_files.py (subprocess upload) and md_translate.py (dialect→Notion-markdown). The block parser stays pure; uploads happen in an isolated post-parse walk. A --engine flag selects between the existing block converter (default) and the new markdown endpoints.

Tech Stack: Python 3.12, notion-client>=2.0.0, python-dotenv, stdlib subprocess/shutil/unittest.mock, the ntn CLI (v0.18.0, already installed + logged in), Notion API 2025-09-03 (blocks) / 2026-03-11 (markdown).

Global Constraints

• All work happens on branch notion-writer-cli-enhancements in repo ~/Project/our-claude-skills.
• Skill code dir (all relative paths below are under it): custom-skills/32-notion-writer/code/scripts/.
• Tests use the existing custom harness (plain assert functions + a run_all() at bottom), run with venv/bin/python test_<name>.py. No pytest.
• The existing test_parser.py has 32 tests and must stay green after every task. (The spec's "28" is a stale count; the real number is 32.)
• Default engine is blocks — existing CLI calls must behave identically. Markdown is opt-in via --engine markdown.
• Markdown write calls use a client built with notion_version="2026-03-11". Schema/property/upsert-lookup calls stay on the default-version client.
• Parser purity is preserved: markdown_to_notion_blocks / _parse_lines make no I/O calls. Uploads live only in ntn_files.py and the materialize_local_media walk.
• client.request(path=..., method=..., body=...) paths omit the v1/ prefix (the SDK adds it): use "pages" and f"pages/{page_id}/markdown".
• Commit after every task. Use git add <explicit files> (never git add -A). End commit messages with the trailer: Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---

File Structure

File	Status	Responsibility
ntn_files.py	create	preflight(), upload(path), materialize_local_media(blocks, base_dir, upload_fn), NtnUploadError
md_translate.py	create	translate(content) + per-construct translators (callout/columns/toggle/mention)
notion_writer.py	modify	IMAGE_RE, create_image_block, extract_local_images, _content_has_local_images, CLI flags, engine routing, two-phase image append
_notion_compat.py	modify	make_client(api_key, notion_version=None), create_page_markdown, append_markdown, replace_markdown, markdown cases in explain_api_error
test_ntn_files.py	create	unit tests for ntn_files (subprocess + walk mocked)
test_md_translate.py	create	unit tests for the translator
test_engine_routing.py	create	unit tests for compat markdown helpers + CLI routing (client mocked)
test_parser.py	modify	image-block tests
requirements.txt	modify	(no dep change expected; touched only if needed)
repo .gitignore	modify	ignore the skill's venv/
code/CLAUDE.md, SKILL.md	modify	document engines, uploads, flags; v1.3.0

---

Task 1: Environment baseline (venv + green existing tests)

Files:

• Create: custom-skills/32-notion-writer/code/scripts/venv/ (not committed)
• Modify: repo .gitignore

Interfaces:

• Consumes: nothing.

• Produces: a working venv/bin/python with deps installed; baseline of 32 green tests.

• Step 1: Create the venv

Run:

cd ~/Project/our-claude-skills/custom-skills/32-notion-writer/code/scripts
python3 -m venv venv

• Step 2: Install dependencies

Run:

venv/bin/python -m pip install -r requirements.txt

Expected: installs python-dotenv and notion-client (and transitive deps) with no error.

• Step 3: Run the existing test suite (baseline)

Run:

venv/bin/python test_parser.py

Expected: ends with ✅ All 32 tests passed.

• Step 4: Ignore the venv in git

Append to the repo-root .gitignore (~/Project/our-claude-skills/.gitignore); create the file if missing. Add this line:

custom-skills/32-notion-writer/code/scripts/venv/

Verify it is ignored:

cd ~/Project/our-claude-skills && git status --short custom-skills/32-notion-writer/code/scripts/

Expected: no venv/ entries appear.

• Step 5: Commit

cd ~/Project/our-claude-skills
git add .gitignore
git commit -m "$(printf 'chore(notion-writer): ignore scripts venv\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 2: ntn_files.py — preflight + upload

Files:

• Create: custom-skills/32-notion-writer/code/scripts/ntn_files.py
• Test: custom-skills/32-notion-writer/code/scripts/test_ntn_files.py

Interfaces:

• Consumes: nothing (stdlib only).

• Produces:

  • class NtnUploadError(Exception) — attributes .path (str), .stderr (str).
  • preflight() -> dict — returns {"workspace_name": str, "workspace_id": str}; raises NtnUploadError if ntn is missing or not logged in. Cached after first success.
  • upload(path) -> str — uploads a local file via ntn files create --plain, returns the file_upload_id. Raises NtnUploadError on failure.

• Step 1: Write the failing tests

Create test_ntn_files.py:

#!/usr/bin/env python3
"""Tests for ntn_files.py — run with `python test_ntn_files.py`."""

import sys
import subprocess
from pathlib import Path
from unittest import mock

sys.path.insert(0, str(Path(__file__).parent))

import ntn_files
from ntn_files import NtnUploadError


def _reset_cache():
    ntn_files._WORKSPACE = None


def test_preflight_missing_ntn():
    _reset_cache()
    with mock.patch("shutil.which", return_value=None):
        try:
            ntn_files.preflight()
            assert False, "expected NtnUploadError"
        except NtnUploadError as e:
            assert "ntn" in str(e).lower()


def test_preflight_returns_workspace():
    _reset_cache()
    fake = subprocess.CompletedProcess(
        args=[], returncode=0,
        stdout='{"bot":{"workspace_name":"D.Intelligence",'
               '"workspace_id":"ws-123"}}',
        stderr="",
    )
    with mock.patch("shutil.which", return_value="/usr/bin/ntn"), \
         mock.patch("subprocess.run", return_value=fake):
        info = ntn_files.preflight()
        assert info["workspace_name"] == "D.Intelligence"
        assert info["workspace_id"] == "ws-123"


def test_upload_returns_id():
    _reset_cache()
    fake = subprocess.CompletedProcess(
        args=[], returncode=0,
        stdout="43833259-72ae-404e-8441-b6577f3159b4\tphoto.png\tuploaded\n",
        stderr="",
    )
    with mock.patch("subprocess.run", return_value=fake):
        upload_id = ntn_files.upload(Path("/tmp/photo.png"))
        assert upload_id == "43833259-72ae-404e-8441-b6577f3159b4"


def test_upload_failure_raises():
    _reset_cache()
    fake = subprocess.CompletedProcess(
        args=[], returncode=1, stdout="", stderr="boom: invalid file",
    )
    with mock.patch("subprocess.run", return_value=fake):
        try:
            ntn_files.upload(Path("/tmp/bad.png"))
            assert False, "expected NtnUploadError"
        except NtnUploadError as e:
            assert e.path == "/tmp/bad.png"
            assert "boom" in e.stderr


def run_all():
    tests = [
        test_preflight_missing_ntn,
        test_preflight_returns_workspace,
        test_upload_returns_id,
        test_upload_failure_raises,
    ]
    for t in tests:
        print(f"\n{t.__name__}")
        t()
    print("\n" + "=" * 50)
    print(f"✅ All {len(tests)} tests passed")
    print("=" * 50)


if __name__ == "__main__":
    run_all()

• Step 2: Run tests to verify they fail

Run: venv/bin/python test_ntn_files.py Expected: FAIL with ModuleNotFoundError: No module named 'ntn_files'.

• Step 3: Write ntn_files.py

Create ntn_files.py:

#!/usr/bin/env python3
"""Local file uploads to Notion via the `ntn` CLI.

Owns all subprocess I/O for the skill. The CLI handles the full File Upload
lifecycle (create -> send bytes -> complete) and prints the upload ID.
"""

from __future__ import annotations

import json
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Dict, List, Optional, Callable, Any

_WORKSPACE: Optional[Dict[str, str]] = None


class NtnUploadError(Exception):
    """Raised when `ntn` is unavailable or a file upload fails."""

    def __init__(self, message: str, path: str = "", stderr: str = ""):
        super().__init__(message)
        self.path = path
        self.stderr = stderr


def preflight() -> Dict[str, str]:
    """Verify `ntn` is installed and logged in; return its target workspace.

    Cached for the process. Raises NtnUploadError with an actionable hint
    on failure. Prints one informational line naming the workspace `ntn`
    targets (file uploads are workspace-scoped).
    """
    global _WORKSPACE
    if _WORKSPACE is not None:
        return _WORKSPACE

    if shutil.which("ntn") is None:
        raise NtnUploadError(
            "ntn CLI not found. Install it with: "
            "curl -fsSL https://ntn.dev | bash"
        )

    result = subprocess.run(
        ["ntn", "api", "v1/users/me"],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        raise NtnUploadError(
            "ntn is not logged in. Run: ntn login\n" + result.stderr.strip()
        )

    try:
        me = json.loads(result.stdout)
        bot = me.get("bot", {})
        info = {
            "workspace_name": bot.get("workspace_name", "unknown"),
            "workspace_id": bot.get("workspace_id", "unknown"),
        }
    except (ValueError, AttributeError):
        info = {"workspace_name": "unknown", "workspace_id": "unknown"}

    print(f'ntn -> workspace "{info["workspace_name"]}"', file=sys.stderr)
    _WORKSPACE = info
    return info


def upload(path: Path) -> str:
    """Upload a local file via `ntn files create --plain`; return its id."""
    path = Path(path)
    with open(path, "rb") as fh:
        result = subprocess.run(
            ["ntn", "files", "create", "--plain"],
            stdin=fh, capture_output=True, text=True,
        )
    if result.returncode != 0:
        raise NtnUploadError(
            f"ntn upload failed for {path}", path=str(path),
            stderr=result.stderr.strip(),
        )
    first_line = result.stdout.strip().splitlines()[0]
    upload_id = first_line.split("\t")[0].strip()
    return upload_id

• Step 4: Run tests to verify they pass

Run: venv/bin/python test_ntn_files.py Expected: ✅ All 4 tests passed.

• Step 5: Commit

cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/ntn_files.py \
        custom-skills/32-notion-writer/code/scripts/test_ntn_files.py
git commit -m "$(printf 'feat(notion-writer): ntn_files upload + preflight\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 3: Image blocks in the parser (pure)

Files:

• Modify: custom-skills/32-notion-writer/code/scripts/notion_writer.py
• Test: custom-skills/32-notion-writer/code/scripts/test_parser.py

Interfaces:

• Consumes: parse_rich_text (existing).

• Produces:

  • IMAGE_RE — compiled regex matching a standalone image line.
  • create_image_block(alt: str, target: str) -> dict — always external shape, caption = parse_rich_text(alt).
  • _parse_lines emits an image block for standalone ![alt](target) lines.

• Step 1: Write the failing tests

Add these test functions to test_parser.py (above run_all), and add their names to the tests list inside run_all:

def test_image_remote_external():
    blocks = markdown_to_notion_blocks("![a chart](https://ex.com/c.png)")
    assert len(blocks) == 1
    b = blocks[0]
    assert b["type"] == "image"
    assert b["image"]["type"] == "external"
    assert b["image"]["external"]["url"] == "https://ex.com/c.png"
    assert b["image"]["caption"][0]["text"]["content"] == "a chart"


def test_image_local_external_shape_preupload():
    blocks = markdown_to_notion_blocks("![local](./pics/x.png)")
    assert len(blocks) == 1
    assert blocks[0]["image"]["external"]["url"] == "./pics/x.png"


def test_image_only_when_standalone():
    # An inline bang-bracket inside prose is NOT an image block.
    blocks = markdown_to_notion_blocks("see ![x](y.png) inline")
    assert blocks[0]["type"] == "paragraph"

Add to the tests list in run_all:

        test_image_remote_external,
        test_image_local_external_shape_preupload,
        test_image_only_when_standalone,

• Step 2: Run tests to verify they fail

Run: venv/bin/python test_parser.py Expected: FAIL — test_image_remote_external errors because the first block is a paragraph (or the ! leaks), not an image.

• Step 3: Add IMAGE_RE and create_image_block

In notion_writer.py, add the constant next to TABLE_SEPARATOR_RE (near line 58):

IMAGE_RE = re.compile(r'^\s*!\[([^\]]*)\]\(([^)\s]+)\)\s*$')

Add the factory alongside the other create_*_block helpers (e.g., after create_divider_block):

def create_image_block(alt: str, target: str) -> Dict[str, Any]:
    """Image block in external shape. Local targets are converted to
    file_upload shape later by ntn_files.materialize_local_media."""
    block: Dict[str, Any] = {
        "object": "block",
        "type": "image",
        "image": {"type": "external", "external": {"url": target}},
    }
    if alt:
        block["image"]["caption"] = parse_rich_text(alt)
    return block

• Step 4: Add the image detector to _parse_lines

In _parse_lines, add this branch before the table detector (before the if _is_table_row(line)... line near line 213), so a standalone image line is caught before other block detectors:

        image_match = IMAGE_RE.match(line)
        if image_match:
            blocks.append(create_image_block(
                image_match.group(1), image_match.group(2)))
            i += 1
            continue

• Step 5: Run tests to verify they pass

Run: venv/bin/python test_parser.py Expected: ✅ All 35 tests passed (32 existing + 3 new).

• Step 6: Commit

cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/notion_writer.py \
        custom-skills/32-notion-writer/code/scripts/test_parser.py
git commit -m "$(printf 'feat(notion-writer): parse standalone images to image blocks\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 4: materialize_local_media walk

Files:

• Modify: custom-skills/32-notion-writer/code/scripts/ntn_files.py
• Test: custom-skills/32-notion-writer/code/scripts/test_ntn_files.py

Interfaces:

• Consumes: create_image_block output shape (block["image"]["external"]["url"]).

• Produces:

  • materialize_local_media(blocks, base_dir, upload_fn=upload) -> list — recursively walks blocks + nested children. For each image block whose external.url is not http(s): and resolves (relative to base_dir, or absolute) to an existing file, calls upload_fn(resolved_path) and rewrites the block to file_upload shape. Missing local file raises NtnUploadError. Remote/external untouched. upload_fn is injectable for testing.

• Step 1: Write the failing tests

Add to test_ntn_files.py (and register in run_all):

def _img(url):
    return {"object": "block", "type": "image",
            "image": {"type": "external", "external": {"url": url}}}


def test_materialize_remote_untouched():
    blocks = [_img("https://ex.com/a.png")]
    out = ntn_files.materialize_local_media(
        blocks, Path("/tmp"), upload_fn=lambda p: "SHOULD_NOT_RUN")
    assert out[0]["image"]["type"] == "external"


def test_materialize_local_uploaded(tmp_path=None):
    import tempfile, os
    d = Path(tempfile.mkdtemp())
    (d / "x.png").write_bytes(b"fake")
    blocks = [_img("x.png")]
    out = ntn_files.materialize_local_media(
        blocks, d, upload_fn=lambda p: "UP123")
    assert out[0]["image"]["type"] == "file_upload"
    assert out[0]["image"]["file_upload"]["id"] == "UP123"


def test_materialize_nested_children():
    import tempfile
    d = Path(tempfile.mkdtemp())
    (d / "y.png").write_bytes(b"fake")
    toggle = {"object": "block", "type": "toggle",
              "toggle": {"rich_text": [], "children": [_img("y.png")]}}
    out = ntn_files.materialize_local_media(
        [toggle], d, upload_fn=lambda p: "UPNESTED")
    child = out[0]["toggle"]["children"][0]
    assert child["image"]["file_upload"]["id"] == "UPNESTED"


def test_materialize_missing_file_raises():
    blocks = [_img("nope.png")]
    try:
        ntn_files.materialize_local_media(
            blocks, Path("/tmp"), upload_fn=lambda p: "x")
        assert False, "expected NtnUploadError"
    except NtnUploadError as e:
        assert "nope.png" in str(e) or "nope.png" in e.path

Register in run_all's tests list:

        test_materialize_remote_untouched,
        test_materialize_local_uploaded,
        test_materialize_nested_children,
        test_materialize_missing_file_raises,

• Step 2: Run tests to verify they fail

Run: venv/bin/python test_ntn_files.py Expected: FAIL — AttributeError: module 'ntn_files' has no attribute 'materialize_local_media'.

• Step 3: Implement the walk

Append to ntn_files.py:

_REMOTE_PREFIXES = ("http://", "https://")


def _resolve_local(url: str, base_dir: Path) -> Optional[Path]:
    """Return the resolved path for a local media url, or None if remote."""
    if url.startswith(_REMOTE_PREFIXES):
        return None
    p = Path(url)
    return p if p.is_absolute() else Path(base_dir) / p


def materialize_local_media(
    blocks: List[Dict[str, Any]],
    base_dir: Path,
    upload_fn: Callable[[Path], str] = upload,
) -> List[Dict[str, Any]]:
    """Walk blocks (and nested children); upload local image files and
    rewrite them to file_upload shape. Remote images are left as-is."""
    for block in blocks:
        if block.get("type") == "image":
            img = block["image"]
            if img.get("type") == "external":
                url = img.get("external", {}).get("url", "")
                resolved = _resolve_local(url, base_dir)
                if resolved is not None:
                    if not resolved.is_file():
                        raise NtnUploadError(
                            f"image references missing file: {url}",
                            path=str(resolved),
                        )
                    upload_id = upload_fn(resolved)
                    caption = img.get("caption")
                    new_img: Dict[str, Any] = {
                        "type": "file_upload",
                        "file_upload": {"id": upload_id},
                    }
                    if caption:
                        new_img["caption"] = caption
                    block["image"] = new_img
        # Recurse into any nested children list.
        btype = block.get("type")
        nested = block.get(btype, {}) if isinstance(block.get(btype), dict) else {}
        if isinstance(nested.get("children"), list):
            materialize_local_media(nested["children"], base_dir, upload_fn)
        if btype == "column_list":
            for col in nested.get("children", []):
                col_children = col.get("column", {}).get("children", [])
                materialize_local_media(col_children, base_dir, upload_fn)
    return blocks

• Step 4: Run tests to verify they pass

Run: venv/bin/python test_ntn_files.py Expected: ✅ All 8 tests passed.

• Step 5: Commit

cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/ntn_files.py \
        custom-skills/32-notion-writer/code/scripts/test_ntn_files.py
git commit -m "$(printf 'feat(notion-writer): materialize local images via upload walk\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 5: md_translate.py — dialect translator

Files:

• Create: custom-skills/32-notion-writer/code/scripts/md_translate.py
• Test: custom-skills/32-notion-writer/code/scripts/test_md_translate.py

Interfaces:

• Consumes: notion_writer.extract_notion_id (existing) for mention resolution.
• Produces: translate(content: str) -> str — converts skill dialect to Notion enhanced markdown. Pure.

Translation rules (everything not listed passes through unchanged): GitHub alert > [!TYPE] block → <callout icon=... color="..._bg"> with tab-indented body; Pandoc ::: columns/column/::: → <columns>/<column> with tab-indented children; <details>/<summary> toggle → children re-indented one tab; inline @[Title](id|url) → <mention-page url="...">Title</mention-page> (unresolvable → plain @Title).

• Step 1: Write the failing tests

Create test_md_translate.py:

#!/usr/bin/env python3
"""Tests for md_translate.py — run with `python test_md_translate.py`."""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

import md_translate


def test_passthrough_basics():
    src = "# Heading\n\n- item\n\n```python\nx=1\n```"
    assert md_translate.translate(src) == src


def test_callout_note():
    out = md_translate.translate("> [!NOTE]\n> Be careful")
    assert '<callout icon="ℹ️" color="blue_bg">' in out
    assert "\tBe careful" in out
    assert "</callout>" in out


def test_callout_warning_color():
    out = md_translate.translate("> [!WARNING]\n> Danger")
    assert 'color="yellow_bg"' in out
    assert 'icon="⚠️"' in out


def test_columns():
    src = "::: columns\n::: column\nLeft\n:::\n::: column\nRight\n:::\n:::"
    out = md_translate.translate(src)
    assert "<columns>" in out
    assert out.count("<column>") == 2
    assert "</columns>" in out
    assert "\tLeft" in out or "\t\tLeft" in out


def test_toggle_children_indented():
    src = "<details>\n<summary>More</summary>\nBody line\n</details>"
    out = md_translate.translate(src)
    assert "<summary>More</summary>" in out
    assert "\tBody line" in out


def test_mention_url():
    out = md_translate.translate(
        "See @[ADR](https://notion.so/X-abcdef0123456789abcdef0123456789).")
    assert "<mention-page url=" in out
    assert ">ADR</mention-page>" in out


def test_mention_invalid_plain():
    out = md_translate.translate("ping @[Bob](not-an-id)")
    assert "@Bob" in out
    assert "mention-page" not in out


def run_all():
    tests = [
        test_passthrough_basics,
        test_callout_note,
        test_callout_warning_color,
        test_columns,
        test_toggle_children_indented,
        test_mention_url,
        test_mention_invalid_plain,
    ]
    for t in tests:
        print(f"\n{t.__name__}")
        t()
    print("\n" + "=" * 50)
    print(f"✅ All {len(tests)} tests passed")
    print("=" * 50)


if __name__ == "__main__":
    run_all()

• Step 2: Run tests to verify they fail

Run: venv/bin/python test_md_translate.py Expected: FAIL with ModuleNotFoundError: No module named 'md_translate'.

• Step 3: Implement md_translate.py

Create md_translate.py:

#!/usr/bin/env python3
"""Translate the notion-writer markdown dialect into Notion enhanced
markdown for the markdown write engine. Pure functions, no I/O."""

from __future__ import annotations

import re
from typing import List

from notion_writer import extract_notion_id, format_id_with_dashes

# Skill alert type -> (emoji, Notion enhanced-markdown background color)
CALLOUT_MAP = {
    "NOTE": ("ℹ️", "blue_bg"),
    "TIP": ("💡", "green_bg"),
    "IMPORTANT": ("☝️", "purple_bg"),
    "WARNING": ("⚠️", "yellow_bg"),
    "CAUTION": ("🚨", "red_bg"),
}
_ALERT_RE = re.compile(r'^\s*\[!(NOTE|TIP|IMPORTANT|WARNING|CAUTION)\]\s*$')
_MENTION_RE = re.compile(r'@\[([^\]]+)\]\(([^)\s]+)\)')


def _translate_mentions(text: str) -> str:
    def repl(m: "re.Match") -> str:
        title, target = m.group(1), m.group(2)
        page_id = extract_notion_id(target)
        if page_id:
            return (f'<mention-page url="{format_id_with_dashes(page_id)}">'
                    f'{title}</mention-page>')
        return f"@{title}"
    return _MENTION_RE.sub(repl, text)


def _indent(lines: List[str]) -> List[str]:
    return ["\t" + ln if ln.strip() else ln for ln in lines]


def translate(content: str) -> str:
    lines = content.split("\n")
    out: List[str] = []
    i = 0
    while i < len(lines):
        line = lines[i]

        # Callout: > [!TYPE] then contiguous > body lines
        if line.lstrip().startswith(">"):
            body_first = line.lstrip()[1:].strip()
            alert = _ALERT_RE.match(body_first)
            if alert:
                emoji, color = CALLOUT_MAP[alert.group(1)]
                i += 1
                body: List[str] = []
                while i < len(lines) and lines[i].lstrip().startswith(">"):
                    body.append(lines[i].lstrip()[1:].lstrip())
                    i += 1
                out.append(f'<callout icon="{emoji}" color="{color}">')
                out.extend(_indent([_translate_mentions(b) for b in body]))
                out.append("</callout>")
                continue

        # Columns: ::: columns / ::: column / :::
        # State machine: "::: column" opens a column; ":::" closes a column
        # when one is open, otherwise closes the wrapper. The Pandoc layout
        # emits N "::: column" opens, N ":::" column-closes, then one final
        # ":::" wrapper-close.
        if line.strip() == "::: columns":
            i += 1
            cols: List[List[str]] = []
            cur: List[str] = None
            while i < len(lines):
                s = lines[i].strip()
                if s == "::: column":
                    if cur is not None:
                        cols.append(cur)
                    cur = []
                    i += 1
                elif s == ":::":
                    if cur is not None:      # close the open column
                        cols.append(cur)
                        cur = None
                        i += 1
                    else:                    # close the wrapper
                        i += 1
                        break
                else:
                    if cur is not None:
                        cur.append(lines[i])
                    i += 1
            out.append("<columns>")
            for col in cols:
                out.append("\t<column>")
                out.extend(_indent(_indent(
                    [_translate_mentions(c) for c in col])))
                out.append("\t</column>")
            out.append("</columns>")
            continue

        # Toggle: <details><summary>..</summary> body </details>
        if line.strip() == "<details>":
            out.append("<details>")
            i += 1
            if i < len(lines) and lines[i].lstrip().startswith("<summary>"):
                out.append(lines[i].strip())
                i += 1
            body = []
            while i < len(lines) and lines[i].strip() != "</details>":
                body.append(_translate_mentions(lines[i]))
                i += 1
            out.extend(_indent(body))
            out.append("</details>")
            if i < len(lines):  # skip closing </details>
                i += 1
            continue

        out.append(_translate_mentions(line))
        i += 1
    return "\n".join(out)

• Step 4: Run tests to verify they pass

Run: venv/bin/python test_md_translate.py Expected: ✅ All 7 tests passed.

• Step 5: Confirm the parser suite still passes (no import cycle)

Run: venv/bin/python test_parser.py Expected: ✅ All 35 tests passed (importing md_translate must not break notion_writer).

• Step 6: Commit

cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/md_translate.py \
        custom-skills/32-notion-writer/code/scripts/test_md_translate.py
git commit -m "$(printf 'feat(notion-writer): dialect->Notion enhanced markdown translator\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 6: _notion_compat.py markdown helpers + version-aware client

Files:

• Modify: custom-skills/32-notion-writer/code/scripts/_notion_compat.py
• Test: custom-skills/32-notion-writer/code/scripts/test_engine_routing.py

Interfaces:

• Consumes: notion_client.Client.request(path, method, body).

• Produces:

  • make_client(api_key, notion_version=None) -> Client (adds optional version).
  • create_page_markdown(client, parent, properties, markdown) -> dict → request("pages", "POST", body={parent, properties, markdown}).
  • append_markdown(client, page_id, markdown) -> dict → request(f"pages/{page_id}/markdown", "PATCH", body={"type":"insert_content","insert_content":{"content":markdown,"position":{"type":"end"}}}).
  • replace_markdown(client, page_id, markdown, allow_deleting=False) -> dict → request(f"pages/{page_id}/markdown", "PATCH", body={"type":"replace_content","replace_content":{"new_str":markdown,"allow_deleting_content":allow_deleting}}).

• Step 1: Write the failing tests

Create test_engine_routing.py:

#!/usr/bin/env python3
"""Tests for markdown transport helpers — run with `python test_engine_routing.py`."""

import sys
from pathlib import Path
from unittest import mock

sys.path.insert(0, str(Path(__file__).parent))

import _notion_compat as compat


def _fake_client():
    c = mock.MagicMock()
    c.request.return_value = {"object": "page", "id": "p1"}
    return c


def test_create_page_markdown():
    c = _fake_client()
    parent = {"type": "data_source_id", "data_source_id": "ds1"}
    compat.create_page_markdown(c, parent, {"Name": {"title": []}}, "# Hi")
    _, kwargs = c.request.call_args
    assert kwargs["path"] == "pages"
    assert kwargs["method"] == "POST"
    assert kwargs["body"]["markdown"] == "# Hi"
    assert kwargs["body"]["parent"] == parent


def test_append_markdown():
    c = _fake_client()
    compat.append_markdown(c, "page-123", "## More")
    _, kwargs = c.request.call_args
    assert kwargs["path"] == "pages/page-123/markdown"
    assert kwargs["method"] == "PATCH"
    assert kwargs["body"]["type"] == "insert_content"
    assert kwargs["body"]["insert_content"]["content"] == "## More"
    assert kwargs["body"]["insert_content"]["position"] == {"type": "end"}


def test_replace_markdown():
    c = _fake_client()
    compat.replace_markdown(c, "page-123", "# Fresh", allow_deleting=True)
    _, kwargs = c.request.call_args
    assert kwargs["path"] == "pages/page-123/markdown"
    assert kwargs["body"]["type"] == "replace_content"
    assert kwargs["body"]["replace_content"]["new_str"] == "# Fresh"
    assert kwargs["body"]["replace_content"]["allow_deleting_content"] is True


def run_all():
    tests = [
        test_create_page_markdown,
        test_append_markdown,
        test_replace_markdown,
    ]
    for t in tests:
        print(f"\n{t.__name__}")
        t()
    print("\n" + "=" * 50)
    print(f"✅ All {len(tests)} tests passed")
    print("=" * 50)


if __name__ == "__main__":
    run_all()

• Step 2: Run tests to verify they fail

Run: venv/bin/python test_engine_routing.py Expected: FAIL — AttributeError: module '_notion_compat' has no attribute 'create_page_markdown'.

• Step 3: Implement the helpers

In _notion_compat.py, replace make_client (lines 19-21) with:

def make_client(api_key: str, notion_version: str = None) -> Client:
    """Build a sync Notion client. Pass notion_version to override the SDK
    default (needed: 2026-03-11 for the markdown content endpoints)."""
    if notion_version:
        return Client(auth=api_key, notion_version=notion_version)
    return Client(auth=api_key)

Add at the end of _notion_compat.py:

def create_page_markdown(client, parent, properties, markdown):
    """POST /v1/pages with the enhanced-markdown body param."""
    return client.request(
        path="pages", method="POST",
        body={"parent": parent, "properties": properties, "markdown": markdown},
    )


def append_markdown(client, page_id, markdown):
    """PATCH /v1/pages/:id/markdown — append at end (insert_content)."""
    return client.request(
        path=f"pages/{page_id}/markdown", method="PATCH",
        body={"type": "insert_content",
              "insert_content": {"content": markdown,
                                 "position": {"type": "end"}}},
    )


def replace_markdown(client, page_id, markdown, allow_deleting=False):
    """PATCH /v1/pages/:id/markdown — replace all content."""
    return client.request(
        path=f"pages/{page_id}/markdown", method="PATCH",
        body={"type": "replace_content",
              "replace_content": {"new_str": markdown,
                                  "allow_deleting_content": allow_deleting}},
    )

Also extend the ValidationError branch of explain_api_error (currently lines ~212-213) so markdown-replace deletion errors are actionable. Replace:

    if code == APIErrorCode.ValidationError:
        return f"Validation error{suffix}: {exc.body.get('message', str(exc))}"

with:

    if code == APIErrorCode.ValidationError:
        msg = exc.body.get('message', str(exc))
        if 'delet' in msg.lower():
            msg += " — re-run with --allow-deleting-content to permit this."
        return f"Validation error{suffix}: {msg}"

• Step 4: Run tests to verify they pass

Run: venv/bin/python test_engine_routing.py Expected: ✅ All 3 tests passed.

• Step 5: Confirm no regression

Run: venv/bin/python test_parser.py Expected: ✅ All 35 tests passed.

• Step 6: Commit

cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/_notion_compat.py \
        custom-skills/32-notion-writer/code/scripts/test_engine_routing.py
git commit -m "$(printf 'feat(notion-writer): markdown endpoint helpers + version-aware client\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 7: Wire the CLI — flags, routing, two-phase image append

Files:

• Modify: custom-skills/32-notion-writer/code/scripts/notion_writer.py
• Test: custom-skills/32-notion-writer/code/scripts/test_engine_routing.py

Interfaces:

• Consumes: ntn_files.preflight/upload/materialize_local_media, md_translate.translate, compat.create_page_markdown/append_markdown/replace_markdown, compat.make_client(..., notion_version=...).

• Produces:

  • extract_local_images(content: str) -> tuple[str, list[tuple[str, str]]] — removes standalone local (!http) image lines, returns (content_without_local_images, [(alt, target), ...]).
  • _content_has_local_images(content: str) -> bool.
  • --engine, --notion-version, --allow-deleting-content argparse flags.
  • MARKDOWN_NOTION_VERSION = "2026-03-11" constant.

• Step 1: Write the failing tests

Add to test_engine_routing.py (register names in run_all). These import from notion_writer and exercise the pure helpers:

import notion_writer


def test_extract_local_images_splits():
    content = ("# Title\n\n![remote](https://ex.com/a.png)\n\n"
               "![local](./b.png)\n\nbody")
    without, imgs = notion_writer.extract_local_images(content)
    assert imgs == [("local", "./b.png")]
    assert "![local](./b.png)" not in without
    assert "![remote](https://ex.com/a.png)" in without  # remote stays


def test_content_has_local_images():
    assert notion_writer._content_has_local_images("![x](./y.png)") is True
    assert notion_writer._content_has_local_images("![x](https://e/y.png)") is False
    assert notion_writer._content_has_local_images("no images") is False

Add to run_all's tests list:

        test_extract_local_images_splits,
        test_content_has_local_images,

• Step 2: Run tests to verify they fail

Run: venv/bin/python test_engine_routing.py Expected: FAIL — AttributeError: module 'notion_writer' has no attribute 'extract_local_images'.

• Step 3: Add imports + helpers + constant

Near the top of notion_writer.py, after import _notion_compat as compat (line 19), add:

import ntn_files
import md_translate

MARKDOWN_NOTION_VERSION = "2026-03-11"

Add these helpers near create_image_block:

def _content_has_local_images(content: str) -> bool:
    for line in content.split('\n'):
        m = IMAGE_RE.match(line)
        if m and not m.group(2).startswith(('http://', 'https://')):
            return True
    return False


def extract_local_images(content: str):
    """Remove standalone LOCAL image lines; keep remote ones inline.
    Returns (content_without_local_images, [(alt, target), ...])."""
    kept, imgs = [], []
    for line in content.split('\n'):
        m = IMAGE_RE.match(line)
        if m and not m.group(2).startswith(('http://', 'https://')):
            imgs.append((m.group(1), m.group(2)))
            continue
        kept.append(line)
    return "\n".join(kept), imgs

• Step 4: Run the new pure-helper tests

Run: venv/bin/python test_engine_routing.py Expected: ✅ All 5 tests passed.

• Step 5: Add the argparse flags

In main(), after the --info argument (around line 798), add:

    parser.add_argument('--engine', choices=['blocks', 'markdown'],
                        default='blocks',
                        help='Write engine: blocks (default) or markdown')
    parser.add_argument('--notion-version', dest='notion_version',
                        help='Override Notion API version')
    parser.add_argument('--allow-deleting-content', dest='allow_deleting',
                        action='store_true',
                        help='Markdown replace may delete child pages/dbs')

• Step 6: Add a base-dir helper + media-client wiring in main()

In main(), right after content is resolved (after the elif args.file: block near line 886), add:

    base_dir = Path(args.file).parent if args.file else Path.cwd()

    def _md_client():
        version = args.notion_version or MARKDOWN_NOTION_VERSION
        return compat.make_client(NOTION_TOKEN, notion_version=version)

    def _upload_blocks_for(images):
        """Build + materialize image blocks for two-phase markdown writes."""
        blocks = [create_image_block(alt, target) for alt, target in images]
        return ntn_files.materialize_local_media(blocks, base_dir, ntn_files.upload)

• Step 7: Route the page-write path through the engine

Replace the page-write block in main() (the if args.page: block, lines ~889-909) with:

    if args.page:
        if not content:
            print("Error: No content provided. Use --file or --stdin")
            sys.exit(1)
        page_id = extract_notion_id(args.page)
        if not page_id:
            print(f"Error: Invalid Notion page URL/ID: {args.page}")
            sys.exit(1)
        formatted_id = format_id_with_dashes(page_id)

        if args.engine == 'markdown':
            if _content_has_local_images(content):
                ntn_files.preflight()
            md_body, local_imgs = extract_local_images(content)
            md_body = md_translate.translate(md_body)
            mc = _md_client()
            try:
                if args.replace:
                    compat.replace_markdown(mc, formatted_id, md_body,
                                            allow_deleting=args.allow_deleting)
                else:
                    compat.append_markdown(mc, formatted_id, md_body)
            except APIResponseError as exc:
                print(f"Error: {compat.explain_api_error(exc, formatted_id)}")
                sys.exit(1)
            if local_imgs:
                append_to_page(notion, page_id, _upload_blocks_for(local_imgs))
            print("✅ Successfully wrote content to page (markdown engine)")
            print(f"   https://notion.so/{formatted_id.replace('-', '')}")
            return

        # blocks engine (default)
        blocks = markdown_to_notion_blocks(content)
        if _content_has_local_images(content):
            ntn_files.preflight()
            blocks = ntn_files.materialize_local_media(blocks, base_dir, ntn_files.upload)
        if not blocks:
            print("No content to write")
            sys.exit(1)
        if args.replace and not clear_page_content(notion, page_id):
            sys.exit(1)
        if append_to_page(notion, page_id, blocks):
            print("✅ Successfully wrote content to page")
            print(f"   https://notion.so/{formatted_id.replace('-', '')}")
        else:
            print("❌ Failed to write content")
            sys.exit(1)
        return

• Step 8: Route the database-row path through the engine

In the if args.database: block, after content_blocks = markdown_to_notion_blocks(content) if content else None (line ~961), insert the markdown-engine branch and guard the blocks-engine upload. Replace that line and the create/upsert tail with:

        if args.engine == 'markdown':
            md_content = content or ""
            if _content_has_local_images(md_content):
                ntn_files.preflight()
            md_body, local_imgs = extract_local_images(md_content)
            md_body = md_translate.translate(md_body)
            parent = compat.build_data_source_parent(data_source_id)
            mc = _md_client()
            try:
                if args.upsert_by and 'existing' in dir() and existing:
                    compat.replace_markdown(mc, existing['id'], md_body,
                                            allow_deleting=args.allow_deleting)
                    update_page_properties(notion, existing['id'], properties)
                    new_id = existing['id']
                else:
                    result = compat.create_page_markdown(mc, parent, properties, md_body)
                    new_id = result['id']
            except APIResponseError as exc:
                print(f"Error: {compat.explain_api_error(exc)}")
                sys.exit(1)
            if local_imgs:
                append_to_page(notion, new_id, _upload_blocks_for(local_imgs))
            print("✅ Successfully wrote database row (markdown engine)")
            print(f"   https://notion.so/{new_id.replace('-', '')}")
            return

        content_blocks = markdown_to_notion_blocks(content) if content else None
        if content_blocks and _content_has_local_images(content or ""):
            ntn_files.preflight()
            content_blocks = ntn_files.materialize_local_media(
                content_blocks, base_dir, ntn_files.upload)

Note: the existing upsert lookup that defines existing runs earlier in this block (the if args.upsert_by: section). Keep that section above this code so existing is in scope. If the markdown branch runs before that section in the current file order, move the markdown branch to just after the upsert lookup. The blocks-engine create/upsert tail below stays unchanged.

• Step 9: Run all suites

Run:

venv/bin/python test_parser.py && \
venv/bin/python test_ntn_files.py && \
venv/bin/python test_md_translate.py && \
venv/bin/python test_engine_routing.py

Expected: every suite prints its ✅ All N tests passed line (35, 8, 7, 5).

• Step 10: Smoke-check the CLI help (no API calls)

Run: venv/bin/python notion_writer.py --help Expected: help text lists --engine, --notion-version, --allow-deleting-content.

• Step 11: Commit

cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/notion_writer.py \
        custom-skills/32-notion-writer/code/scripts/test_engine_routing.py
git commit -m "$(printf 'feat(notion-writer): --engine routing + two-phase image append\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 8: Documentation (CLAUDE.md + SKILL.md, v1.3.0)

Files:

• Modify: custom-skills/32-notion-writer/code/CLAUDE.md
• Modify: custom-skills/32-notion-writer/SKILL.md

Interfaces: none (docs only).

• Step 1: Document the new capabilities in code/CLAUDE.md

Add a "File uploads" subsection under Markdown Support describing ![alt](local.png) auto-upload via ntn (requires ntn installed + ntn login; workspace-scoped). Add an "Engines" subsection: --engine blocks (default) vs --engine markdown (native endpoints, 2026-03-11, dialect auto-translated, local images appended at page end). Document --notion-version and --allow-deleting-content. Add example:

# Markdown engine, create a row from a doc with callouts/columns
python notion_writer.py -d DB_URL -t "Notes" --engine markdown -f notes.md

# Blocks engine with a local image (auto-uploaded via ntn)
python notion_writer.py -p PAGE_URL -f post.md   # post.md contains ![chart](./chart.png)

• Step 2: Bump the version footer in code/CLAUDE.md

Change the footer line to *Version 1.3.0 | Claude Code | 2026-06-27* and add a changelog entry:

- 1.3.0 — Local file/image uploads via the `ntn` CLI (`![](local)` → file_upload image blocks). New `--engine markdown` path writing through Notion's native enhanced-markdown endpoints with a dialect translator. Added `--notion-version` and `--allow-deleting-content`.

• Step 3: Mirror the summary into SKILL.md

Under the "Supported Markdown" / capabilities area of SKILL.md, add a one-paragraph note on engines and image upload, with the --engine markdown example above. Keep it short — SKILL.md is the trigger surface.

• Step 4: Commit

cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/CLAUDE.md \
        custom-skills/32-notion-writer/SKILL.md
git commit -m "$(printf 'docs(notion-writer): document engines + file uploads (v1.3.0)\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"

---

Task 9: Live smoke test (both engines)

Files: none (manual verification against the live API).

Interfaces: uses the real NOTION_API_KEY (1Password) and the "Working with AI" data source f8f19ede-32bd-43ac-9f60-0651f6f40afe.

• Step 1: Connection + token

Run:

cd ~/Project/our-claude-skills/custom-skills/32-notion-writer/code/scripts
NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
  venv/bin/python notion_writer.py --test

Expected: ✅ Connected successfully!. If the 1Password item path differs, use the working credential path from SKILL.md's "Credential handling".

• Step 2: Blocks engine — local image upload

Create /tmp/nw_img.md with a heading and ![smoke](./logo.png) (place any small PNG at ./logo.png in the scripts dir). Run an upsert into the AI DB:

NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
  venv/bin/python notion_writer.py -d f8f19ede-32bd-43ac-9f60-0651f6f40afe \
  -t "[smoke] blocks image" --upsert-by "Name" -f /tmp/nw_img.md

Expected: prints the ntn -> workspace line and a success URL; the row shows an embedded image (not a broken link). Verify the ntn workspace line matches the token's workspace — if not, the image won't attach (documented constraint).

• Step 3: Markdown engine — callout + columns + appended image

Create /tmp/nw_md.md:

# Smoke: markdown engine

> [!NOTE]
> Rendered as a Notion callout.

::: columns
::: column
Left column
:::
::: column
Right column
:::
:::

![smoke](./logo.png)

Run:

NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
  venv/bin/python notion_writer.py -d f8f19ede-32bd-43ac-9f60-0651f6f40afe \
  -t "[smoke] markdown engine" --upsert-by "Name" --engine markdown -f /tmp/nw_md.md

Expected: success URL; the page shows a callout, two columns, and the image at the end. This confirms the open pipe-table/version items — if the callout or columns render as raw text, check the translator's tab indentation; if the API rejects the body, check the 2026-03-11 version and the data-source parent.

• Step 4: Markdown engine — replace

Run the same command as Step 3 but with -p <the page URL from Step 3>, --engine markdown, --replace, and a trivial /tmp/nw_md2.md (# Replaced\nNew body.):

NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
  venv/bin/python notion_writer.py -p PAGE_URL_FROM_STEP_3 \
  --engine markdown --replace -f /tmp/nw_md2.md

Expected: page content replaced via replace_content. If it errors about deleting child content, re-run with --allow-deleting-content.

• Step 5: Record results

If all four steps pass, the feature is verified end-to-end. If the pipe-table passthrough failed in Step 3, open a follow-up per the spec's parking lot (<table> rewrite). No commit (manual test task).

---

Notes for the implementer

• Run from the scripts dir. All venv/bin/python ... commands assume cwd = custom-skills/32-notion-writer/code/scripts/.
• Line numbers in "Modify" hints are from the current notion_writer.py (1013 lines) and _notion_compat.py (256 lines); if they've drifted, locate by the quoted anchor code instead.
• Don't reformat untouched code. Keep diffs minimal and additive.
• After Task 7, the four suites total 55 tests (35 + 8 + 7 + 5). Keep them all green through Tasks 8-9.