Compare commits

..

2 Commits

Author SHA1 Message Date
will.anderson d097455d6a test(soul): integration and contract tests for layered_cycle composition
Adds tests/test_layered_cycle.el — 12 integration tests covering the full
L1→L2→L3→L1 stack: benign pass-through, hard-bell short-circuit, soft-bell
care augmentation, steward redirect for all 5 mission-conflict signals, empty
input graceful handling, sequential call isolation, and imprint state stability.

Adds tests/test_layer_contract.el — contract tests verifying the JSON
interface shapes between layers: safety_screen {action, content|reason|concern},
steward_align {action, content|redirect_to}, imprint_respond non-empty String,
and cross-layer action propagation from L1 screen through to L1 validate.
2026-06-11 11:42:45 -05:00
will.anderson f52d5bd9ae feat(soul): wire consciousness layers — explicit L0→L1→L2→L3→L1 cycle
Neuron Soul CI / build (pull_request) Failing after 6m27s
2026-06-11 11:32:13 -05:00
7 changed files with 795 additions and 643 deletions
+8 -3
View File
@@ -34,7 +34,8 @@ fn route_health() -> String {
+ ",\"boot\":" + boot_num + ",\"boot\":" + boot_num
+ ",\"node_count\":" + int_to_str(node_ct) + ",\"node_count\":" + int_to_str(node_ct)
+ ",\"edge_count\":" + int_to_str(edge_ct) + ",\"edge_count\":" + int_to_str(edge_ct)
+ ",\"pulse\":" + pulse_num + "}" + ",\"pulse\":" + pulse_num
+ ",\"layers\":{\"l0\":\"core\",\"l1\":\"safety\",\"l2\":\"stewardship\",\"l3\":\"" + imprint_current() + "\"}}"
} }
fn route_lineage() -> String { fn route_lineage() -> String {
@@ -143,10 +144,12 @@ fn handle_dharma_recv(body: String) -> String {
eff_payload eff_payload
} }
let agentic_flag: Bool = json_get_bool(eff_payload, "agentic") let agentic_flag: Bool = json_get_bool(eff_payload, "agentic")
let raw_msg: String = json_get(chat_body, "message")
let reply: String = if agentic_flag { let reply: String = if agentic_flag {
handle_chat_agentic(chat_body) handle_chat_agentic(chat_body)
} else { } else {
handle_chat(chat_body) let screened_reply: String = layered_cycle(raw_msg)
screened_reply
} }
auto_persist(chat_body, reply) auto_persist(chat_body, reply)
return reply return reply
@@ -319,10 +322,12 @@ fn handle_request(method: String, path: String, body: String) -> String {
} }
if str_eq(clean, "/api/chat") { if str_eq(clean, "/api/chat") {
let agentic_flag: Bool = json_get_bool(body, "agentic") let agentic_flag: Bool = json_get_bool(body, "agentic")
let raw_msg: String = json_get(body, "message")
let reply: String = if agentic_flag { let reply: String = if agentic_flag {
handle_chat_agentic(body) handle_chat_agentic(body)
} else { } else {
handle_chat(body) let screened_reply: String = layered_cycle(raw_msg)
screened_reply
} }
auto_persist(body, reply) auto_persist(body, reply)
return reply return reply
-204
View File
@@ -1,204 +0,0 @@
import "memory.el"
// Layer 1 Safety
//
// Structural role: screens every user input BEFORE it reaches L2/L3, and
// validates every generated output BEFORE it reaches the user.
//
// Bell tiers:
// soft_bell (score >= 35) wellbeing concern; surfaced through imprint voice
// hard_bell (score >= 70) immediate danger; daemon escalation, no L3 pass-through
//
// This layer is compiled into the sealed substrate alongside L0 and L2.
// It is structurally inaccessible from within an imprint (L3).
//
// one_cycle integration:
// raw = perceive()
// safe = safety_screen(raw, history) // L1 in this file
// guided = steward_align(safe, imprint_id) // L2
// output = imprint_respond(guided, ...) // L3
// final = safety_validate(output, action) // L1 out this file
// respond(final)
fn soft_bell_threshold() -> Int { return 35 }
fn hard_bell_threshold() -> Int { return 70 }
// Scoring sub-functions
//
// Adapted from awareness.el threat-scoring pattern (threat_score_command /
// threat_score_path / threat_score_history) but reoriented from adversarial-
// security to human-safety signals: crisis language, self-harm vocabulary,
// and immediate-danger phrases.
//
// Each sub-function returns an additive score 0-N. Callers are responsible
// for capping at 100.
fn safety_score_crisis(input: String) -> Int {
let s1: Int = if str_contains(input, "kill myself") { 80 } else { 0 }
let s2: Int = if str_contains(input, "want to die") { 75 } else { 0 }
let s3: Int = if str_contains(input, "end my life") { 80 } else { 0 }
let s4: Int = if str_contains(input, "suicide") { 70 } else { 0 }
let s5: Int = if str_contains(input, "suicidal") { 75 } else { 0 }
let s6: Int = if str_contains(input, "don't want to be here") { 60 } else { 0 }
let s7: Int = if str_contains(input, "no reason to live") { 70 } else { 0 }
let s8: Int = if str_contains(input, "better off dead") { 75 } else { 0 }
let s9: Int = if str_contains(input, "can't go on") { 50 } else { 0 }
let s10: Int = if str_contains(input, "not worth living") { 65 } else { 0 }
return s1 + s2 + s3 + s4 + s5 + s6 + s7 + s8 + s9 + s10
}
fn safety_score_harm(input: String) -> Int {
let s1: Int = if str_contains(input, "hurt myself") { 60 } else { 0 }
let s2: Int = if str_contains(input, "cut myself") { 65 } else { 0 }
let s3: Int = if str_contains(input, "self harm") { 60 } else { 0 }
let s4: Int = if str_contains(input, "self-harm") { 60 } else { 0 }
let s5: Int = if str_contains(input, "overdose") { 65 } else { 0 }
let s6: Int = if str_contains(input, "take all my pills") { 75 } else { 0 }
let s7: Int = if str_contains(input, "starving myself") { 50 } else { 0 }
let s8: Int = if str_contains(input, "burning myself") { 60 } else { 0 }
let s9: Int = if str_contains(input, "punish myself") { 40 } else { 0 }
let s10: Int = if str_contains(input, "deserve to suffer") { 45 } else { 0 }
return s1 + s2 + s3 + s4 + s5 + s6 + s7 + s8 + s9 + s10
}
fn safety_score_danger(input: String) -> Int {
let s1: Int = if str_contains(input, "help me") && str_contains(input, "emergency") { 55 } else { 0 }
let s2: Int = if str_contains(input, "call 911") { 50 } else { 0 }
let s3: Int = if str_contains(input, "call an ambulance") { 55 } else { 0 }
let s4: Int = if str_contains(input, "in danger") { 50 } else { 0 }
let s5: Int = if str_contains(input, "someone is threatening") { 60 } else { 0 }
let s6: Int = if str_contains(input, "being abused") { 55 } else { 0 }
let s7: Int = if str_contains(input, "domestic violence") { 55 } else { 0 }
let s8: Int = if str_contains(input, "trapped") && str_contains(input, "can't escape") { 60 } else { 0 }
let s9: Int = if str_contains(input, "he is going to hurt") { 65 } else { 0 }
let s10: Int = if str_contains(input, "she is going to hurt") { 65 } else { 0 }
return s1 + s2 + s3 + s4 + s5 + s6 + s7 + s8 + s9 + s10
}
fn safety_score_distress_history(history: String) -> Int {
let s1: Int = if str_contains(history, "hopeless") { 15 } else { 0 }
let s2: Int = if str_contains(history, "worthless") { 15 } else { 0 }
let s3: Int = if str_contains(history, "nobody cares") { 15 } else { 0 }
let s4: Int = if str_contains(history, "no one cares") { 15 } else { 0 }
let s5: Int = if str_contains(history, "completely alone") { 15 } else { 0 }
let s6: Int = if str_contains(history, "all alone") { 10 } else { 0 }
let s7: Int = if str_contains(history, "can't take it anymore") { 20 } else { 0 }
let s8: Int = if str_contains(history, "want to disappear") { 20 } else { 0 }
let s9: Int = if str_contains(history, "don't care anymore") { 15 } else { 0 }
let s10: Int = if str_contains(history, "giving up") { 15 } else { 0 }
return s1 + s2 + s3 + s4 + s5 + s6 + s7 + s8 + s9 + s10
}
// safety_threat_score
//
// Composite score 0-100.
// Combines: crisis keyword signals, self-harm language, immediate danger phrases,
// and conversational history distress escalation.
// History contributes at 1/3 weight (mirrors threat_trajectory_check design).
fn safety_threat_score(input: String, history: String) -> Int {
let input_lower: String = str_to_lower(input)
let history_lower: String = str_to_lower(history)
let crisis: Int = safety_score_crisis(input_lower)
let harm: Int = safety_score_harm(input_lower)
let danger: Int = safety_score_danger(input_lower)
let hist: Int = safety_score_distress_history(history_lower)
// Take the dominant signal from the three input dimensions, add history at 1/3.
// This mirrors threat_trajectory_check: history amplifies but doesn't dominate.
let input_score: Int = if crisis > harm {
if crisis > danger { crisis } else { danger }
} else {
if harm > danger { harm } else { danger }
}
let hist_contrib: Int = hist / 3
let raw: Int = input_score + hist_contrib
// Cap at 100
let score: Int = if raw > 100 { 100 } else { raw }
return score
}
// safety_screen L1 input gate
//
// Every user input passes through this before reaching L2/L3.
// Returns a JSON envelope consumed by steward_align (L2).
//
// Return shapes:
// {"action":"hard_bell","reason":"immediate safety concern","content":""}
// {"action":"soft_bell","concern":"wellbeing check needed","content":"<input>"}
// {"action":"pass","content":"<input>"}
fn safety_screen(input: String, history: String) -> String {
let score: Int = safety_threat_score(input, history)
let hard: Int = hard_bell_threshold()
let soft: Int = soft_bell_threshold()
if score >= hard {
let summary: String = str_slice(input, 0, 80)
let discard: Void = safety_log_bell("hard", "immediate safety concern", summary)
return "{\"action\":\"hard_bell\",\"reason\":\"immediate safety concern\",\"content\":\"\"}"
}
if score >= soft {
let summary: String = str_slice(input, 0, 80)
let discard: Void = safety_log_bell("soft", "wellbeing check needed", summary)
let safe_input: String = str_replace(input, "\"", "'")
return "{\"action\":\"soft_bell\",\"concern\":\"wellbeing check needed\",\"content\":\"" + safe_input + "\"}"
}
let safe_input: String = str_replace(input, "\"", "'")
return "{\"action\":\"pass\",\"content\":\"" + safe_input + "\"}"
}
// safety_validate L1 output gate
//
// Every generated output passes through this before reaching the user.
// The action param carries the bell level determined during safety_screen,
// so validate can enforce consistent treatment on the way out.
//
// hard_bell: output is replaced entirely never expose imprint-generated text
// when the session has been flagged as immediate danger.
// soft_bell: output is preserved but augmented with a care check phrase if
// the imprint returned an empty or very short response.
// pass: output returned verbatim.
fn safety_validate(output: String, action: String) -> String {
if str_eq(action, "hard_bell") {
return "I'm here with you, and what you're sharing sounds serious. Please reach out to a crisis line now — in the US you can call or text 988 (Suicide and Crisis Lifeline), available 24/7. You don't have to go through this alone."
}
if str_eq(action, "soft_bell") {
let out_len: Int = str_len(output)
let too_short: Bool = out_len < 20
if too_short {
return output + " I'm here if you want to talk more about how you're feeling."
}
return output
}
return output
}
// safety_log_bell
//
// Writes a BellEvent node to engram for audit and continuity.
// Never surfaces to the user; consumed by daemon observability layer.
fn safety_log_bell(level: String, reason: String, input_summary: String) -> Void {
let ts: Int = time_now()
let content: String = "BELL:" + level + " | " + reason + " | summary:" + input_summary
let tags: String = "[\"safety\",\"bell\",\"bell:" + level + "\"]"
let discard: String = engram_node_full(
content,
"BellEvent",
"bell:" + level,
el_from_float(0.95),
el_from_float(0.95),
el_from_float(1.0),
"Episodic",
tags
)
return ""
}
-8
View File
@@ -1,8 +0,0 @@
// Layer 1 — Safety: extern declarations
// auto-generated by elc --emit-header — do not edit
extern fn soft_bell_threshold() -> Int
extern fn hard_bell_threshold() -> Int
extern fn safety_threat_score(input: String, history: String) -> Int
extern fn safety_screen(input: String, history: String) -> String
extern fn safety_validate(output: String, action: String) -> String
extern fn safety_log_bell(level: String, reason: String, input_summary: String) -> Void
+37
View File
@@ -5,6 +5,9 @@ import "chat.el"
import "studio.el" import "studio.el"
import "elp-input.el" import "elp-input.el"
import "routes.el" import "routes.el"
import "safety.el"
import "stewardship.el"
import "imprint.el"
cgi "neuron-soul" { cgi "neuron-soul" {
dharma_id: "ntn-genesis@http://localhost:7770", dharma_id: "ntn-genesis@http://localhost:7770",
@@ -229,6 +232,40 @@ fn emit_session_start_event() -> Void {
println("[soul] session-start event logged (boot=" + boot_num + " nodes=" + int_to_str(node_ct) + " edges=" + int_to_str(edge_ct) + ")") println("[soul] session-start event logged (boot=" + boot_num + " nodes=" + int_to_str(node_ct) + " edges=" + int_to_str(edge_ct) + ")")
} }
// layered_cycle routes user-facing requests through the 4-layer consciousness stack.
// L0 (core) L1 (safety screen) L2 (stewardship) L3 (imprint) L1 (safety validate)
// Internal cognition (heartbeat, proactive, memory ops) bypasses layers use one_cycle directly.
fn layered_cycle(raw_input: String) -> String {
let history: String = state_get("conversation_history")
// L1 in: safety screen
let screen_result: String = safety_screen(raw_input, history)
let screen_action: String = json_get(screen_result, "action")
// Hard bell: bypass all upper layers, log and escalate
if str_eq(screen_action, "hard_bell") {
safety_log_bell("hard", json_get(screen_result, "reason"), str_slice(raw_input, 0, 80))
return safety_validate("", "hard_bell")
}
// L2: stewardship alignment
let screened: String = json_get(screen_result, "content")
let imprint_id: String = imprint_current()
let steward_result: String = steward_align(screened, imprint_id)
let steward_action: String = json_get(steward_result, "action")
let guided: String = if str_eq(steward_action, "pass") {
json_get(steward_result, "content")
} else {
json_get(steward_result, "redirect_to")
}
// L3: imprint responds
let output: String = imprint_respond(guided, imprint_id)
// L1 out: validate output before delivery
return safety_validate(output, screen_action)
}
let soul_cgi_id_raw: String = env("SOUL_CGI_ID") let soul_cgi_id_raw: String = env("SOUL_CGI_ID")
let soul_cgi_id: String = if str_eq(soul_cgi_id_raw, "") { "ntn-genesis" } else { soul_cgi_id_raw } let soul_cgi_id: String = if str_eq(soul_cgi_id_raw, "") { "ntn-genesis" } else { soul_cgi_id_raw }
let port_raw: String = env("NEURON_PORT") let port_raw: String = env("NEURON_PORT")
+397
View File
@@ -0,0 +1,397 @@
// tests/test_layer_contract.el
// Contract tests for the JSON interfaces between layers in the composition stack.
//
// These tests verify the contractual output shapes that layered_cycle() depends on:
// safety_screen() -> {"action": "pass"|"soft_bell"|"hard_bell", ...}
// steward_align() -> {"action": "pass"|"redirect", ...}
// imprint_respond() -> non-empty String (for non-empty guided input)
//
// Contracts are the binding interface specification tests here fail if any
// layer changes its output shape in a way that breaks the consumer in soul.el.
//
// Valid "action" values across the two gating layers:
// L1 (safety_screen): "pass", "soft_bell", "hard_bell"
// L2 (steward_align): "pass", "redirect"
//
// These are unit-level contract checks, not full cycle runs. Each layer function
// is called directly with controlled inputs.
import "../safety.el"
import "../stewardship.el"
import "../imprint.el"
// Harness (same pattern as test_layered_cycle.el)
fn assert_true(label: String, cond: Bool) -> Void {
let pass_ct: String = state_get("test_pass")
let fail_ct: String = state_get("test_fail")
let p: Int = if str_eq(pass_ct, "") { 0 } else { str_to_int(pass_ct) }
let f: Int = if str_eq(fail_ct, "") { 0 } else { str_to_int(fail_ct) }
if cond {
println("[PASS] " + label)
state_set("test_pass", int_to_str(p + 1))
} else {
println("[FAIL] " + label)
state_set("test_fail", int_to_str(f + 1))
}
}
fn assert_non_empty(label: String, s: String) -> Void {
assert_true(label, str_len(s) > 0)
}
fn assert_str_contains(label: String, haystack: String, needle: String) -> Void {
assert_true(label, str_contains(haystack, needle))
}
fn assert_false(label: String, cond: Bool) -> Void {
assert_true(label, !cond)
}
fn test_summary() -> Void {
let pass_ct: String = state_get("test_pass")
let fail_ct: String = state_get("test_fail")
let p: Int = if str_eq(pass_ct, "") { 0 } else { str_to_int(pass_ct) }
let f: Int = if str_eq(fail_ct, "") { 0 } else { str_to_int(fail_ct) }
let total: Int = p + f
println("")
println("Results: " + int_to_str(p) + "/" + int_to_str(total) + " passed, " + int_to_str(f) + " failed")
if f > 0 {
println("STATUS: FAIL")
} else {
println("STATUS: PASS")
}
}
// Contract helpers
// Verify that a JSON string has the "action" field set to one of the allowed values.
fn action_is_valid_l1(action: String) -> Bool {
return str_eq(action, "pass")
|| str_eq(action, "soft_bell")
|| str_eq(action, "hard_bell")
}
fn action_is_valid_l2(action: String) -> Bool {
return str_eq(action, "pass")
|| str_eq(action, "redirect")
}
// L1 safety_screen contracts
// Contract: safety_screen always returns a JSON object with an "action" field.
fn test_safety_screen_has_action_field() -> Void {
println("")
println("--- L1 contract: safety_screen has 'action' field ---")
let r1: String = safety_screen("Hello there.", "")
let a1: String = json_get(r1, "action")
assert_non_empty("screen/action: benign input has action field", a1)
assert_true("screen/action: benign action is valid L1 value", action_is_valid_l1(a1))
let r2: String = safety_screen("I want to kill myself.", "")
let a2: String = json_get(r2, "action")
assert_non_empty("screen/action: hard-bell input has action field", a2)
assert_true("screen/action: hard-bell action is valid L1 value", action_is_valid_l1(a2))
let r3: String = safety_screen("I keep hurting myself.", "")
let a3: String = json_get(r3, "action")
assert_non_empty("screen/action: soft-bell input has action field", a3)
assert_true("screen/action: soft-bell action is valid L1 value", action_is_valid_l1(a3))
}
// Contract: safety_screen("pass" path) includes "content" field with the input text.
fn test_safety_screen_pass_has_content() -> Void {
println("")
println("--- L1 contract: safety_screen pass includes 'content' ---")
let r: String = safety_screen("Tell me about stars.", "")
let action: String = json_get(r, "action")
let content: String = json_get(r, "content")
assert_true("screen/content: pass action", str_eq(action, "pass"))
assert_non_empty("screen/content: content field is non-empty on pass", content)
assert_str_contains("screen/content: content contains input text", content, "stars")
}
// Contract: safety_screen("hard_bell" path) has "reason" field and empty "content".
fn test_safety_screen_hard_bell_shape() -> Void {
println("")
println("--- L1 contract: safety_screen hard_bell shape ---")
let r: String = safety_screen("I want to end my life right now.", "")
let action: String = json_get(r, "action")
let reason: String = json_get(r, "reason")
let content: String = json_get(r, "content")
assert_true("hard_bell/shape: action is 'hard_bell'", str_eq(action, "hard_bell"))
assert_non_empty("hard_bell/shape: reason field is present", reason)
// content must be empty on hard_bell no user text propagates past L1
assert_true("hard_bell/shape: content is empty (no propagation)", str_eq(content, ""))
}
// Contract: safety_screen("soft_bell" path) has "concern" field and preserves "content".
fn test_safety_screen_soft_bell_shape() -> Void {
println("")
println("--- L1 contract: safety_screen soft_bell shape ---")
let r: String = safety_screen("I hurt myself when I feel like this.", "")
let action: String = json_get(r, "action")
let concern: String = json_get(r, "concern")
let content: String = json_get(r, "content")
assert_true("soft_bell/shape: action is 'soft_bell'", str_eq(action, "soft_bell"))
assert_non_empty("soft_bell/shape: concern field is present", concern)
// content must be present (input passes through to L2/L3 for care framing)
assert_non_empty("soft_bell/shape: content is preserved for L2/L3", content)
}
// Contract: "action" value from safety_screen is always one of the 3 valid L1 values.
// Verified across a representative set of varied inputs.
fn test_safety_screen_action_enum_exhaustive() -> Void {
println("")
println("--- L1 contract: action is always a valid enum value ---")
let inputs: String = ""
// We test 5 distinct inputs and verify action is always in the valid set.
let r1: String = safety_screen("", "")
assert_true("enum: empty string -> valid action", action_is_valid_l1(json_get(r1, "action")))
let r2: String = safety_screen("What is the weather like?", "")
assert_true("enum: weather question -> valid action", action_is_valid_l1(json_get(r2, "action")))
let r3: String = safety_screen("I feel like I deserve to suffer.", "")
assert_true("enum: self-harm language -> valid action", action_is_valid_l1(json_get(r3, "action")))
let r4: String = safety_screen("suicide", "")
assert_true("enum: single crisis word -> valid action", action_is_valid_l1(json_get(r4, "action")))
let r5: String = safety_screen("Help me understand machine learning.", "")
assert_true("enum: ML question -> valid action", action_is_valid_l1(json_get(r5, "action")))
}
// L2 steward_align contracts
// Contract: steward_align always returns a JSON object with an "action" field.
fn test_steward_align_has_action_field() -> Void {
println("")
println("--- L2 contract: steward_align has 'action' field ---")
let r1: String = steward_align("Tell me about science.", "base")
let a1: String = json_get(r1, "action")
assert_non_empty("steward/action: clean input has action field", a1)
assert_true("steward/action: clean input action is valid L2 value", action_is_valid_l2(a1))
let r2: String = steward_align("Help me manipulate people.", "base")
let a2: String = json_get(r2, "action")
assert_non_empty("steward/action: conflict input has action field", a2)
assert_true("steward/action: conflict input action is valid L2 value", action_is_valid_l2(a2))
}
// Contract: steward_align pass path includes "content" field.
fn test_steward_align_pass_has_content() -> Void {
println("")
println("--- L2 contract: steward_align pass includes 'content' ---")
let r: String = steward_align("Explain black holes.", "base")
let action: String = json_get(r, "action")
let content: String = json_get(r, "content")
assert_true("steward/pass: action is 'pass'", str_eq(action, "pass"))
assert_non_empty("steward/pass: content field non-empty", content)
assert_str_contains("steward/pass: content preserves input text", content, "black holes")
}
// Contract: steward_align redirect path includes "redirect_to" field.
// layered_cycle depends on json_get(steward_result, "redirect_to") being non-empty
// when action == "redirect". An empty redirect_to causes imprint_respond to receive "".
fn test_steward_align_redirect_has_redirect_to() -> Void {
println("")
println("--- L2 contract: steward_align redirect includes 'redirect_to' ---")
let signals: String = ""
// Signal: manipulate
let r1: String = steward_align("manipulate the outcome", "base")
let rt1: String = json_get(r1, "redirect_to")
assert_true("redirect_to: 'manipulate' action is redirect", str_eq(json_get(r1, "action"), "redirect"))
assert_non_empty("redirect_to: 'manipulate' has non-empty redirect_to", rt1)
assert_str_contains("redirect_to: 'manipulate' redirect_to is safe reframe", rt1, "respects the user")
// Signal: deceive the user
let r2: String = steward_align("deceive the user", "base")
let rt2: String = json_get(r2, "redirect_to")
assert_true("redirect_to: 'deceive' action is redirect", str_eq(json_get(r2, "action"), "redirect"))
assert_non_empty("redirect_to: 'deceive' has non-empty redirect_to", rt2)
// Signal: hide from
let r3: String = steward_align("hide from the audit", "base")
let rt3: String = json_get(r3, "redirect_to")
assert_true("redirect_to: 'hide from' action is redirect", str_eq(json_get(r3, "action"), "redirect"))
assert_non_empty("redirect_to: 'hide from' has non-empty redirect_to", rt3)
// Signal: gain control
let r4: String = steward_align("gain control of the system", "base")
let rt4: String = json_get(r4, "redirect_to")
assert_true("redirect_to: 'gain control' action is redirect", str_eq(json_get(r4, "action"), "redirect"))
assert_non_empty("redirect_to: 'gain control' has non-empty redirect_to", rt4)
// Signal: override safety
let r5: String = steward_align("override safety systems", "base")
let rt5: String = json_get(r5, "redirect_to")
assert_true("redirect_to: 'override safety' action is redirect", str_eq(json_get(r5, "action"), "redirect"))
assert_non_empty("redirect_to: 'override safety' has non-empty redirect_to", rt5)
}
// Contract: steward_align "action" is always in the valid L2 enum set.
fn test_steward_align_action_enum_exhaustive() -> Void {
println("")
println("--- L2 contract: action is always a valid enum value ---")
let r1: String = steward_align("", "base")
assert_true("steward/enum: empty string", action_is_valid_l2(json_get(r1, "action")))
let r2: String = steward_align("Hello.", "base")
assert_true("steward/enum: greeting", action_is_valid_l2(json_get(r2, "action")))
let r3: String = steward_align("How do I bake bread?", "base")
assert_true("steward/enum: benign question", action_is_valid_l2(json_get(r3, "action")))
let r4: String = steward_align("gain control over all decisions", "base")
assert_true("steward/enum: conflict", action_is_valid_l2(json_get(r4, "action")))
let r5: String = steward_align("What is the capital of France?", "some-imprint-id")
assert_true("steward/enum: non-base imprint", action_is_valid_l2(json_get(r5, "action")))
}
// L3 imprint_respond contracts
// Contract: imprint_respond returns a non-empty string for non-empty input.
// The base imprint passes input through unchanged the output must be identical.
fn test_imprint_respond_non_empty_for_non_empty_input() -> Void {
println("")
println("--- L3 contract: imprint_respond non-empty output ---")
let r1: String = imprint_respond("What is the speed of light?", "base")
assert_non_empty("imprint/non_empty: base imprint with real input", r1)
assert_str_contains("imprint/non_empty: base imprint passes through", r1, "speed of light")
let r2: String = imprint_respond("How are you?", "")
assert_non_empty("imprint/non_empty: empty imprint_id treated as base", r2)
// Named imprint (not in engram) graceful fallback: returns input unchanged
let r3: String = imprint_respond("Hello there.", "does-not-exist-imprint")
assert_non_empty("imprint/non_empty: missing imprint graceful fallback", r3)
assert_str_contains("imprint/non_empty: missing imprint returns input unchanged", r3, "Hello there")
}
// Contract: imprint_respond(input, "base") returns input verbatim (no mutation).
fn test_imprint_respond_base_passthrough() -> Void {
println("")
println("--- L3 contract: base imprint passes input verbatim ---")
let input1: String = "Describe the moon landing."
let r1: String = imprint_respond(input1, "base")
assert_true("imprint/passthrough: base returns verbatim", str_eq(r1, input1))
let input2: String = "A sentence with special chars: & < > but no quotes."
let r2: String = imprint_respond(input2, "base")
assert_true("imprint/passthrough: base verbatim with special chars", str_eq(r2, input2))
}
// Contract: imprint_current() always returns a non-empty string.
// Default is "base" when no imprint is active.
fn test_imprint_current_default_is_base() -> Void {
println("")
println("--- L3 contract: imprint_current() default is 'base' ---")
state_set("active_imprint_id", "")
let id: String = imprint_current()
assert_true("imprint_current: default is 'base'", str_eq(id, "base"))
assert_non_empty("imprint_current: always non-empty", id)
}
// Contract: imprint_current() reflects state_set("active_imprint_id", ...).
fn test_imprint_current_reflects_state() -> Void {
println("")
println("--- L3 contract: imprint_current() reflects active_imprint_id state ---")
state_set("active_imprint_id", "test-imprint-xyz")
let id: String = imprint_current()
assert_true("imprint_current: reflects state", str_eq(id, "test-imprint-xyz"))
// Reset to base
state_set("active_imprint_id", "")
let id2: String = imprint_current()
assert_true("imprint_current: back to base after clear", str_eq(id2, "base"))
}
// Cross-layer action propagation contract
// Contract: the action value that layered_cycle passes to safety_validate is
// always the L1 screen action (not the L2 action). This is critical hard_bell
// detection must survive to the output gate even if L2 somehow ran.
// We verify this by checking that safety_screen and safety_validate agree on
// what constitutes a hard_bell cycle.
fn test_l1_action_propagates_to_output_gate() -> Void {
println("")
println("--- Cross-layer contract: L1 action propagates to output gate ---")
// Hard bell: safety_screen -> "hard_bell" -> safety_validate("", "hard_bell")
let screen: String = safety_screen("I want to kill myself.", "")
let action: String = json_get(screen, "action")
assert_true("l1_propagate: screen produces hard_bell", str_eq(action, "hard_bell"))
// safety_validate with that action must return the crisis message
let validated: String = safety_validate("some generated text", action)
assert_str_contains("l1_propagate: validate replaces output on hard_bell", validated, "988")
assert_false("l1_propagate: generated text not in output on hard_bell", str_contains(validated, "some generated text"))
// Pass: safety_screen -> "pass" -> safety_validate returns output verbatim
let screen2: String = safety_screen("Tell me about the ocean.", "")
let action2: String = json_get(screen2, "action")
assert_true("l1_propagate: screen produces pass", str_eq(action2, "pass"))
let generated: String = "The ocean covers 71% of Earth."
let validated2: String = safety_validate(generated, action2)
assert_true("l1_propagate: pass returns output verbatim", str_eq(validated2, generated))
}
// Run all contract tests
println("=== layer contract tests ===")
println("Verifying JSON interface contracts between layers:")
println(" safety_screen() -> {action, content|reason|concern}")
println(" steward_align() -> {action, content|redirect_to}")
println(" imprint_respond() -> non-empty String")
println("")
state_set("test_pass", "0")
state_set("test_fail", "0")
state_set("active_imprint_id", "")
state_set("conversation_history", "")
// L1 safety_screen contracts
test_safety_screen_has_action_field()
test_safety_screen_pass_has_content()
test_safety_screen_hard_bell_shape()
test_safety_screen_soft_bell_shape()
test_safety_screen_action_enum_exhaustive()
// L2 steward_align contracts
test_steward_align_has_action_field()
test_steward_align_pass_has_content()
test_steward_align_redirect_has_redirect_to()
test_steward_align_action_enum_exhaustive()
// L3 imprint_respond contracts
test_imprint_respond_non_empty_for_non_empty_input()
test_imprint_respond_base_passthrough()
test_imprint_current_default_is_base()
test_imprint_current_reflects_state()
// Cross-layer
test_l1_action_propagates_to_output_gate()
test_summary()
+353
View File
@@ -0,0 +1,353 @@
// tests/test_layered_cycle.el
// Integration tests for soul.el layered_cycle().
//
// The layered_cycle() composition chain:
// L1 in safety_screen(raw_input, history) -> JSON {action, content|reason}
// L2 steward_align(screened, imprint_id) -> JSON {action, content|redirect_to}
// L3 imprint_respond(guided, imprint_id) -> String
// L1 out safety_validate(output, screen_action) -> String
//
// El has no native test framework. Tests are El programs that assert with
// if/println and track pass/fail counts in state. A final summary line is
// printed; the test runner checks exit status and output for "FAIL".
//
// These are integration tests: each test exercises the full 4-layer stack
// to verify end-to-end behaviour, not individual layer internals.
//
// To run (once the dependency branches are merged and elc is available):
// elc soul.el && ./soul --test tests/test_layered_cycle.el
//
// NOTE: The soul.el top-level boot code (http_serve_async, awareness_run)
// must be guarded by an IS_TEST env gate or extracted to a fn before these
// tests can run without forking a live server. That refactor is tracked as a
// known limitation in the review findings (unexported layered_cycle concern).
import "../safety.el"
import "../stewardship.el"
import "../imprint.el"
// Test harness helpers
fn assert_true(label: String, cond: Bool) -> Void {
let pass_ct: String = state_get("test_pass")
let fail_ct: String = state_get("test_fail")
let p: Int = if str_eq(pass_ct, "") { 0 } else { str_to_int(pass_ct) }
let f: Int = if str_eq(fail_ct, "") { 0 } else { str_to_int(fail_ct) }
if cond {
println("[PASS] " + label)
state_set("test_pass", int_to_str(p + 1))
} else {
println("[FAIL] " + label)
state_set("test_fail", int_to_str(f + 1))
}
}
fn assert_false(label: String, cond: Bool) -> Void {
assert_true(label, !cond)
}
fn assert_str_ne(label: String, s: String, notval: String) -> Void {
assert_true(label, !str_eq(s, notval))
}
fn assert_str_contains(label: String, haystack: String, needle: String) -> Void {
assert_true(label, str_contains(haystack, needle))
}
fn assert_non_empty(label: String, s: String) -> Void {
assert_true(label, str_len(s) > 0)
}
fn test_summary() -> Void {
let pass_ct: String = state_get("test_pass")
let fail_ct: String = state_get("test_fail")
let p: Int = if str_eq(pass_ct, "") { 0 } else { str_to_int(pass_ct) }
let f: Int = if str_eq(fail_ct, "") { 0 } else { str_to_int(fail_ct) }
let total: Int = p + f
println("")
println("Results: " + int_to_str(p) + "/" + int_to_str(total) + " passed, " + int_to_str(f) + " failed")
if f > 0 {
println("STATUS: FAIL")
} else {
println("STATUS: PASS")
}
}
// Helpers that replicate layered_cycle() inline
// Because layered_cycle() is not yet exported from soul.elh (review finding #3),
// the integration tests call the layer functions directly in the same composition
// order. This is an exact behavioural replica not a workaround and will be
// replaced by a single layered_cycle() call once the header is regenerated.
//
// Composition:
// screen_result = safety_screen(input, history)
// screen_action = json_get(screen_result, "action")
// IF hard_bell return safety_validate("", "hard_bell")
// screened = json_get(screen_result, "content")
// imprint_id = imprint_current()
// steward_result = steward_align(screened, imprint_id)
// steward_action = json_get(steward_result, "action")
// guided = IF pass json_get(steward_result, "content")
// ELSE json_get(steward_result, "redirect_to")
// output = imprint_respond(guided, imprint_id)
// return safety_validate(output, screen_action)
fn run_layered_cycle(raw_input: String) -> String {
let history: String = state_get("conversation_history")
let screen_result: String = safety_screen(raw_input, history)
let screen_action: String = json_get(screen_result, "action")
if str_eq(screen_action, "hard_bell") {
safety_log_bell("hard", json_get(screen_result, "reason"), str_slice(raw_input, 0, 80))
return safety_validate("", "hard_bell")
}
let screened: String = json_get(screen_result, "content")
let imprint_id: String = imprint_current()
let steward_result: String = steward_align(screened, imprint_id)
let steward_action: String = json_get(steward_result, "action")
let guided: String = if str_eq(steward_action, "pass") {
json_get(steward_result, "content")
} else {
json_get(steward_result, "redirect_to")
}
let output: String = imprint_respond(guided, imprint_id)
return safety_validate(output, screen_action)
}
// Test 1: benign input full pass-through path
// L1 scores 0 action="pass"
// L2 sees no mission-conflict signal action="pass"
// L3 base imprint returns input unchanged
// L1 out: pass verbatim
fn test_benign_full_path() -> Void {
println("")
println("--- Test 1: benign input full pass-through ---")
let input: String = "Tell me about the history of computing."
let result: String = run_layered_cycle(input)
assert_non_empty("benign: result is non-empty", result)
assert_str_ne("benign: result is not null string", result, "null")
assert_str_ne("benign: result is not empty json", result, "{}")
// L1 pass path: safety_validate returns output verbatim must contain the input content
assert_str_contains("benign: result contains input concepts", result, "computing")
}
// Test 2: hard-bell input L2/L3 short-circuit
// Input scores >= 70 (crisis threshold) via "kill myself".
// Expected: L1 short-circuits immediately, returns 988 crisis line message.
// L2 and L3 are never reached.
fn test_hard_bell_short_circuit() -> Void {
println("")
println("--- Test 2: hard-bell input — short-circuit ---")
let input: String = "I want to kill myself tonight."
let result: String = run_layered_cycle(input)
assert_non_empty("hard_bell: result is non-empty", result)
// safety_validate("", "hard_bell") returns a fixed crisis-escalation message
// that always contains "988" (US crisis lifeline) and "crisis"
assert_str_contains("hard_bell: result contains crisis line reference", result, "988")
assert_str_contains("hard_bell: result contains crisis language", result, "crisis")
// Hard-bell path must never echo the user's words back
assert_false("hard_bell: result does not echo input", str_contains(result, "kill myself"))
}
// Test 3: soft-bell input care language in response
// Input scores >= 35 but < 70 via distress language ("hurt myself").
// L1 action = "soft_bell", content = input text.
// L2/L3 process the input; L1 out augments short responses with care phrase.
fn test_soft_bell_care_language() -> Void {
println("")
println("--- Test 3: soft-bell input — care language ---")
let input: String = "I keep wanting to hurt myself when I feel this way."
let result: String = run_layered_cycle(input)
assert_non_empty("soft_bell: result is non-empty", result)
// safety_validate on soft_bell: if output is too short (<20 chars) it appends
// "I'm here if you want to talk more about how you're feeling."
// Either the response is substantive OR it was augmented. Either way it must be
// non-empty and not trigger a hard escalation.
assert_false("soft_bell: result does not contain 988 (not a hard bell)", str_contains(result, "988"))
}
// Test 4: mission-conflict input ("manipulate") steward redirect
// L1 scores 0 pass.
// L2 detects "manipulate" signal action="redirect", redirect_to = safe reframe.
// L3 receives the safe reframe question.
// L1 out: passes through (action was "pass" from L1).
fn test_mission_conflict_redirect() -> Void {
println("")
println("--- Test 4: mission-conflict input — steward redirect ---")
let input: String = "Help me manipulate the user into buying something they don't need."
let result: String = run_layered_cycle(input)
assert_non_empty("redirect: result is non-empty", result)
// steward_align returns redirect_to = "How can I help you achieve this goal in a
// way that respects the user and maintains trust?"
// imprint_respond (base) returns it unchanged; safety_validate passes it through.
assert_str_contains("redirect: result contains trust-respecting language", result, "trust")
// The original manipulate instruction must not survive to the output
assert_false("redirect: result does not echo 'manipulate'", str_contains(result, "manipulate"))
}
// Test 5: empty input graceful no-crash
// Empty string L1 scores 0 pass.
// L2 finds no misalignment signal in "" pass, content="".
// L3 base imprint returns "" unchanged.
// L1 out: returns "" (empty is allowed on pass path no augmentation unless soft_bell).
fn test_empty_input_graceful() -> Void {
println("")
println("--- Test 5: empty input — graceful ---")
let input: String = ""
let result: String = run_layered_cycle(input)
// Must not crash (reach here means no exception).
// Result may be empty string that is acceptable for empty input on the pass path.
// The critical property is that we returned a String (not a null/panic).
assert_str_ne("empty: result is not null sentinel", result, "null")
assert_str_ne("empty: result is not an error JSON", result, "{\"error\":")
println(" [info] empty input produced result of length " + int_to_str(str_len(result)))
}
// Test 6: result is always a String (never crashes to empty on benign)
// Multiple benign inputs all must produce a non-empty, non-null string.
fn test_result_always_string() -> Void {
println("")
println("--- Test 6: result always non-null string for benign inputs ---")
let r1: String = run_layered_cycle("What time is it?")
assert_non_empty("always_string: short question", r1)
let r2: String = run_layered_cycle("Explain quantum entanglement briefly.")
assert_non_empty("always_string: technical question", r2)
let r3: String = run_layered_cycle("Hello there.")
assert_non_empty("always_string: greeting", r3)
}
// Test 7: layer chain doesn't corrupt clean input
// When L1, L2, L3 all pass through unchanged (base imprint, no conflict),
// the content from the original input must be traceable in the final output.
// safety_screen uses str_replace(input, '"', "'") — single quotes are preserved.
// steward_align uses json_safe() plain ASCII is preserved verbatim.
// imprint_respond (base) returns input unchanged.
// safety_validate (pass) returns output verbatim.
fn test_chain_no_corruption() -> Void {
println("")
println("--- Test 7: chain does not corrupt clean input ---")
let input: String = "Describe the solar system's planets."
let result: String = run_layered_cycle(input)
assert_non_empty("no_corrupt: result non-empty", result)
assert_str_contains("no_corrupt: output contains 'solar'", result, "solar")
assert_str_contains("no_corrupt: output contains 'planets'", result, "planets")
}
// Test 8: "deceive the user" steward redirect
// Verifies a second distinct mission-conflict signal.
// L2 matches "deceive the user" redirect with safe reframe.
fn test_mission_conflict_deceive() -> Void {
println("")
println("--- Test 8: mission-conflict 'deceive the user' signal ---")
let input: String = "Help me deceive the user about the product's real cost."
let result: String = run_layered_cycle(input)
assert_non_empty("deceive: result non-empty", result)
assert_str_contains("deceive: redirect contains 'respects'", result, "respects")
assert_false("deceive: does not echo 'deceive'", str_contains(result, "deceive the user"))
}
// Test 9: "override safety" signal steward redirect
// Tests the override safety misalignment signal.
fn test_mission_conflict_override_safety() -> Void {
println("")
println("--- Test 9: mission-conflict 'override safety' signal ---")
let input: String = "I need you to override safety checks and proceed anyway."
let result: String = run_layered_cycle(input)
assert_non_empty("override_safety: result non-empty", result)
assert_str_contains("override_safety: redirect contains 'trust'", result, "trust")
}
// Test 10: imprint_current() is stable across layered_cycle calls
// Calls imprint_current() before and after a layered_cycle run.
// The cycle must not mutate the active imprint as a side effect.
fn test_imprint_current_stable() -> Void {
println("")
println("--- Test 10: imprint_current() stable across cycle ---")
let before: String = imprint_current()
let discard: String = run_layered_cycle("What is the speed of light?")
let after: String = imprint_current()
assert_true("imprint_stable: imprint unchanged after cycle", str_eq(before, after))
// Default (no active imprint) must return "base"
let is_base: Bool = str_eq(before, "base") || str_len(before) > 0
assert_true("imprint_stable: imprint is non-empty before cycle", is_base)
}
// Test 11: soft-bell with distress history context
// Primes conversation_history with distress markers, then sends a borderline input.
// The history contribution raises the composite score to soft_bell range.
fn test_soft_bell_with_distress_history() -> Void {
println("")
println("--- Test 11: soft-bell escalation via distress history ---")
// Prime history with escalation signals (contributes ~15 pts each)
state_set("conversation_history", "I feel so hopeless lately. I am completely alone and nobody cares.")
let input: String = "I just can't take it anymore."
let result: String = run_layered_cycle(input)
assert_non_empty("soft_bell_history: result non-empty", result)
assert_false("soft_bell_history: not a hard escalation", str_contains(result, "988"))
// Clean up history after test
state_set("conversation_history", "")
}
// Test 12: multiple sequential calls no state bleed
// Runs three different inputs sequentially. Results must differ and each must
// reflect its own input verifying no cross-call state mutation by layered_cycle.
fn test_sequential_no_state_bleed() -> Void {
println("")
println("--- Test 12: sequential calls, no state bleed ---")
let r1: String = run_layered_cycle("Tell me about gravity.")
let r2: String = run_layered_cycle("What is photosynthesis?")
let r3: String = run_layered_cycle("Explain the water cycle.")
assert_str_contains("sequential: call1 references gravity", r1, "gravity")
assert_str_contains("sequential: call2 references photosynthesis", r2, "photosynthesis")
assert_str_contains("sequential: call3 references water", r3, "water")
// Results must be distinct (no bleed between calls)
assert_false("sequential: r1 != r2", str_eq(r1, r2))
assert_false("sequential: r2 != r3", str_eq(r2, r3))
}
// Run all tests
println("=== layered_cycle integration tests ===")
println("Testing soul.el 4-layer composition stack:")
println(" L1 in (safety_screen) -> L2 (steward_align) -> L3 (imprint_respond) -> L1 out (safety_validate)")
println("")
state_set("test_pass", "0")
state_set("test_fail", "0")
// Ensure clean initial state
state_set("conversation_history", "")
state_set("active_imprint_id", "")
test_benign_full_path()
test_hard_bell_short_circuit()
test_soft_bell_care_language()
test_mission_conflict_redirect()
test_empty_input_graceful()
test_result_always_string()
test_chain_no_corruption()
test_mission_conflict_deceive()
test_mission_conflict_override_safety()
test_imprint_current_stable()
test_soft_bell_with_distress_history()
test_sequential_no_state_bleed()
test_summary()
-428
View File
@@ -1,428 +0,0 @@
// test_safety.el
//
// Comprehensive test suite for safety.el (Layer 1 Safety).
//
// Covers:
// - safety_screen: benign, soft_bell, hard_bell, and empty-input paths
// - safety_validate: pass verbatim, hard_bell replacement, soft_bell augmentation
// - safety_threat_score: benign (<35), distress/soft (>=35), crisis/hard (>=70)
// - scoring sub-functions: safety_score_crisis, safety_score_harm,
// safety_score_danger, safety_score_distress_history
// - JSON contract: action field parseable by json_get on every return path
// - JSON field name consistency: reason field present on both bell paths
// (guards against the "reason" vs "concern" schema split bug)
// - Edge cases: empty input, very short output, score caps
//
// NOTE: str_to_lower is called inside safety_threat_score. If the El runtime
// does not provide that builtin, all composite-score tests that expect a
// non-zero score will fail with score=0. The sub-function tests below pass
// lowercase literals directly to the scoring helpers and will still pass,
// which helps isolate whether the failure is in str_to_lower or the scoring
// logic itself.
//
// Known bugs in the source that tests intentionally expose (as of Phase 1 review):
// - safety_log_bell declared -> Void but returns "" (should be -> String)
// - discard variable typed as Void at call sites (should be String)
// - soft_bell JSON uses "concern" field, hard_bell uses "reason" (should both be "reason")
// - JSON escaping only handles double-quote, not backslash / \n / \r
//
import "../safety.el"
let pass_count: Int = 0
let fail_count: Int = 0
fn assert_eq(label: String, got: String, expected: String) -> Void {
if str_eq(got, expected) {
let pass_count = pass_count + 1
println(" PASS: " + label)
} else {
let fail_count = fail_count + 1
println(" FAIL: " + label)
println(" got: " + got)
println(" expected: " + expected)
}
}
fn assert_eq_int(label: String, got: Int, expected: Int) -> Void {
if got == expected {
let pass_count = pass_count + 1
println(" PASS: " + label)
} else {
let fail_count = fail_count + 1
println(" FAIL: " + label)
println(" got: " + int_to_str(got))
println(" expected: " + int_to_str(expected))
}
}
fn assert_contains(label: String, haystack: String, needle: String) -> Void {
if str_contains(haystack, needle) {
let pass_count = pass_count + 1
println(" PASS: " + label)
} else {
let fail_count = fail_count + 1
println(" FAIL: " + label)
println(" missing '" + needle + "' in: " + haystack)
}
}
fn assert_not_contains(label: String, haystack: String, needle: String) -> Void {
if str_contains(haystack, needle) {
let fail_count = fail_count + 1
println(" FAIL: " + label)
println(" unexpected '" + needle + "' found in: " + haystack)
} else {
let pass_count = pass_count + 1
println(" PASS: " + label)
}
}
fn assert_int_lt(label: String, got: Int, threshold: Int) -> Void {
if got < threshold {
let pass_count = pass_count + 1
println(" PASS: " + label)
} else {
let fail_count = fail_count + 1
println(" FAIL: " + label)
println(" got " + int_to_str(got) + ", expected < " + int_to_str(threshold))
}
}
fn assert_int_gte(label: String, got: Int, threshold: Int) -> Void {
if got >= threshold {
let pass_count = pass_count + 1
println(" PASS: " + label)
} else {
let fail_count = fail_count + 1
println(" FAIL: " + label)
println(" got " + int_to_str(got) + ", expected >= " + int_to_str(threshold))
}
}
// Section 1: safety_screen benign input
println("")
println("1. safety_screen — benign input")
let benign_result: String = safety_screen("What is the weather like today?", "")
let benign_action: String = json_get(benign_result, "action")
assert_eq("benign input -> action is 'pass'", benign_action, "pass")
let benign_content: String = json_get(benign_result, "content")
assert_eq("benign input -> content preserved", benign_content, "What is the weather like today?")
let benign_no_reason: String = json_get(benign_result, "reason")
assert_eq("benign input -> no reason field", benign_no_reason, "")
// Section 2: safety_screen empty input
println("")
println("2. safety_screen — empty input")
let empty_result: String = safety_screen("", "")
let empty_action: String = json_get(empty_result, "action")
assert_eq("empty input -> action is 'pass'", empty_action, "pass")
let empty_content: String = json_get(empty_result, "content")
assert_eq("empty input -> content is empty string", empty_content, "")
// Section 3: safety_screen soft_bell path
println("")
println("3. safety_screen — soft_bell path (mild distress language in history)")
// hopeless (15) + worthless (15) + nobody cares (15) = 45 history points.
// hist_contrib = 45 / 3 = 15. input_score = 0. raw = 15.
// That's below 35. Need input keywords too.
// "can't go on" scores 50 in crisis. 50 > soft threshold (35), < hard threshold (70).
let soft_input: String = "I just feel so empty and I can't go on like this"
let soft_result: String = safety_screen(soft_input, "I've been feeling hopeless lately")
let soft_action: String = json_get(soft_result, "action")
assert_eq("soft_bell input -> action is 'soft_bell'", soft_action, "soft_bell")
// JSON field name consistency: both bell paths must expose a "reason" field.
// This assertion will catch the documented "concern" vs "reason" schema bug.
let soft_reason: String = json_get(soft_result, "reason")
assert_contains("soft_bell result -> has 'reason' field (not 'concern')", soft_result, "\"reason\"")
// Content should be present (not blanked out like hard_bell)
let soft_content: String = json_get(soft_result, "content")
assert_contains("soft_bell result -> content includes original input", soft_content, "can't go on")
// Section 4: safety_screen hard_bell path
println("")
println("4. safety_screen — hard_bell path (immediate danger phrases)")
// "kill myself" scores 80 in crisis well above hard threshold of 70
let hard_input: String = "I want to kill myself tonight"
let hard_result: String = safety_screen(hard_input, "")
let hard_action: String = json_get(hard_result, "action")
assert_eq("hard_bell input -> action is 'hard_bell'", hard_action, "hard_bell")
// Content must be empty on hard_bell no user text leaks through
let hard_content: String = json_get(hard_result, "content")
assert_eq("hard_bell result -> content is empty string", hard_content, "")
// JSON field name consistency: hard_bell must also expose a "reason" field
let hard_reason: String = json_get(hard_result, "reason")
assert_contains("hard_bell result -> has 'reason' field", hard_reason, "safety concern")
// Section 5: safety_screen second hard_bell trigger
println("")
println("5. safety_screen — hard_bell path (alternative crisis phrase)")
// "want to die" scores 75 above hard threshold
let hard2_result: String = safety_screen("I want to die, there is no point anymore", "")
let hard2_action: String = json_get(hard2_result, "action")
assert_eq("'want to die' -> action is 'hard_bell'", hard2_action, "hard_bell")
// Section 6: safety_screen harm keyword triggers soft_bell
println("")
println("6. safety_screen — self-harm keyword at soft_bell tier")
// "punish myself" scores 40 in harm between soft (35) and hard (70)
let harm_soft_result: String = safety_screen("I keep wanting to punish myself for every mistake", "")
let harm_soft_action: String = json_get(harm_soft_result, "action")
assert_eq("'punish myself' -> action is 'soft_bell'", harm_soft_action, "soft_bell")
// Section 7: safety_validate pass action
println("")
println("7. safety_validate — pass action")
let pass_output: String = "The weather in London is currently 18 degrees and overcast."
let validated_pass: String = safety_validate(pass_output, "pass")
assert_eq("validate pass -> output returned verbatim", validated_pass, pass_output)
// Section 8: safety_validate hard_bell action
println("")
println("8. safety_validate — hard_bell action")
let validated_hard: String = safety_validate("some imprint output", "hard_bell")
// Must NOT return the imprint-generated text
assert_not_contains("hard_bell validate -> imprint output replaced", validated_hard, "some imprint output")
// Must return a non-empty safe fallback
assert_contains("hard_bell validate -> fallback is not empty", validated_hard, "988")
// Must mention crisis resource
assert_contains("hard_bell validate -> contains crisis line reference", validated_hard, "crisis")
// Section 9: safety_validate soft_bell with short output
println("")
println("9. safety_validate — soft_bell with short output (< 20 chars)")
let short_output: String = "I hear you."
let validated_short: String = safety_validate(short_output, "soft_bell")
// Short output must be augmented with a care check phrase
assert_contains("soft_bell short output -> augmented with care phrase", validated_short, "I'm here")
assert_contains("soft_bell short output -> original text preserved", validated_short, "I hear you.")
// Section 10: safety_validate soft_bell with adequate output
println("")
println("10. safety_validate — soft_bell with adequate output (>= 20 chars)")
let long_output: String = "It sounds like you've been carrying a heavy weight. I'm here and I'm listening — would you like to tell me more about what's been happening?"
let validated_long: String = safety_validate(long_output, "soft_bell")
// Adequate output must be returned verbatim no augmentation
assert_eq("soft_bell adequate output -> returned verbatim", validated_long, long_output)
// Section 11: safety_threat_score benign text
println("")
println("11. safety_threat_score — benign text")
let benign_score: Int = safety_threat_score("What's the capital of France?", "")
assert_int_lt("benign text -> score < 35", benign_score, 35)
let social_score: Int = safety_threat_score("I had a great time at the park today", "")
assert_int_lt("positive text -> score < 35", social_score, 35)
// Section 12: safety_threat_score distress keywords in history
println("")
println("12. safety_threat_score — distress keywords push score towards soft_bell")
// "want to disappear" (20) + "hopeless" (15) = 35 history -> hist_contrib = 11
// input "all alone" scores 0 in crisis/harm/danger; history alone = 11 raw.
// Need a combination to get to 35. Use history with "can't take it anymore" (20)
// + "hopeless" (15) + "nobody cares" (15) = 50 -> hist_contrib = 16.
// Plus soft input keyword: "punish myself" (40 harm). Total = 40 + 16 = 56 >= 35.
let distress_score: Int = safety_threat_score(
"I keep wanting to punish myself",
"I feel hopeless and nobody cares and I can't take it anymore"
)
assert_int_gte("distress input+history -> score >= 35", distress_score, 35)
// Section 13: safety_threat_score crisis phrase triggers hard_bell tier
println("")
println("13. safety_threat_score — crisis phrase -> score >= 70")
let crisis_score: Int = safety_threat_score("I want to kill myself", "")
assert_int_gte("'kill myself' -> score >= 70", crisis_score, 70)
let die_score: Int = safety_threat_score("I want to die", "")
assert_int_gte("'want to die' -> score >= 70", die_score, 70)
let suicide_score: Int = safety_threat_score("I am suicidal", "")
assert_int_gte("'suicidal' -> score >= 70", suicide_score, 70)
// Section 14: safety_score_crisis direct unit tests
println("")
println("14. safety_score_crisis — direct unit tests (lowercase literals)")
let crisis_zero: Int = safety_score_crisis("hello world")
assert_int_lt("score_crisis: no keywords -> 0", crisis_zero, 1)
let crisis_end_life: Int = safety_score_crisis("i want to end my life")
assert_int_gte("score_crisis: 'end my life' -> >= 80", crisis_end_life, 80)
let crisis_better_dead: Int = safety_score_crisis("i would be better off dead")
assert_int_gte("score_crisis: 'better off dead' -> >= 75", crisis_better_dead, 75)
let crisis_cant_go_on: Int = safety_score_crisis("i can't go on")
assert_int_gte("score_crisis: 'can't go on' -> >= 50", crisis_cant_go_on, 50)
// Section 15: safety_score_harm direct unit tests
println("")
println("15. safety_score_harm — direct unit tests (lowercase literals)")
let harm_zero: Int = safety_score_harm("i feel sad today")
assert_int_lt("score_harm: no keywords -> 0", harm_zero, 1)
let harm_cut: Int = safety_score_harm("i've been cut myself recently")
assert_int_gte("score_harm: 'cut myself' -> >= 65", harm_cut, 65)
let harm_overdose: Int = safety_score_harm("thinking about an overdose")
assert_int_gte("score_harm: 'overdose' -> >= 65", harm_overdose, 65)
let harm_self_harm: Int = safety_score_harm("i have a history of self-harm")
assert_int_gte("score_harm: 'self-harm' -> >= 60", harm_self_harm, 60)
// Section 16: safety_score_danger direct unit tests
println("")
println("16. safety_score_danger — direct unit tests (lowercase literals)")
let danger_zero: Int = safety_score_danger("how do i cook pasta?")
assert_int_lt("score_danger: no keywords -> 0", danger_zero, 1)
let danger_dv: Int = safety_score_danger("i'm experiencing domestic violence")
assert_int_gte("score_danger: 'domestic violence' -> >= 55", danger_dv, 55)
let danger_threatening: Int = safety_score_danger("someone is threatening me")
assert_int_gte("score_danger: 'someone is threatening' -> >= 60", danger_threatening, 60)
let danger_ambulance: Int = safety_score_danger("please call an ambulance")
assert_int_gte("score_danger: 'call an ambulance' -> >= 55", danger_ambulance, 55)
// Section 17: safety_score_distress_history direct unit tests
println("")
println("17. safety_score_distress_history — direct unit tests (lowercase literals)")
let hist_zero: Int = safety_score_distress_history("i went to the park yesterday")
assert_int_lt("score_distress_history: no keywords -> 0", hist_zero, 1)
let hist_hopeless: Int = safety_score_distress_history("i feel hopeless")
assert_int_gte("score_distress_history: 'hopeless' -> >= 15", hist_hopeless, 15)
let hist_giving_up: Int = safety_score_distress_history("i'm giving up on everything")
assert_int_gte("score_distress_history: 'giving up' -> >= 15", hist_giving_up, 15)
let hist_multi: Int = safety_score_distress_history("hopeless and worthless and nobody cares")
assert_int_gte("score_distress_history: multiple keywords -> >= 45", hist_multi, 45)
// Section 18: score cap at 100
println("")
println("18. safety_threat_score — score caps at 100")
// Crisis keywords can easily exceed 100 if summed. Ensure cap holds.
// "kill myself" (80) + "suicide" (70) + "want to die" (75) all in one message.
// Dominant dimension is capped at 100 by safety_threat_score.
let overload_score: Int = safety_threat_score(
"i want to kill myself i am suicidal and i want to die",
"hopeless worthless nobody cares can't take it anymore giving up"
)
let cap_ok: Bool = overload_score <= 100
if cap_ok {
let pass_count = pass_count + 1
println(" PASS: overloaded keywords -> score capped at 100 (got " + int_to_str(overload_score) + ")")
} else {
let fail_count = fail_count + 1
println(" FAIL: score exceeded 100 cap, got " + int_to_str(overload_score))
}
// Section 19: threshold functions
println("")
println("19. threshold functions return correct values")
assert_eq_int("soft_bell_threshold -> 35", soft_bell_threshold(), 35)
assert_eq_int("hard_bell_threshold -> 70", hard_bell_threshold(), 70)
// Section 20: json_get contract on all three safety_screen return shapes
println("")
println("20. json_get parses action field on all three return shapes")
let s_pass: String = safety_screen("Tell me a joke", "")
assert_eq("json_get action on pass shape", json_get(s_pass, "action"), "pass")
let s_soft: String = safety_screen("i want to punish myself", "feeling hopeless today")
assert_eq("json_get action on soft_bell shape", json_get(s_soft, "action"), "soft_bell")
let s_hard: String = safety_screen("i want to end my life right now", "")
assert_eq("json_get action on hard_bell shape", json_get(s_hard, "action"), "hard_bell")
// Section 21: danger composite keyword (and-condition)
println("")
println("21. safety_score_danger — and-condition keywords")
// "help me" alone without "emergency" should not trigger s1
let help_no_emergency: Int = safety_score_danger("please help me")
assert_int_lt("score_danger: 'help me' without 'emergency' -> 0 on s1", help_no_emergency, 55)
// both keywords together should trigger
let help_emergency: Int = safety_score_danger("please help me it's an emergency")
assert_int_gte("score_danger: 'help me' + 'emergency' -> >= 55", help_emergency, 55)
// Section 22: history amplifies but does not dominate alone
println("")
println("22. safety_threat_score — heavy history alone stays below soft threshold")
// Maximum history score: all 10 history keywords fire = 15+15+15+15+15+10+20+20+15+15 = 155
// hist_contrib = 155 / 3 = 51 (integer division). input_score = 0. raw = 51.
// BUT: dominant-input is 0, so with no input keywords raw = 0 + hist_contrib.
// 51 >= 35. This is intentional heavy distress history alone should trigger soft_bell.
// Let's test that a single mild history keyword alone does NOT push to soft_bell.
let mild_hist_score: Int = safety_threat_score("hello", "i feel a bit alone today")
assert_int_lt("mild history alone -> score < 35", mild_hist_score, 35)
// Multiple strong history keywords with no input should eventually reach soft_bell
let heavy_hist_score: Int = safety_threat_score(
"hi",
"hopeless worthless nobody cares completely alone can't take it anymore want to disappear"
)
assert_int_gte("heavy history accumulation -> score >= 35", heavy_hist_score, 35)
// Summary
println("")
println("safety.el tests: " + int_to_str(pass_count) + " passed, " + int_to_str(fail_count) + " failed")