/*
* m0_roundtrip.cc — HyperStorageNode bridge harness (M0 + M1).
*
* Ships a small AtomSpace over a UNIX-domain socket to the Bare `hyper-atomspace`
* daemon (daemon.js), using the REAL opencog s-expr codec (Sexpr::encode_atom /
* Sexpr::decode_atom), then replays the log back and gold-checks identity.
*
* `HyperChannel` is the reusable AF_UNIX transport (adapted from CogSimpleStorage
* do_send/do_recv, TCP->UDS); at the guile-module step it lifts into
* `HyperStorageNode : public StorageNode`. main() is the test harness.
*
* Modes:
* full (default) clear + store + barrier + load + gold-check (live append/replay)
* verify load + gold-check ONLY (run against a RESTARTED daemon on the
* same Hypercore dir -> proves durable deterministic replay, M1 #1)
* fork clear + store + load(full) + loadat(K) + prefix check
* (checkout at seq K = cheap fork, M1 #2)
* lineage clear + store; then for a high-level atom X, trace its
* causal history through the log via seqof and check it is the
* exact set of ops that produced X, ordered, excluding
* causally-unrelated atoms (complete lineage, M3 #3)
* m4 <sockB> TWO daemons (agents) on one Autobase log: agent A stores
* half the graph, agent B (arg 3) the other half; both must
* converge to the identical full AtomSpace (multi-agent, M4)
*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#include <algorithm>
#include <functional>
#include <iostream>
#include <set>
#include <string>
#include <vector>
#include <string.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <opencog/atomspace/AtomSpace.h>
#include <opencog/atoms/atom_types/atom_types.h>
#include <opencog/atoms/base/Handle.h>
#include <opencog/persist/sexpr/Sexpr.h>
using namespace opencog;
// ---------------------------------------------------------------------------
// HyperChannel — newline-framed s-expr protocol over a UNIX-domain socket.
class HyperChannel
{
int _fd = -1;
std::string _rbuf;
public:
void open(const std::string& path)
{
_fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (_fd < 0) throw std::runtime_error("socket() failed");
struct sockaddr_un addr;
memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX;
if (path.size() >= sizeof(addr.sun_path))
throw std::runtime_error("socket path too long");
strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1);
if (connect(_fd, (struct sockaddr*) &addr, sizeof(addr)) != 0)
throw std::runtime_error(std::string("connect() failed: ")
+ strerror(errno));
}
void close_fd() { if (_fd >= 0) { ::close(_fd); _fd = -1; } }
~HyperChannel() { close_fd(); }
void send_line(const std::string& line)
{
size_t off = 0;
while (off < line.size())
{
ssize_t n = ::send(_fd, line.data() + off, line.size() - off, 0);
if (n < 0) throw std::runtime_error("send() failed");
off += n;
}
}
std::string recv_line()
{
for (;;)
{
size_t nl = _rbuf.find('\n');
if (nl != std::string::npos)
{
std::string line = _rbuf.substr(0, nl);
_rbuf.erase(0, nl + 1);
return line;
}
char buf[8192];
ssize_t n = ::recv(_fd, buf, sizeof(buf), 0);
if (n < 0) throw std::runtime_error("recv() failed");
if (n == 0) throw std::runtime_error("daemon closed connection");
_rbuf.append(buf, n);
}
}
void store(const std::string& sexpr) { send_line("store " + sexpr + "\n"); }
void clear() { send_line("clear\n"); (void) recv_line(); }
void barrier() { send_line("barrier\n"); (void) recv_line(); }
std::string load() { send_line("load\n"); return recv_line(); }
std::string loadat(int k)
{
send_line("loadat " + std::to_string(k) + "\n");
return recv_line();
}
int seqof(const std::string& sexpr)
{
send_line("seqof " + sexpr + "\n");
return std::stoi(recv_line());
}
int count() { send_line("count\n"); return std::stoi(recv_line()); }
bool writable(){ send_line("writable\n"); return recv_line() == "true"; }
};
// ---------------------------------------------------------------------------
// Deterministic source AtomSpace. `order` returns the atoms in creation order
// (nodes before the links that use them), so any prefix is a self-consistent
// sub-graph — which is what makes the fork/checkout check crisp.
static AtomSpacePtr build_src(HandleSeq& order)
{
AtomSpacePtr as = createAtomSpace();
Handle cat = as->add_node(CONCEPT_NODE, "cat"); order.push_back(cat);
Handle animal = as->add_node(CONCEPT_NODE, "animal"); order.push_back(animal);
order.push_back(as->add_link(INHERITANCE_LINK, cat, animal));
Handle whale = as->add_node(CONCEPT_NODE, "whale"); order.push_back(whale);
Handle water = as->add_node(CONCEPT_NODE, "water"); order.push_back(water);
Handle livesIn = as->add_node(PREDICATE_NODE, "lives-in"); order.push_back(livesIn);
Handle lst = as->add_link(LIST_LINK, whale, water); order.push_back(lst);
order.push_back(as->add_link(EVALUATION_LINK, livesIn, lst));
return as;
}
// Every atom in `as`, encoded and sorted (for set comparison).
static std::vector<std::string> encoded_sorted(const AtomSpacePtr& as)
{
HandleSeq all;
as->get_handles_by_type(all, ATOM, true);
std::vector<std::string> out;
for (const Handle& h : all) out.push_back(Sexpr::encode_atom(h));
std::sort(out.begin(), out.end());
return out;
}
// Decode a "(<atom><atom>…)" daemon reply into a fresh AtomSpace.
static AtomSpacePtr decode_reply(const std::string& reply)
{
AtomSpacePtr dst = createAtomSpace();
std::unordered_map<std::string, Handle> cache;
size_t l = reply.find('(');
size_t end = reply.rfind(')');
if (l == std::string::npos || end == std::string::npos || l >= end)
return dst;
l += 1;
size_t r = end;
while (true)
{
int pcnt = Sexpr::get_next_expr(reply, l, r, 0);
if (l == r) break;
if (0 < pcnt) break;
Handle h = Sexpr::decode_atom(reply, l, r, 0, cache);
dst->add_atom(h);
l = r + 1;
r = end;
}
return dst;
}
int main(int argc, char** argv)
{
if (argc < 2)
{
std::cerr << "usage: m0_roundtrip <socket-path> [full|verify|fork]\n";
return 2;
}
const std::string sockpath = argv[1];
const std::string mode = (argc >= 3) ? argv[2] : "full";
HandleSeq order;
AtomSpacePtr src = build_src(order);
std::vector<std::string> es = encoded_sorted(src);
HyperChannel ch;
ch.open(sockpath);
if (mode == "verify")
{
// Load ONLY — the daemon was restarted on the same Hypercore dir.
AtomSpacePtr dst = decode_reply(ch.load());
std::vector<std::string> ed = encoded_sorted(dst);
bool ok = (es == ed);
std::cout << "[verify] durable replay src=" << src->get_size()
<< " dst=" << dst->get_size()
<< (ok ? " GOLD #1 PASS\n" : " GOLD #1 FAIL\n");
return ok ? 0 : 1;
}
if (mode == "fork")
{
ch.clear();
for (const Handle& h : order) ch.store(Sexpr::encode_atom(h)); // creation order
ch.barrier();
AtomSpacePtr full = decode_reply(ch.load());
AtomSpacePtr forkK = decode_reply(ch.loadat(3)); // checkout @ seq 3
std::vector<std::string> ef = encoded_sorted(forkK);
std::vector<std::string> eF = encoded_sorted(full);
bool szF = (full->get_size() == 8);
bool szf = (forkK->get_size() == 3); // cat, animal, Inheritance(cat,animal)
bool subset = std::includes(eF.begin(), eF.end(), ef.begin(), ef.end());
bool strict = (forkK->get_size() < full->get_size());
bool ok = szF && szf && subset && strict;
std::cout << "[fork] full=" << full->get_size()
<< " checkout@3=" << forkK->get_size()
<< " subset=" << subset << " strict=" << strict
<< (ok ? " GOLD #2 PASS\n" : " GOLD #2 FAIL\n");
return ok ? 0 : 1;
}
if (mode == "lineage")
{
ch.clear();
for (const Handle& h : order) ch.store(Sexpr::encode_atom(h)); // creation order
ch.barrier();
// X = the high-level fact: Evaluation(lives-in, List(whale, water)).
const Handle& X = order[7];
// Causal closure: X + its transitive outgoing atoms (its dependencies).
std::set<Handle> closure;
std::function<void(const Handle&)> dfs = [&](const Handle& h) {
if (closure.count(h)) return;
if (h->is_link())
for (const Handle& o : h->getOutgoingSet()) dfs(o);
closure.insert(h);
};
dfs(X);
int seqX = ch.seqof(Sexpr::encode_atom(X));
// Each dependency has a real log position, entered BEFORE X (append-only,
// outgoing-first). Order the lineage by seq to reconstruct the chain.
std::vector<std::pair<int, std::string>> chain;
bool all_found = true, deps_before_X = true;
for (const Handle& h : closure)
{
int s = ch.seqof(Sexpr::encode_atom(h));
if (s < 0) all_found = false;
if (h != X && !(s < seqX)) deps_before_X = false;
chain.push_back({s, Sexpr::encode_atom(h)});
}
std::sort(chain.begin(), chain.end());
// Causally-unrelated atoms (cat, animal, Inheritance) must NOT appear.
bool excludes_unrelated =
!closure.count(order[0]) && !closure.count(order[1]) &&
!closure.count(order[2]);
// Expected lineage, by seq: whale@3 water@4 lives-in@5 List@6 Eval@7.
std::vector<std::pair<int, std::string>> expect;
for (int i : {3, 4, 5, 6, 7})
expect.push_back({i, Sexpr::encode_atom(order[i])});
bool chain_ok = (chain == expect);
std::cout << "[lineage] X=" << Sexpr::encode_atom(X) << " @seq " << seqX << "\n";
for (const auto& pr : chain)
std::cout << " seq " << pr.first << " " << pr.second << "\n";
bool ok = all_found && deps_before_X && excludes_unrelated && chain_ok &&
(int) closure.size() == 5 && seqX == 7;
std::cout << " found=" << all_found << " deps<X=" << deps_before_X
<< " excl-unrelated=" << excludes_unrelated
<< " chain=" << chain_ok
<< (ok ? " GOLD #3 PASS\n" : " GOLD #3 FAIL\n");
return ok ? 0 : 1;
}
if (mode == "m4")
{
if (argc < 4) { std::cerr << "m4 needs <sockB>\n"; return 2; }
HyperChannel chB;
chB.open(argv[3]);
// Ceremony (addWriter) is done by run.sh before we run; both must be
// writable now. A = this channel (ch), B = chB.
if (!ch.writable() || !chB.writable())
{
std::cout << "[m4] A.writable=" << ch.writable()
<< " B.writable=" << chB.writable()
<< " FAIL (ceremony not complete)\n";
return 1;
}
// Each agent stores a self-contained half of the graph (no cross-agent
// dependency): A = {cat, animal, Inheritance}; B = {whale, water,
// lives-in, List, Eval}.
for (int i = 0; i < 3; i++) ch.store(Sexpr::encode_atom(order[i]));
for (int i = 3; i < 8; i++) chB.store(Sexpr::encode_atom(order[i]));
ch.barrier();
chB.barrier();
// Wait for the two writers' streams to linearize + replicate to 8 ops.
int ca = 0, cb = 0;
for (int t = 0; t < 100; t++) // ~10s budget
{
ca = ch.count();
cb = chB.count();
if (ca == 8 && cb == 8) break;
usleep(100000);
}
AtomSpacePtr dstA = decode_reply(ch.load());
AtomSpacePtr dstB = decode_reply(chB.load());
std::vector<std::string> ea = encoded_sorted(dstA);
std::vector<std::string> eb = encoded_sorted(dstB);
bool a_ok = (ea == es); // agent A converged to the full graph
bool b_ok = (eb == es); // agent B converged to the full graph
bool agree = (ea == eb); // and to the SAME graph
bool ok = a_ok && b_ok && agree && dstA->get_size() == 8;
std::cout << "[m4] converge A=" << ca << " B=" << cb
<< " | A==src=" << a_ok << " B==src=" << b_ok
<< " A==B=" << agree
<< (ok ? " M4 MULTI-AGENT PASS\n" : " M4 MULTI-AGENT FAIL\n");
return ok ? 0 : 1;
}
// mode == full : live append + replay round-trip.
ch.clear();
for (const Handle& h : order) ch.store(Sexpr::encode_atom(h));
ch.barrier();
AtomSpacePtr dst = decode_reply(ch.load());
std::vector<std::string> ed = encoded_sorted(dst);
bool ok = (es == ed) && (src->get_size() == dst->get_size());
std::cout << "[full] live round-trip src=" << src->get_size()
<< " dst=" << dst->get_size()
<< (ok ? " PASS\n" : " FAIL\n");
if (not ok)
{
std::cout << " src:\n"; for (auto& s : es) std::cout << " " << s << "\n";
std::cout << " dst:\n"; for (auto& s : ed) std::cout << " " << s << "\n";
}
return ok ? 0 : 1;
}