rose-ash/plans/business-logic-fed-flows.md

# Business logic as composition — a content-addressed DAG over pluggable substrates

**Vision (elevated 2026-07-02):** business logic IS art-dag. An object's behavior is a
**content-addressed DAG** (lib/artdag), declared on its **type** alongside content grammar +
allowed relations. Everything else is a pluggable ADAPTER — the same fold/adapter principle as
render-vs-execute-vs-deps, applied to execution/communication/deployment:

- **Behavior = an artdag DAG** — the invariant, content-addressed (`artdag/dag`, analyze/plan/
  optimize/schedule). Business logic, art media pipelines, workflows — all the same abstraction.
- **Execution = an injected RUNNER** (`artdag/run dag RUNNER cache`; `artdag/op-table-runner`).
  Substrates are runners on a **capability ladder**, same DAG throughout — a node DECLARES the
  capabilities it needs, a runner ADVERTISES what it supports, and the match is checked at bind
  time (fail fast, not a mystery at run time). artdag/analyze computes a DAG's required-capability
  set → its MINIMUM runner. So "simple in SX / durable in Erlang / distributed in celery-sx" is a
  DERIVED property of the DAG, not a human judgment; a trivial rule is a one-node DAG needing only
  `{effect}` and runs synchronously with zero ceremony. Business logic and media pipelines are the
  SAME structure (a content-addressed op-DAG); they differ only in the capabilities their nodes
  require, hence the runner. The ladder (capabilities each runner adds):
  - **op-table / execute-fold runner** — `{effect, branch, each}` — synchronous, local, in-request. Covers P0.
  - **Erlang runner** — `+ suspend/resume/wait` (durable), deterministic replay (flow-on-erlang).
  - **celery-sx runner** — `+ parallel, retry, offload` — distributed/durable task executor, "Celery the way it should have
    been" on erlang-on-sx, ZERO packages. It's a LEAN GLUE of parts we already have, not a
    reimplementation: broker = lib/persist KV (durable enqueue/claim/ack/visibility-timeout) ·
    worker pool = the er-scheduler / Erlang processes · result backend = content-addressed results
    (artdag keys by content-id → dedup/memoization FREE — Celery bolts this on badly) · retries/
    replay = flow-on-erlang · scheduling/fan-out/chords = artdag/schedule (minikanren CLP(FD)) +
    the DAG's topo batches · the plug point = artdag/op-table-runner. The genuinely-new code is
    small (~a few hundred lines): a durable queue + a worker loop (pull node → runner → write
    content-addressed result) + retry/backoff. **BUILD WHEN A DAG DEMANDS IT** — heavy compute,
    long-running/retryable tasks, or fan-out across machines — NOT for the synchronous P0.
  - **real-Celery over artdag/L1** — `+ gpu/heavy-compute` — the existing Python media pipeline (JAX/IPFS) as a runner.
- **Communication = an injected TRANSPORT** (`artdag/federation`, transport injected). Substrates:
  fed-sx (ActivityPub/next/), internal HMAC HTTP (services), IPFS (content-addressed). Because
  content-ids are global, a result computed on one instance is reusable on another by id.
- **Deployment = PLACEMENT** — a subdomain service, a fed-sx peer, an L1 worker: just where a
  runner runs. Not the essence.
- **State change → triggers a DAG** (over a transport) → executed by a runner → effects (data) a
  driver dispatches. fed-sx + Erlang is ONE adapter set (durable/federated), not THE architecture.

So: the TYPE carries content-grammar + allowed-relations + a **behavior DAG (+ triggers)**; the
object's state changes emit activities; the platform picks runner/transport/placement per context.

**Design (decided 2026-07-02; corrected after review):**
- **Activity log = every OBSERVABLE object-level state change** — the event source. NOT just CID
  deltas: relations write `edge:*` rows, NOT the record, so a relation change does NOT shift the
  CID. Two ActivityPub-faithful event classes: content/status change → a CID-carrying
  `Create`/`Update`; relation change → an `Add`/`Remove` referencing the edge.
- **Verbs are TRANSITIONS, not raw deltas.** `on-publish` = draft→published (fire-once), not every
  CID delta of a published post. Emitter picks the verb (Create/Update/Add/Remove/Delete). Global
  fire-once is the emitter's job (+ a durable inbox for federation); the seam only dedups in-call.
