swarm: durable, inspectable work-tracking journal (worktrack.el)

Single-writer append-only JSONL journal keyed by correlation ID: swarm +
worker + convergence records, reconstructable into a status report. Optional
engram mirror via POST /api/node when ENGRAM_URL is set. Coordinator is the
only writer (workers return structured results), which is race-free and
enforces Swarm containment rule 3 by construction.

Also prototype now_millis/now_ns in el_runtime.h (defined in el_runtime.c but
unprototyped — blocked any El program needing a real ms clock under clang 21).

Test proves durability + inspectability end-to-end.

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)
+}