adeelahmad · Oct 18, 2026 · Oct 18, 2026 · Oct 18, 2026 · Oct 18, 2026
diff --git a/DEVLOG.md b/DEVLOG.md
@@ -619,7 +619,7 @@ Todo:
 - [x] Canvas editor in the web app (pipelines and workflows): drag nodes, connect ports, node settings, I/O, save versions
 - [x] Attach workflows from the pipeline editor
 - [x] Content-type mapping on the Pipelines page (By content type)
-- [ ] Asset converter steps (video → audio, document → images) as pipeline steps
+- [x] Asset converter steps (video → audio, document → images) as pipeline steps (the convert step, conversions.py)
 - [ ] Workflow node types: HTTP call, template render, entity filter
 - [x] Docs: processing.md section on workflows
 

diff --git a/README.md b/README.md
@@ -149,7 +149,8 @@ listed under [In progress](#in-progress), and what comes next under [Planned](#p
   [specs from transcripts](docs/notes.md#specs-from-transcripts), [notes](docs/notes.md),
   [evidence PDFs](docs/api.md#evidence-pdfs), [published feeds](docs/feeds.md);
   [checks](docs/requirements.md#learn-from-it))
-- **Automate it.** Pipelines and workflows on a canvas, routines on a schedule, batch runs over a collection, and
+- **Automate it.** Pipelines and workflows on a canvas (with Convert steps that make a video's soundtrack an audio
+  asset and a document's pages image assets), routines on a schedule, batch runs over a collection, and
   notifications to webhooks, Slack, Discord and Matterbridge. Everything that happens in Lens (new assets, emails,
   calendar events, files, feed posts, what people say to Lens, sensor readings, notes, graph changes, runs, budget
   warnings) is kept as a stream of activities in W3C Activity Streams form, which agents read through the API, MCP and
@@ -165,7 +166,8 @@ listed under [In progress](#in-progress), and what comes next under [Planned](#p
   in the chat it told you in; agents can decline an approval but never give one. Each automation has limits and budgets,
   and agents make, try and run them over MCP and in the assistant, or draft one from a sentence; the studio to draw them
   in the web app is planned. ([Processing](docs/processing.md#workflows-and-the-canvas),
-  [notifications](docs/notifications.md), [automations](docs/automations.md),
+  [converting assets](docs/processing.md#converting-assets), [notifications](docs/notifications.md),
+  [automations](docs/automations.md),
   [approvals](docs/automations.md#approvals), [timeline](docs/timeline.md); [checks](docs/requirements.md#automate-it))
 - **See what it costs.** An activity ledger of every model call, with reports, charts and budgets that cap spending.
   Setup asks for one overall budget after a cloud model, with a soft limit that warns and a hard limit that stops paid

diff --git a/docs/processing.md b/docs/processing.md
@@ -671,6 +671,8 @@ audio skip themselves for imported text. Steps:
   template unless the step names another prompt template ([Specs from transcripts](notes.md#specs-from-transcripts)).
 - `report` with a template: renders an HTML page into the namespace's reports. These pages may not run scripts.
 - `export`: renders a template to a file (Markdown, text, HTML...) and can copy it to a storage source.
+- `convert`: makes new assets from one, a video's soundtrack as audio or a document's pages as images
+  ([Converting assets](#converting-assets)).
 
 Any step can carry a condition: `min_minutes`, `max_minutes`, `source` (audio or transcript), `languages`.
 
@@ -688,6 +690,26 @@ New pipelines, workflows, templates and routines don't need a name typed: the we
 steps, what a workflow runs on, a template's kind, a routine's schedule and actions), numbered when it's taken
 ("Prompt template 2"). The pencil next to the name on its page renames it later, as does `PATCH` on its address.
 
+### Converting assets
+
+A Convert step makes new assets from the one it runs on. Default pipelines have none; add one in the pipeline editor
+or on the canvas (Step library → Convert).
+
+- **To audio** (`{"type": "convert", "to": "audio"}`): a video's soundtrack becomes an audio asset, titled
+  "<the video's title> (audio)". It is m4a unless `format` says `mp3`, `opus` or `flac`; AAC sound in m4a (or MP3
+  sound in mp3) is copied as it is, not encoded again. Audio, documents and videos with no sound are skipped.
+- **To images** (`{"type": "convert", "to": "images"}`): each page the Read step drew becomes an image asset,
+  titled "<title>, page N", the same JPEG the asset's page shows (documents.page_pixels on its longest side). Put a
+  Read step before it; without drawn pages it skips. A multi-page TIFF's frames work too; at most 200 pages are made
+  into images from one document.
+
+New assets are kept in the same namespace and collection, encrypted where its files are, and each says which asset
+(and page) it was made from: `converted_from` on `GET /api/v1/recordings/{id}`, and a link on its page. They then run
+the pipeline their content type gives them, as an upload does; `pipeline` names another pipeline by its id, or
+`"none"` keeps them as they are until a pipeline is run on them. Running the step again keeps what it made before;
+`force` makes their files again. An asset a Convert step made isn't converted again, so a pipeline can't feed itself,
+and deleting it deletes its file.
+
 ## Workflows and the canvas
 
 A pipeline's steps make derived files (a transcript or a page's text, shots, text on screen, faces); workflows turn them into metadata. A

diff --git a/docs/requirements.md b/docs/requirements.md
@@ -179,6 +179,7 @@ request, to another check that proves the same thing; loosening a requirement or
 | Requirement | Checks |
 |---|---|
 | Pipelines and workflows on a canvas | [test_workflows.py](https://github.com/adeelahmad/lense/blob/main/fastapi_backend/tests/api/test_workflows.py) `test_workflows_run_in_pipelines_drawn_as_graphs`; [workflow-canvas.test.ts](https://github.com/adeelahmad/lense/blob/main/nextjs-frontend/__tests__/workflow-canvas.test.ts); [journeys.mjs](https://github.com/adeelahmad/lense/blob/main/ci/clicks/journeys.mjs) `pipeline-create`, `workflow-create` |
+| Convert steps make new assets: a video's soundtrack as audio, a document's pages as images, each run through its own pipeline | [test_conversions.py](https://github.com/adeelahmad/lense/blob/main/fastapi_backend/tests/api/test_conversions.py) `test_a_videos_soundtrack_becomes_an_audio_asset`, `test_a_documents_pages_become_image_assets`, `test_settings_are_checked`; [pipeline-convert.test.ts](https://github.com/adeelahmad/lense/blob/main/nextjs-frontend/__tests__/pipeline-convert.test.ts) `is added from the library ready to run: to audio unless images are picked` |
 | A Classify node in workflows chooses from labels, content types or topics, and acting on a sure answer keeps something | [test_classify.py](https://github.com/adeelahmad/lense/blob/main/fastapi_backend/tests/api/test_classify.py) `test_a_workflow_classifies`; [workflow-flow.test.ts](https://github.com/adeelahmad/lense/blob/main/nextjs-frontend/__tests__/workflow-flow.test.ts) `chooses from labels, content types or topics, and acting on a sure answer keeps something` |
 | Routines on a schedule | [test_routines.py](https://github.com/adeelahmad/lense/blob/main/fastapi_backend/tests/api/test_routines.py) `test_routines_are_checked_and_run_when_due`, `test_schedules`; [journeys.mjs](https://github.com/adeelahmad/lense/blob/main/ci/clicks/journeys.mjs) `routine-create`; [baseline.json](https://github.com/adeelahmad/lense/blob/main/ci/clicks/baseline.json) `routine-create` ≤ 2 clicks |
 | Batch runs over a collection | [test_batches.py](https://github.com/adeelahmad/lense/blob/main/fastapi_backend/tests/api/test_batches.py) `test_collections_and_batches` |

diff --git a/fastapi_backend/app/api/v1/routes/recordings.py b/fastapi_backend/app/api/v1/routes/recordings.py
@@ -324,6 +324,7 @@ def get_recording(rid: int, acl: Acl, db: Db, cfg: Cfg) -> Recording:
     ]
     d["jobs"] = jobs.list_jobs(db, recording=rid, limit=5)
     d["attached_to"] = _attached_to(acl, db, r.get("attached_to"))
+    d["converted_from"] = _attached_to(acl, db, r.get("converted_from"))
     if r.get("enrichment") and acl.member(r):  # where a photo was taken: members only
         d["enrichment"] = {k: v for k, v in r["enrichment"].items() if k != "photo_fp"}
     d["nsfw"] = _nsfw(db, cfg, r, rid)
@@ -350,14 +351,15 @@ def mark_recording_nsfw(rid: int, body: RecordingNsfwUpdate, acl: Acl, user: Wri
 
 
 def _attached_to(acl: Access, db: DB, link: dict[str, Any] | None) -> dict[str, Any] | None:
-    """The email an attachment came from, with its title, when the caller may see that email."""
+    """The email an attachment came from (or the asset a convert step made this one from), with its title, when the
+    caller may see it."""
     if not link or not link.get("resource"):
         return None
     try:
         email = acl.recording(int(link["resource"]))
     except HTTPException:
         return None  # deleted since, or not theirs to see
-    return {"resource": int(link["resource"]), "file": link.get("file"), "title": email.get("title")}
+    return {**link, "resource": int(link["resource"]), "title": email.get("title")}
 
 
 @router.patch("/{rid}")

diff --git a/fastapi_backend/app/domain/conversions.py b/fastapi_backend/app/domain/conversions.py
@@ -0,0 +1,211 @@
+"""The convert step: new assets made from one (docs/processing.md#converting-assets).
+
+`{type: convert, to: audio}` makes a video's soundtrack an audio asset of its own, as m4a (AAC, copied as it is when
+the video's sound is AAC already), mp3, opus or flac (`format`). `{type: convert, to: images}` makes each page a
+document's Read step drew (or each frame of a TIFF) an image asset of its own, the JPEG Lens shows for that page.
+
+A new asset lives in the namespace and collection of the one it was made from, says which asset and page it was made
+from (converted_from), and is kept encrypted like an upload. It then runs the pipeline its content type gives it, as an
+upload does, or the one the step names (`pipeline`), or none (`pipeline: "none"`): it's kept as it is until a pipeline
+is run on it. Running the step again keeps the assets it made before (forced, it makes their files again). An asset
+made by a convert step isn't converted again, so a pipeline that converts can't feed itself.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import pathlib
+import re
+import subprocess
+
+from . import documents, ingest, jobs, keyring, store, video
+
+R = store.R
+TO = ("audio", "images")
+# format: (codec that is copied as it is, ffmpeg's encoder and its options)
+AUDIO = {
+    "m4a": ("aac", ["-c:a", "aac", "-b:a", "128k"]),
+    "mp3": ("mp3", ["-c:a", "libmp3lame", "-q:a", "4"]),
+    "opus": ("opus", ["-c:a", "libopus", "-b:a", "64k"]),
+    "flac": ("flac", ["-c:a", "flac"]),
+}
+MAX_PAGES = 200  # images made from one document at most
+
+
+def check(spec):
+    """A convert step's settings (pipelines.validate_steps): ValueError naming the first problem."""
+    if spec.get("to") not in TO:
+        raise ValueError("convert step: convert to audio or images")
+    fmt = spec.get("format")
+    if fmt is not None and (spec["to"] != "audio" or fmt not in AUDIO):
+        if spec["to"] != "audio":
+            raise ValueError("convert step: images are JPEGs, with no format to choose")
+        raise ValueError(f"convert step: the audio format is one of {', '.join(AUDIO)}")
+    p = spec.get("pipeline")
+    if p is not None and p != "none" and (isinstance(p, bool) or not isinstance(p, int)):
+        raise ValueError('convert step: pipeline is a pipeline\'s id, or "none" to keep the new assets as they are')
+
+
+def made_from(db, rid, to):
+    """The assets a convert step made from this one before: {page or None: id}."""
+    rows = db.rows(
+        "SELECT record::id(id) AS id, converted_from FROM recording WHERE converted_from.resource = $r AND converted_from.to = $t",
+        r=int(rid),
+        t=to,
+    )
+    return {(r["converted_from"] or {}).get("page"): r["id"] for r in rows}
+
+
+def step(db, cfg, rid, say, spec):
+    rec = db.one(
+        "SELECT space, collection, path, remote, source, title, recorded_at, duration_ms, converted_from FROM $r",
+        r=R("recording", rid),
+    )
+    if not rec:
+        raise KeyError(rid)
+    if rec.get("converted_from"):
+        raise jobs.Skip("it was made by a convert step itself, so it isn't converted again")
+    if spec["to"] == "audio":
+        return _audio(db, cfg, rid, rec, say, spec)
+    return _images(db, cfg, rid, rec, say, spec)
+
+
+def _stem(rec, rid):
+    stem = pathlib.PurePosixPath(str(rec.get("path") or "")).stem or str(rec.get("title") or "")
+    return re.sub(r"[^\w .()-]+", "_", stem).strip(" .")[:100] or f"asset-{int(rid)}"
+
+
+def _folder(db, cfg, rec, rid):
+    ns = store.space_names(db).get(rec["space"]) or str(rec["space"])
+    d = pathlib.Path(cfg["data_dir"]) / "uploads" / ns / f"converted-{int(rid)}"
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def _audio_codec(path):
+    out = subprocess.run(
+        ["ffprobe", "-v", "error", "-format_whitelist", ingest.MEDIA_DEMUXERS, "-select_streams", "a:0"]
+        + ["-show_entries", "stream=codec_name", "-of", "json", str(path)],
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+    streams = json.loads(out.stdout or "{}").get("streams") or []
+    return (streams[0] if streams else {}).get("codec_name")
+
+
+def _audio(db, cfg, rid, rec, say, spec):
+    if rec.get("source") != "audio":
+        raise jobs.Skip("only a video has a soundtrack to make into audio")
+    path = ingest.audio_path(db, cfg, rec)
+    if not path or not os.path.exists(path):
+        raise jobs.Skip("there is no media file")
+    media = video.probe_media(path)
+    if media["kind"] != "video":
+        raise jobs.Skip("it's audio already")
+    if not media.get("has_audio"):
+        raise jobs.Skip("the video has no sound")
+    have = made_from(db, rid, "audio").get(None)
+    if have and not spec.get("force"):
+        return say(f"kept the audio made from it before (asset {have}); force the step to make it again")
+    fmt = spec.get("format") or "m4a"
+    same, encode = AUDIO[fmt]
+    codec = ["-c:a", "copy"] if _audio_codec(path) == same else encode
+    dest = _folder(db, cfg, rec, rid) / f"{_stem(rec, rid)}.{fmt}"
+    tmp = dest.with_name(f".part-{dest.name}")
+    argv = ["ffmpeg", "-nostdin", "-v", "error", "-y", "-format_whitelist", ingest.MEDIA_DEMUXERS, "-i", str(path)]
+    # no picture, no tags: the same video makes the same file, byte for byte
+    argv += ["-map", "0:a:0", "-vn", "-sn", "-dn", "-map_metadata", "-1", "-fflags", "+bitexact", "-flags:a", "+bitexact", *codec, str(tmp)]
+    try:
+        out = subprocess.run(argv, capture_output=True, timeout=ingest.DECODE_SECONDS)
+        if out.returncode != 0:
+            raise RuntimeError("ffmpeg: " + out.stderr.decode(errors="replace")[-300:])
+        os.replace(tmp, dest)
+    finally:
+        tmp.unlink(missing_ok=True)
+    keyring.protect(db, cfg, rec["space"], dest)  # encrypted like an upload, where its namespace's files are
+    title = f"{rec.get('title') or _stem(rec, rid)} (audio)"
+    more = {"duration_ms": media["duration_ms"] or rec.get("duration_ms")}
+    new = _keep(db, cfg, rid, rec, dest, "audio", title, {"to": "audio"}, have, spec, more)
+    how = "copied" if codec[-1] == "copy" else "encoded"
+    say(f"made its soundtrack an audio asset ({fmt}, {how}): asset {new}")
+
+
+def _images(db, cfg, rid, rec, say, spec):
+    if rec.get("source") not in documents.KINDS:
+        raise jobs.Skip("only a document's pages (or a TIFF's frames) are made into images")
+    pages = [p for p in documents.pages(db, rid) if p.get("image")]
+    if not pages:
+        raise jobs.Skip("its pages haven't been drawn: put a Read step before this one")
+    if rec["source"] == "image" and len(pages) < 2:
+        raise jobs.Skip("it's one image already")
+    if len(pages) > MAX_PAGES:
+        say(f"only its first {MAX_PAGES} of {len(pages)} pages are made into images")
+        pages = pages[:MAX_PAGES]
+    have = made_from(db, rid, "images")
+    d, frames, stem = _folder(db, cfg, rec, rid), video.frames_dir(cfg, rid), _stem(rec, rid)
+    made = kept = 0
+    for p in pages:
+        old = have.get(p["idx"])
+        if old and not spec.get("force"):
+            kept += 1
+            continue
+        src = frames / p["image"]
+        if not src.exists():
+            continue
+        dest = d / f"{stem}-page-{p['idx'] + 1:04d}.jpg"
+        keyring.keep(db, cfg, rec["space"], dest, keyring.read_plain(db, cfg, src))  # encrypted as it's written, if wanted
+        name = p.get("label") or str(p["idx"] + 1)
+        title = f"{rec.get('title') or stem}, page {name}"
+        _keep(db, cfg, rid, rec, dest, "image", title, {"to": "images", "page": p["idx"]}, old, spec, {"media": {"kind": "image"}})
+        made += 1
+    say(f"made {made} page(s) into image assets" + (f"; kept the {kept} made before (force the step to make them again)" if kept else ""))
+
+
+def _keep(db, cfg, rid, rec, dest, source, title, link, old, spec, more):
+    """The converted file (kept as it's to be, encrypted or not) as an asset: a new one, or the one made before when
+    forced; queued for its pipeline."""
+    st, fp = dest.stat(), ingest.fingerprint(dest, db=db, cfg=cfg)
+    row = {
+        "path": str(dest),
+        "source": source,
+        "size": keyring.plain_size(db, cfg, dest),
+        "mtime": st.st_mtime,
+        "fingerprint": fp,
+        "fp_key": f"{rec['space']}:{fp}",
+        **more,
+    }
+    if old:
+        was = (db.one("SELECT path FROM $r", r=R("recording", old)) or {}).get("path")
+        db.q("UPDATE $r MERGE $d", r=R("recording", old), d={**row, "status": "new"})
+        if was and was != str(dest) and keyring.owned(cfg, was):  # made again in another format: the old file goes
+            pathlib.Path(was).unlink(missing_ok=True)
+        new = old
+    else:
+        known = db.one("SELECT record::id(id) AS id, path FROM recording WHERE fp_key = $k", k=row["fp_key"])
+        if known:  # the same file is an asset here already: the copy just made isn't kept
+            if known.get("path") != str(dest):
+                dest.unlink(missing_ok=True)
+            return known["id"]
+        new = db.next_id("recording")
+        db.q(
+            "CREATE $r CONTENT $d",
+            r=R("recording", new),
+            d=store.clean(
+                {
+                    **row,
+                    "space": rec["space"],
+                    "collection": rec.get("collection"),
+                    "title": title[:200],
+                    "recorded_at": rec.get("recorded_at") or store.now(),
+                    "converted_from": {"resource": int(rid), **link},
+                    "status": "new",
+                    "created_at": store.now(),
+                }
+            ),
+        )
+    then = spec.get("pipeline")
+    if then != "none":
+        jobs.enqueue(db, new, None, by=f"convert:{int(rid)}", pipeline=then)
+    return new