- **Triggers = declared subscriptions** — a type declares named triggers; flows fire only on
  matching ones. Log complete, execution precise.
- **Flows split by CAPABILITY (DERIVED, not a human "complexity" call):** every flow is a
  content-addressed op-DAG. A node declares the capabilities it needs (a `wait` needs `suspend`, a
  fan-out needs `parallel`, a heavy op needs `offload`); a runner advertises what it supports;
  artdag/analyze computes the DAG's required set → its MINIMUM runner; bind-time checks required ⊆
  runner (fail fast). So the sync/durable/distributed choice falls out of the DAG — a `{effect}`-only
  DAG runs synchronously (no ceremony); a DAG with a `wait` node auto-requires the Erlang runner.
  Business logic and media pipelines are the SAME structure — they differ only in node capabilities.
- **Effects are DATA; a DRIVER dispatches them** (perform no IO — a blocking call deadlocks the
  er-scheduler). The host is the driver for P0.
- **The type carries its whole contract:** fields+grammar (content) · allowed relations (external)
  · **behavior bindings** (triggers → DAG). All composition, all editable in the type-def editor.
- **CANONICAL ACTIVITY = the seam shape** `{:verb :actor :object <cid> :object-type :delta :prev
  :ts :id}`; each runner adapter MARSHALS it to its substrate (e.g. the Erlang runner → next/'s
  `[{type,…},{object,[…]}]` proplist). NOTE: P0.1's host/blog--publish-activity currently emits the
  next/-Erlang shape ({:type :object-dict}); P0.4 reconciles it to the canonical seam shape + a
  marshaller.

**Reference (one durable-runner substrate, verified):** `next/tests/triggers_e2e.sh` = 10/10 —
next/'s trigger_registry + flow_dispatch + blog_publish_digest (suspend/resume/guard/dedup). This
is what the ERLANG RUNNER ADAPTER wraps (Phase RA), not what P0 wires directly.

## The ADAPTER SEAM (design first — the contract every substrate plugs into)
The invariant is an **activity** (state-change event) + a **behavior DAG**. Everything between is
a swappable adapter — each a dict-of-functions (SX-native, like the fold domains). Six contracts:

1. **Activity** (content-addressed event): `{:verb "create"|"update"|"add"|"remove"|"delete"
   :actor :object <cid> :object-type :delta :prev <cid> :ts}`.
2. **Behavior binding** (declared on a TYPE): `{:on {:verb :object-type :guard} :dag <ref>}`. The
   TYPE carries content-grammar + allowed-relations + these bindings. NO :runner hint — the runner
   is DERIVED: the DAG's required-capability set (artdag/analyze over its nodes; a node declares
   `:needs` e.g. `#{suspend}`) selects the minimum runner that advertises them.
3. **Trigger-registry adapter** — `{:register! (fn spec dag hint) :match (fn activity -> [binding])}`.
   Impls: a local SX matcher / next/'s Erlang trigger_registry.
4. **Runner adapter** — `{:capabilities <set>  :run (fn dag env -> {:status "done"|"suspended"|
   "failed" :effects :resume :error})}`. :capabilities is what it ADVERTISES (op-table `{effect,
   branch, each}`; Erlang `+ suspend`; celery-sx `+ parallel, retry, offload`); the binder checks
   dag-required ⊆ runner-caps and fails fast otherwise. env = `{:activity :actor :ctx :effects
   :binding}` (:effects = injected external-read interfaces, deterministic for replay). `:results`
   is runner-INTERNAL. Durability = the runner's, not the DAG's. A durable runner that SUSPENDS is
   wired at construction to the transport's INBOUND channel and injects its out-of-band completion
   there (pump drains it).
5. **Transport adapter** — `{:emit (fn activity) :deliver (fn -> [activity])}`. `:emit` = outbound
   (log/publish), `:deliver` = inbound (peers + async runner completions). Impls: in-process /
   fed-sx (next/) / HTTP / IPFS. Content-ids global → results move by id.
6. **Effect driver** — `{:dispatch (fn effect -> [activity])}`. Perform the effect-as-data; may emit
   NEW activities (loop closure).

