diff --git a/CHANGELOG.md b/CHANGELOG.md
index 978b32119fef..10ec480f76a8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,7 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - JSON API: `getinfo` now returns `num_peers` `num_pending_channels`,
   `num_active_channels` and `num_inactive_channels` fields.
 - JSON API: use `\n\n` to terminate responses, for simplified parsing (pylightning now relies on this)
-- Plugins: Added plugins to `lightningd` and implemented the option passthrough.
+- Plugins: Added plugins to `lightningd`, including option passthrough and JSON-RPC passthrough.
 
 ### Changed
 
diff --git a/common/utils.c b/common/utils.c
index f6d28c8b783d..216544b4740c 100644
--- a/common/utils.c
+++ b/common/utils.c
@@ -49,3 +49,15 @@ void clean_tmpctx(void)
 	while ((p = tal_first(tmpctx)) != NULL)
 		tal_free(p);
 }
+
+void tal_arr_remove_(void *p, size_t elemsize, size_t n)
+{
+    // p is a pointer-to-pointer for tal_resize.
+    char *objp = *(char **)p;
+    size_t len = tal_bytelen(objp);
+    assert(len % elemsize == 0);
+    assert((n + 1) * elemsize <= len);
+    memmove(objp + elemsize * n, objp + elemsize * (n+1),
+                      len - (elemsize * (n+1)));
+    tal_resize((char **)p, len - elemsize);
+}
diff --git a/common/utils.h b/common/utils.h
index e43c82c225e5..5cff73610468 100644
--- a/common/utils.h
+++ b/common/utils.h
@@ -29,6 +29,15 @@ u8 *tal_hexdata(const tal_t *ctx, const void *str, size_t len);
 	(tal_resize((p), tal_count(*(p))+1), (*p) + tal_count(*(p))-1)
 #endif
 
+/**
+ * Remove an element from an array
+ *
+ * This will shift the elements past the removed element, changing
+ * their position in memory, so only use this for arrays of pointers.
+ */
+#define tal_arr_remove(p, n) tal_arr_remove_((p), sizeof(**p), (n))
+void tal_arr_remove_(void *p, size_t elemsize, size_t n);
+
 /* Use the POSIX C locale. */
 void setup_locale(void);
 
