From d4e82d3d5644de33fe202e45a5e9633a54eaaaea Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:37:35 -0500 Subject: [PATCH] swarm: convergence strategies + failure threshold, hardened El JSON usage - vote/merge/reduce/collect convergence proven end-to-end; failure threshold aborts a swarm below min_success_ratio (integer per-mille) and completes when failures are within tolerance, with worker.failed + swarm.aborted tracked durably. - worked around three El runtime/codegen semantics surfaced during the build: json_set inserts RAW (use json_set_str for string values); json_set cannot update an existing key (vote tallies via list rescanning); json_array_get keeps quotes (use json_array_get_string). Also: float division is unreliable (swarm uses integer math), and a let-rebind in a deeply nested if/else does not propagate outward (accumulators kept at one block level). test_convergence: 8/8; test_swarm: 12/12. --- lang/swarm/ccr.el | 2 +- lang/swarm/containment.el | 2 +- lang/swarm/swarm.el | 99 ++++++++++++++++++---------- lang/swarm/tests/test_convergence.el | 55 ++++++++++++++++ lang/swarm/worktrack.el | 10 ++- 5 files changed, 132 insertions(+), 36 deletions(-) create mode 100644 lang/swarm/tests/test_convergence.el diff --git a/lang/swarm/ccr.el b/lang/swarm/ccr.el index af70641..784bdc6 100644 --- a/lang/swarm/ccr.el +++ b/lang/swarm/ccr.el @@ -88,7 +88,7 @@ fn ccr_retrieve_scoped(blueprint: String, knowledge_refs: String, input_item: St let m: Int = json_array_len(knowledge_refs) let i = 0 while i < m { - let ref: String = json_array_get(knowledge_refs, i) + let ref: String = json_array_get_string(knowledge_refs, i) let hit: String = primitive_attend(ref, 3) let acc = acc + "# ref:" + ref + "\n" + hit + "\n" let i = i + 1 diff --git a/lang/swarm/containment.el b/lang/swarm/containment.el index a824d91..75bd3ee 100644 --- a/lang/swarm/containment.el +++ b/lang/swarm/containment.el @@ -117,7 +117,7 @@ fn containment_guard_open(token: String, corr_id: String) -> String { if str_eq(reason, "") { return "" } - let p: String = json_set("{}", "reason", reason) + let p: String = json_set_str("{}", "reason", reason) worktrack_append("containment.violation", corr_id, "open", p) return reason } diff --git a/lang/swarm/swarm.el b/lang/swarm/swarm.el index 5414c24..7ac21df 100644 --- a/lang/swarm/swarm.el +++ b/lang/swarm/swarm.el @@ -37,11 +37,18 @@ fn swarm_worker_entry(envelope_json: String) -> String { // swarm, so it proceeds. Its only outward edge is this returned result // (the vertical worker->coordinator path). let out: String = swarm_run_blueprint(ctx) + // A worker reports failed iff its blueprint signalled failure. This is the + // vertical status edge the coordinator reads during convergence (§4.3, §7). + let bstatus: String = json_get_string(out, "blueprint_status") + let status: String = "completed" + if str_eq(bstatus, "failed") { + let status = "failed" + } let kv: [String] = el_list_empty() let kv = el_list_append(kv, "worker_id") let kv = el_list_append(kv, worker_id) let kv = el_list_append(kv, "status") - let kv = el_list_append(kv, "completed") + let kv = el_list_append(kv, status) let res: String = json_build_object(kv) return json_set(res, "output", out) } @@ -52,13 +59,41 @@ fn swarm_worker_entry(envelope_json: String) -> String { // json_get_string(ctx,"blueprint"). Idempotent: reads ctx, writes only its // returned output (§7.3). fn swarm_run_blueprint(ctx: String) -> String { + let blueprint: String = json_get_string(ctx, "blueprint") let input_item: String = json_get_string(ctx, "input") let knowledge: String = json_get_string(ctx, "knowledge") + + // classify — deterministic verdict for the `vote` convergence strategy: + // verdict is "long" if the input has >4 chars, else "short". + if str_eq(blueprint, "classify") { + let verdict: String = "short" + if str_len(input_item) > 4 { + let verdict = "long" + } + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "verdict") + let kv = el_list_append(kv, verdict) + let kv = el_list_append(kv, "blueprint_status") + let kv = el_list_append(kv, "ok") + return json_build_object(kv) + } + + // faildemo — a worker that fails on inputs beginning with "x" (exercises the + // failure threshold + partial convergence path). Idempotent, side-effect-free. + if str_eq(blueprint, "faildemo") { + let st: String = "ok" + if str_starts_with(input_item, "x") { + let st = "failed" + } + return json_set_str("{}", "blueprint_status", st) + } + + // default (analyze_item): the CCR execution cycle think -> intend -> act. let instruction: String = "process input: " + input_item let thought: String = primitive_think(knowledge, instruction) let intent: String = primitive_intend(thought) let effect: String = primitive_act(intent, input_item) - return effect + return json_set_str(effect, "blueprint_status", "ok") } // ── native-thread fan-out, bounded by concurrency, order-preserving ────────── @@ -130,15 +165,16 @@ fn swarm_converge_merge(results: [String]) -> String { let merged = merged + out let i = i + 1 } - return json_set("{}", "merged", merged) + return json_set_str("{}", "merged", merged) } // swarm_converge_vote — tally a field across worker outputs, pick the majority. // Each worker output is expected to carry a "verdict" string field. fn swarm_converge_vote(results: [String]) -> String { let n: Int = el_list_len(results) - // count occurrences by scanning; first-past-the-post - let tally: String = "{}" + // Collect verdicts (no mutable tally: json_set can't update an existing key + // and there is no el_list_set). Then count each verdict by rescanning. + let verdicts: [String] = el_list_empty() let i = 0 while i < n { let out: String = json_get_raw(el_list_get(results, i), "output") @@ -146,34 +182,31 @@ fn swarm_converge_vote(results: [String]) -> String { if str_eq(v, "") { let i = i + 1 } else { - let cur: String = json_get_string(tally, v) - let c: Int = 0 - if str_eq(cur, "") { - let c = 1 - } else { - let c = str_to_int(cur) + 1 - } - let tally = json_set(tally, v, int_to_str(c)) + let verdicts = el_list_append(verdicts, v) let i = i + 1 } } - // pick the max + // pick the verdict with the highest count (first-past-the-post) + let vn: Int = el_list_len(verdicts) let best: String = "" let bestc = 0 - let j = 0 - while j < n { - let out: String = json_get_raw(el_list_get(results, j), "output") - let v: String = json_get_string(out, "verdict") - if str_eq(v, "") { - let j = j + 1 - } else { - let c: Int = str_to_int(json_get_string(tally, v)) - if c > bestc { - let bestc = c - let best = v + let a = 0 + while a < vn { + let cand: String = el_list_get(verdicts, a) + // count occurrences of cand + let c = 0 + let b = 0 + while b < vn { + if str_eq(el_list_get(verdicts, b), cand) { + let c = c + 1 } - let j = j + 1 + let b = b + 1 } + if c > bestc { + let bestc = c + let best = cand + } + let a = a + 1 } let kv: [String] = el_list_empty() let kv = el_list_append(kv, "winner") @@ -260,7 +293,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let n: Int = json_array_len(inputs_json) // swarm.created - let cp: String = json_set("{}", "blueprint", blueprint) + let cp: String = json_set_str("{}", "blueprint", blueprint) let cp2: String = json_set(cp, "input_count", int_to_str(n)) worktrack_append("swarm.created", corr_id, corr_id, cp2) @@ -269,7 +302,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let i = 0 while i < n { let worker_id: String = corr_id + "/worker-" + int_to_str(i) - let input_item: String = json_array_get(inputs_json, i) + let input_item: String = json_array_get_string(inputs_json, i) let wtoken: String = containment_worker_token(corr_id, worker_id) let ctx: String = ccr_compile(blueprint, knowledge_refs, input_item, corr_id, worker_id, wtoken) // envelope: only this worker's compiled context + its closed token @@ -283,7 +316,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let env2: String = json_set(env1, "ctx", ctx) let envelopes = el_list_append(envelopes, env2) - let sp: String = json_set("{}", "input", input_item) + let sp: String = json_set_str("{}", "input", input_item) worktrack_append("worker.started", corr_id, worker_id, sp) let i = i + 1 } @@ -301,9 +334,9 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let st: String = json_get_string(res, "status") if str_eq(st, "completed") { let succ = succ + 1 - worktrack_append("worker.completed", corr_id, wid, json_set("{}", "status", "completed")) + worktrack_append("worker.completed", corr_id, wid, json_set_str("{}", "status", "completed")) } else { - worktrack_append("worker.failed", corr_id, wid, json_set("{}", "error", json_get_string(res, "error"))) + worktrack_append("worker.failed", corr_id, wid, json_set_str("{}", "error", json_get_string(res, "error"))) } let r = r + 1 } @@ -321,7 +354,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con } if str_eq(status, "aborted") { - let ap: String = json_set("{}", "reason", "success ratio below min_success_ratio") + let ap: String = json_set_str("{}", "reason", "success ratio below min_success_ratio") worktrack_append("swarm.aborted", corr_id, corr_id, ap) let rep: String = worktrack_swarm_report(corr_id) let ok: [String] = el_list_empty() @@ -335,7 +368,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con // ── converge ── let merged: String = swarm_converge(strategy, results) - let dp: String = json_set("{}", "strategy", strategy) + let dp: String = json_set_str("{}", "strategy", strategy) worktrack_append("swarm.completed", corr_id, corr_id, dp) let rep2: String = worktrack_swarm_report(corr_id) diff --git a/lang/swarm/tests/test_convergence.el b/lang/swarm/tests/test_convergence.el new file mode 100644 index 0000000..0340a1b --- /dev/null +++ b/lang/swarm/tests/test_convergence.el @@ -0,0 +1,55 @@ +// test_convergence.el — convergence strategies + failure threshold / abort. + +fn assert_true(label: String, cond: Bool, fails: Int) -> Int { + if cond { print(" ok " + label); return fails } + print(" FAIL " + label); return fails + 1 +} + +fn main() -> Int { + let fails = 0 + let refs: String = "[]" + + // ── vote: classify 5 inputs; 3 "long" (>4 chars) vs 2 "short" -> winner long ── + let inputs: String = "[\"alpha\",\"bravo\",\"hi\",\"charlie\",\"ok\"]" + let cfg_v: String = "{\"concurrency\":\"3\",\"strategy\":\"vote\",\"min_success_ratio\":\"1.0\"}" + let rv: String = swarm_run("classify", refs, inputs, cfg_v) + let merged_v: String = json_get_raw(rv, "merged") + let winner: String = json_get_string(merged_v, "winner") + let votes: Int = str_to_int(json_get_string(merged_v, "votes")) + let fails = assert_true("vote winner = long", str_eq(winner, "long"), fails) + let fails = assert_true("vote count = 3", votes == 3, fails) + + // ── merge: outputs joined ── + let cfg_m: String = "{\"concurrency\":\"2\",\"strategy\":\"merge\",\"min_success_ratio\":\"1.0\"}" + let rm: String = swarm_run("analyze_item", refs, "[\"a\",\"b\",\"c\"]", cfg_m) + let merged_m: String = json_get_raw(rm, "merged") + let joined: String = json_get_string(merged_m, "merged") + let fails = assert_true("merge produced a joined string", str_contains(joined, "|"), fails) + + // ── reduce: count accumulates ── + let cfg_r: String = "{\"concurrency\":\"4\",\"strategy\":\"reduce\",\"min_success_ratio\":\"1.0\"}" + let rr: String = swarm_run("analyze_item", refs, "[\"a\",\"b\",\"c\",\"d\"]", cfg_r) + let merged_r: String = json_get_raw(rr, "merged") + let rcount: Int = str_to_int(json_get_string(merged_r, "count")) + let fails = assert_true("reduce count = 4", rcount == 4, fails) + + // ── failure threshold: 2 of 5 fail (x-prefixed); ratio 3/5=0.6 < 0.8 -> aborted ── + let fin: String = "[\"a\",\"xb\",\"c\",\"xd\",\"e\"]" + let cfg_f: String = "{\"concurrency\":\"5\",\"strategy\":\"collect\",\"min_success_ratio\":\"0.8\"}" + let rf: String = swarm_run("faildemo", refs, fin, cfg_f) + let fstatus: String = json_get_string(rf, "status") + let fails = assert_true("swarm aborted below min_success_ratio (0.6<0.8)", str_eq(fstatus, "aborted"), fails) + let corr_f: String = json_get_string(rf, "corr_id") + let failed_n: Int = worktrack_count_kind(corr_f, "worker.failed") + let aborted_n: Int = worktrack_count_kind(corr_f, "swarm.aborted") + let fails = assert_true("tracked 2 worker.failed", failed_n == 2, fails) + let fails = assert_true("tracked swarm.aborted", aborted_n == 1, fails) + + // ── same failures tolerated when min_success_ratio=0.5 (0.6>=0.5) -> completed ── + let cfg_ok: String = "{\"concurrency\":\"5\",\"strategy\":\"collect\",\"min_success_ratio\":\"0.5\"}" + let rok: String = swarm_run("faildemo", refs, fin, cfg_ok) + let fails = assert_true("swarm completes when failures within tolerance", str_eq(json_get_string(rok, "status"), "completed"), fails) + + if fails == 0 { print("PASS test_convergence"); return 0 } + print("FAIL test_convergence (" + int_to_str(fails) + ")"); return 1 +} diff --git a/lang/swarm/worktrack.el b/lang/swarm/worktrack.el index 2f37337..5b75704 100644 --- a/lang/swarm/worktrack.el +++ b/lang/swarm/worktrack.el @@ -24,6 +24,14 @@ // 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. @@ -112,7 +120,7 @@ fn worktrack_mirror_engram(rec: String, corr_id: String, kind: String, subject: 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 body2: String = json_set_str(body, "_auth", key) let resp: String = http_post(url + "/api/node", body2) return true }