**Engine** (`behavior/make-engine {:triggers :runner :transport :driver :effects? :ctx-of?}`) →
`behavior/process(engine, activity)` / `behavior/pump(engine)`:
```
step(activity):  if activity.id already :seen → skip (dedup)      ; in-call cycle guard
  emit(transport, activity)                                       ; log
  for b in match(triggers, activity):
     r = run(runner, b.dag, {:activity :actor :ctx :effects :binding})
     case r.status:
       "done"      → for eff in r.effects: for a in dispatch(driver, eff): step(a)   ; loop closes
       "suspended" → record (+ r.resume); a durable runner completes out-of-band → inbound → pump
       "failed"    → record (+ r.error) for flow-level retry / dead-letter
pump():  for a in transport.deliver(): step(a)                    ; inbound drain, shared :seen
```
Every stage injected → swap runner (sync→Erlang→celery-sx), transport (in-proc→fed-sx), registry,
driver — DAG + engine unchanged. **Global** idempotency = emitter fire-once + a durable inbox.

**Build order:** (a) **DONE** — seam as SX contracts + reference engine, tested substrate-agnostic:
:status branch (done/suspended/failed), injected env (:ctx + :effects), dedup by activity :id,
behavior/pump for inbound, async-completion loop (suspend → out-of-band inject → pump). **behavior
10/10.** (b) P0 supplies the REAL SYNCHRONOUS adapters; durable/federated runners+transports are
later adapter phases (RA/TA below).

## P0 — publish workflow on the seam (SYNCHRONOUS, all-SX)
Prove: live host publish → the seam engine (in-process transport · local-SX trigger registry ·
sync op-table runner over an SX publish-DAG · host driver) → the effect surfaces. NO Erlang, NO
fed-sx yet — those are adapter phases (RA/TA). Every piece swaps later; the DAG + engine don't.

- [x] **P0.1 — publish-activity contract (SX side).** host/blog--publish-activity + post-category.
  blog 200/200. NOTE: emits the next/-Erlang shape today; P0.4 reconciles to the canonical seam shape.
- [ ] **P0.2 — the publish-DAG + op-table runner + the CAPABILITY check.** Author the publish
  workflow as an SX artdag DAG (validate → publish → notify/digest) whose nodes need only
  `{effect, branch}`, + the op-table runner (advertises `{effect, branch, each}`). ACCEPTANCE
  (the hypothesis test): (a) the workflow expresses NATURALLY as a DAG — if it's forced, that's the
  signal the node vocabulary / representation needs rethinking BEFORE RA/TA; (b) artdag/analyze
  computes the DAG's required-capability set = `{effect, branch}`; (c) the binder confirms required
  ⊆ op-table caps → runs; (d) run it → the expected effect-as-data. Then flip one node to `wait`
  and confirm the bind FAILS FAST against the op-table runner (would require the Erlang runner, RA)
  — proving durability is derived, not chosen.
- [ ] **P0.3 — wire the seam on the live host.** A local-SX trigger registry (on-publish →
  publish-DAG), an in-process transport (a KV-backed log), the host as effect driver. In edit-submit
  detect the draft→published TRANSITION (prev≠published & new=published — fire-once), build the
  activity, behavior/process it (in the handler BODY, not a render). ACCEPTANCE: publish a post on
  the LIVE host → the effect surfaces (a /flows page + a durable record).
- [ ] **P0.4 — canonical activity + reconcile.** Move host/blog--publish-activity to the seam shape
  {:verb :actor :object <cid> :object-type :delta :ts :id}; keep the category/slug fields the DAG reads.

## P1 — types DECLARE behavior (generalize)
- [ ] The type carries :behavior = [{:on {:verb :object-type :guard} :dag <ref>}] — edited in the
  type-def editor beside grammar + relations. NO runner hint (the runner is DERIVED from the DAG's
  required capabilities).
- [ ] host/blog--engine-for(object) builds the seam engine from the object's type bindings +
  registers the triggers, selecting each DAG's MINIMUM runner via artdag/analyze (required-caps) —
  a `{effect}`-only DAG runs synchronously; a `wait` node pulls in the Erlang runner automatically.
- [ ] The type-def editor can SHOW each behavior's required capabilities + which runner it resolves
  to (the derived sync/durable/distributed classification, visible not hand-set).

## P2 — state-change → activity emission (ALL events, not just publish)
- [ ] Wire the host write path: put!/set-comp!/edit-submit → Create/Update; relate!/unrelate!/tag →
  Add/Remove. Emit canonical activities into the transport log. Define the delta summary.

