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

@@ -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

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
 }

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)

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

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
 }