diff --git a/contrib/plugins/fail/failtimeout.py b/contrib/plugins/fail/failtimeout.py
new file mode 100755
index 000000000000..9477c23f9ccd
--- /dev/null
+++ b/contrib/plugins/fail/failtimeout.py
@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""An example plugin that fails to answer to `getmanifest`
+
+Used to test the `getmanifest` timeout.
+"""
+import json
+import sys
+import time
+
+
+def json_getmanifest(request):
+    # Timeout is 10 seconds, so wait 11
+    time.sleep(11)
+    return {
+        "options": [
+        ],
+        "rpcmethods": [
+        ]
+    }
+
+
+methods = {
+    'getmanifest': json_getmanifest,
+}
+
+
+partial = ""
+for l in sys.stdin:
+    try:
+        partial += l
+        request = json.loads(partial)
+    except Exception:
+        continue
+
+    result = None
+    method = methods[request['method']]
+    params = request['params']
+    try:
+        if isinstance(params, dict):
+            result = method(request, **params)
+        else:
+            result = method(request, *params)
+        result = {
+            "jsonrpc": "2.0",
+            "result": result,
+            "id": request['id']
+        }
+    except Exception as e:
+        result = {
+            "jsonrpc": "2.0",
+            "error": "Error while processing {}".format(request['method']),
+            "id": request['id']
+        }
+
+    json.dump(result, fp=sys.stdout)
+    sys.stdout.write('\n')
+    sys.stdout.flush()
+    partial = ""
diff --git a/lightningd/jsonrpc.c b/lightningd/jsonrpc.c
index 17b88550e5ef..f29aeec62831 100644
--- a/lightningd/jsonrpc.c
+++ b/lightningd/jsonrpc.c
@@ -347,11 +347,8 @@ static const struct json_command *find_cmd(const struct jsonrpc *rpc,
 {
 	struct json_command **commands = rpc->commands;
 
-	/* commands[i] can be NULL if the plugin that registered it
-	 * was killed, commands[i]->name can be NULL in test code. */
 	for (size_t i = 0; i < tal_count(commands); i++)
-		if (commands[i] && commands[i]->name &&
-		    json_tok_streq(buffer, tok, commands[i]->name))
+		if (json_tok_streq(buffer, tok, commands[i]->name))
 			return commands[i];
 	return NULL;
 }
@@ -731,8 +728,7 @@ bool jsonrpc_command_add(struct jsonrpc *rpc, struct json_command *command)
 
 	/* Check that we don't clobber a method */
 	for (size_t i = 0; i < count; i++)
-		if (rpc->commands[i] != NULL &&
-		    streq(rpc->commands[i]->name, command->name))
+		if (streq(rpc->commands[i]->name, command->name))
 			return false;
 
 	*tal_arr_expand(&rpc->commands) = command;
@@ -741,12 +737,12 @@ bool jsonrpc_command_add(struct jsonrpc *rpc, struct json_command *command)
 
 void jsonrpc_command_remove(struct jsonrpc *rpc, const char *method)
 {
-	// FIXME: Currently leaves NULL entries in the table, if we
-	// restart plugins we should shift them out.
 	for (size_t i=0; i<tal_count(rpc->commands); i++) {
 		struct json_command *cmd = rpc->commands[i];
-		if (cmd && streq(cmd->name, method)) {
-			rpc->commands[i] = tal_free(cmd);
+		if (streq(cmd->name, method)) {
+			tal_arr_remove(&rpc->commands, i);
+			tal_free(cmd);
+			break;
 		}
 	}
 }
diff --git a/lightningd/plugin.c b/lightningd/plugin.c
index 2336055e475c..121ddf42f294 100644
--- a/lightningd/plugin.c
+++ b/lightningd/plugin.c
@@ -9,6 +9,7 @@
 #include <ccan/tal/str/str.h>
 #include <ccan/utf8/utf8.h>
 #include <common/memleak.h>
+#include <common/timeout.h>
 #include <dirent.h>
 #include <errno.h>
 #include <lightningd/json.h>
@@ -19,6 +20,14 @@
 #include <sys/types.h>
 #include <unistd.h>
 
+/* How many seconds may the plugin take to reply to the `getmanifest
+ * call`? This is the maximum delay to `lightningd --help` and until
+ * we can start the main `io_loop` to communicate with peers. If this
+ * hangs we can't do much, so we put an upper bound on the time we're
+ * willing to wait. Plugins shouldn't do any initialization in the
+ * `getmanifest` call anyway, that's what `init `is for. */
+#define PLUGIN_MANIFEST_TIMEOUT 10
+
 struct plugin {
 	struct list_node list;
 
@@ -33,9 +42,10 @@ struct plugin {
 	char *buffer;
 	size_t used, len_read;
 
-	/* Stuff we write */
-	struct list_head output;
-	const char *outbuf;
+	/* Our json_streams. Since multiple streams could start
+	 * returning data at once, we always service these in order,
+	 * freeing once empty. */
+	struct json_stream **js_arr;
 
 	struct log *log;
 
@@ -43,6 +53,10 @@ struct plugin {
 	struct list_head plugin_opts;
 
 	const char **methods;
+
+	/* Timer to add a timeout to some plugin RPC calls. Used to
+	 * guarantee that `getmanifest` doesn't block indefinitely. */
+	const struct oneshot *timeout_timer;
 };
 
 struct plugin_request {
@@ -56,6 +70,7 @@ struct plugin_request {
 	const char *json_params;
 	const char *response;
 	const jsmntok_t *resulttok, *errortok, *toks;
+	struct json_stream *stream;
 
 	/* The response handler to be called on success or error */
 	void (*cb)(const struct plugin_request *, void *);
@@ -73,11 +88,8 @@ struct plugins {
 
 	/* RPC interface to bind JSON-RPC methods to */
 	struct jsonrpc *rpc;
-};
 
-struct json_output {
-	struct list_node list;
-	const char *json;
+	struct timers timers;
 };
 
 /* Represents a pending JSON-RPC request that was forwarded to a
@@ -112,6 +124,7 @@ struct plugins *plugins_new(const tal_t *ctx, struct log_book *log_book,
 	p->log_book = log_book;
 	p->log = new_log(p, log_book, "plugin-manager");
 	p->rpc = rpc;
+	timers_init(&p->timers, time_mono());
 	return p;
 }
 
@@ -122,7 +135,8 @@ void plugin_register(struct plugins *plugins, const char* path TAKES)
 	list_add_tail(&plugins->plugins, &p->list);
 	p->plugins = plugins;
 	p->cmd = tal_strdup(p, path);
-	p->outbuf = NULL;
+	p->js_arr = tal_arr(p, struct json_stream *, 0);
+	p->used = 0;
 
 	p->log = new_log(p, plugins->log_book, "plugin-%s",
 			 path_basename(tmpctx, p->cmd));
@@ -182,7 +196,62 @@ static void PRINTF_FMT(2,3) plugin_kill(struct plugin *plugin, char *fmt, ...)
 	io_wake(plugin);
 	kill(plugin->pid, SIGKILL);
 	list_del(&plugin->list);
-	tal_free(plugin);
+}
+
+/**
+ * Create the header of a JSON-RPC request and return open stream.
+ *
+ * This is a partial request, missing the params element, which the
+ * caller needs to add. We can't open it yet since we don't know
+ * whether it is supposed to be an object (name-value pairs) or an
+ * array.
+ */
+static struct plugin_request *
+plugin_request_new_(struct plugin *plugin, const char *method,
+		    void (*cb)(const struct plugin_request *, void *),
+		    void *arg)
+{
+	static u64 next_request_id = 0;
+	struct plugin_request *req = tal(plugin, struct plugin_request);
+	u64 request_id = next_request_id++;
+
+	req->id = request_id;
+	req->method = tal_strdup(req, method);
+	req->cb = cb;
+	req->arg = arg;
+	req->plugin = plugin;
+
+	/* We will not concurrently drain, if we do we must set the
+	 * writer to non-NULL */
+	req->stream = new_json_stream(req, NULL);
+
+	/* Add to map so we can find it later when routing the response */
+	uintmap_add(&plugin->plugins->pending_requests, req->id, req);
+
+	json_object_start(req->stream, NULL);
+	json_add_string(req->stream, "jsonrpc", "2.0");
+	json_add_string(req->stream, "method", method);
+	json_add_u64(req->stream, "id", request_id);
+	return req;
+}
+
+#define plugin_request_new(plugin, method, cb, arg)                            \
+	plugin_request_new_(                                                   \
+	    (plugin), (method),                                                \
+	    typesafe_cb_preargs(void, void *, (cb), (arg),                     \
+				const struct plugin_request *),                \
+	    (arg))
+
+/**
+ * Given a request, send it to the plugin.
+ */
+static void plugin_request_queue(struct plugin_request *req)
+{
+	/* Finish the `params` object and submit the request */
+	json_object_end(req->stream); /* root element */
+	json_stream_append(req->stream, "\n\n");
+	*tal_arr_expand(&req->plugin->js_arr) = req->stream;
+	io_wake(req->plugin);
 }
 
 /**
@@ -284,77 +353,55 @@ static struct io_plan *plugin_read_json(struct io_conn *conn UNUSED,
 			       &plugin->len_read, plugin_read_json, plugin);
 }
 
-static struct io_plan *plugin_write_json(struct io_conn *conn UNUSED,
+/* Mutual recursion */
+static struct io_plan *plugin_write_json(struct io_conn *conn,
+					 struct plugin *plugin);
+
+static struct io_plan *plugin_stream_complete(struct io_conn *conn, struct json_stream *js, struct plugin *plugin)
+{
+	assert(tal_count(plugin->js_arr) > 0);
+	/* Remove js and shift all remainig over */
+	tal_arr_remove(&plugin->js_arr, 0);
+
+	return plugin_write_json(conn, plugin);
+}
+
+static struct io_plan *plugin_write_json(struct io_conn *conn,
 					 struct plugin *plugin)
 {
-	struct json_output *out;
-	if (plugin->outbuf)
-		plugin->outbuf = tal_free(plugin->outbuf);
-
-	out = list_pop(&plugin->output, struct json_output, list);
-	if (!out) {
-		if (plugin->stop) {
-			return io_close(conn);
-		} else {
-			return io_out_wait(plugin->stdin_conn, plugin,
-					   plugin_write_json, plugin);
-		}
+	if (tal_count(plugin->js_arr)) {
+		return json_stream_output(plugin->js_arr[0], plugin->stdin_conn, plugin_stream_complete, plugin);
+	} else if (plugin->stop) {
+		return io_close(conn);
 	}
 
-	/* We have a message we'd like to send */
-	plugin->outbuf = tal_steal(plugin, out->json);
-	tal_free(out);
-	return io_write(conn, plugin->outbuf, strlen(plugin->outbuf),
-			plugin_write_json, plugin);
+	return io_out_wait(conn, plugin, plugin_write_json, plugin);
 }
 
-static void plugin_request_send_(
-    struct plugin *plugin, const char *method TAKES, const char *params TAKES,
-    void (*cb)(const struct plugin_request *, void *), void *arg)
+/**
+ * Finalizer for both stdin and stdout connections.
+ *
+ * Takes care of final cleanup, once the plugin is definitely dead.
+ */
+static void plugin_conn_finish(struct io_conn *conn, struct plugin *plugin)
 {
-	static u64 next_request_id = 0;
-	struct plugin_request *req = tal(plugin, struct plugin_request);
-	struct json_output *out = tal(plugin, struct json_output);
-	u64 request_id = next_request_id++;
+	if (conn == plugin->stdin_conn)
+		plugin->stdin_conn = NULL;
 
-	req->id = request_id;
-	req->method = tal_strdup(req, method);
-	req->json_params = tal_strdup(req, params);
-	req->cb = cb;
-	req->arg = arg;
-	req->plugin = plugin;
+	else if (conn == plugin->stdout_conn)
+		plugin->stdout_conn = NULL;
 
-	/* Add to map so we can find it later when routing the response */
-	uintmap_add(&plugin->plugins->pending_requests, req->id, req);
-
-	/* Wrap the request in the JSON-RPC request object. Terminate
-	 * with an empty line that serves as a hint that the JSON
-	 * object is done. */
-	out->json = tal_fmt(out, "{"
-			    "\"jsonrpc\": \"2.0\", "
-			    "\"method\": \"%s\", "
-			    "\"params\" : %s, "
-			    "\"id\" : %" PRIu64 " }\n\n",
-			    method, params, request_id);
-
-	/* Queue and notify the writer */
-	list_add_tail(&plugin->output, &out->list);
-	io_wake(plugin);
+	if (plugin->stdin_conn == NULL && plugin->stdout_conn == NULL)
+		tal_free(plugin);
 }
 
-#define plugin_request_send(plugin, method, params, cb, arg)                   \
-	plugin_request_send_(                                                  \
-	    (plugin), (method), (params),                                      \
-	    typesafe_cb_preargs(void, void *, (cb), (arg),                     \
-				const struct plugin_request *),                \
-	    (arg))
-
 static struct io_plan *plugin_stdin_conn_init(struct io_conn *conn,
 					      struct plugin *plugin)
 {
 	/* We write to their stdin */
 	/* We don't have anything queued yet, wait for notification */
 	plugin->stdin_conn = conn;
+	io_set_finish(conn, plugin_conn_finish, plugin);
 	return io_wait(plugin->stdin_conn, plugin, plugin_write_json, plugin);
 }
 
@@ -363,6 +410,7 @@ static struct io_plan *plugin_stdout_conn_init(struct io_conn *conn,
 {
 	/* We read from their stdout */
 	plugin->stdout_conn = conn;
+	io_set_finish(conn, plugin_conn_finish, plugin);
 	return io_read_partial(plugin->stdout_conn, plugin->buffer,
 			       tal_bytelen(plugin->buffer), &plugin->len_read,
 			       plugin_read_json, plugin);
@@ -489,6 +537,7 @@ static void plugin_rpcmethod_dispatch(struct command *cmd, const char *buffer,
 	struct plugin_rpc_request *request;
 	struct plugins *plugins = cmd->ld->plugins;
 	struct plugin *plugin;
+	struct plugin_request *req;
 
 	if (cmd->mode == CMD_USAGE) {
 		cmd->usage = "[params]";
@@ -533,7 +582,9 @@ static void plugin_rpcmethod_dispatch(struct command *cmd, const char *buffer,
 	assert(request->plugin);
 
 	tal_steal(request->plugin, request);
-	plugin_request_send(request->plugin, request->method, request->params, plugin_rpcmethod_cb, request);
+	req = plugin_request_new(request->plugin, request->method, plugin_rpcmethod_cb, request);
+	json_stream_append_fmt(req->stream, ", \"params\": %s", request->params);
+	plugin_request_queue(req);
 
 	command_still_pending(cmd);
 }
@@ -618,6 +669,12 @@ static bool plugin_rpcmethods_add(const struct plugin_request *req)
 	return true;
 }
 
+static void plugin_manifest_timeout(struct plugin *plugin)
+{
+	log_broken(plugin->log, "The plugin failed to respond to \"getmanifest\" in time, terminating.");
+	fatal("Can't recover from plugin failure, terminating.");
+}
+
 /**
  * Callback for the plugin_manifest request.
  */
@@ -637,6 +694,8 @@ static void plugin_manifest_cb(const struct plugin_request *req, struct plugin *
 
 	if (!plugin_opts_add(req) || !plugin_rpcmethods_add(req))
 		plugin_kill(plugin, "Failed to register options or methods");
+	/* Reset timer, it'd kill us otherwise. */
+	tal_free(plugin->timeout_timer);
 }
 
 /* If this is a valid plugin return full path name, otherwise NULL */
@@ -664,7 +723,11 @@ static const char *plugin_fullpath(const tal_t *ctx, const char *dir,
 	fullname = path_join(ctx, dir, basename);
 	if (stat(fullname, &st) != 0)
 		return tal_free(fullname);
-	if (!(st.st_mode & (S_IXUSR|S_IXGRP|S_IXOTH)))
+	if (!(st.st_mode & (S_IXUSR|S_IXGRP|S_IXOTH)) || st.st_mode & S_IFDIR)
+		return tal_free(fullname);
+
+	/* Ignore directories, they have exec mode, but aren't executable. */
+	if (st.st_mode & S_IFDIR)
 		return tal_free(fullname);
 	return fullname;
 }
@@ -706,6 +769,8 @@ void plugins_init(struct plugins *plugins)
 	struct plugin *p;
 	char **cmd;
 	int stdin, stdout;
+	struct timer *expired;
+	struct plugin_request *req;
 	plugins->pending_manifests = 0;
 	uintmap_init(&plugins->pending_requests);
 
@@ -719,22 +784,31 @@ void plugins_init(struct plugins *plugins)
 		if (p->pid == -1)
 			fatal("error starting plugin '%s': %s", p->cmd,
 			      strerror(errno));
-
-		list_head_init(&p->output);
 		p->buffer = tal_arr(p, char, 64);
-		p->used = 0;
 		p->stop = false;
 
 		/* Create two connections, one read-only on top of p->stdin, and one
 		 * write-only on p->stdout */
 		io_new_conn(p, stdout, plugin_stdout_conn_init, p);
 		io_new_conn(p, stdin, plugin_stdin_conn_init, p);
-		plugin_request_send(p, "getmanifest", "[]", plugin_manifest_cb, p);
+		req = plugin_request_new(p, "getmanifest", plugin_manifest_cb, p);
+		json_array_start(req->stream, "params");
+		json_array_end(req->stream);
+		plugin_request_queue(req);
 		plugins->pending_manifests++;
+		p->timeout_timer = new_reltimer(
+		    &plugins->timers, p, time_from_sec(PLUGIN_MANIFEST_TIMEOUT),
+		    plugin_manifest_timeout, p);
 		tal_free(cmd);
 	}
-	if (plugins->pending_manifests > 0)
-		io_loop(NULL, NULL);
+
+	while (plugins->pending_manifests > 0) {
+		void *v = io_loop(&plugins->timers, &expired);
+		if (v == plugins)
+			break;
+		if (expired)
+			timer_expired(plugins, expired);
+	}
 }
 
 static void plugin_config_cb(const struct plugin_request *req,
@@ -749,19 +823,25 @@ static void plugin_config_cb(const struct plugin_request *req,
 static void plugin_config(struct plugin *plugin)
 {
 	struct plugin_opt *opt;
-	bool first = true;
-	const char *name, *sep;
-	char *conf = tal_fmt(tmpctx, "{\n  \"options\": {");
+	const char *name;
+	struct plugin_request *req;
+
+	/* No writer since we don't flush concurrently. */
+	req = plugin_request_new(plugin, "init", plugin_config_cb, plugin);
+	json_object_start(req->stream, "params"); /* start of .params */
+
+	/* Add .params.options */
+	json_object_start(req->stream, "options");
 	list_for_each(&plugin->plugin_opts, opt, list) {
 		/* Trim the `--` that we added before */
 		name = opt->name + 2;
-		/* Separator between elements in the same object */
-		sep = first?"":",";
-		first = false;
-		tal_append_fmt(&conf, "%s\n    \"%s\": \"%s\"", sep, name, opt->value);
+		json_add_string(req->stream, name, opt->value);
 	}
-	tal_append_fmt(&conf, "\n  }\n}");
-	plugin_request_send(plugin, "init", conf, plugin_config_cb, plugin);
+	json_object_end(req->stream); /* end of .params.options */
+
+	json_object_end(req->stream); /* end of .params */
+
+	plugin_request_queue(req);
 }
 
 void plugins_config(struct plugins *plugins)
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
index 3aa1ccb41d3a..8b85269ad358 100644
--- a/tests/test_plugin.py
+++ b/tests/test_plugin.py
@@ -77,3 +77,18 @@ def test_plugin_disable(node_factory):
                                                     'helloworld.py')]))
     with pytest.raises(RpcError):
         n.rpc.hello(name='Sun')
+
+
+def test_failing_plugins():
+    fail_plugins = [
+        'contrib/plugins/fail/failtimeout.py',
+        'contrib/plugins/fail/doesnotexist.py',
+    ]
+
+    for p in fail_plugins:
+        with pytest.raises(subprocess.CalledProcessError):
+            subprocess.check_output([
+                'lightningd/lightningd',
+                '--plugin={}'.format(p),
+                '--help',
+            ])