## RA — the ERLANG (durable) RUNNER adapter   ← the old "fed-sx spike", now an adapter
- [ ] A seam runner wrapping next/'s flow_dispatch + flow_store: marshal the canonical activity →
  next/'s proplist, dispatch to a named flow (blog_publish_digest), map the result back to
  {:status done|suspended :effects :resume}. On suspend, wire the resumed completion → the transport
  inbound (pump). Baseline: next/tests/triggers_e2e.sh 10/10. RISK: er-scheduler context
  (project_fed_prims_http_listen_scheduler); keep flows effect-as-data (no blocking). SPIKE the
  SX→erlang-on-sx call in isolation first.

## TA — the FED-SX TRANSPORT adapter   ← federation proper
- [ ] A seam transport over next/ delivery: :emit → outbox → peers; :deliver → inbox. A remote
  peer's engine fires its own triggers. Everything works over fed-sx. RISK: next/ delivery M2
  blockers (er-scheduler context). Needs the ACTOR MODEL real (:actor was a P0 placeholder —
  peer_actors / follower_graph / per-author identity).

## RX — celery-sx runner (DEMAND-DRIVEN, not scheduled)
Build the distributed/durable runner adapter the moment a real DAG needs heavy compute /
long-running-retryable tasks / cross-machine fan-out (the artdag/JAX media case, or federated
flows that can't run in-request). New code is small — glue persist (durable queue: enqueue/claim/
ack/visibility-timeout) + er-scheduler (worker loop: pull node → op-table-runner → content-
addressed result) + artdag/schedule (fan-out) + retry/backoff. Slots in at artdag/op-table-runner
alongside the synchronous + Erlang runners. Zero packages. Do NOT pre-build; the op-table runner
covers everything until a DAG's cost/latency/placement forces the substrate.

## P4 — close the loop
- [ ] Flow effects mutate objects back durably (a flow's DescribeEffects → host writes / new
  activities), so business logic can change state, which federates, which triggers more flows.

## Progress log (newest first)
- 2026-07-02 — folded in CAPABILITY-TYPED nodes / CAPABILITY-ADVERTISING runners. A node declares
  `:needs` (wait→suspend, fan-out→parallel, heavy→offload); a runner advertises `:capabilities`
  (op-table {effect,branch,each}; Erlang +suspend; celery-sx +parallel,retry,offload); artdag/analyze
  computes a DAG's min-runner; the binder checks required ⊆ runner-caps (fail fast). So the sync/
  durable/distributed split is DERIVED from the DAG, not a human call — a {effect}-only DAG runs with
  zero ceremony; a wait node auto-requires Erlang. Removed the :runner hint from the binding. P0.2
  gains the hypothesis test: does the publish workflow express naturally as a DAG, and does flipping
  a node to `wait` fail-fast against the op-table runner? Clarifies "business logic = art-dag" — same
  op-DAG structure, differing only in node capabilities, hence runner. (Insight: they're two
  instances of one thing, suitable to run very differently.)
- 2026-07-02 — whole-plan coherence review. The reframe (artdag+seam) had left the middle stale:
  P0.2/P0.3 still described the Erlang-bridge-first path; P0.1's activity didn't match the seam
  contract; the seam section predated the enrichment. Fixed: P0 rewritten around the seam + SX
  op-table runner (all-SX, no Erlang/fed-sx); the Erlang/fed-sx path DEMOTED to explicit adapter
  phases RA (durable runner) + TA (fed-sx transport); canonical activity shape + P0.4 reconcile;
  seam contract refreshed to behavior.sx (status/dedup/pump/async, behavior 10/10); stray bits cleared.
- 2026-07-02 — seam DONE + reviewed twice. lib/host/behavior.sx (engine + 4 adapters); enriched
  substrate-agnostic (status/env/dedup/pump); 2nd review corrected the async-completion contract
  (construction-wired inbound + pump, not env :emit) + proved it. behavior 10/10, conformance 580.
- 2026-07-02 — P0.1 done. host/blog--publish-activity + host/blog--post-category; the publish
  contract in SX, 200/200. Verified next/ triggers e2e baseline 10/10. Roadmap anchored. NEXT:
  P0.2 the dispatch bridge (in-process: serve.sh loads next/ kernel + registers the on-publish
  trigger; host emits the activity via the erlang-on-sx bridge to pipeline:apply_triggers).