Files
rose-ash/plans/business-logic-fed-flows.md
giles f240c46fa8 plan: account for celery-sx as the distributed/durable runner adapter
celery-sx = one more runner on artdag/op-table-runner, not a Celery port: broker=persist KV,
workers=er-scheduler, result backend=content-addressed (dedup free), retries/replay=flow-on-
erlang, fan-out=artdag/schedule. ~few hundred lines of glue, zero packages, 'Celery the way it
should have been' on erlang-on-sx. DEMAND-DRIVEN (RX) — build when a DAG needs heavy compute /
long-running-retryable / cross-machine fan-out; the synchronous op-table runner covers P0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-02 13:35:44 +00:00

137 lines
11 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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 just runners — a ladder by capability, same DAG throughout (**durability is a
runner capability, not a DAG feature**):
- **op-table / execute-fold runner** — synchronous, local, in-request. Covers P0.
- **Erlang runner** — durable: suspend/resume/`wait`, deterministic replay (flow-on-erlang).
- **celery-sx runner** — 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** — 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.
**Prior narrower framing (kept below as the concrete first slice):** wire the live host's publish
onto next/'s Erlang trigger→flow machinery. That's now understood as *one adapter choice* — a good
concrete spike, but P0 should keep the DAG + the state-change event substrate-CLEAN so runners and
transports swap trivially.
**Design (decided 2026-07-02; corrected after review):**
- **Activity log = every OBSERVABLE object-level state change** — the federated event source.
NOT just CID deltas: verified that relations write `edge:*` rows, NOT the record, so a relation
change does NOT shift the CID. So the log has TWO event classes (ActivityPub-faithful):
content/status change → a CID-carrying `Create`/`Update` (the record's canon incl. :status → the
CID); relation change (relate/unrelate/tag) → an `Add`/`Remove` activity referencing the edge.
(CID delta is one class, not the whole log — this is the fix to the original "every CID delta".)
- **Verbs are TRANSITIONS, not raw deltas.** `on-publish` = the draft→published transition
(fire-once), not every CID delta of a published post. The emitter picks the verb: `Create` on
first publish, `Update` on subsequent content edits, `Add`/`Remove` on relations, `Delete` on
unpublish/delete. Triggers are scoped to the transition, so re-editing doesn't re-fire on-publish.
- **Triggers = declared subscriptions** — a type declares named triggers (on-publish, on-relate,
…); flows fire only on matching ones. (fed-sx `DefineTrigger`: activity-type → flow-name + guard
+ actor-scope.) Log complete, execution precise.
- **Flows = hybrid, split by DURABILITY not "complexity":** SYNCHRONOUS declarative logic authored
as **SX composition** (the execute-fold: effect/alt/each — eager, one-pass, NO suspend; live in
/workflow-demo). Anything needing a timer / suspend-resume / human-in-the-loop is a named
**Erlang flow** (next/flow/*.erl — flow_spec:sequence/branch/wait, effect-as-data, deterministic
replay). The execute-fold canNOT express `wait`; that's the escape-hatch boundary.
- **Effects are DATA; a DRIVER dispatches them.** Flows return effect descriptions (digest_sent,
a DigestSent activity) — they perform no IO (a blocking call deadlocks the er-scheduler). For P0
the HOST is the driver (dispatch the effect → a durable record / append the follow-up activity /
show it). The driver closing the loop back to object state is P4.
- **Federated execution = Erlang** (`next/` fed-sx Milestone-1 kernel: trigger_registry +
flow_dispatch + pipeline post-append fan-out). Authoring stays SX; the fed-sx activity is the
bridge. flow-on-sx (Scheme, lib/flow) remains for purely-local durable logic.
- **The type carries its whole contract:** fields+grammar (content) · allowed relations
(external) · triggers+flows (behavior). All composition, all editable in the type-def editor.
**Verified baseline:** `next/tests/triggers_e2e.sh` = 10/10 — publish activity → trigger →
blog_publish_digest flow (urgent/newsletter-suspend/draft-skip/guard-reject/dedup). This is the
reference P0 wires the live host onto.
## P0 — publish workflow, end-to-end (spike)
Prove: live host publishes a post → fed-sx activity → on-publish trigger → blog_publish_digest.
- [x] **P0.1 — the publish-activity contract (SX side).** host/blog--publish-activity(slug):
a post record → the fed-sx activity {:type "create" :actor "site" :id <CID> :object {:type
"article" :category … :slug …}}. category from field-value "category", else first tag, else
"urgent". + host/blog--post-category. blog 200/200 (3 tests). DONE 2026-07-02.
- [ ] **P0.2 — the dispatch bridge.** Detect the draft→published TRANSITION (edit-submit /a publish
action where prev status ≠ "published" and new = "published") — fire-once, not on every edit —
and emit host/blog--publish-activity into the trigger machinery. RUNTIME CHOICE for P0: in-process
— serve.sh loads next/ kernel + registers the on-publish trigger; host emits via the erlang-on-sx
bridge (erlang-eval-ast pipeline:apply_triggers). RISK: run this in the handler BODY (never a
render/quasiquote — VmSuspended), and the flow must stay effect-as-data (no blocking, or the
er-scheduler deadlocks). SPIKE the SX→erlang-on-sx call in isolation first. (P3 swaps this for
real fed-sx delivery over next/kernel/http_server.erl.)
- [ ] **P0.3 — the effect is dispatched (host = driver).** pipeline:apply_triggers returns the
flow's effect-as-data ({digest_sent, Emails, DigestObject}); the HOST driver dispatches it —
P0: append a DigestSent record / activity + show it on a /flows page (or on the post). ACCEPTANCE:
publish a post on the LIVE host → the DigestSent surfaces, driven by the real flow. (Marshalling:
the SX activity dict ↔ the erlang proplist term is the fiddly part — factor host/blog--activity->erl.)
## P1 — types declare behavior (generalize)
- [ ] A Composition field / the type carries a :triggers list (on-publish → flow-name + guard) —
edited in the type-def editor, like grammar + relations.
- [ ] A fold turns a type's declared behavior into DefineTrigger + flow registrations at boot.
- [ ] SYNCHRONOUS flows authored as SX composition (execute-fold: effect/alt/each — one-pass, no
suspend) → dispatched to the engine; anything needing wait/suspend/human-gate references a named
Erlang flow. (Optional bigger: extend the SX flow vocabulary with a `wait` that compiles to
flow_spec — only if authoring durable flows in SX proves worth it.)
## P2 — state-change → activity emission (the CID-delta event source)
- [ ] Every content mutation (new CID) appends a state-change activity to the log. Define the
activity envelope (verb, actor, object CID, prev CID, delta summary).
- [ ] Wire the host's write path (put!/set-comp!/edit-submit) to the append.
## P3 — federation proper
- [ ] Activities cross peers via next/ delivery (http_server / outbox / follower_graph). A remote
service's trigger_registry fires the flow. Everything works over fed-sx.
RISK: next/ delivery had M2 blockers (http-listen handler runs off the er-scheduler context →
gen_server:call can't complete; project_fed_prims_http_listen_scheduler). Confirm delivery is
green before depending on it. Also: the ACTOR MODEL (:actor "site" is a P0 placeholder) is
foundational here — peer_actors / follower_graph / per-author identity underpin who federates to
whom. Deferred through P0P2, but P3 needs it real.
## 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 — 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).