swarm: local-swarm integration harness + one-flip primitive seam + telemetry
- primitive_seam.el: SWARM_PRIMITIVE_SEAM selects stub (default, hermetic) vs decorated (reshape's dharma-bus primitives). Every seam call is an afferent signal; telemetry (seam_mode + afferent tick) rides the vertical result path. - primitive_binding.el: THE ONE FLIP POINT — bound_think/attend/learn today fall back to the stub; when the reshape's decorated primitives land, flip one line each and set SWARM_PRIMITIVE_SEAM=decorated. No other change anywhere. - swarm.el: default blueprint routes think through the seam; the @manager aggregates afferent counters from worker results (containment-safe, no shared bus register) and journals a swarm.telemetry record; telemetry in the return. - harness_local_swarm.el: 17/17 GREEN on :8901 with the stub — 8 native-thread workers at concurrency 4, reduce+vote convergence, CCR scoping+non-leak, all three containment rules (incl. live Rule-2 denial), durable work-tracking, afferent telemetry observed. Runs identically under seam=decorated today (binding fallback), proving the flip path executes. Engram writes stay opt-in (durable journal is the substrate); daemon healthy.
This commit is contained in:
@@ -31,6 +31,8 @@ SWARM_MODULES="
|
|||||||
swarm/worktrack.el
|
swarm/worktrack.el
|
||||||
swarm/containment.el
|
swarm/containment.el
|
||||||
swarm/primitives.el
|
swarm/primitives.el
|
||||||
|
swarm/primitive_binding.el
|
||||||
|
swarm/primitive_seam.el
|
||||||
swarm/ccr.el
|
swarm/ccr.el
|
||||||
swarm/swarm.el
|
swarm/swarm.el
|
||||||
"
|
"
|
||||||
|
|||||||
@@ -0,0 +1,39 @@
|
|||||||
|
// primitive_binding.el — THE ONE FLIP POINT.
|
||||||
|
//
|
||||||
|
// This file is the single seam between the swarm and the real agentic
|
||||||
|
// primitives. Binding the reshape's decorated primitives is a one-line change
|
||||||
|
// HERE and nothing else changes anywhere in the swarm.
|
||||||
|
//
|
||||||
|
// The api-reshape agent (wt/api-reshape) is wiring the primitives as DECORATED
|
||||||
|
// El on the dharma_* event bus over the engram — think/attend/learn/ground/assert
|
||||||
|
// become decorated fns that emit afferent events onto the bus. The moment they
|
||||||
|
// land, flip `bound_think` (and its siblings) to call them.
|
||||||
|
//
|
||||||
|
// TODAY (stub fallback, compiles + runs now against :8901):
|
||||||
|
// fn bound_think(...) { return primitive_think(ctx, instruction) }
|
||||||
|
//
|
||||||
|
// THE FLIP (when reshape's decorated primitives land — one line each):
|
||||||
|
// fn bound_think(...) { return think(ctx, instruction) } // decorated, on dharma bus
|
||||||
|
//
|
||||||
|
// Keep the stub as fallback: `bound_think` is only reached when the seam mode is
|
||||||
|
// "decorated" (SWARM_PRIMITIVE_SEAM=decorated). Until you flip these bodies AND
|
||||||
|
// set that env, the harness runs entirely on the hermetic stub.
|
||||||
|
|
||||||
|
// bound_think — decorated `think` over a worker's compiled context.
|
||||||
|
fn bound_think(ctx: String, instruction: String) -> String {
|
||||||
|
// FLIP HERE -> `return think(ctx, instruction)` once the decorated primitive lands.
|
||||||
|
return primitive_think(ctx, instruction)
|
||||||
|
}
|
||||||
|
|
||||||
|
// bound_attend — decorated retrieval over the dharma bus (falls back to the
|
||||||
|
// HTTP/engram attend today).
|
||||||
|
fn bound_attend(query: String, limit: Int) -> String {
|
||||||
|
// FLIP HERE -> `return attend(query, limit)` once decorated.
|
||||||
|
return primitive_attend(query, limit)
|
||||||
|
}
|
||||||
|
|
||||||
|
// bound_learn — decorated write onto the bus (falls back to opt-in engram write).
|
||||||
|
fn bound_learn(corr_id: String, observation: String) -> String {
|
||||||
|
// FLIP HERE -> `return learn(corr_id, observation)` once decorated.
|
||||||
|
return primitive_learn(corr_id, observation)
|
||||||
|
}
|
||||||
@@ -0,0 +1,55 @@
|
|||||||
|
// primitive_seam.el — the configurable primitive seam + telemetry.
|
||||||
|
//
|
||||||
|
// One switch selects where a worker's primitive invocation goes:
|
||||||
|
// SWARM_PRIMITIVE_SEAM=stub (default) — hermetic in-process think.
|
||||||
|
// SWARM_PRIMITIVE_SEAM=decorated — the reshape's decorated
|
||||||
|
// primitives on the dharma bus
|
||||||
|
// (see primitive_binding.el).
|
||||||
|
//
|
||||||
|
// Every seam invocation is an AFFERENT signal — a primitive call travelling
|
||||||
|
// toward the manager. The seam stamps telemetry onto each thought (seam_mode +
|
||||||
|
// one afferent tick) so the coordinator can aggregate afferent counters across
|
||||||
|
// the swarm without any shared mutable state (containment-safe: counts ride the
|
||||||
|
// vertical result path, not a shared bus register).
|
||||||
|
|
||||||
|
// seam_mode — "stub" (default) or "decorated".
|
||||||
|
fn seam_mode() -> String {
|
||||||
|
let m: String = env("SWARM_PRIMITIVE_SEAM")
|
||||||
|
if str_eq(m, "decorated") {
|
||||||
|
return "decorated"
|
||||||
|
}
|
||||||
|
return "stub"
|
||||||
|
}
|
||||||
|
|
||||||
|
// seam_think — route a worker's `think` through the configured seam and stamp
|
||||||
|
// telemetry. Returns the thought JSON augmented with:
|
||||||
|
// seam_mode : which side of the seam served this call
|
||||||
|
// afferent : "1" — one afferent primitive signal was emitted
|
||||||
|
fn seam_think(ctx: String, instruction: String) -> String {
|
||||||
|
let mode: String = seam_mode()
|
||||||
|
let thought: String = ""
|
||||||
|
if str_eq(mode, "decorated") {
|
||||||
|
let thought = bound_think(ctx, instruction)
|
||||||
|
} else {
|
||||||
|
let thought = primitive_think(ctx, instruction)
|
||||||
|
}
|
||||||
|
let t1: String = json_set_str(thought, "seam_mode", mode)
|
||||||
|
let t2: String = json_set_str(t1, "afferent", "1")
|
||||||
|
return t2
|
||||||
|
}
|
||||||
|
|
||||||
|
// seam_attend / seam_learn — same seam for the other primitives (used when a
|
||||||
|
// blueprint retrieves or writes through the bus).
|
||||||
|
fn seam_attend(query: String, limit: Int) -> String {
|
||||||
|
if str_eq(seam_mode(), "decorated") {
|
||||||
|
return bound_attend(query, limit)
|
||||||
|
}
|
||||||
|
return primitive_attend(query, limit)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn seam_learn(corr_id: String, observation: String) -> String {
|
||||||
|
if str_eq(seam_mode(), "decorated") {
|
||||||
|
return bound_learn(corr_id, observation)
|
||||||
|
}
|
||||||
|
return primitive_learn(corr_id, observation)
|
||||||
|
}
|
||||||
+34
-5
@@ -88,12 +88,17 @@ fn swarm_run_blueprint(ctx: String) -> String {
|
|||||||
return json_set_str("{}", "blueprint_status", st)
|
return json_set_str("{}", "blueprint_status", st)
|
||||||
}
|
}
|
||||||
|
|
||||||
// default (analyze_item): the CCR execution cycle think -> intend -> act.
|
// default (analyze_item): the CCR execution cycle think -> intend -> act,
|
||||||
|
// with `think` routed through the CONFIGURABLE PRIMITIVE SEAM. Telemetry
|
||||||
|
// (seam_mode + afferent tick) rides the worker's returned output.
|
||||||
let instruction: String = "process input: " + input_item
|
let instruction: String = "process input: " + input_item
|
||||||
let thought: String = primitive_think(knowledge, instruction)
|
let thought: String = seam_think(knowledge, instruction)
|
||||||
let intent: String = primitive_intend(thought)
|
let intent: String = primitive_intend(thought)
|
||||||
let effect: String = primitive_act(intent, input_item)
|
let effect: String = primitive_act(intent, input_item)
|
||||||
return json_set_str(effect, "blueprint_status", "ok")
|
let e1: String = json_set_str(effect, "blueprint_status", "ok")
|
||||||
|
let e2: String = json_set_str(e1, "seam_mode", json_get_string(thought, "seam_mode"))
|
||||||
|
let e3: String = json_set_str(e2, "afferent", json_get_string(thought, "afferent"))
|
||||||
|
return e3
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── native-thread fan-out, bounded by concurrency, order-preserving ──────────
|
// ── native-thread fan-out, bounded by concurrency, order-preserving ──────────
|
||||||
@@ -324,14 +329,28 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con
|
|||||||
// ── native-thread fan-out (bounded) ──
|
// ── native-thread fan-out (bounded) ──
|
||||||
let results: [String] = swarm_fanout("swarm_worker_entry", envelopes, concurrency)
|
let results: [String] = swarm_fanout("swarm_worker_entry", envelopes, concurrency)
|
||||||
|
|
||||||
// record per-worker terminal status
|
// record per-worker terminal status + aggregate AFFERENT telemetry.
|
||||||
|
// Afferent counters (primitive signals travelling toward the @manager)
|
||||||
|
// are summed from the vertical result path — no shared bus register,
|
||||||
|
// so the aggregation is containment-safe.
|
||||||
let succ = 0
|
let succ = 0
|
||||||
|
let afferent = 0
|
||||||
|
let seam_mode_seen: String = "stub"
|
||||||
let rn: Int = el_list_len(results)
|
let rn: Int = el_list_len(results)
|
||||||
let r = 0
|
let r = 0
|
||||||
while r < rn {
|
while r < rn {
|
||||||
let res: String = el_list_get(results, r)
|
let res: String = el_list_get(results, r)
|
||||||
let wid: String = json_get_string(res, "worker_id")
|
let wid: String = json_get_string(res, "worker_id")
|
||||||
let st: String = json_get_string(res, "status")
|
let st: String = json_get_string(res, "status")
|
||||||
|
let out: String = json_get_raw(res, "output")
|
||||||
|
let aff: Int = str_to_int(json_get_string(out, "afferent"))
|
||||||
|
let afferent = afferent + aff
|
||||||
|
let sm: String = json_get_string(out, "seam_mode")
|
||||||
|
if str_eq(sm, "") {
|
||||||
|
let seam_mode_seen = seam_mode_seen
|
||||||
|
} else {
|
||||||
|
let seam_mode_seen = sm
|
||||||
|
}
|
||||||
if str_eq(st, "completed") {
|
if str_eq(st, "completed") {
|
||||||
let succ = succ + 1
|
let succ = succ + 1
|
||||||
worktrack_append("worker.completed", corr_id, wid, json_set_str("{}", "status", "completed"))
|
worktrack_append("worker.completed", corr_id, wid, json_set_str("{}", "status", "completed"))
|
||||||
@@ -345,6 +364,15 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con
|
|||||||
let vg: String = json_set("{}", "success_count", int_to_str(succ))
|
let vg: String = json_set("{}", "success_count", int_to_str(succ))
|
||||||
worktrack_append("swarm.converging", corr_id, corr_id, vg)
|
worktrack_append("swarm.converging", corr_id, corr_id, vg)
|
||||||
|
|
||||||
|
// swarm.telemetry — afferent counters observed by the @manager.
|
||||||
|
let tkv: [String] = el_list_empty()
|
||||||
|
let tkv = el_list_append(tkv, "seam_mode")
|
||||||
|
let tkv = el_list_append(tkv, seam_mode_seen)
|
||||||
|
let telem0: String = json_build_object(tkv)
|
||||||
|
let telem1: String = json_set_str(telem0, "afferent_think", int_to_str(afferent))
|
||||||
|
let telemetry: String = json_set_str(telem1, "results_received", int_to_str(rn))
|
||||||
|
worktrack_append("swarm.telemetry", corr_id, corr_id, telemetry)
|
||||||
|
|
||||||
// ── failure threshold (Swarm §4.3), integer per-mille math ──
|
// ── failure threshold (Swarm §4.3), integer per-mille math ──
|
||||||
// require succ/n >= min_success_ratio <=> succ*1000 >= permille*n
|
// require succ/n >= min_success_ratio <=> succ*1000 >= permille*n
|
||||||
let permille: Int = ratio_to_permille(json_get_string(config_json, "min_success_ratio"))
|
let permille: Int = ratio_to_permille(json_get_string(config_json, "min_success_ratio"))
|
||||||
@@ -379,7 +407,8 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con
|
|||||||
let ok2 = el_list_append(ok2, "completed")
|
let ok2 = el_list_append(ok2, "completed")
|
||||||
let out1: String = json_build_object(ok2)
|
let out1: String = json_build_object(ok2)
|
||||||
let out2: String = json_set(out1, "report", rep2)
|
let out2: String = json_set(out1, "report", rep2)
|
||||||
return json_set(out2, "merged", merged)
|
let out3: String = json_set(out2, "merged", merged)
|
||||||
|
return json_set(out3, "telemetry", telemetry)
|
||||||
}
|
}
|
||||||
// ── denied: caller was a worker trying to open a swarm (Rule 2) ──
|
// ── denied: caller was a worker trying to open a swarm (Rule 2) ──
|
||||||
let dkv: [String] = el_list_empty()
|
let dkv: [String] = el_list_empty()
|
||||||
|
|||||||
@@ -0,0 +1,88 @@
|
|||||||
|
// 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
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user