el/lang/swarm/tests/harness_local_swarm.el

// harness_local_swarm.el — LOCAL-SWARM INTEGRATION HARNESS.
//
// Proves the FULL local-swarm mechanics end-to-end, TODAY, on the isolated
// engram clone (:8901), with the primitive seam pointed at the hermetic stub.
// The moment the api-reshape agent lands the decorated primitives on the
// dharma bus, binding is ONE flip (primitive_binding.el) + SWARM_PRIMITIVE_SEAM=
// decorated — this same harness then runs the bound path with no other change.
//
// The @manager (the coordinator) fans out N native El worker threads at real
// concurrency, each given a CCR-scoped engram slice, each invoking the primitive
// seam (think over its slice), enforces all three containment rules, converges
// (vote AND reduce), work-tracks durably, and observes afferent telemetry.
//
// Run with the sandbox env sourced (ENGRAM_URL=:8901) to also exercise CCR
// retrieval against the real (isolated) mind; runs fully without it too.

fn ok(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
    print("== LOCAL-SWARM INTEGRATION HARNESS (seam=" + seam_mode() + ") ==")

    // 8 independent slices, real concurrency of 4 (2 waves of native pthreads).
    let inputs: String = "[\"billing\",\"payments\",\"ledger\",\"invoicing\",\"tax\",\"payroll\",\"audit\",\"fx\"]"
    let refs: String = "[\"Volatility-Based Decomposition\"]"

    // ── A) fan-out / converge at real concurrency (reduce) ──
    let cfg_r: String = "{\"concurrency\":\"4\",\"strategy\":\"reduce\",\"min_success_ratio\":\"1.0\"}"
    let rr: String = swarm_run("analyze_item", refs, inputs, cfg_r)
    let fails = ok("swarm completed at concurrency=4 over 8 native-thread workers", str_eq(json_get_string(rr, "status"), "completed"), fails)
    let corr: String = json_get_string(rr, "corr_id")
    let merged_r: String = json_get_raw(rr, "merged")
    let fails = ok("reduce converged all 8 worker outputs", str_to_int(json_get_string(merged_r, "count")) == 8, fails)

    // ── B) afferent telemetry observed by the @manager ──
    let telem: String = json_get_raw(rr, "telemetry")
    let aff: Int = str_to_int(json_get_string(telem, "afferent_think"))
    let seen_mode: String = json_get_string(telem, "seam_mode")
    let fails = ok("afferent think-signals counted = 8 (one per worker)", aff == 8, fails)
    let fails = ok("telemetry records the active seam mode", str_eq(seen_mode, seam_mode()), fails)
    let telem_recs: Int = worktrack_count_kind(corr, "swarm.telemetry")
    let fails = ok("telemetry durably journalled", telem_recs == 1, fails)

    // ── C) CCR scoping + non-leak per worker ──
    let wt: String = containment_worker_token(corr, corr + "/worker-3")
    let ctx3: String = ccr_compile("analyze_item", refs, "invoicing", corr, corr + "/worker-3", wt)
    let fails = ok("CCR context bounded within token budget", ccr_within_budget(ctx3), fails)
    let fails = ok("CCR context carries THIS slice", str_eq(json_get_string(ctx3, "input"), "invoicing"), fails)
    let leaks: Bool = str_contains(ctx3, "payroll") || str_contains(ctx3, "audit")
    let fails = ok("CCR context does NOT leak sibling slices (security boundary)", !leaks, fails)

    // ── D) all three containment rules ──
    let deny: String = containment_check_open(wt)
    let fails = ok("Rule 2: worker token may not OPEN a swarm", !str_eq(deny, ""), fails)
    let denyj: String = containment_check_join(wt, "other-swarm")
    let fails = ok("Rule 1: worker token may not JOIN another swarm", !str_eq(denyj, ""), fails)
    let lat: String = containment_check_lateral(wt, "sibling-9")
    let fails = ok("Rule 3: worker->worker lateral edge rejected", !str_eq(lat, ""), fails)
    let ver: String = containment_check_lateral(wt, "")
    let fails = ok("Rule 3: worker->manager vertical edge allowed", str_eq(ver, ""), fails)
    // enforced live: a worker-token caller is denied opening a real swarm
    let wcfg: String = json_set(cfg_r, "caller_token", wt)
    let denied: String = swarm_run("analyze_item", refs, inputs, wcfg)
    let fails = ok("Rule 2 enforced live: worker-caller swarm denied", str_eq(json_get_string(denied, "status"), "denied"), fails)

    // ── E) vote convergence strategy at concurrency ──
    let cfg_v: String = "{\"concurrency\":\"8\",\"strategy\":\"vote\",\"min_success_ratio\":\"1.0\"}"
    let rv: String = swarm_run("classify", refs, inputs, cfg_v)
    let winner: String = json_get_string(json_get_raw(rv, "merged"), "winner")
    // billing/payments/ledger/invoicing/payroll/audit = long(>4); tax/fx = short -> long wins
    let fails = ok("vote converged (winner=long)", str_eq(winner, "long"), fails)

    // ── F) durable, inspectable work-tracking ──
    let started: Int = worktrack_count_kind(corr, "worker.started")
    let completed: Int = worktrack_count_kind(corr, "worker.completed")
    let fails = ok("work-tracking journal: 8 started + 8 completed", (started == 8) && (completed == 8), fails)

    print("")
    if fails == 0 {
        print("HARNESS GREEN — full local-swarm mechanics proven with seam=" + seam_mode())
        return 0
    }
    print("HARNESS FAIL (" + int_to_str(fails) + ")")
    return 1
}