beam subsystem hack for karuta execution under internal nodeHEADmaster

13 files changed, 710 insertions, 15 deletions

diff --git a/docs/roadmap.org b/docs/roadmap.org
index c9d14cc..f36bdbd 100644
--- a/docs/roadmap.org
+++ b/docs/roadmap.org
@@ -131,7 +131,10 @@ These do not block the steps above.
 - Programs that change state have no result protocol.
 - Once [[file:../lib/kernel/Merkle.ml][Merkle]] chunking depends only on the tuples a tree holds,
   maintenance can be checked by comparing addresses instead of tuples.
-- =Path.update= keeps the target it looks up and never releases it, so
-  each call leaks one reference to a counted namespace. This came with
-  the lifecycle rework and needs a decision.
+- Storing a relation reads it whole. [[file:../lib/kernel/Multigroup.ml][Multigroup]] admits a relation by
+  draining it with =Relation.read= before the write begins, and
+  [[file:../lib/kernel/SubstantialRelation.ml][SubstantialRelation]] builds its tree from that list. Streaming the
+  cursor into the tree instead would read during the write, holding
+  the writer lock while the relation is computed, and =with_batch=
+  would still keep every node it makes until the batch ends.
 - The signatures in [[file:object-system.org][the object system]] predate the code.
diff --git a/flake.nix b/flake.nix
index df66d9a..34cdc6f 100644
--- a/flake.nix
+++ b/flake.nix
@@ -55,7 +55,10 @@
         });
 
       devShells = forAllSystems (pkgs:
-        let ocamlPackages = pkgs.ocamlPackages; in
+        let
+          ocamlPackages = pkgs.ocamlPackages;
+          beamPackages = pkgs.beam27Packages;
+        in
         {
           default = pkgs.mkShell {
             inputsFrom = [ self.packages.${pkgs.system}.default ];
@@ -65,7 +68,7 @@
               utop
               qcheck-core
               qcheck-alcotest
-            ]);
+            ]) ++ [ beamPackages.erlang pkgs.erlang-language-platform ];
           };
         });
     };
diff --git a/lib/kernel/Initialization.ml b/lib/kernel/Initialization.ml
index 54bb76d..86c6882 100644
--- a/lib/kernel/Initialization.ml
+++ b/lib/kernel/Initialization.ml
@@ -20,12 +20,11 @@ let namespace entries =
    evaluators, and whatever else it registers beside them. *)
 let mount root name =
   match Path.lookup root Path.(name @/ this) with
-  | Ok mounted -> Ok mounted
+  | Ok mounted -> Ok (Protocols.Handle.release mounted)
   | Error _ ->
       let open Utilities.Result in
       let* mounted = namespace ["evaluator", Namespace.make ()] in
-      let* () = bind root (name, mounted) in
-      Ok mounted
+      bind root (name, mounted)
 
 module Make (S : Abstract.Storage.STORAGE) = struct
   module BM = BranchManager.Make (S)
diff --git a/lib/kernel/Initialization.mli b/lib/kernel/Initialization.mli
index a6e6ae0..02db00f 100644
--- a/lib/kernel/Initialization.mli
+++ b/lib/kernel/Initialization.mli
@@ -1,7 +1,6 @@
 (** a subsystem's root, \[name], holding \[name]\evaluator for the
-    evaluators it registers. Mounting it again gives the one there. *)
-val mount :
-  Protocols.Handle.t -> string -> (Protocols.Handle.t, Concepts.Condition.condition) result
+    evaluators it registers. Mounting it again keeps the one there. *)
+val mount : Protocols.Handle.t -> string -> (unit, Concepts.Condition.condition) result
 
 module Make (S : Abstract.Storage.STORAGE) : sig
   (** the root of the object tree, holding \system: its branches, of
diff --git a/lib/kernel/Path.ml b/lib/kernel/Path.ml
index 512cfb0..2a9b053 100644
--- a/lib/kernel/Path.ml
+++ b/lib/kernel/Path.ml
@@ -22,6 +22,10 @@ module Error = struct
     condition "overlapping-change"
       "A batch of changes both replaced an object and changed inside it"
       ("path" |=| Concepts.Value.String (to_string path))
+
+  let unheld path =
+    condition "path-unheld" "The object at the path could not be held"
+      ("path" |=| Concepts.Value.String (to_string path))
 end
 
 let walking ?missing f g handle path =
@@ -42,7 +46,11 @@ let walking ?missing f g handle path =
         | Some elem -> f x handle (fun () -> walk elem xs) )
   in
   Handle.with_autorelease (fun () ->
-      walk handle path
+      begin match path with
+      | [] -> (
+        match Handle.copy handle with Some copy -> g copy | None -> Error (Error.unheld path) )
+      | _ -> walk handle path
+      end
       |> Result.map_error
            Concepts.Condition.(complement ("path" |=| Concepts.Value.String (to_string path))) )
 
