adeelahmad · Oct 15, 2026 · Oct 15, 2026
diff --git a/docs/automations.md b/docs/automations.md
@@ -21,6 +21,7 @@ merged is called built.
 | Telling a person: the `notify` action and `channels.tell` (their own chat, a notification target, a Matterbridge room) | built |
 | The model and the agent: the `ask_model`, `call_tool` and `ask_agent` actions, and prompts with outside text marked | built |
 | Calling a web address: the `http_request` action, with credentials in headers kept sealed | built |
+| Running Lens work: the `run_workflow`, `run_pipeline` and `run_routine` actions | built |
 | Settings → Automations: how often followers look and how long the event log is kept | built |
 | The Timeline page with filters, grouping, review keys and new items, and Home's Latest card ([Timeline](timeline.md)) | built |
 | Storing and running automations, schedules and timers, joins, the other actions, approvals, variables, the studio | planned |
@@ -292,6 +293,9 @@ or removed does nothing. What it writes is by the automation itself, `automation
 | `call_tool` | calls one of the assistant's tools (`tool`, with `args`) as the automation | read: a change becomes an approval | `{tool, result, summary, approvals}` |
 | `ask_agent` | hands what happened (`about`, by default the event) to the agent with your `instructions`, limited to some `tools` if you like, for up to `max_steps` rounds (6, at most 12) | read: changes become approvals | `{answer, chat, approvals, steps}` |
 | `http_request` | sends a `method` (GET, POST or PUT) to a `url`, with `headers` and a `body` (text, or an object sent as JSON) | out | `{status, ok, body, json, content_type, truncated, method, url, headers}`: the answer, as outside text |
+| `run_workflow` | runs a recording `workflow` (its published `version`, or the one named) on an `asset` as a job | lens | `{job, workflow, version, asset}` |
+| `run_pipeline` | processes an `asset` again with a `pipeline` (or the one its namespace uses), keeping its outputs | lens | `{job, pipeline, asset}` |
+| `run_routine` | asks the scheduler to run a `routine` now (`propose_only` to only propose its changes) | lens, so it asks | `{routine, requested, propose_only}` |
 
 #### Telling a person
 
@@ -345,6 +349,16 @@ key ([Encryption](encryption.md)), bound to the automation and to the scheme, ho
 sealed value can't be copied into another automation or sent to another host; it is opened only to send. A locked
 vault stops the request until it is unlocked.
 
+#### Running Lens work
+
+`run_workflow` and `run_pipeline` take an `asset`: an asset id, or the activity's object (`lens:recording/<id>`, which
+is the default, `{{ event.object.id }}`). On the asset the activity is about, in the automation's namespaces, they run
+without asking; on any other, they ask first. They need the automation's person to be able to process assets in that
+namespace, queue a job (after whatever the asset's job still has to do) that shows as by "automation: <name>", and pass
+on the job's id. `run_routine` asks the scheduler to start a routine's run within half a minute, as Run in Settings
+does: routines are an admin's, the routine must work only in the automation's namespaces, a run over one of its budgets
+is refused, and it always asks first. The run's id shows in the routine's runs.
+
 ## Refine later
 
 - More operators: `in`, `matches` (bounded regular expressions), `between`, times against now, `changed`.
@@ -369,3 +383,5 @@ vault stops the request until it is unlocked.
 - Calling a web address: secrets in the address itself (a Slack or Discord webhook URL) sealed like headers; PATCH and
   DELETE (DELETE would count as deleting); OAuth sign-in for an API; form-encoded bodies; retries with backoff;
   networks allowed for automations apart from notifications.
+- Running Lens work: a routine run's id passed on (today the scheduler starts it later); waiting for the job and passing
+  on its outputs; roles on collections counting for `process`; running a pipeline on every asset a join collected.
diff --git a/fastapi_backend/app/domain/actions.py b/fastapi_backend/app/domain/actions.py
@@ -14,7 +14,7 @@
 (`actor`). Params marked `"format": "prompt"` are prompts: the engine renders them with `render_prompt`, which wraps
 what the run read from outside Lens as data. `log_line` ships here; the other actions are registered by a module per
 group, loaded with this one: actions_messages.py (notify), actions_ai.py (ask_model, call_tool, ask_agent),
-actions_web.py (http_request).
+actions_web.py (http_request), actions_run.py (run_workflow, run_pipeline, run_routine).
 """
 
 from __future__ import annotations
@@ -243,6 +243,55 @@ def reach(db, intent, account):
     return readable, {s for s in readable if auth.allows(roles, s, "editor")}
 
 
+def allowed(db, account, space, perm):
+    """Whether the person's role in a namespace now has a permission (roles.py: edit, process, …)."""
+    return perm in auth.permissions(db, account).get(space, ())
+
+
+# ---------- the asset an action is about ----------
+EVENT_OBJECT = "{{ event.object.id }}"
+ASSET_RX = re.compile(r"^(?:lens:recording/)?(\d{1,12})$")
+
+
+def asset_id(v):
+    """An asset's id from a param: an id, or an activity's object id (lens:recording/<id>). ValueError otherwise."""
+    m = ASSET_RX.match(str("" if v is None else v).strip())
+    if not m:
+        raise ValueError(f"{v} isn't an asset: give an asset id, or run it on an activity about an asset")
+    return int(m.group(1))
+
+
+def asset_subject(p, key="asset"):
+    """The object an action on an asset changes, as activities name it, for resolve_risk; None when it can't tell."""
+    try:
+        return f"lens:recording/{asset_id(p.get(key))}"
+    except ValueError:
+        return None
+
+
+def asset_param(title="Asset"):
+    return {"title": title, "default": EVENT_OBJECT, "description": "an asset id, or the activity's object (lens:recording/<id>)"}
+
+
+def on_asset(p, key="asset"):
+    """The asset an action names, in words."""
+    a = p.get(key)
+    return "the asset it's about" if a in (None, EVENT_OBJECT) else f"asset {a}"
+
+
+def asset_for(db, intent, account, rid, perm):
+    """The asset {id, space, title, collection}, when it is in one of the automation's namespaces and the person's
+    role there now has `perm`. ValueError naming the problem."""
+    rec = db.one("SELECT space, title, collection FROM $r", r=store.R("recording", int(rid)))
+    if not rec:
+        raise ValueError(f"there's no asset {rid}")
+    if intent["spaces"] and rec["space"] not in intent["spaces"]:
+        raise ValueError(f"asset {rid} isn't in this automation's namespaces")
+    if not allowed(db, account, rec["space"], perm):
+        raise ValueError(f"{account['name']} can't {perm} assets in that namespace")
+    return {"id": int(rid), **rec}
+
+
 # ---------- prompts ----------
 def prompt_params(spec):
     """The names of an action's params that are prompts (format "prompt")."""
@@ -342,4 +391,4 @@ def _log_line(db, cfg, intent, user, dry):
 )
 
 
-from . import actions_ai, actions_messages, actions_web  # noqa: E402, F401 - they register their actions; loaded with the catalog
+from . import actions_ai, actions_messages, actions_run, actions_web  # noqa: E402, F401 - they register their actions; loaded with the catalog
diff --git a/fastapi_backend/app/domain/actions_run.py b/fastapi_backend/app/domain/actions_run.py
@@ -0,0 +1,172 @@
+"""Automation actions that run Lens's own work (docs/automations.md, Actions): `run_workflow`, `run_pipeline` and
+`run_routine`.
+
+Each changes something inside Lens (effect `lens`): a workflow or a pipeline on the asset the run is about, in the
+automation's namespaces, runs without asking; on any other asset, and a routine always, it asks first. The work is
+queued by the automation (its jobs and runs are by "automation: <name>"), with its person's permission to process
+assets in that namespace; a routine is an admin's to run, as in Settings.
+
+`asset` takes the activity's object id (`lens:recording/<id>`, the default `{{ event.object.id }}`) or an asset id.
+"""
+
+from __future__ import annotations
+
+from . import actions, budgets, jobs, pipelines, routines, workflows
+
+
+def _int(p, key, what):
+    try:
+        return int(p[key])
+    except (KeyError, TypeError, ValueError):
+        raise ValueError(f"{what} is an id") from None
+
+
+# ---------- run_workflow ----------
+def _run_workflow(db, cfg, intent, user, dry):
+    p = intent["params"]
+    acc = actions.person(db, intent, user)
+    wid = _int(p, "workflow", "the workflow")
+    rec = actions.asset_for(db, intent, acc, actions.asset_id(p.get("asset")), "process")
+    try:
+        w = workflows.get(db, wid, p.get("version"))
+    except KeyError:
+        raise ValueError(f"there's no workflow {wid}" + (f" version {p['version']}" if p.get("version") else "")) from None
+    if w["scope"] != "recording":
+        raise ValueError(f"the workflow {w['name']} organises the graph; run it with a routine")
+    out = {"workflow": wid, "version": w["version"], "asset": rec["id"]}
+    if dry:
+        return {**out, "job": None, "would": f"run the workflow {w['name']} on {rec.get('title') or rec['id']}"}
+    step = {"type": "workflow", "workflow": wid, "version": w["version"]}
+    return {**out, "job": jobs.add_steps(db, rec["id"], [step], by=actions.actor(db, intent)["email"])}
+
+
+# ---------- run_pipeline ----------
+def _run_pipeline(db, cfg, intent, user, dry):
+    p = intent["params"]
+    acc = actions.person(db, intent, user)
+    rec = actions.asset_for(db, intent, acc, actions.asset_id(p.get("asset")), "process")
+    pid = _int(p, "pipeline", "the pipeline") if p.get("pipeline") is not None else None
+    name = "its namespace's pipeline"
+    if pid is not None:
+        try:
+            name = f"the pipeline {pipelines.get(db, pid)['name']}"
+        except KeyError:
+            raise ValueError(f"there's no pipeline {pid}") from None
+    out = {"pipeline": pid, "asset": rec["id"]}
+    if dry:
+        return {**out, "job": None, "would": f"run {name} on {rec.get('title') or rec['id']}"}
+    try:
+        jid = jobs.enqueue(db, rec["id"], None, by=actions.actor(db, intent)["email"], pipeline=pid)
+    except KeyError:
+        raise ValueError(f"there's no pipeline {pid}") from None
+    return {**out, "job": jid}
+
+
+# ---------- run_routine ----------
+def _run_routine(db, cfg, intent, user, dry):
+    p = intent["params"]
+    acc = actions.person(db, intent, user)
+    rid = _int(p, "routine", "the routine")
+    try:
+        r = routines.get(db, rid)
+    except KeyError:
+        raise ValueError(f"there's no routine {rid}") from None
+    if not acc["admin"]:
+        raise ValueError(f"routines are run by admins, and {acc['name']} isn't one")
+    spaces = {int(s.split(":")[1]) for s in routines.refs(db, r) if s.startswith("space:")}
+    if intent["spaces"] and not spaces <= set(intent["spaces"]):
+        raise ValueError(f"the routine {r['name']} works in namespaces outside this automation's")
+    propose = bool(p.get("propose_only"))
+    out = {"routine": rid, "propose_only": propose}
+    if dry:
+        return {**out, "requested": False, "would": f"run the routine {r['name']}" + (", proposing its changes" if propose else "")}
+    verdict = budgets.check(db, cfg, routines.refs(db, r))
+    if not verdict["go"]:
+        raise ValueError(verdict["why"])
+    try:
+        routines.request_run(db, rid, actions.actor(db, intent)["email"], propose)
+    except ValueError as e:
+        raise ValueError(f"the routine {r['name']}: {e}") from None
+    return {**out, "requested": True}
+
+
+actions.register(
+    actions.ActionSpec(
+        name="run_workflow",
+        label="Run a workflow",
+        group="Run",
+        description="Runs a workflow on an asset as a job: at once on the asset the activity is about, else asking first.",
+        params={
+            "type": "object",
+            "properties": {
+                "workflow": {"type": "integer", "title": "Workflow"},
+                "version": {"type": "integer", "title": "Version", "description": "empty: the published one"},
+                "asset": actions.asset_param(),
+            },
+        },
+        required=("workflow",),
+        effect="lens",
+        subject=actions.asset_subject,
+        output={
+            "type": "object",
+            "properties": {
+                "job": {"type": "integer"},
+                "workflow": {"type": "integer"},
+                "version": {"type": "integer"},
+                "asset": {"type": "integer"},
+            },
+        },
+        perform=_run_workflow,
+        describe=lambda p: f"Run workflow {p.get('workflow')} on {actions.on_asset(p)}",
+    )
+)
+
+actions.register(
+    actions.ActionSpec(
+        name="run_pipeline",
+        label="Run a pipeline",
+        group="Run",
+        description="Processes an asset again with a pipeline (or its namespace's own), keeping its outputs.",
+        params={
+            "type": "object",
+            "properties": {
+                "pipeline": {"type": "integer", "title": "Pipeline", "description": "empty: the one its namespace uses"},
+                "asset": actions.asset_param(),
+            },
+        },
+        effect="lens",
+        subject=actions.asset_subject,
+        output={
+            "type": "object",
+            "properties": {"job": {"type": "integer"}, "pipeline": {"type": "integer"}, "asset": {"type": "integer"}},
+        },
+        perform=_run_pipeline,
+        describe=lambda p: f"Run {'pipeline ' + str(p['pipeline']) if p.get('pipeline') else 'its pipeline'} on {actions.on_asset(p)}",
+    )
+)
+
+actions.register(
+    actions.ActionSpec(
+        name="run_routine",
+        label="Run a routine",
+        group="Run",
+        description="Asks the scheduler to run a routine now (an admin's); always asks you first.",
+        params={
+            "type": "object",
+            "properties": {
+                "routine": {"type": "integer", "title": "Routine"},
+                "propose_only": {"type": "boolean", "title": "Only propose its changes", "default": False},
+            },
+        },
+        required=("routine",),
+        effect="lens",
+        subject=lambda p: None,
+        output={
+            "type": "object",
+            "description": "the scheduler starts the run within half a minute; its id shows in the routine's runs",
+            "properties": {"routine": {"type": "integer"}, "requested": {"type": "boolean"}, "propose_only": {"type": "boolean"}},
+        },
+        perform=_run_routine,
+        describe=lambda p: f"Run routine {p.get('routine')}" + (", proposing its changes" if p.get("propose_only") else ""),
+    )
+)