diff --git a/.gitignore b/.gitignore
index 6985cf1b..5ea5efc1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -12,3 +12,5 @@ Cargo.lock
 
 # MSVC Windows builds of rustc generate these, which store debugging information
 *.pdb
+
+.vscode/
diff --git a/Cargo.lock b/Cargo.lock
index 04d142a7..b78638fd 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -600,7 +600,7 @@ checksum = "fdf6721fb0140e4f897002dd086c06f6c27775df19cfe1fccb21181a48fd2c98"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -678,6 +678,22 @@ dependencies = [
  "tower-service",
 ]
 
+[[package]]
+name = "bacalhau"
+version = "0.1.0"
+dependencies = [
+ "handlebars",
+ "reqwest",
+ "section",
+ "serde",
+ "serde_json",
+ "serde_yaml",
+ "stub",
+ "tokio",
+ "tokio-stream",
+ "tokio-util",
+]
+
 [[package]]
 name = "backtrace"
 version = "0.3.69"
@@ -767,7 +783,7 @@ dependencies = [
  "proc-macro-crate 2.0.1",
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
  "syn_derive",
 ]
 
@@ -927,7 +943,7 @@ dependencies = [
  "heck",
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -1397,7 +1413,7 @@ checksum = "53b153fd91e4b0147f4aced87be237c98248656bb01050b96bf3ee89220a8ddb"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -1508,6 +1524,20 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "handlebars"
+version = "4.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "faa67bab9ff362228eb3d00bd024a4965d8231bbb7921167f0cfa66c6626b225"
+dependencies = [
+ "log",
+ "pest",
+ "pest_derive",
+ "serde",
+ "serde_json",
+ "thiserror",
+]
+
 [[package]]
 name = "hashbrown"
 version = "0.12.3"
@@ -2135,6 +2165,7 @@ name = "myceliald"
 version = "0.2.0"
 dependencies = [
  "anyhow",
+ "bacalhau",
  "base64 0.21.5",
  "clap",
  "common",
@@ -2406,7 +2437,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -2547,6 +2578,51 @@ version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
+[[package]]
+name = "pest"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f200d8d83c44a45b21764d1916299752ca035d15ecd46faca3e9a2a2bf6ad06"
+dependencies = [
+ "memchr",
+ "thiserror",
+ "ucd-trie",
+]
+
+[[package]]
+name = "pest_derive"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bcd6ab1236bbdb3a49027e920e693192ebfe8913f6d60e294de57463a493cfde"
+dependencies = [
+ "pest",
+ "pest_generator",
+]
+
+[[package]]
+name = "pest_generator"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a31940305ffc96863a735bef7c7994a00b325a7138fdbc5bda0f1a0476d3275"
+dependencies = [
+ "pest",
+ "pest_meta",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.48",
+]
+
+[[package]]
+name = "pest_meta"
+version = "2.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a7ff62f5259e53b78d1af898941cdcdccfae7385cf7d793a6e55de5d05bb4b7d"
+dependencies = [
+ "once_cell",
+ "pest",
+ "sha2",
+]
+
 [[package]]
 name = "pin-project"
 version = "1.1.3"
@@ -2564,7 +2640,7 @@ checksum = "4359fd9c9171ec6e8c62926d6faaf553a8dc3f64e1507e76da7911b4f6a04405"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -2684,9 +2760,9 @@ dependencies = [
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.71"
+version = "1.0.76"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75cb1540fadbd5b8fbccc4dddad2734eba435053f725621c070711a14bb5f4b8"
+checksum = "95fc56cda0b5c3325f5fbbd7ff9fda9e02bb00bb3dac51252d2f1bfa1cb8cc8c"
 dependencies = [
  "unicode-ident",
 ]
@@ -2733,9 +2809,9 @@ dependencies = [
 
 [[package]]
 name = "quote"
-version = "1.0.33"
+version = "1.0.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5267fca4496028628a95160fc423a33e8b2e6af8a5302579e322e4b520293cae"
+checksum = "291ec9ab5efd934aaf503a6466c5d5251535d108ee747472c3977cc5acc868ef"
 dependencies = [
  "proc-macro2",
 ]
@@ -3077,7 +3153,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rust-embed-utils",
- "syn 2.0.42",
+ "syn 2.0.48",
  "walkdir",
 ]
 
@@ -3266,22 +3342,22 @@ checksum = "a3f0bf26fd526d2a95683cd0f87bf103b8539e2ca1ef48ce002d67aad59aa0b4"
 
 [[package]]
 name = "serde"
-version = "1.0.193"
+version = "1.0.195"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "25dd9975e68d0cb5aa1120c288333fc98731bd1dd12f561e468ea4728c042b89"
+checksum = "63261df402c67811e9ac6def069e4786148c4563f4b50fd4bf30aa370d626b02"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.193"
+version = "1.0.195"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43576ca501357b9b071ac53cdc7da8ef0cbd9493d8df094cd821777ea6e894d3"
+checksum = "46fe8f8603d81ba86327b23a2e9cdf49e1255fb94a4c5f297f6ee0547178ea2c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -3326,6 +3402,19 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_yaml"
+version = "0.9.30"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1bf28c79a99f70ee1f1d83d10c875d2e70618417fda01ad1785e027579d9d38"
+dependencies = [
+ "indexmap 2.1.0",
+ "itoa",
+ "ryu",
+ "serde",
+ "unsafe-libyaml",
+]
+
 [[package]]
 name = "server"
 version = "0.2.0"
@@ -3815,7 +3904,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -3845,9 +3934,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.42"
+version = "2.0.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b7d0a2c048d661a1a59fcd7355baa232f7ed34e0ee4df2eef3c1c1c0d3852d8"
+checksum = "0f3531638e407dfc0814761abb7c00a5b54992b849452a0646b7f65c9f770f3f"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3863,7 +3952,7 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -3947,7 +4036,7 @@ checksum = "01742297787513b79cf8e29d1056ede1313e2420b7b3b15d0a768b4921f549df"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -4041,7 +4130,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -4207,7 +4296,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -4241,6 +4330,12 @@ version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825"
 
+[[package]]
+name = "ucd-trie"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed646292ffc8188ef8ea4d1e0e0150fb15a5c2e12ad9b8fc191ae7a8a7f3c4b9"
+
 [[package]]
 name = "unicase"
 version = "2.7.0"
@@ -4289,6 +4384,12 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39ec24b3121d976906ece63c9daad25b85969647682eee313cb5779fdd69e14e"
 
+[[package]]
+name = "unsafe-libyaml"
+version = "0.2.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ab4c90930b95a82d00dc9e9ac071b4991924390d46cbd0dfe566148667605e4b"
+
 [[package]]
 name = "untrusted"
 version = "0.7.1"
@@ -4343,7 +4444,7 @@ checksum = "f49e7f3f3db8040a100710a11932239fd30697115e2ba4107080d8252939845e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -4404,7 +4505,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
  "wasm-bindgen-shared",
 ]
 
@@ -4438,7 +4539,7 @@ checksum = "f0eb82fcb7930ae6219a7ecfd55b217f5f0893484b7a13022ebb2b2bf20b5283"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
@@ -4716,7 +4817,7 @@ checksum = "9ce1b18ccd8e73a9321186f97e46f9f04b778851177567b1975109d26a08d2a6"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.42",
+ "syn 2.0.48",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 5e3a409f..d60075d7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,6 +6,7 @@ members = [
     "pipe/runtime",
     "pipe/arrow_msg",
     "pipe/section",
+    "pipe/section/section_impls/bacalhau",
     "pipe/section/section_impls/hello_world",
     "pipe/section/section_impls/sqlite_connector",
     "pipe/section/section_impls/excel_connector",
diff --git a/common/src/lib.rs b/common/src/lib.rs
index 9048d7c1..1aee2c52 100644
--- a/common/src/lib.rs
+++ b/common/src/lib.rs
@@ -53,6 +53,7 @@ pub enum Destination {
     Sqlite_Connector(SqliteConnectorConfig),
     Snowflake(SnowflakeConfig),
     Sqlite_Physical_Replication(SqlitePhysicalReplicationDestinationConfig),
+    Bacalhau(BacalhauDestinationConfig),
     Hello_World(HelloWorldDestinationConfig),
     Kafka(KafkaDestinationConfig),
     Postgres_Connector(PostgresConnectorDestinationConfig),
@@ -115,6 +116,14 @@ pub struct SqlitePhysicalReplicationSourceConfig {
     // database path
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct BacalhauDestinationConfig {
+    #[serde(flatten)]
+    pub common_attrs: CommonAttrs,
+    pub job: String,
+    pub jobstore: String,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct HelloWorldSourceConfig {
     #[serde(flatten)]
diff --git a/console/components/Flow/index.tsx b/console/components/Flow/index.tsx
index 8cfa1690..3324b277 100644
--- a/console/components/Flow/index.tsx
+++ b/console/components/Flow/index.tsx
@@ -40,6 +40,7 @@ import {
   KafkaSourceNode,
   SnowflakeSourceNode,
   SnowflakeDestinationNode,
+  BacalhauDestinationNode,
   HelloWorldSourceNode,
   HelloWorldDestinationNode,
   ExcelSourceNode,
@@ -73,11 +74,10 @@ const useStyles = createStyles((theme) => ({
     marginBottom: theme.spacing.md,
 
     "&:not(:last-of-type)": {
-      borderBottom: `${rem(1)} solid ${
-        theme.colorScheme === "dark"
-          ? theme.colors.dark[4]
-          : theme.colors.gray[3]
-      }`,
+      borderBottom: `${rem(1)} solid ${theme.colorScheme === "dark"
+        ? theme.colors.dark[4]
+        : theme.colors.gray[3]
+        }`,
     },
   },
 
@@ -184,6 +184,7 @@ const nodeTypes = {
   snowflake_destination: SnowflakeDestinationNode,
   hello_world_source: HelloWorldSourceNode,
   hello_world_destination: HelloWorldDestinationNode,
+  bacalhau_destination: BacalhauDestinationNode,
   postgres_connector_source: PostgresConnectorSourceNode,
   postgres_connector_destination: PostgresConnectorDestinationNode,
   mysql_connector_destination: MysqlConnectorDestinationNode,
@@ -282,7 +283,7 @@ function NavbarSearch(props: NavbarSearchProps) {
         onDragStart={(event) => onDragStart(event, null, source, source)}
         draggable
       >
-        Mycelial Server 
+        Mycelial Server
       </div>,
     ];
   };
@@ -341,7 +342,7 @@ async function getConfigs(token: string) {
     });
     const result = await response.json();
     return result;
-  } catch (error) {}
+  } catch (error) { }
 }
 
 const dagreGraph = new dagre.graphlib.Graph();
@@ -564,7 +565,7 @@ function Flow() {
 
       if (edge.data?.id) {
         let payload = {
-          configs: [{id: edge.data.id, pipe: pipe}]
+          configs: [{ id: edge.data.id, pipe: pipe }]
         }
         try {
           const response = await fetch("/api/pipe", {
@@ -582,7 +583,7 @@ function Flow() {
         }
       } else {
         let payload = {
-          configs: [{pipe: pipe}]
+          configs: [{ pipe: pipe }]
         }
         const response = await fetch("/api/pipe", {
           method: "POST",
@@ -605,7 +606,7 @@ function Flow() {
                   id: id,
                 },
               }
-            } 
+            }
             return ed;
           });
         })
diff --git a/console/components/nodes.tsx b/console/components/nodes.tsx
index 0dbd37cd..69d6ec4b 100644
--- a/console/components/nodes.tsx
+++ b/console/components/nodes.tsx
@@ -16,6 +16,107 @@ import styles from "@/components/Flow/Flow.module.css";
 import { ClientContext } from "./context/clientContext";
 import { ClientContextType } from "./@types/client";
 
+
+const BacalhauDestinationNode: FC<NodeProps> = memo(({ id, data, selected }) => {
+  const instance = useReactFlow();
+  const { clients } = useContext(ClientContext) as ClientContextType;
+
+  let initialValues = useMemo(() => {
+    return {
+      client: data.client ? data.client : "-",
+      job: data.job ? data.job : "job",
+      jobstore: data.jobstore ? data.jobstore : "jobstore",
+    };
+  }, []);
+
+  const handleChange = useCallback((name: string, value: string) => {
+    instance.setNodes((nodes) =>
+      nodes.map((node) => {
+        if (node.id === id) {
+          node.data = {
+            ...node.data,
+            [name]: value,
+          };
+        }
+
+        return node;
+      }),
+    );
+  }, []);
+
+  useEffect(() => {
+    handleChange("client", initialValues.client);
+    handleChange("job", initialValues.job);
+    handleChange("jobstore", initialValues.jobstore);
+  }, []);
+
+  let classNames = `${styles.customNode} `;
+  if (selected) {
+    classNames = classNames + `${styles.selected}`;
+  }
+
+  const removeNode = useCallback((id: string) => {
+    const node = instance.getNode(id);
+    if (node === undefined) {
+      return;
+    }
+    instance.deleteElements({
+      edges: getConnectedEdges([node], []),
+      nodes: [node],
+    });
+  }, []);
+
+  return (
+    <div className={classNames}>
+      <div className=" grid grid-cols-1 gap-x-6 gap-y-2">
+        <h2 className="text-slate-400 font-normal">HelloWorld Destination</h2>
+        <button
+          onClick={() => {
+            if (confirm("Are you sure you want to delete this node?")) {
+              removeNode(id);
+            }
+          }}
+          type="button"
+          className="absolute right-1 top-1 rounded bg-red-200 text-white shadow-sm hover:bg-red-600 focus-visible:outline focus-visible:outline-2 focus-visible:outline-offset-2 focus-visible:outline-red-800"
+          title="delete"
+        >
+          <XMarkIcon className="h-5 w-5" aria-hidden="true" />
+        </button>
+        <Select
+          name="client"
+          label="Client"
+          placeholder="Pick one"
+          defaultValue={initialValues.client}
+          options={(clients || []).map((c) => c.id)}
+          onChange={(value) => {
+            handleChange("client", value || "");
+          }}
+        />
+        <TextInput
+          name="jobstore"
+          label="Job Store"
+          placeholder={initialValues.jobstore}
+          defaultValue={initialValues.jobstore}
+          onChange={(event) => {
+            handleChange("jobstore", event.currentTarget.value)
+          }}
+        />
+        <TextInput
+          name="job"
+          label="Job"
+          placeholder={initialValues.job}
+          defaultValue={initialValues.job}
+          onChange={(event) => {
+            handleChange("job", event.currentTarget.value)
+          }}
+        />
+
+        <Handle type="target" position={Position.Left} id={id} />
+      </div>
+    </div>
+  );
+});
+
 const HelloWorldDestinationNode: FC<NodeProps> = memo(({ id, data, selected }) => {
   const instance = useReactFlow();
   const { clients } = useContext(ClientContext) as ClientContextType;
@@ -1401,7 +1502,7 @@ const PostgresConnectorDestinationNode: FC<NodeProps> = memo(({ id, data, select
 
   let initialValues = useMemo(() => {
     return {
-      url: data.url? data.url: "postgres://root:root@127.0.0.1:5432/test",
+      url: data.url ? data.url : "postgres://root:root@127.0.0.1:5432/test",
       client: data.client ? data.client : "-",
     };
   }, []);
@@ -1485,7 +1586,7 @@ const MysqlConnectorDestinationNode: FC<NodeProps> = memo(({ id, data, selected
 
   let initialValues = useMemo(() => {
     return {
-      url: data.url? data.url: "mysql://root:root@127.0.0.1:3306/test",
+      url: data.url ? data.url : "mysql://root:root@127.0.0.1:3306/test",
       client: data.client ? data.client : "-",
     };
   }, []);
@@ -1574,6 +1675,7 @@ export {
   KafkaDestination,
   SnowflakeSourceNode,
   SnowflakeDestinationNode,
+  BacalhauDestinationNode,
   HelloWorldSourceNode,
   HelloWorldDestinationNode,
   PostgresConnectorSourceNode,
diff --git a/myceliald/Cargo.toml b/myceliald/Cargo.toml
index 054c9114..6ff565bf 100644
--- a/myceliald/Cargo.toml
+++ b/myceliald/Cargo.toml
@@ -11,18 +11,19 @@ clap = { version = "4", features = ["derive", "env"] }
 reqwest = { version = "0.11", features = ["json"] }
 anyhow = "1"
 base64 = { version = "0.21" }
-serde = { version = "1", features = ["derive"]}
+serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-sqlx = { version = "0.7", features = ["sqlite"]}
+sqlx = { version = "0.7", features = ["sqlite"] }
 log = "0.4"
 pretty_env_logger = "0.5"
 toml = "0.7"
 common = { path = "../common" }
 thiserror = "1"
-pipe = { version = "0.2", path = "../pipe/runtime/", package="runtime" }
+pipe = { version = "0.2", path = "../pipe/runtime/", package = "runtime" }
 section = { version = "0.2", path = "../pipe/section/" }
 
 ## sections
+bacalhau = { path = "../pipe/section/section_impls/bacalhau/" }
 hello_world = { path = "../pipe/section/section_impls/hello_world/" }
 sqlite_connector = { path = "../pipe/section/section_impls/sqlite_connector/" }
 excel_connector = { path = "../pipe/section/section_impls/excel_connector/" }
diff --git a/myceliald/config.example.toml b/myceliald/config.example.toml
index d401f2da..3f96df3b 100644
--- a/myceliald/config.example.toml
+++ b/myceliald/config.example.toml
@@ -67,6 +67,13 @@ poll_interval = 5
 # Define all data DESTINATIONS (data stores and directory paths)
 # accessible by the Node that should be exposed as DESTINATIONS to the Sqlite Physical Replication Server
 
+
+[[destinations]]
+display_name = "Bacalhau"
+jobstore = "c:/temp/jobstore"
+type = "bacalhau"
+job = "jobname"
+
 [[destinations]]
 type = "sqlite_physical_replication"
 display_name = "Sqlite Physical Replication Movie"
diff --git a/myceliald/src/constructors/bacalhau.rs b/myceliald/src/constructors/bacalhau.rs
new file mode 100644
index 00000000..a63f1cf8
--- /dev/null
+++ b/myceliald/src/constructors/bacalhau.rs
@@ -0,0 +1,28 @@
+use std::path::Path;
+
+use pipe::{config::Map, types::DynSection};
+use section::{command_channel::SectionChannel, SectionError};
+
+pub fn destination_ctor<S: SectionChannel>(
+    config: &Map,
+) -> Result<Box<dyn DynSection<S>>, SectionError> {
+    let job = config
+        .get("job")
+        .ok_or("bacalhau section requires 'job'")?
+        .as_str()
+        .ok_or("'job' should be a string")?;
+    let jobstore = config
+        .get("jobstore")
+        .ok_or("bacalhau section requires 'jobstore'")?
+        .as_str()
+        .ok_or("'jobstore' should be a string")?;
+
+    // Verify jobstore folder exists
+    if !Path::new(jobstore).exists() {
+        return Err(format!("'jobstore' path '{}' does not exist", jobstore).into());
+    }
+
+    Ok(Box::new(bacalhau::destination::Bacalhau::new(
+        job, jobstore,
+    )))
+}
diff --git a/myceliald/src/constructors/mod.rs b/myceliald/src/constructors/mod.rs
index 52edbe0e..a2ba2ed1 100644
--- a/myceliald/src/constructors/mod.rs
+++ b/myceliald/src/constructors/mod.rs
@@ -1,3 +1,4 @@
+pub mod bacalhau;
 pub mod excel_connector;
 pub mod hello_world;
 pub mod kafka_connector;
diff --git a/myceliald/src/runtime.rs b/myceliald/src/runtime.rs
index 03d64897..61af561a 100644
--- a/myceliald/src/runtime.rs
+++ b/myceliald/src/runtime.rs
@@ -15,6 +15,10 @@ use section::command_channel::SectionChannel;
 /// Setup & populate registry
 fn setup_registry<S: SectionChannel>() -> Registry<S> {
     let arr: &[(&str, Constructor<S>)] = &[
+        (
+            "bacalhau_destination",
+            constructors::bacalhau::destination_ctor,
+        ),
         (
             "hello_world_destination",
             constructors::hello_world::destination_ctor,
diff --git a/pipe/section/section_impls/bacalhau/Cargo.toml b/pipe/section/section_impls/bacalhau/Cargo.toml
new file mode 100644
index 00000000..6b850667
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/Cargo.toml
@@ -0,0 +1,20 @@
+[package]
+name = "bacalhau"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+section = { version = "0.2", path = "../../../section/" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+reqwest = { version = "0.11.22", features = ["json"] }
+tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1.14"
+tokio-util = "0.7.10"
+handlebars = "4.4"
+serde_yaml = "0.9"
+
+[dev-dependencies]
+stub = { path = "../../section_impls/stub/" }
diff --git a/pipe/section/section_impls/bacalhau/src/api.rs b/pipe/section/section_impls/bacalhau/src/api.rs
new file mode 100644
index 00000000..0f9a13c7
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/api.rs
@@ -0,0 +1,49 @@
+use crate::StdError;
+use reqwest;
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+struct SubmitErrorResponse {
+    error: String,
+    #[allow(dead_code)]
+    message: String,
+}
+
+#[derive(Deserialize)]
+struct SubmitResponse {
+    #[serde(rename = "JobID")]
+    job_id: String,
+
+    #[serde(rename = "EvaluationID")]
+    _evaluation_id: String,
+
+    #[serde(rename = "Warnings")]
+    _warnings: Option<Vec<String>>,
+}
+
+pub async fn submit(body: &str) -> Result<String, StdError> {
+    // For now we'll assume a local instance and using the newer version
+    // of the jobs API.
+    let api_url = "http://127.0.0.1:20000/api/v1/orchestrator/jobs";
+    let client = reqwest::Client::new();
+    let response = client
+        .post(api_url)
+        .header("Content-Type", "application/json")
+        .body(body.to_owned())
+        .send()
+        .await?;
+
+    let success = &response.status().is_success();
+    let resp_body = &response.bytes().await?;
+
+    match success {
+        true => {
+            let sresp: SubmitResponse = serde_json::from_slice(resp_body.as_ref()).unwrap();
+            Ok(sresp.job_id)
+        }
+        _ => {
+            let err: SubmitErrorResponse = serde_json::from_slice(resp_body.as_ref()).unwrap();
+            Err(err.error.into())
+        }
+    }
+}
diff --git a/pipe/section/section_impls/bacalhau/src/destination.rs b/pipe/section/section_impls/bacalhau/src/destination.rs
new file mode 100644
index 00000000..d2c463eb
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/destination.rs
@@ -0,0 +1,136 @@
+use crate::jobstore::JobStore;
+use crate::{api::submit, BacalhauPayload};
+
+use section::message::ValueView;
+use section::{
+    command_channel::{Command, SectionChannel},
+    futures::{self, FutureExt, Sink, SinkExt, Stream, StreamExt},
+    message::Chunk,
+    pretty_print::pretty_print,
+    section::Section,
+    SectionError, SectionMessage,
+};
+use std::collections::HashMap;
+use std::pin::{pin, Pin};
+
+use std::future::Future;
+
+type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+#[derive(Debug)]
+pub struct Bacalhau {
+    pub job: String,
+    pub jobstore: JobStore,
+}
+
+impl Bacalhau {
+    pub fn new(job: impl Into<String>, jobstore: impl Into<String>) -> Self {
+        Self {
+            job: job.into(),
+            jobstore: JobStore::new(jobstore).expect("should be able to create jobstore"),
+        }
+    }
+
+    pub async fn submit_job(&self, data: &HashMap<String, String>) -> Result<String, StdError> {
+        // We'll either get an Err from the call to render, or we'll get either
+        // an Ok or Err from the call to submit.
+        match self.jobstore.render(self.job.clone(), data) {
+            Ok(output) => submit(&output).await,
+            Err(m) => Err(m),
+        }
+    }
+
+    async fn enter_loop<Input, Output, SectionChan>(
+        self,
+        input: Input,
+        _output: Output,
+        mut section_channel: SectionChan,
+    ) -> Result<(), SectionError>
+    where
+        Input: Stream<Item = SectionMessage> + Send + 'static,
+        Output: Sink<SectionMessage, Error = SectionError> + Send + 'static,
+        SectionChan: SectionChannel + Send + 'static,
+    {
+        let mut input = pin!(input.fuse());
+        //let output = pin!(output);
+        loop {
+            futures::select_biased! {
+                cmd = section_channel.recv().fuse() => {
+                    if let Command::Stop = cmd? {
+                        return Ok(())
+                    }
+                }
+                stream = input.next() => {
+                    let mut msg = match stream {
+                        Some(msg) => msg,
+                        None => Err("input stream closed")?
+                    };
+
+
+                    loop {
+                        futures::select! {
+                            chunk = msg.next().fuse() => {
+                                let frame = match chunk? {
+                                    None => break,
+                                    Some(Chunk::DataFrame(df)) => df,
+                                    Some(_ch) => continue,
+                                };
+
+
+                                let columns = frame.columns();
+
+
+                                section_channel.log(format!("got dataframe chunk from {}:\n{}", msg.origin(), pretty_print(&*frame))).await?;
+
+                                let mut c = columns.into_iter().nth(0).unwrap();
+                                match c.nth(0).unwrap() {
+                                    ValueView::Str(s) => {
+                                        section_channel.log(format!("got string: {}", s)).await?;
+                                    }
+                                    _ => continue,
+                                }
+
+
+                                //ValueView::Str(child.as_string::<i32>().value(value_offset)),
+                                let args: HashMap<String, String> = HashMap::new();
+                                //args.insert("target".into(), "".into());
+                                self.submit_job(&args).await?;
+
+                                msg.ack().await;
+                            }
+                        }
+
+                    }
+
+                    // // let payload: crate::BacalhauPayload = msg.into();
+                    // // let origin = &msg.origin();
+                    // // self.submit_job(&payload.message).await?;
+
+                    // // section_channel.log(&format!("Message from '{:?}' received! {:?}", origin, payload)).await?;
+                    // output.send(msg).await?;
+                    //         }
+                    //         cmd = section_channel.recv().fuse() => {
+                    //             if let Command::Stop = cmd? {
+                    //                 return Ok(())
+                    //             }
+                    //         }
+
+                },
+            }
+        }
+    }
+}
+
+impl<Input, Output, SectionChan> Section<Input, Output, SectionChan> for Bacalhau
+where
+    Input: Stream<Item = SectionMessage> + Send + 'static,
+    Output: Sink<SectionMessage, Error = SectionError> + Send + 'static,
+    SectionChan: SectionChannel + Send + 'static,
+{
+    type Error = SectionError;
+    type Future = Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send + 'static>>;
+
+    fn start(self, input: Input, output: Output, command: SectionChan) -> Self::Future {
+        Box::pin(async move { self.enter_loop(input, output, command).await })
+    }
+}
diff --git a/pipe/section/section_impls/bacalhau/src/jobstore.rs b/pipe/section/section_impls/bacalhau/src/jobstore.rs
new file mode 100644
index 00000000..da7c0d37
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/jobstore.rs
@@ -0,0 +1,103 @@
+use crate::StdError;
+use std::{collections::HashMap, fs, path::PathBuf};
+
+use handlebars::Handlebars;
+use serde::Serialize;
+use serde_json::{json, Map};
+use serde_yaml::Value;
+
+pub struct JobStore {
+    templates: Handlebars<'static>,
+}
+
+impl std::fmt::Debug for JobStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("JobStore").finish()
+    }
+}
+
+impl JobStore {
+    pub fn new(jobstore: impl Into<String>) -> Result<Self, StdError> {
+        let path = PathBuf::from(jobstore.into());
+        let entries = path.read_dir()?;
+
+        let mut handlebars = Handlebars::new();
+        for e in entries {
+            let entry = e?;
+            let meta = entry.metadata()?;
+            if !meta.is_file() {
+                continue;
+            }
+
+            let entryname = entry.file_name();
+            let fname = entryname.to_str().unwrap();
+            if fname.ends_with("yaml") || fname.ends_with("yml") {
+                let p = entry.path();
+                let name = p.file_stem().unwrap();
+
+                handlebars.register_template_string(
+                    name.to_str().unwrap(),
+                    &*fs::read_to_string(entry.path())
+                        .expect("should have been able to read the file"),
+                )?;
+            }
+        }
+
+        Ok(Self {
+            templates: handlebars,
+        })
+    }
+
+    pub fn render(&self, name: String, data: &HashMap<String, String>) -> Result<String, StdError> {
+        let text = self.templates.render(&name, &json!(data))?;
+        let job: Value = serde_yaml::from_str(&text).unwrap();
+
+        let request = PutJobRequest {
+            job,
+            idempotency_token: None,
+            namespace: None,
+            headers: None,
+        };
+
+        Ok(serde_json::to_string(&request).unwrap())
+    }
+}
+
+#[derive(Serialize)]
+struct PutJobRequest {
+    #[serde(rename = "Job")]
+    job: Value,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "IdempotencyToken")]
+    idempotency_token: Option<String>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "Namespace")]
+    namespace: Option<String>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "Headers")]
+    headers: Option<Map<String, serde_json::Value>>,
+}
+
+// #[cfg(test)]
+// mod tests {
+//     use super::*;
+
+//     #[tokio::test]
+//     async fn test_jobstore() -> Result<(), StdError> {
+//         let path: PathBuf = [env!("CARGO_MANIFEST_DIR"), "testdata"].iter().collect();
+
+//         let mut args: HashMap<String, String> = HashMap::new();
+//         args.insert("image".into(), "ubuntu".into());
+
+//         let j = JobStore::new(path.to_str().unwrap())?;
+//         let output = j.render(String::from("process"), &args)?;
+
+//         let r = submit(&output).await;
+//         println!("{:?}", r);
+
+//         Ok(())
+//     }
+// }
diff --git a/pipe/section/section_impls/bacalhau/src/lib.rs b/pipe/section/section_impls/bacalhau/src/lib.rs
new file mode 100644
index 00000000..f8aa98a3
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/src/lib.rs
@@ -0,0 +1,69 @@
+use std::sync::Arc;
+
+use section::message::{Ack, Chunk, Column, DataFrame, DataType, Message, ValueView};
+pub mod api;
+pub mod destination;
+pub mod jobstore;
+
+type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct BacalhauPayload {
+    pub filepath: String,
+}
+
+impl BacalhauPayload {
+    fn new(filepath: String) -> Self {
+        Self { filepath }
+    }
+}
+
+impl DataFrame for BacalhauPayload {
+    fn columns(&self) -> Vec<section::message::Column<'_>> {
+        vec![Column::new(
+            "filepath",
+            DataType::Str,
+            Box::new(std::iter::once(ValueView::from(&self.filepath))),
+        )]
+    }
+}
+
+pub struct BacalhauMessage {
+    origin: Arc<str>,
+    payload: Option<Box<dyn DataFrame>>,
+    ack: Option<Ack>,
+}
+
+impl BacalhauMessage {
+    fn new(origin: Arc<str>, payload: impl DataFrame, ack: Option<Ack>) -> Self {
+        Self {
+            origin,
+            payload: Some(Box::new(payload)),
+            ack,
+        }
+    }
+}
+
+impl std::fmt::Debug for BacalhauMessage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("BacalhauMessage")
+            .field("origin", &self.origin)
+            .field("payload", &self.payload)
+            .finish()
+    }
+}
+
+impl Message for BacalhauMessage {
+    fn origin(&self) -> &str {
+        self.origin.as_ref()
+    }
+
+    fn next(&mut self) -> section::message::Next<'_> {
+        let v = self.payload.take().map(Chunk::DataFrame);
+        Box::pin(async move { Ok(v) })
+    }
+
+    fn ack(&mut self) -> Ack {
+        self.ack.take().unwrap_or(Box::pin(async {}))
+    }
+}
diff --git a/pipe/section/section_impls/bacalhau/testdata/process.yaml b/pipe/section/section_impls/bacalhau/testdata/process.yaml
new file mode 100644
index 00000000..de2ee443
--- /dev/null
+++ b/pipe/section/section_impls/bacalhau/testdata/process.yaml
@@ -0,0 +1,10 @@
+Name: Extractor
+Type: batch
+Namespace: default
+Count: 1
+Tasks:
+- Name: extract
+  Engine:
+    Type: process
+    Params:
+      Name: date