@@ -55,8 +63,9 @@ let update handle path key reference value =
   let open Utilities.Result in
   let open Protocols in
   let* target = lookup handle path in
-  let* registry = Handle.require Registry.from target in
-  Registry.update registry key reference value
+  Handle.releasing target (fun () ->
+      let* registry = Handle.require Registry.from target in
+      Registry.update registry key reference value )
 
 (* The changes at this object, and those beneath each child they reach. *)
 let split changes =
diff --git a/subsystem/dune b/subsystem/dune
new file mode 100644
index 0000000..0148f5e
--- /dev/null
+++ b/subsystem/dune
@@ -0,0 +1,6 @@
+(include_subdirs qualified)
+
+(library
+ (name subsystem)
+ (public_name rnt.subsystem)
+ (libraries rnt batteries unix))
diff --git a/subsystem/karuta/Beam.ml b/subsystem/karuta/Beam.ml
new file mode 100644
index 0000000..82ae827
--- /dev/null
+++ b/subsystem/karuta/Beam.ml
@@ -0,0 +1,414 @@
+open Rnt
+
+module Error = struct
+  open Concepts.Condition
+
+  let closed () = condition "beam-closed" "The BEAM node is gone" empty
+  let taken () = condition "beam-taken" "An evaluator is already registered as karuta:beam" empty
+
+  let malformed () =
+    condition "malformed-beam-program"
+      "A beam program is not {Module, Function, Beam, #{Name => Domain}} of binaries" empty
+
+  let not_a_program () =
+    condition "not-a-beam-program" "The evaluator was given a program of another kind" empty
+
+  let inputs () = condition "beam-program-inputs" "A beam program takes no inputs" empty
+
+  let failed message =
+    condition "beam-failed" "The BEAM node failed to answer a call"
+      ("message" |=| Concepts.Value.String message)
+
+  let unexpected () =
+    condition "unexpected-beam-message" "The BEAM node said something out of turn" empty
+
+  let unnamed () = condition "unnamed-beam-program" "A program was defined at an empty path" empty
+
+  let program_taken path =
+    condition "beam-program-taken" "A program is already defined at the path"
+      ("path" |=| Concepts.Value.String (Kernel.Path.to_string path))
+
+  let unknown_cursor id =
+    condition "unknown-beam-cursor" "The BEAM node named a cursor it does not hold"
+      ("cursor" |=| Concepts.Value.Integer id)
+end
+
+module type TRANSPORT = sig
+  type t
+
+  val send : t -> string -> (unit, Concepts.Condition.condition) result
+  val receive : t -> (string, Concepts.Condition.condition) result
+  val stop : t -> unit
+end
+
+module Port = struct
+  type t = {input: in_channel; output: out_channel}
+
+  let send {output; _} message =
+    let header = Bytes.create 4 in
+    Bytes.set_int32_be header 0 (Int32.of_int (String.length message));
+    try
+      output_bytes output header;
+      output_string output message;
+      Ok (flush output)
+    with Sys_error _ -> Error (Error.closed ())
+
+  let receive {input; _} =
+    try
+      let header = really_input_string input 4 in
+      Ok (really_input_string input (Int32.to_int (String.get_int32_be header 0) land 0xffff_ffff))
+    with End_of_file | Sys_error _ -> Error (Error.closed ())
+
+  let start ?(erl = "erl") directories =
+    let boot =
+      "try [{module, _} = code:load_abs(filename:rootname(F)) || D <- init:get_plain_arguments(), \
+       F <- filelib:wildcard(D ++ \"/*.beam\")], rnt:start() catch C:R:S -> \
+       io:format(standard_error, \"~p:~p~n~p~n\", [C, R, S]), halt(1) end."
+    in
+    match
+      Unix.open_process_args erl
+        (Array.of_list ([erl; "-noshell"; "-noinput"; "-eval"; boot; "-extra"] @ directories))
+    with
+    | input, output -> Ok {input; output}
+    | exception Unix.Unix_error _ -> Error (Error.closed ())
+
+  let stop {input; output} = ignore (Unix.close_process (input, output))
+end
+
+let name = "karuta"
+
+type program =
+  { module_: string;
+    function_: string;
+    beam: string;
+    heading: Protocols.Schematics.relation_description }
+
+type Kernel.Program.kind += Beam of program
+
+let decode source =
+  let domain _ = function
+    | Concepts.Value.String domain -> Some {Protocols.Schematics.domain; provenance= []}
+    | Concepts.Value.Integer _ -> None
+  in
+  match Translator.read 4 source with
+  | Some ([module_; function_; beam], attributes)
+    when String.length module_ <= 0xffff && String.length function_ <= 0xffff ->
+      let heading = BatMap.String.filter_map domain attributes in
+      if BatMap.String.cardinal heading = BatMap.String.cardinal attributes then
+        Ok (Beam {module_; function_; beam; heading})
+      else Error (Error.malformed ())
+  | _ -> Error (Error.malformed ())
+
+(* Both ends speak over one pipe, and each serves the other's requests
+   while it waits for its own answer, so a BEAM program RNT is reading
+   can call back into RNT, and so on, as deep as the calls nest.
+
+   RNT asks the node:
+   - h                                 run every loaded rnt_init/0  -> o | e
+   - i Id:64 M:16 Module F:16 Function B:32 Beam
+                                       start Module:Function() from Beam
+   - f Id:64 Limit:32                  the next batch of it         -> b | e
+   - c Id:64                           forget it
+   The node asks RNT:
+   - D Path Evaluator:16 Source        define a program at Path     -> o | e
+   - X Path                            invoke the program at Path   -> r Id:64 | e
+   - E Evaluator:16 Source             invoke a program of no path  -> r Id:64 | e
+   - N Id:64 Limit:32                  the next batch of its result -> b | e
+   - C Id:64                           release that result
+   A Path is Count:16 then each segment as Length:16 Segment, and a
+   batch b is Exhausted:8 then each tuple as Size:32 term_to_binary. *)
+module Make (T : TRANSPORT) = struct
+  (* [root] is borrowed, not held: the evaluator lives under it, so
+     holding it would keep both alive forever. [cursors] are the
+     results the node invoked and has not finished reading. *)
+  type node =
+    { transport: T.t;
+      root: Protocols.Handle.t;
+      cursors: (int, Protocols.Handle.t * Protocols.Handle.t) Hashtbl.t;
+      mutable next: int }
+
+  let node transport root = {transport; root; cursors= Hashtbl.create 8; next= 0}
+
+  module Request = struct
+    type t =
+      | Define of string list * string * string
+      | Invoke of string list
+      | Evaluate of string * string
+      | Next of int * int
+      | Close of int
+
+    exception Short
+
+    let of_message message =
+      let length = String.length message in
+      let at = ref 1 in
+      let take n =
+        if !at + n > length then raise Short;
+        let p = !at in
+        at := p + n;
+        p
+      in
+      let u16 () = String.get_uint16_be message (take 2) in
+      let u32 () = Int32.to_int (String.get_int32_be message (take 4)) land 0xffff_ffff in
+      let u64 () = Int64.to_int (String.get_int64_be message (take 8)) in
+      let sized () =
+        let n = u16 () in
+        String.sub message (take n) n
+      in
+      let path () = List.init (u16 ()) (fun _ -> sized ()) in
+      let finished request = if !at = length then Some request else None in
+      match message.[0] with
+      | 'D' ->
+          let path = path () in
+          let evaluator = sized () in
+          Some (Define (path, evaluator, String.sub message !at (length - !at)))
+      | 'E' ->
+          let evaluator = sized () in
+          Some (Evaluate (evaluator, String.sub message !at (length - !at)))
+      | 'X' -> finished (Invoke (path ()))
+      | 'N' ->
+          let id = u64 () in
+          finished (Next (id, u32 ()))
+      | 'C' -> finished (Close (u64 ()))
+      | _ -> None
+      | exception Short -> None
+  end
+
+  let batch_message {Protocols.Cursor.tuples; exhausted} =
+    let buffer = Buffer.create 64 in
+    Buffer.add_char buffer 'b';
+    Buffer.add_char buffer (if exhausted then '\001' else '\000');
+    BatFingerTree.iter
+      (fun tuple ->
+        let term = Translator.encode tuple in
+        Buffer.add_int32_be buffer (Int32.of_int (String.length term));
+        Buffer.add_string buffer term )
+      tuples;
+    Buffer.contents buffer
+
+  let cursor_message id =
+    let buffer = Buffer.create 9 in
+    Buffer.add_char buffer 'r';
+    Buffer.add_int64_be buffer (Int64.of_int id);
+    Buffer.contents buffer
+
+  let forget node id =
+    Hashtbl.find_opt node.cursors id
+    |> Option.iter (fun (relation, cursor) ->
+        Hashtbl.remove node.cursors id;
+        Protocols.Handle.release cursor;
+        Protocols.Handle.release relation )
+
+  let define node path evaluator source =
+    let open Utilities.Result in
+    match List.rev path with
+    | [] -> Error (Error.unnamed ())
+    | key :: directory -> (
+        let* program = Kernel.Program.load node.root ~evaluator ~source in
+        match
+          Kernel.Path.update node.root
+            (Kernel.Path.of_list (List.rev directory))
+            key None (Some program)
+        with
+        | Ok true -> Ok "o"
+        | Ok false ->
+            Protocols.Handle.release program;
+            Error (Error.program_taken (Kernel.Path.of_list path))
+        | Error c -> Protocols.Handle.release program; Error c )
+
+  (* Runs [program], which it releases, keeping a cursor on the result
+     for the node to read. *)
+  let invoke node program =
+    let open Utilities.Result in
+    let* relation =
+      Protocols.Handle.releasing program (fun () ->
+          let* executable = Protocols.Executable.require program in
+          Protocols.Executable.invoke executable [] ~within:node.root )
+    in
+    match Protocols.Enumerable.cursor relation with
+    | Error c -> Protocols.Handle.release relation; Error c
+    | Ok cursor ->
+        let id = node.next in
+        node.next <- id + 1;
+        Hashtbl.replace node.cursors id (relation, cursor);
+        Ok (cursor_message id)
+
+  let fetch node id limit =
+    let open Utilities.Result in
+    let* _, cursor =
+      Hashtbl.find_opt node.cursors id |> Option.to_result ~none:(Error.unknown_cursor id)
+    in
+    let* scan = Protocols.Cursor.require cursor in
+    let* batch = Protocols.Cursor.fetch scan limit in
+    if batch.exhausted then forget node id;
+    Ok (batch_message batch)
+
+  let serve node request =
+    let answer = function
+      | Ok message -> T.send node.transport message
+      | Error c -> T.send node.transport ("e" ^ Concepts.Condition.to_string_hum c)
+    in
+    match request with
+    | Request.Define (path, evaluator, source) -> answer (define node path evaluator source)
+    | Request.Invoke path ->
+        answer (Result.bind (Kernel.Path.lookup node.root (Kernel.Path.of_list path)) (invoke node))
+    | Request.Evaluate (evaluator, source) ->
+        answer (Result.bind (Kernel.Program.load node.root ~evaluator ~source) (invoke node))
+    | Request.Next (id, limit) -> answer (fetch node id limit)
+    | Request.Close id -> Ok (forget node id)
+
+  let batch_of message =
+    let open Utilities.Result in
+    let length = String.length message in
+    let rec tuples at read =
+      if at = length then Ok (List.rev read)
+      else if at + 4 > length then Error (Error.unexpected ())
+      else
+        let size = Int32.to_int (String.get_int32_be message at) land 0xffff_ffff in
+        if at + 4 + size > length then Error (Error.unexpected ())
+        else
+          let* tuple = Translator.decode (String.sub message (at + 4) size) in
+          tuples (at + 4 + size) (tuple :: read)
+    in
+    if length < 2 then Error (Error.unexpected ())
+    else
+      let* read = tuples 2 [] in
+      Ok {Protocols.Cursor.tuples= BatFingerTree.of_list read; exhausted= message.[1] <> '\000'}
+
+  (* The node's answer to what RNT last asked, serving whatever the
+     node asks in between. *)
+  let rec await node =
+    let open Utilities.Result in
+    let* message = T.receive node.transport in
+    match if message = "" then ' ' else message.[0] with
+    | 'b' -> Result.map (fun batch -> `Batch batch) (batch_of message)
+    | 'o' when message = "o" -> Ok `Done
+    | 'e' -> Error (Error.failed (String.sub message 1 (String.length message - 1)))
+    | _ -> (
+      match Request.of_message message with
+      | None -> Error (Error.unexpected ())
+      | Some request ->
+          let* () = serve node request in
+          await node )
+
+  (* Lets the node's rnt_init/0 functions define their programs. *)
+  let handshake node =
+    let open Utilities.Result in
+    let* () = T.send node.transport "h" in
+    let* reply = await node in
+    match reply with `Done -> Ok () | `Batch _ -> Error (Error.unexpected ())
+
+  module Poll = struct
+    let invoke_message id {module_; function_; beam; _} =
+      let buffer = Buffer.create (64 + String.length beam) in
+      Buffer.add_char buffer 'i';
+      Buffer.add_int64_be buffer (Int64.of_int id);
+      List.iter
+        (fun name ->
+          Buffer.add_uint16_be buffer (String.length name);
+          Buffer.add_string buffer name )
+        [module_; function_];
+      Buffer.add_int32_be buffer (Int32.of_int (String.length beam));
+      Buffer.add_string buffer beam;
+      Buffer.contents buffer
+
+    let fetch_message id limit =
+      let buffer = Buffer.create 13 in
+      Buffer.add_char buffer 'f';
+      Buffer.add_int64_be buffer (Int64.of_int id);
+      Buffer.add_int32_be buffer (Int32.of_int (max 0 (min limit 0xffff_ffff)));
+      Buffer.contents buffer
+
+    let close_message id =
+      let buffer = Buffer.create 9 in
+      Buffer.add_char buffer 'c';
+      Buffer.add_int64_be buffer (Int64.of_int id);
+      Buffer.contents buffer
+
+    class t node =
+      object (self)
+        inherit Kernel.Lifecycle.counted
+        inherit Kernel.Identity.of_id
+        method to_string = name
+
+        method fetch limit =
+          let open Utilities.Result in
+          let* () = T.send node.transport (fetch_message (Oo.id self) limit) in
+          let* reply = await node in
+          match reply with `Batch batch -> Ok batch | `Done -> Error (Error.unexpected ())
+
+        method destroy = ignore (T.send node.transport (close_message (Oo.id self)))
+        method protocols : Protocols.Handle.protocol list = Protocols.[Cursor.make self]
+      end
+
+    let start node program =
+      let poll = new t node in
+      T.send node.transport (invoke_message (Oo.id poll) program)
+      |> Result.map (fun () -> Protocols.Handle.make poll)
+  end
+
+  class evaluator node =
+    object (self)
+      inherit Kernel.Lifecycle.counted
+      inherit Kernel.Identity.of_id
+      method to_string = name
+
+      method destroy =
+        Hashtbl.to_seq_keys node.cursors |> List.of_seq |> List.iter (forget node);
+        T.stop node.transport
+
+      method parse source = decode source
+
+      method eval :
+          Kernel.Program.kind ->
+          Protocols.Handle.t list ->
+          within:Protocols.Handle.t ->
+          (Protocols.Handle.t, Concepts.Condition.condition) result =
+        fun kind inputs ~within:_ ->
+          match kind, inputs with
+          | Beam program, [] ->
+              Ok
+                (Kernel.EphemeralRelation.instantiate program.heading
+                   ~modes:(fun () -> Ok [Concepts.Mode.mode [] Concepts.Cardinality.Countable])
+                   (Kernel.EphemeralRelation.temporary program.heading)
+                   (fun () -> Poll.start node program) )
+          | Beam _, _ -> Error (Error.inputs ())
+          | _ -> Error (Error.not_a_program ())
+
+      method protocols : Protocols.Handle.protocol list = [Protocols.Evaluator.make self]
+    end
+end
+
+module Node = Make (Port)
+
+(* \karuta\program, where the node can define programs of its own. *)
+let programs root =
+  let open Utilities.Result in
+  let* found = Kernel.Path.find root Kernel.Path.(name @/ "program" @/ this) in
+  match found with
+  | Some programs -> Ok (Protocols.Handle.release programs)
+  | None -> (
+      let programs = Kernel.Namespace.make () in
+      match Kernel.Path.update root Kernel.Path.(name @/ this) "program" None (Some programs) with
+      | Ok true -> Ok ()
+      | Ok false -> Ok (Protocols.Handle.release programs)
+      | Error c -> Protocols.Handle.release programs; Error c )
+
+let register root directories =
+  let open Utilities.Result in
+  let* port = Port.start directories in
+  let node = Node.node port root in
+  let evaluator = Protocols.Handle.make (new Node.evaluator node) in
+  let bound =
+    let* () = Kernel.Initialization.mount root name in
+    let* () = programs root in
+    Kernel.Path.update root Kernel.Path.(name @/ "evaluator" @/ this) "beam" None (Some evaluator)
+  in
+  match bound with
+  | Ok true -> Node.handshake node
+  | Ok false ->
+      Protocols.Handle.release evaluator;
+      Error (Error.taken ())
+  | Error c ->
+      Protocols.Handle.release evaluator;
+      Error c
diff --git a/subsystem/karuta/Translator.ml b/subsystem/karuta/Translator.ml
new file mode 100644
index 0000000..5dec2e7
--- /dev/null
+++ b/subsystem/karuta/Translator.ml
@@ -0,0 +1,107 @@
+open Rnt
+
+module Error = struct
+  open Concepts.Condition
+
+  let malformed () =
+    condition "malformed-beam-tuple"
+      "A term is not {Type, #{Name => Value}} of binaries, and integers that fit an int" empty
+end
+
+let binary buffer s =
+  Buffer.add_uint8 buffer 109;
+  Buffer.add_int32_be buffer (Int32.of_int (String.length s));
+  Buffer.add_string buffer s
+
+let integer buffer n =
+  let rec digits m = if m = 0 then [] else (m land 0xff) :: digits (m lsr 8) in
+  let digits = digits (abs n) in
+  Buffer.add_uint8 buffer 110;
+  Buffer.add_uint8 buffer (List.length digits);
+  Buffer.add_uint8 buffer (if n < 0 then 1 else 0);
+  List.iter (Buffer.add_uint8 buffer) digits
+
+let encode {Concepts.Tuple.type_; attributes} =
+  let buffer = Buffer.create 64 in
+  Buffer.add_uint8 buffer 131;
+  Buffer.add_uint8 buffer 104;
+  Buffer.add_uint8 buffer 2;
+  binary buffer type_;
+  Buffer.add_uint8 buffer 116;
+  Buffer.add_int32_be buffer (Int32.of_int (BatMap.String.cardinal attributes));
+  BatMap.String.iter
+    (fun name value ->
+      binary buffer name;
+      match value with
+      | Concepts.Value.String s -> binary buffer s
+      | Concepts.Value.Integer n -> integer buffer n )
+    attributes;
+  Buffer.contents buffer
+
+exception Malformed
+
+let read arity bytes =
+  let position = ref 0 in
+  let take n =
+    if !position + n > String.length bytes then raise Malformed;
+    let at = !position in
+    position := at + n;
+    at
+  in
+  let u8 () = String.get_uint8 bytes (take 1) in
+  let i32 () = Int32.to_int (String.get_int32_be bytes (take 4)) in
+  let u32 () = i32 () land 0xffff_ffff in
+  let expect byte = if u8 () <> byte then raise Malformed in
+  let binary () =
+    let n = u32 () in
+    String.sub bytes (take n) n
+  in
+  let big n =
+    let negative = u8 () = 1 in
+    if n > 8 then raise Malformed;
+    let at = take n in
+    let rec magnitude i m =
+      if i < 0 then m
+      else
+        magnitude (i - 1)
+          (Int64.logor (Int64.shift_left m 8) (Int64.of_int (String.get_uint8 bytes (at + i))))
+    in
+    let m = magnitude (n - 1) 0L in
+    let limit = if negative then Int64.neg (Int64.of_int min_int) else Int64.of_int max_int in
+    if Int64.compare m 0L < 0 || Int64.compare m limit > 0 then raise Malformed;
+    Int64.to_int (if negative then Int64.neg m else m)
+  in
+  let value () =
+    match u8 () with
+    | 109 -> Concepts.Value.String (binary ())
+    | 97 -> Concepts.Value.Integer (u8 ())
+    | 98 -> Concepts.Value.Integer (i32 ())
+    | 110 -> Concepts.Value.Integer (big (u8 ()))
+    | _ -> raise Malformed
+  in
+  let rec attributes n map =
+    if n = 0 then map
+    else begin
+      expect 109;
+      let name = binary () in
+      let value = value () in
+      attributes (n - 1) (BatMap.String.add name value map)
+    end
+  in
+  match
+    expect 131;
+    expect 104;
+    expect arity;
+    let names = List.init (arity - 1) (fun _ -> expect 109; binary ()) in
+    expect 116;
+    let attributes = attributes (u32 ()) BatMap.String.empty in
+    if !position <> String.length bytes then raise Malformed;
+    names, attributes
+  with
+  | read -> Some read
+  | exception Malformed -> None
+
+let decode bytes =
+  match read 2 bytes with
+  | Some ([type_], attributes) -> Ok {Concepts.Tuple.type_; attributes}
+  | _ -> Error (Error.malformed ())
diff --git a/test/dune b/test/dune
index d58e758..bb946a8 100644
--- a/test/dune
+++ b/test/dune
@@ -2,4 +2,4 @@
 
 (test
  (name test_rnt)
- (libraries rnt alcotest qcheck-core qcheck-alcotest batteries))
+ (libraries rnt subsystem alcotest qcheck-core qcheck-alcotest batteries))
diff --git a/test/test_rnt.ml b/test/test_rnt.ml
index c6fcfd5..299c6ca 100644
--- a/test/test_rnt.ml
+++ b/test/test_rnt.ml
@@ -7,6 +7,7 @@ let all_suites =
   @ Unit.Kernel.Session.suites ()
   @ Unit.Kernel.Branch.suites ()
   @ Unit.Evaluators.FOL.suites ()
+  @ Unit.Karuta.Translator.suites ()
   @ Integration.Registration.suites ()
 
 let () = Alcotest.run "RNT" all_suites
diff --git a/test/unit/karuta/Translator.ml b/test/unit/karuta/Translator.ml
new file mode 100644
index 0000000..5e1beef
--- /dev/null
+++ b/test/unit/karuta/Translator.ml
@@ -0,0 +1,57 @@
+open Rnt
+module Translator = Subsystem.Karuta.Translator
+module Gen = QCheck2.Gen
+
+let value =
+  Gen.oneof
+    [ Gen.map (fun s -> Concepts.Value.String s) Gen.string_small;
+      Gen.map
+        (fun n -> Concepts.Value.Integer n)
+        (Gen.oneof [Gen.int; Gen.oneof_list [min_int; max_int]]) ]
+
+let tuple =
+  Gen.map2
+    (fun type_ attributes -> {Concepts.Tuple.type_; attributes= BatMap.String.of_list attributes})
+    Gen.string_small
+    (Gen.list_small (Gen.pair Gen.string_small value))
+
+let print tuple = tuple.Concepts.Tuple.type_ ^ " " ^ Concepts.Tuple.to_string tuple
+
+let same (a : Concepts.Tuple.t) (b : Concepts.Tuple.t) =
+  a.type_ = b.type_ && BatMap.String.equal Concepts.Value.equal a.attributes b.attributes
+
+let decodes_to_itself =
+  QCheck2.Test.make ~name:"a tuple decodes to itself" ~print tuple (fun tuple ->
+      match Translator.decode (Translator.encode tuple) with
+      | Ok decoded -> same tuple decoded
+      | Error _ -> false )
+
+let a_cut_term_is_refused =
+  QCheck2.Test.make ~name:"a cut term is refused" ~print tuple (fun tuple ->
+      let bytes = Translator.encode tuple in
+      List.init (String.length bytes) (fun n -> Translator.decode (String.sub bytes 0 n))
+      |> List.for_all Result.is_error )
+
+let erlang_integers_decode =
+  QCheck2.Test.make ~name:"an integer as term_to_binary writes it decodes to itself"
+    ~print:string_of_int
+    (Gen.oneof [Gen.int_range 0 255; Gen.int_range (-0x8000_0000) 0x7fff_ffff])
+    (fun n ->
+      let term = Buffer.create 32 in
+      Buffer.add_string term
+        "\131\104\002\109\000\000\000\001t\116\000\000\000\001\109\000\000\000\001n";
+      if n >= 0 && n < 256 then begin
+        Buffer.add_uint8 term 97; Buffer.add_uint8 term n
+      end
+      else begin
+        Buffer.add_uint8 term 98;
+        Buffer.add_int32_be term (Int32.of_int n)
+      end;
+      match Translator.decode (Buffer.contents term) with
+      | Ok {attributes; _} -> BatMap.String.bindings attributes = ["n", Concepts.Value.Integer n]
+      | Error _ -> false )
+
+let suites () =
+  [ ( "karuta/translator",
+      List.map QCheck_alcotest.to_alcotest
+        [decodes_to_itself; a_cut_term_is_refused; erlang_integers_decode] ) ]
diff --git a/tools/driver/dune b/tools/driver/dune
new file mode 100644
index 0000000..429a4fc
--- /dev/null
+++ b/tools/driver/dune
@@ -0,0 +1,3 @@
+(executable
+ (name main)
+ (libraries rnt subsystem batteries unix))
diff --git a/tools/driver/main.ml b/tools/driver/main.ml
new file mode 100644
index 0000000..662032c
--- /dev/null
+++ b/tools/driver/main.ml
@@ -0,0 +1,94 @@
+(* Boots a BEAM node on a directory of .beam images (Karuta's runtime
+   plus compiled modules), lets their rnt_init/0 register programs, and
+   reads them back through the kernel.
+
+   dune exec ./tools/driver/main.exe -- <image directory> *)
+
+open Rnt
+module Beam = Subsystem.Karuta.Beam
+
+let ( let* ) = Result.bind
+let program = Kernel.Path.("karuta" @/ "program" @/ this)
+
+(* \system\evaluator\fol, so programs can name the "fol" evaluator. *)
+let mount_fol root =
+  let evaluators = Kernel.Namespace.make () and system = Kernel.Namespace.make () in
+  let* _ =
+    Kernel.Path.update evaluators Kernel.Path.this "fol" None (Some (Evaluators.FOL.make ()))
+  in
+  let* _ = Kernel.Path.update system Kernel.Path.this "evaluator" None (Some evaluators) in
+  let* _ = Kernel.Path.update root Kernel.Path.this "system" None (Some system) in
+  Ok ()
+
+(* {Module, Function, Beam, #{Attribute => <<"string">>}} as term_to_binary. *)
+let image directory module_ function_ heading =
+  let beam =
+    In_channel.with_open_bin (Filename.concat directory (module_ ^ ".beam")) In_channel.input_all
+  in
+  let buffer = Buffer.create 64 in
+  let binary s =
+    Buffer.add_uint8 buffer 109;
+    Buffer.add_int32_be buffer (Int32.of_int (String.length s));
+    Buffer.add_string buffer s
+  in
+  List.iter (Buffer.add_uint8 buffer) [131; 104; 4];
+  List.iter binary [module_; function_; beam];
+  Buffer.add_uint8 buffer 116;
+  Buffer.add_int32_be buffer (Int32.of_int (List.length heading));
+  List.iter (fun name -> binary name; binary "string") heading;
+  Buffer.contents buffer
+
+let print title relation =
+  Protocols.Handle.releasing relation (fun () ->
+      let* _, tuples = Protocols.Relation.read relation in
+      Printf.printf "%s\n" title;
+      List.iter
+        (fun {Concepts.Tuple.type_; attributes} ->
+          Printf.printf "  %s {%s}\n" type_
+            ( BatMap.String.bindings attributes
+            |> List.map (fun (k, v) -> k ^ " = " ^ Concepts.Value.to_string v)
+            |> String.concat ", " ) )
+        tuples;
+      Ok () )
+
+let run directory root =
+  let* () = mount_fol root in
+  (* Boots erl on the images; the handshake runs every rnt_init/0. *)
+  let* () = Beam.register root [directory] in
+  let* listing = Kernel.Path.lookup root program in
+  let* names =
+    Protocols.Handle.releasing listing (fun () ->
+        Result.bind
+          (Protocols.Handle.require Protocols.Directory.from listing)
+          Protocols.Directory.list )
+  in
+  Printf.printf "\\karuta\\program holds: %s\n" (String.concat ", " (BatFingerTree.to_list names));
+  (* 1. The FOL program librarian defined from inside the node. *)
+  let* authors = Kernel.Path.lookup root Kernel.Path.("karuta" @/ "program" @/ "authors" @/ this) in
+  let* relation =
+    Protocols.Handle.releasing authors (fun () ->
+        let* executable = Protocols.Executable.require authors in
+        Protocols.Executable.invoke executable [] ~within:root )
+  in
+  let* () = print "FOL \\karuta\\program\\authors (project author over main):" relation in
+  (* 2. A beam program that itself issues a FOL plan back to RNT. *)
+  let* titles =
+    Kernel.Program.load root ~evaluator:"karuta:beam"
+      ~source:(image directory "librarian" "titles" ["title"])
+  in
+  let* relation =
+    Protocols.Handle.releasing titles (fun () ->
+        let* executable = Protocols.Executable.require titles in
+        Protocols.Executable.invoke executable [] ~within:root )
+  in
+  print "librarian:titles (BEAM -> RNT fol -> BEAM main):" relation
+
+let () =
+  Sys.set_signal Sys.sigpipe Sys.Signal_ignore;
+  let root = Kernel.Namespace.make () in
+  let outcome = Protocols.Handle.releasing root (fun () -> run Sys.argv.(1) root) in
+  match outcome with
+  | Ok () -> ()
+  | Error c ->
+      prerr_endline (Concepts.Condition.to_string_hum c);
+      exit 1