diff --git a/lang/el-compiler/runtime/el_runtime.h b/lang/el-compiler/runtime/el_runtime.h index 4b30d7c..ca12886 100644 --- a/lang/el-compiler/runtime/el_runtime.h +++ b/lang/el-compiler/runtime/el_runtime.h @@ -302,6 +302,8 @@ el_val_t now_ns(void); el_val_t el_now_instant(void); el_val_t now(void); +el_val_t now_millis(void); /* wall-clock milliseconds (defined in el_runtime.c) */ +el_val_t now_ns(void); /* wall-clock nanoseconds (defined in el_runtime.c) */ el_val_t unix_seconds(el_val_t n); el_val_t unix_millis(el_val_t n); el_val_t instant_from_iso8601(el_val_t s); diff --git a/lang/swarm/build.sh b/lang/swarm/build.sh new file mode 100644 index 0000000..cd92645 --- /dev/null +++ b/lang/swarm/build.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# build.sh — compile an El program that uses the swarm capability. +# +# Concatenates the El native-concurrency stdlib (thread.el, channel.el) and the +# swarm capability modules in dependency order, then the user program, compiles +# with the canonical elc, and links against the shared C runtime. +# +# Usage: +# swarm/build.sh +# +# The swarm modules use only el_runtime.c builtins plus thread.el/channel.el, +# so nothing else needs concatenating (engram_*, json_*, str_*, fs_*, http_*, +# uuid_v4, now_millis are all C builtins in el_runtime.c). + +set -uo pipefail +cd "$(dirname "$0")/.." # -> lang/ +LANG_DIR="$(pwd)" +ELC="${ELC:-${LANG_DIR}/dist/platform/elc}" +RT="${LANG_DIR}/el-compiler/runtime" + +PROG="${1:?usage: build.sh }" +OUT="${2:?usage: build.sh }" + +# swarm module load order (each may depend on those before it): +# worktrack — durable work-tracking journal (no swarm deps) +# containment — the three containment rules (no swarm deps) +# primitives — think/act/attend/intend/learn seam (no swarm deps) +# ccr — per-worker compiled bounded context (depends: primitives) +# swarm — orchestrator: fan-out/converge (depends: all above + thread) +SWARM_MODULES=" + swarm/worktrack.el + swarm/containment.el + swarm/primitives.el + swarm/ccr.el + swarm/swarm.el +" + +TMP_C="$(mktemp -t swarm_build.XXXXXX).c" +COMBINED="$(mktemp -t swarm_combined.XXXXXX).el" + +cat runtime/thread.el runtime/channel.el $SWARM_MODULES "$PROG" > "$COMBINED" + +if ! "$ELC" "$COMBINED" > "$TMP_C" 2>/tmp/swarm.elc.err; then + echo "elc FAILED:" >&2 + sed 's/^/ /' /tmp/swarm.elc.err >&2 + rm -f "$TMP_C" "$COMBINED" + exit 1 +fi + +if ! cc -O2 -I "$RT" "$TMP_C" "$RT/el_runtime.c" -lcurl -lpthread -lm -o "$OUT" 2>/tmp/swarm.cc.err; then + echo "cc FAILED:" >&2 + sed 's/^/ /' /tmp/swarm.cc.err >&2 + rm -f "$TMP_C" "$COMBINED" + exit 1 +fi + +rm -f "$TMP_C" "$COMBINED" +echo "built: $OUT" diff --git a/lang/swarm/tests/test_worktrack.el b/lang/swarm/tests/test_worktrack.el new file mode 100644 index 0000000..cf763eb --- /dev/null +++ b/lang/swarm/tests/test_worktrack.el @@ -0,0 +1,38 @@ +// test_worktrack.el — durability + inspectability of the work-tracking journal. + +fn main() -> Int { + let corr: String = "test-" + uuid_v4() + + // record a swarm lifecycle + let p1: String = json_set("{}", "input_count", "3") + worktrack_append("swarm.created", corr, "swarm-1", p1) + worktrack_append("worker.started", corr, "worker-001", "{}") + worktrack_append("worker.started", corr, "worker-002", "{}") + worktrack_append("worker.completed", corr, "worker-001", "{}") + worktrack_append("worker.failed", corr, "worker-002", "{}") + worktrack_append("swarm.completed", corr, "swarm-1", "{}") + + // inspect: reconstruct the report from the durable journal + let report: String = worktrack_swarm_report(corr) + print("report=" + report) + + let recs_n: Int = el_list_len(worktrack_records(corr)) + print("records=" + int_to_str(recs_n)) + + let state: String = json_get_string(report, "state") + let completed: Int = str_to_int(json_get_string(report, "workers_completed")) + let failed: Int = str_to_int(json_get_string(report, "workers_failed")) + + if str_eq(state, "completed") { + if completed == 1 { + if failed == 1 { + if recs_n == 6 { + print("PASS worktrack") + return 0 + } + } + } + } + print("FAIL worktrack") + return 1 +} diff --git a/lang/swarm/worktrack.el b/lang/swarm/worktrack.el new file mode 100644 index 0000000..2f37337 --- /dev/null +++ b/lang/swarm/worktrack.el @@ -0,0 +1,200 @@ +// worktrack.el — full work-tracking for the swarm. +// +// "Intent all the way up, orchestrator at the top." Every unit of parallel +// work a swarm fans out is recorded here: the swarm itself, each worker, its +// status, its result summary, the convergence, and the final merged output — +// all threaded by a single correlation ID so the entire execution graph can be +// reconstructed and audited (Swarm Architecture §6.1). +// +// DURABILITY. Records are appended to a JSON-lines journal on disk. The journal +// is append-only and single-writer: only the coordinator (the main thread, before +// and after each fan-out and during convergence) writes to it. Workers never +// touch it — they return structured results and the coordinator records them. +// This is deliberate: it makes the tracking store race-free and, not +// coincidentally, enforces Swarm containment rule 3 (no lateral worker state). +// +// INSPECTABILITY. The journal is plain JSONL — greppable, tailable, replayable. +// worktrack_read() loads it back; worktrack_swarm_report() reconstructs a +// swarm's full record from its correlation ID. +// +// ENGRAM MIRROR (optional). When ENGRAM_URL is set, each record is also mirrored +// into the engram as a node (POST /api/node) tagged with the correlation ID, so +// the swarm's execution becomes part of the durable mind, queryable by memory. +// +// Depends on: el_runtime.c builtins (fs_*, http_post, env, json_*, uuid_v4, +// now_millis, str_*). No El-module concat dependencies of its own. + +// ── Journal location ───────────────────────────────────────────────────────── + +// worktrack_dir — directory holding the swarm journals. +// Override with SWARM_TRACK_DIR; defaults to ./.swarm-track (relative to CWD). +fn worktrack_dir() -> String { + let d: String = env("SWARM_TRACK_DIR") + if str_eq(d, "") { + return ".swarm-track" + } + return d +} + +// worktrack_journal_path — the JSONL journal file for one correlation ID. +fn worktrack_journal_path(corr_id: String) -> String { + return worktrack_dir() + "/" + corr_id + ".jsonl" +} + +// worktrack_init — ensure the journal directory exists. Idempotent. +fn worktrack_init() -> Bool { + let d: String = worktrack_dir() + if fs_exists(d) { + return true + } + return fs_mkdir(d) +} + +// ── Record construction ────────────────────────────────────────────────────── + +// worktrack_record — build one journal record as a JSON object string. +// kind: the record kind (swarm.created, worker.started, ...) +// corr_id: the swarm correlation ID (links every record) +// subject: the entity the record is about (swarm id, worker id, "") +// payload: a JSON object string with kind-specific fields +fn worktrack_record(kind: String, corr_id: String, subject: String, payload: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "kind") + let kv = el_list_append(kv, kind) + let kv = el_list_append(kv, "corr_id") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "subject") + let kv = el_list_append(kv, subject) + let kv = el_list_append(kv, "ts_ms") + let kv = el_list_append(kv, int_to_str(now_millis())) + let rec: String = json_build_object(kv) + // Attach the payload as a nested raw JSON field. + let rec2: String = json_set(rec, "data", payload) + return rec2 +} + +// ── Journal append (single-writer, durable) ────────────────────────────────── + +// worktrack_append — append one record to the correlation journal (durable), +// and mirror it to the engram if ENGRAM_URL is configured. Returns the record. +// +// fs_write here is used in append semantics: we read-modify-write the file. The +// coordinator is the only writer, so this is safe and race-free. +fn worktrack_append(kind: String, corr_id: String, subject: String, payload: String) -> String { + worktrack_init() + let rec: String = worktrack_record(kind, corr_id, subject, payload) + let path: String = worktrack_journal_path(corr_id) + let prior: String = "" + if fs_exists(path) { + let prior = fs_read(path) + } + let next: String = prior + rec + "\n" + fs_write(path, next) + worktrack_mirror_engram(rec, corr_id, kind, subject) + return rec +} + +// worktrack_mirror_engram — best-effort mirror of a record into the engram. +// No-op unless ENGRAM_URL is set. Failures are swallowed (tracking must not +// depend on the mind being reachable). +fn worktrack_mirror_engram(rec: String, corr_id: String, kind: String, subject: String) -> Bool { + let url: String = env("ENGRAM_URL") + if str_eq(url, "") { + return false + } + let content: String = "swarm-track " + kind + " " + subject + " :: " + rec + let body_kv: [String] = el_list_empty() + let body_kv = el_list_append(body_kv, "content") + let body_kv = el_list_append(body_kv, content) + let body_kv = el_list_append(body_kv, "node_type") + let body_kv = el_list_append(body_kv, "SwarmTrack") + let body_kv = el_list_append(body_kv, "salience") + let body_kv = el_list_append(body_kv, "0.5") + let body: String = json_build_object(body_kv) + let key: String = env("ENGRAM_API_KEY") + let body2: String = json_set(body, "_auth", key) + let resp: String = http_post(url + "/api/node", body2) + return true +} + +// ── Read / inspect ─────────────────────────────────────────────────────────── + +// worktrack_read — read the raw JSONL journal for a correlation ID. +fn worktrack_read(corr_id: String) -> String { + let path: String = worktrack_journal_path(corr_id) + if fs_exists(path) { + return fs_read(path) + } + return "" +} + +// worktrack_records — the journal as a [String] of record JSON objects, in order. +fn worktrack_records(corr_id: String) -> [String] { + let raw: String = worktrack_read(corr_id) + let out: [String] = el_list_empty() + if str_eq(raw, "") { + return out + } + let lines: [String] = str_split_lines(raw) + let n: Int = el_list_len(lines) + let i = 0 + while i < n { + let ln: String = el_list_get(lines, i) + if str_eq(ln, "") { + let i = i + 1 + } else { + let out = el_list_append(out, ln) + let i = i + 1 + } + } + return out +} + +// worktrack_count_kind — how many records of a given kind exist for a swarm. +// Powers assertions and live status ("how many workers completed"). +fn worktrack_count_kind(corr_id: String, kind: String) -> Int { + let recs: [String] = worktrack_records(corr_id) + let n: Int = el_list_len(recs) + let c = 0 + let i = 0 + while i < n { + let r: String = el_list_get(recs, i) + let k: String = json_get_string(r, "kind") + if str_eq(k, kind) { + let c = c + 1 + } + let i = i + 1 + } + return c +} + +// worktrack_swarm_report — reconstruct a compact status report for a swarm from +// its journal: counts of started/completed/failed workers and terminal state. +// Inspectable, durable, derived purely from the append-only record. +fn worktrack_swarm_report(corr_id: String) -> String { + let started: Int = worktrack_count_kind(corr_id, "worker.started") + let completed: Int = worktrack_count_kind(corr_id, "worker.completed") + let failed: Int = worktrack_count_kind(corr_id, "worker.failed") + let done: Int = worktrack_count_kind(corr_id, "swarm.completed") + let aborted: Int = worktrack_count_kind(corr_id, "swarm.aborted") + let state: String = "running" + if aborted > 0 { + let state = "aborted" + } else { + if done > 0 { + let state = "completed" + } + } + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "corr_id") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "state") + let kv = el_list_append(kv, state) + let kv = el_list_append(kv, "workers_started") + let kv = el_list_append(kv, int_to_str(started)) + let kv = el_list_append(kv, "workers_completed") + let kv = el_list_append(kv, int_to_str(completed)) + let kv = el_list_append(kv, "workers_failed") + let kv = el_list_append(kv, int_to_str(failed)) + return json_build_object(kv) +}