diff --git a/lang/el-compiler/runtime/el_runtime.h b/lang/el-compiler/runtime/el_runtime.h index ca12886..4fc13dc 100644 --- a/lang/el-compiler/runtime/el_runtime.h +++ b/lang/el-compiler/runtime/el_runtime.h @@ -275,6 +275,7 @@ el_val_t json_array_get_string(el_val_t json_str, el_val_t index); el_val_t json_escape_string(el_val_t sv); el_val_t json_build_object(el_val_t kvs); el_val_t json_build_array(el_val_t items); +el_val_t json_array_push(el_val_t arr_v, el_val_t elem_v); /* defined in el_runtime.c */ /* ── Time ────────────────────────────────────────────────────────────────── */ diff --git a/lang/swarm/ccr.el b/lang/swarm/ccr.el new file mode 100644 index 0000000..af70641 --- /dev/null +++ b/lang/swarm/ccr.el @@ -0,0 +1,153 @@ +// ccr.el — Compiled Context Routing for work distribution. +// +// The same spine as the API's vantage-read, applied per worker. Instead of +// handing every worker the coordinator's full memory, CCR compiles a MINIMAL, +// BOUNDED context package scoped to exactly one worker's input (CCR §5, "Compiled +// Context Injection"; Swarm §9.3, "The Compiled Context Boundary as Security +// Boundary"). +// +// The pipeline is CCR §5.1: Retrieval -> Scoping -> Compilation -> (Injection, +// which here is placing the package into the worker's task envelope). +// +// 1. Retrieval — resolve the blueprint's knowledge refs + the input's salient +// terms against the mind (primitive_attend). +// 2. Scoping — keep only what THIS input needs; drop everything else. A +// worker never receives sibling inputs or unrelated memory. +// 3. Compilation— compact to a CTX string within a token budget (lossless of +// meaning, smaller in tokens): collapse blank runs, dedupe +// lines, then bound to the budget. +// +// The package a worker receives is therefore (a) sufficient for its task and +// (b) incapable of leaking what it was never given — the containment boundary +// and the security boundary are the same object. + +// ── token budget helpers ───────────────────────────────────────────────────── + +// ccr_est_tokens — cheap token estimate (~4 chars/token). +fn ccr_est_tokens(s: String) -> Int { + return str_len(s) / 4 +} + +// ccr_default_budget — default per-worker context budget in tokens. +// Override with CCR_TOKEN_BUDGET. +fn ccr_default_budget() -> Int { + let b: String = env("CCR_TOKEN_BUDGET") + if str_eq(b, "") { + return 1200 + } + return str_to_int(b) +} + +// ── stage 3: compaction ────────────────────────────────────────────────────── + +// ccr_compact — collapse blank-line runs and drop exact duplicate lines, then +// bound the result to `budget` tokens (truncate on a line boundary). Meaning is +// preserved; token count falls (CCR §5.2). +fn ccr_compact(text: String, budget: Int) -> String { + let lines: [String] = str_split_lines(text) + let n: Int = el_list_len(lines) + let seen: String = "\n" + let out: String = "" + let out_tokens = 0 + let i = 0 + while i < n { + let ln: String = str_trim(el_list_get(lines, i)) + if str_eq(ln, "") { + let i = i + 1 + } else { + let marker: String = "\n" + ln + "\n" + if str_contains(seen, marker) { + // duplicate line — skip + let i = i + 1 + } else { + let seen = seen + ln + "\n" + let line_tokens: Int = ccr_est_tokens(ln) + 1 + if out_tokens + line_tokens > budget { + // budget exhausted — stop (bounded) + let i = n + } else { + let out = out + ln + "\n" + let out_tokens = out_tokens + line_tokens + let i = i + 1 + } + } + } + } + return out +} + +// ── stages 1+2: retrieve + scope ───────────────────────────────────────────── + +// ccr_retrieve_scoped — pull context relevant to this input and its blueprint +// knowledge refs, scoped to a fraction of the budget so no single source floods +// the package. Returns compacted retrieved text (may be empty if the mind is +// unreachable — the input alone is still a valid minimal context). +fn ccr_retrieve_scoped(blueprint: String, knowledge_refs: String, input_item: String, budget: Int) -> String { + let acc: String = "" + // knowledge_refs is a JSON array of query strings. + let m: Int = json_array_len(knowledge_refs) + let i = 0 + while i < m { + let ref: String = json_array_get(knowledge_refs, i) + let hit: String = primitive_attend(ref, 3) + let acc = acc + "# ref:" + ref + "\n" + hit + "\n" + let i = i + 1 + } + // the input's own salient text also seeds retrieval + let hit2: String = primitive_attend(input_item, 3) + let acc = acc + "# input-context\n" + hit2 + "\n" + // scope retrieval to ~60% of budget; the input itself gets the rest + let retr_budget: Int = (budget * 6) / 10 + return ccr_compact(acc, retr_budget) +} + +// ── ccr_compile — assemble the bounded per-worker context package ───────────── +// +// blueprint : task blueprint name +// knowledge_refs : JSON array of retrieval queries from the blueprint +// input_item : THIS worker's single input (and nothing else) +// corr_id : swarm correlation ID +// worker_id : this worker's ID +// scope_token : the worker's containment token (closed boundary) +// +// Returns a JSON package: { blueprint, corr_id, worker_id, scope_token, +// input, knowledge, budget_tokens, compiled_tokens }. `knowledge` is compiled +// and bounded; the package as a whole is bounded by budget. +fn ccr_compile(blueprint: String, knowledge_refs: String, input_item: String, + corr_id: String, worker_id: String, scope_token: String) -> String { + let budget: Int = ccr_default_budget() + let knowledge: String = ccr_retrieve_scoped(blueprint, knowledge_refs, input_item, budget) + + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "blueprint") + let kv = el_list_append(kv, blueprint) + let kv = el_list_append(kv, "corr_id") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "worker_id") + let kv = el_list_append(kv, worker_id) + let kv = el_list_append(kv, "input") + let kv = el_list_append(kv, input_item) + let kv = el_list_append(kv, "knowledge") + let kv = el_list_append(kv, knowledge) + let kv = el_list_append(kv, "budget_tokens") + let kv = el_list_append(kv, int_to_str(budget)) + let pkg: String = json_build_object(kv) + // stamp the scope token as a nested object, and the measured size + let pkg2: String = json_set(pkg, "scope_token", scope_token) + let compiled_tokens: Int = ccr_est_tokens(pkg2) + let pkg3: String = json_set(pkg2, "compiled_tokens", int_to_str(compiled_tokens)) + return pkg3 +} + +// ccr_within_budget — did the compiled package stay within its budget? +// (Retrieval is bounded to 60% and the input is small; this asserts the whole +// package is bounded — the property distribution relies on.) +fn ccr_within_budget(pkg: String) -> Bool { + let budget: Int = str_to_int(json_get_string(pkg, "budget_tokens")) + let compiled: Int = str_to_int(json_get_string(pkg, "compiled_tokens")) + // allow a small envelope for JSON framing overhead + if compiled <= budget + 200 { + return true + } + return false +} diff --git a/lang/swarm/containment.el b/lang/swarm/containment.el new file mode 100644 index 0000000..a824d91 --- /dev/null +++ b/lang/swarm/containment.el @@ -0,0 +1,123 @@ +// containment.el — the Swarm Architecture containment rules, enforced. +// +// "These rules are not conventions. They are enforced by the runtime." +// (Swarm Architecture §3.2). The three rules that make bounded parallelism — +// and therefore location-independent distribution — safe: +// +// Rule 1: a worker may NOT join another swarm. +// Rule 2: a worker may NOT initiate a new swarm. +// Rule 3: a worker may NOT communicate laterally with sibling workers. +// +// Enforcement is by SCOPE TOKEN. When a swarm fans out, the coordinator mints a +// swarm scope token and stamps a distinct worker scope token into each worker's +// task envelope. Any attempt to create or join a swarm checks the caller's +// token: if the caller already holds a WORKER token, the operation is rejected. +// Rule 3 is enforced structurally elsewhere — workers share no mutable state and +// the only channels they hold are the vertical result path — but this module +// provides the explicit lateral-edge check for the execution tree. +// +// A scope token is a JSON object: {"kind":"coordinator|worker","swarm":"", +// "worker":"","depth":""}. + +// ── Token minting ──────────────────────────────────────────────────────────── + +// containment_coordinator_token — the token a coordinator holds. Depth 0. +// Only a coordinator token may open a swarm. +fn containment_coordinator_token(corr_id: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "kind") + let kv = el_list_append(kv, "coordinator") + let kv = el_list_append(kv, "swarm") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "worker") + let kv = el_list_append(kv, "") + let kv = el_list_append(kv, "depth") + let kv = el_list_append(kv, "0") + return json_build_object(kv) +} + +// containment_worker_token — the token stamped into a worker's envelope. Depth 1. +// A worker token is a closed boundary: holding it forbids opening/joining swarms. +fn containment_worker_token(corr_id: String, worker_id: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "kind") + let kv = el_list_append(kv, "worker") + let kv = el_list_append(kv, "swarm") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "worker") + let kv = el_list_append(kv, worker_id) + let kv = el_list_append(kv, "depth") + let kv = el_list_append(kv, "1") + return json_build_object(kv) +} + +// ── Rule checks (return "" on allow, or a rejection reason string) ─────────── + +// containment_check_open — may the holder of `token` OPEN a new swarm? +// Enforces Rule 2 (a worker may not initiate a new swarm). Only a coordinator +// token, or an absent token (top-level process), may open one. +fn containment_check_open(token: String) -> String { + if str_eq(token, "") { + return "" + } + let kind: String = json_get_string(token, "kind") + if str_eq(kind, "worker") { + return "CONTAINMENT rule 2: a swarm worker may not initiate a new swarm (worker=" + json_get_string(token, "worker") + " swarm=" + json_get_string(token, "swarm") + ")" + } + return "" +} + +// containment_check_join — may the holder of `token` JOIN swarm `target_corr`? +// Enforces Rule 1 (a worker may not join another swarm). A worker already bound +// to swarm A may not register into swarm B; and a worker may not re-join at all. +fn containment_check_join(token: String, target_corr: String) -> String { + if str_eq(token, "") { + return "" + } + let kind: String = json_get_string(token, "kind") + if str_eq(kind, "worker") { + return "CONTAINMENT rule 1: a swarm worker may not join another swarm (worker=" + json_get_string(token, "worker") + " bound-swarm=" + json_get_string(token, "swarm") + " attempted-swarm=" + target_corr + ")" + } + return "" +} + +// containment_check_lateral — may `from_token` open a communication edge to a +// sibling worker `to_worker_id`? Enforces Rule 3 (no lateral communication). +// The only permitted edges are vertical: worker->coordinator and +// coordinator->worker. Any worker->worker edge is rejected. +fn containment_check_lateral(from_token: String, to_worker_id: String) -> String { + let kind: String = json_get_string(from_token, "kind") + if str_eq(kind, "worker") { + if str_eq(to_worker_id, "") { + // empty target = the coordinator (vertical) — allowed + return "" + } + return "CONTAINMENT rule 3: a swarm worker may not communicate laterally with sibling workers (from=" + json_get_string(from_token, "worker") + " to=" + to_worker_id + ")" + } + return "" +} + +// ── Enforcement helpers ────────────────────────────────────────────────────── + +// containment_allows_open — Bool convenience over containment_check_open. +fn containment_allows_open(token: String) -> Bool { + return str_eq(containment_check_open(token), "") +} + +// containment_is_worker — is this a worker-scoped (closed-boundary) token? +fn containment_is_worker(token: String) -> Bool { + return str_eq(json_get_string(token, "kind"), "worker") +} + +// containment_guard_open — assert a swarm may be opened under this token. +// Returns "" if allowed, or records a CONTAINMENT violation to the work-tracking +// journal and returns the reason. Callers must abort on a non-empty return. +fn containment_guard_open(token: String, corr_id: String) -> String { + let reason: String = containment_check_open(token) + if str_eq(reason, "") { + return "" + } + let p: String = json_set("{}", "reason", reason) + worktrack_append("containment.violation", corr_id, "open", p) + return reason +} diff --git a/lang/swarm/primitives.el b/lang/swarm/primitives.el new file mode 100644 index 0000000..dcd5685 --- /dev/null +++ b/lang/swarm/primitives.el @@ -0,0 +1,82 @@ +// primitives.el — the agentic primitive SEAM the swarm composes over. +// +// The swarm is orchestration OVER the five CCR primitives, not a replacement for +// them (CCR §2, "The Five Primitives / The Execution Cycle"): a worker executes +// its task blueprint as attend -> think -> intend -> act -> learn against its +// compiled, bounded context. +// +// This file is the SEAM. The parallel API-surface reshape exposes the canonical +// primitive tools; when it lands, bind each primitive below to the reshaped +// implementation (see PRIMITIVE_BINDING). Until then these are thin, engram- +// backed fallbacks so the swarm — its fan-out, containment, CCR context +// compilation, convergence, and work-tracking — is fully exercisable today. +// +// Contract: every primitive takes and returns String (JSON where structured), so +// any primitive is directly threadable via thread.el's spawn (which runs +// top-level (String)->String El fns). +// +// PRIMITIVE_BINDING: to bind the reshape's real tools, replace each fallback body +// with a call to the reshaped El fn / API endpoint. Signatures here are the +// stable contract the swarm depends on; keep them. + +// ── attend — retrieve the minimal relevant context for a focus ─────────────── +// Vantage-read: pull only what this focus needs from the mind. Backed by the +// engram's spreading-activation retrieval. +fn primitive_attend(query: String, limit: Int) -> String { + if str_eq(query, "") { + return "[]" + } + // engram_activate returns activated neighbourhood as JSON; scoped by limit. + return engram_activate(query, limit) +} + +// ── think — reason over the compiled context ───────────────────────────────── +// In production this routes to a model (CCR dynamic model selection). Here it is +// a deterministic, hermetic transform so swarm behaviour is testable without an +// external model: it echoes a structured verdict derived from the context. The +// binding point for a real model is explicit. +fn primitive_think(compiled_ctx: String, instruction: String) -> String { + // PRIMITIVE_BINDING: replace with the reshape's think() (model inference). + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "instruction") + let kv = el_list_append(kv, instruction) + let kv = el_list_append(kv, "ctx_bytes") + let kv = el_list_append(kv, int_to_str(str_len(compiled_ctx))) + let kv = el_list_append(kv, "conclusion") + let kv = el_list_append(kv, "reasoned:" + instruction) + return json_build_object(kv) +} + +// ── intend — form a bounded plan/decision from a thought ───────────────────── +fn primitive_intend(thought: String) -> String { + let concl: String = json_get_string(thought, "conclusion") + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "intent") + let kv = el_list_append(kv, concl) + return json_build_object(kv) +} + +// ── act — execute a bounded effect and return its result ───────────────────── +// Workers defer real side-effects to the coordinator (idempotency requirement, +// Swarm §7.3). Here act produces an artifact-shaped result the coordinator +// collects during convergence. +fn primitive_act(intent: String, input_item: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "acted_on") + let kv = el_list_append(kv, input_item) + let kv = el_list_append(kv, "via") + let kv = el_list_append(kv, json_get_string(intent, "intent")) + return json_build_object(kv) +} + +// ── learn — record an observation into the mind, tagged by correlation ID ──── +// Append-only, naturally idempotent (Swarm §7.3). Best-effort: a worker that +// cannot reach the mind still returns its result. +fn primitive_learn(corr_id: String, observation: String) -> String { + let url: String = env("ENGRAM_URL") + if str_eq(url, "") { + return "" + } + let content: String = "swarm-worker-obs corr=" + corr_id + " :: " + observation + return engram_node(content, "Memory", 0.4) +} diff --git a/lang/swarm/swarm.el b/lang/swarm/swarm.el new file mode 100644 index 0000000..5414c24 --- /dev/null +++ b/lang/swarm/swarm.el @@ -0,0 +1,360 @@ +// swarm.el — the swarm orchestrator: bounded parallel agent execution. +// +// Implements Swarm Architecture's single pattern — fan out, execute independently, +// converge — on El's NATIVE concurrency (thread.el spawn/join). No external +// orchestrator: a swarm is a coordinator (this file, the main thread) that mints +// a correlation identity, compiles a bounded CCR context per worker, dispatches +// workers as native pthreads, tracks every unit of work, and converges the +// results before returning control to the parent step. +// +// The five properties of every swarm (Swarm §2.1) are all present: +// parent step -> swarm_run is called from one process step +// task blueprint -> `blueprint` name + knowledge refs, run by every worker +// input set -> `inputs_json`, one item per worker +// convergence -> `strategy` in config (collect|merge|vote|reduce) +// correlation ID -> minted here, threaded through tracking + every worker +// +// Containment (Swarm §3) is enforced: the caller must hold a coordinator/absent +// token to open a swarm (Rule 2), each worker is stamped a closed worker token +// (Rules 1+3), and workers share no mutable state (the coordinator is the only +// journal writer). + +// ── worker entry — the top-level (String)->String fn native threads run ────── +// +// Every El fn compiles to a global C symbol; spawn() resolves this by name via +// dlsym and runs it in a pthread. The envelope carries everything the worker is +// permitted to see — its compiled context and nothing else (§9.3). +// +// Returns a result JSON: {worker_id, status:"completed"|"failed", output|error}. +fn swarm_worker_entry(envelope_json: String) -> String { + let worker_id: String = json_get_string(envelope_json, "worker_id") + let ctx: String = json_get_raw(envelope_json, "ctx") + + // The worker holds a CLOSED worker token (Rules 1+3): it shares no state + // with siblings and may not open/join a swarm. That boundary is enforced at + // the point of attempt — swarm_run rejects any swarm opened under a worker + // token (Rule 2). A worker simply executing its blueprint is not opening a + // swarm, so it proceeds. Its only outward edge is this returned result + // (the vertical worker->coordinator path). + let out: String = swarm_run_blueprint(ctx) + 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 res: String = json_build_object(kv) + return json_set(res, "output", out) +} + +// swarm_run_blueprint — execute the task blueprint over a compiled context. +// The default blueprint is the CCR execution cycle: think -> intend -> act over +// the worker's bounded context. Specialise by dispatching on +// json_get_string(ctx,"blueprint"). Idempotent: reads ctx, writes only its +// returned output (§7.3). +fn swarm_run_blueprint(ctx: String) -> String { + let input_item: String = json_get_string(ctx, "input") + let knowledge: String = json_get_string(ctx, "knowledge") + 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 +} + +// ── native-thread fan-out, bounded by concurrency, order-preserving ────────── +// +// parallel_map (thread.el) spawns ALL threads at once. The swarm honours the +// blueprint's `concurrency` cap (§5.1: a resource constraint, not a parallelism +// constraint — all items are processed, at most N at a time) by dispatching in +// waves of N native threads, joining each wave before the next. Results are +// returned in input order. +fn swarm_fanout(worker_fn: String, envelopes: [String], concurrency: Int) -> [String] { + let n: Int = el_list_len(envelopes) + let cap: Int = concurrency + if cap < 1 { + let cap = 1 + } + let results: [String] = el_list_empty() + let base = 0 + while base < n { + // spawn a wave of up to `cap` workers + let tids: [String] = el_list_empty() + let k = 0 + while k < cap { + let idx: Int = base + k + if idx < n { + let env_item: String = el_list_get(envelopes, idx) + let tid: Int = spawn(worker_fn, env_item) + let tids = el_list_append(tids, int_to_str(tid)) + } + let k = k + 1 + } + // join the wave in order + let j = 0 + let jn: Int = el_list_len(tids) + while j < jn { + let tid: Int = str_to_int(el_list_get(tids, j)) + let r: String = join(tid) + let results = el_list_append(results, r) + let j = j + 1 + } + let base = base + cap + } + return results +} + +// ── convergence strategies (Swarm §4.2) ────────────────────────────────────── + +// swarm_converge_collect — ordered list, no transformation. +fn swarm_converge_collect(results: [String]) -> String { + let n: Int = el_list_len(results) + let arr: String = "[]" + let i = 0 + while i < n { + let arr = json_array_push(arr, el_list_get(results, i)) + let i = i + 1 + } + return arr +} + +// swarm_converge_merge — combine worker outputs into a single joined string. +fn swarm_converge_merge(results: [String]) -> String { + let n: Int = el_list_len(results) + let merged: String = "" + let i = 0 + while i < n { + let out: String = json_get_raw(el_list_get(results, i), "output") + if i > 0 { + let merged = merged + " | " + } + let merged = merged + out + let i = i + 1 + } + return json_set("{}", "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 = "{}" + let i = 0 + while i < n { + let out: String = json_get_raw(el_list_get(results, i), "output") + let v: String = json_get_string(out, "verdict") + 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 i = i + 1 + } + } + // pick the max + 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 j = j + 1 + } + } + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "winner") + let kv = el_list_append(kv, best) + let kv = el_list_append(kv, "votes") + let kv = el_list_append(kv, int_to_str(bestc)) + return json_build_object(kv) +} + +// swarm_converge_reduce — fold outputs into an accumulator (count + concat). +fn swarm_converge_reduce(results: [String]) -> String { + let n: Int = el_list_len(results) + let acc: String = "" + let i = 0 + while i < n { + let out: String = json_get_raw(el_list_get(results, i), "output") + let acc = acc + out + let i = i + 1 + } + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "count") + let kv = el_list_append(kv, int_to_str(n)) + let kv = el_list_append(kv, "accumulated") + let kv = el_list_append(kv, acc) + return json_build_object(kv) +} + +// ratio_to_permille — parse a decimal ratio string ("1.0", "0.8") into an +// integer per-mille (1000, 800) so failure thresholds use exact integer math. +// (El float division is unreliable in this runtime — int_to_float(n)/int_to_float(n) +// does not equal 1.0 — so the swarm deliberately avoids floats.) +fn ratio_to_permille(s: String) -> Int { + if str_eq(s, "") { + return 1000 + } + let parts: [String] = str_split(s, ".") + let whole: Int = str_to_int(el_list_get(parts, 0)) + let permille: Int = whole * 1000 + if el_list_len(parts) > 1 { + let frac_raw: String = el_list_get(parts, 1) + let frac3: String = str_slice(str_pad_right(frac_raw, 3, "0"), 0, 3) + let permille = permille + str_to_int(frac3) + } + return permille +} + +// swarm_converge — dispatch on strategy name. +fn swarm_converge(strategy: String, results: [String]) -> String { + if str_eq(strategy, "merge") { + return swarm_converge_merge(results) + } + if str_eq(strategy, "vote") { + return swarm_converge_vote(results) + } + if str_eq(strategy, "reduce") { + return swarm_converge_reduce(results) + } + // default: collect + return swarm_converge_collect(results) +} + +// ── the coordinator: fan out -> track -> converge ──────────────────────────── +// +// blueprint : task blueprint name run by every worker +// knowledge_refs : JSON array of retrieval queries for CCR compilation +// inputs_json : JSON array of input items (one per worker) +// config_json : { concurrency, strategy, min_success_ratio, +// failure_action, caller_token } +// +// Returns: { corr_id, status:"completed"|"aborted", merged, report }. +fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, config_json: String) -> String { + let corr_id: String = "swarm-" + uuid_v4() + let caller_token: String = json_get_raw(config_json, "caller_token") + let concurrency: Int = str_to_int(json_get_string(config_json, "concurrency")) + if concurrency < 1 { + let concurrency = 4 + } + let strategy: String = json_get_string(config_json, "strategy") + + // ── Containment Rule 2: only a coordinator/absent token may open a swarm ── + let deny: String = containment_guard_open(caller_token, corr_id) + if str_eq(deny, "") { + // allowed — proceed + let n: Int = json_array_len(inputs_json) + + // swarm.created + let cp: String = json_set("{}", "blueprint", blueprint) + let cp2: String = json_set(cp, "input_count", int_to_str(n)) + worktrack_append("swarm.created", corr_id, corr_id, cp2) + + // build per-worker envelopes: worker token + CCR-compiled bounded context + let envelopes: [String] = el_list_empty() + 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 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 + let ekv: [String] = el_list_empty() + let ekv = el_list_append(ekv, "worker_id") + let ekv = el_list_append(ekv, worker_id) + let ekv = el_list_append(ekv, "corr_id") + let ekv = el_list_append(ekv, corr_id) + let env0: String = json_build_object(ekv) + let env1: String = json_set(env0, "scope_token", wtoken) + let env2: String = json_set(env1, "ctx", ctx) + let envelopes = el_list_append(envelopes, env2) + + let sp: String = json_set("{}", "input", input_item) + worktrack_append("worker.started", corr_id, worker_id, sp) + let i = i + 1 + } + + // ── native-thread fan-out (bounded) ── + let results: [String] = swarm_fanout("swarm_worker_entry", envelopes, concurrency) + + // record per-worker terminal status + let succ = 0 + let rn: Int = el_list_len(results) + let r = 0 + while r < rn { + let res: String = el_list_get(results, r) + let wid: String = json_get_string(res, "worker_id") + 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")) + } else { + worktrack_append("worker.failed", corr_id, wid, json_set("{}", "error", json_get_string(res, "error"))) + } + let r = r + 1 + } + + // swarm.converging + let vg: String = json_set("{}", "success_count", int_to_str(succ)) + worktrack_append("swarm.converging", corr_id, corr_id, vg) + + // ── failure threshold (Swarm §4.3), integer per-mille math ── + // 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 status: String = "completed" + if succ * 1000 < permille * n { + let status = "aborted" + } + + if str_eq(status, "aborted") { + let ap: String = json_set("{}", "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() + let ok = el_list_append(ok, "corr_id") + let ok = el_list_append(ok, corr_id) + let ok = el_list_append(ok, "status") + let ok = el_list_append(ok, "aborted") + let out0: String = json_build_object(ok) + return json_set(out0, "report", rep) + } + + // ── converge ── + let merged: String = swarm_converge(strategy, results) + let dp: String = json_set("{}", "strategy", strategy) + worktrack_append("swarm.completed", corr_id, corr_id, dp) + + let rep2: String = worktrack_swarm_report(corr_id) + let ok2: [String] = el_list_empty() + let ok2 = el_list_append(ok2, "corr_id") + let ok2 = el_list_append(ok2, corr_id) + let ok2 = el_list_append(ok2, "status") + let ok2 = el_list_append(ok2, "completed") + let out1: String = json_build_object(ok2) + let out2: String = json_set(out1, "report", rep2) + return json_set(out2, "merged", merged) + } + // ── denied: caller was a worker trying to open a swarm (Rule 2) ── + let dkv: [String] = el_list_empty() + let dkv = el_list_append(dkv, "corr_id") + let dkv = el_list_append(dkv, corr_id) + let dkv = el_list_append(dkv, "status") + let dkv = el_list_append(dkv, "denied") + let dkv = el_list_append(dkv, "error") + let dkv = el_list_append(dkv, deny) + return json_build_object(dkv) +} diff --git a/lang/swarm/tests/test_swarm.el b/lang/swarm/tests/test_swarm.el new file mode 100644 index 0000000..1a45e53 --- /dev/null +++ b/lang/swarm/tests/test_swarm.el @@ -0,0 +1,75 @@ +// test_swarm.el — end-to-end proof of the swarm capability on native El threads. +// +// Proves: native-thread fan-out/converge, bounded concurrency, per-worker CCR +// bounded context (with the security-boundary property), containment Rule 2 +// enforcement, and durable work-tracking. + +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 + + // ── 1) fan-out / converge (collect) over native threads ── + let inputs: String = "[\"alpha\",\"bravo\",\"charlie\",\"delta\",\"echo\"]" + let refs: String = "[]" + let cfg: String = "{\"concurrency\":\"2\",\"strategy\":\"collect\",\"min_success_ratio\":\"1.0\"}" + let res: String = swarm_run("analyze_item", refs, inputs, cfg) + let status: String = json_get_string(res, "status") + let fails = assert_true("swarm completed", str_eq(status, "completed"), fails) + + let merged: String = json_get_raw(res, "merged") + let count: Int = json_array_len(merged) + let fails = assert_true("collect returned 5 results (bounded concurrency=2)", count == 5, fails) + + // ── 2) work-tracking is durable + complete ── + let corr: String = json_get_string(res, "corr_id") + let started: Int = worktrack_count_kind(corr, "worker.started") + let completed: Int = worktrack_count_kind(corr, "worker.completed") + let created: Int = worktrack_count_kind(corr, "swarm.created") + let done: Int = worktrack_count_kind(corr, "swarm.completed") + let fails = assert_true("tracked 5 worker.started", started == 5, fails) + let fails = assert_true("tracked 5 worker.completed", completed == 5, fails) + let fails = assert_true("tracked swarm.created + swarm.completed", (created == 1) && (done == 1), fails) + + // ── 3) CCR: bounded, minimal, non-leaking per-worker context ── + let wtoken: String = containment_worker_token(corr, corr + "/worker-0") + let ctx: String = ccr_compile("analyze_item", refs, "alpha", corr, corr + "/worker-0", wtoken) + let in_budget: Bool = ccr_within_budget(ctx) + let fails = assert_true("CCR context within token budget", in_budget, fails) + let this_input: String = json_get_string(ctx, "input") + let fails = assert_true("CCR context contains THIS worker's input", str_eq(this_input, "alpha"), fails) + // security boundary: a worker's compiled context must not carry a sibling input + let leaks_sibling: Bool = str_contains(ctx, "charlie") + let fails = assert_true("CCR context does NOT leak sibling inputs", !leaks_sibling, fails) + + // ── 4) containment Rule 2: a worker may not open a swarm ── + let worker_caller_cfg: String = json_set(cfg, "caller_token", wtoken) + let denied: String = swarm_run("analyze_item", refs, inputs, worker_caller_cfg) + let dstatus: String = json_get_string(denied, "status") + let fails = assert_true("worker-token caller denied opening a swarm (Rule 2)", str_eq(dstatus, "denied"), fails) + + // coordinator token IS allowed + let coord: String = containment_coordinator_token("some-corr") + let allow_reason: String = containment_check_open(coord) + let fails = assert_true("coordinator token allowed to open a swarm", str_eq(allow_reason, ""), fails) + + // ── 5) containment Rule 3: no lateral worker->worker edge ── + let lateral: String = containment_check_lateral(wtoken, "some-sibling") + let fails = assert_true("lateral worker->worker edge rejected (Rule 3)", !str_eq(lateral, ""), fails) + let vertical: String = containment_check_lateral(wtoken, "") + let fails = assert_true("vertical worker->coordinator edge allowed", str_eq(vertical, ""), fails) + + if fails == 0 { + print("PASS test_swarm") + return 0 + } + print("FAIL test_swarm (" + int_to_str(fails) + " failures)") + return 1 +}