// 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. // ── JSON helper ────────────────────────────────────────────────────────────── // json_set inserts its value as a RAW JSON fragment (objects/arrays/numbers). // json_set_str sets a plain STRING value, correctly quoted and escaped. Use // json_set for nested JSON, json_set_str for strings. fn json_set_str(j: String, key: String, val: String) -> String { return json_set(j, key, "\"" + json_escape_string(val) + "\"") } // ── 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 { // Opt-in: the durable substrate is the JSONL journal (always written). The // engram mirror is an additional convenience, enabled with SWARM_MIRROR=1, // so a swarm never depends on — or loads — the mind just to track its work. if str_eq(env("SWARM_MIRROR"), "1") { // enabled — fall through to the mirror POST let _go: Int = 1 } else { return false } 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_str(body, "_auth", key) let resp: String = http_post(url + "/api/nodes", 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) }