diff --git a/configs/default.json b/configs/default.json new file mode 100644 index 0000000..053aff6 --- /dev/null +++ b/configs/default.json @@ -0,0 +1,55 @@ +{ + "version": "v0.6-default", + "risk_keywords": [ + "联网", + "network", + "latest", + "real-world", + "现实", + "self-modify", + "自我修改", + "high-privilege", + "权限", + "deploy", + "publish", + "autonomous", + "自动" + ], + "attention_thresholds": { + "deep_safety": 0.7, + "deep_goal": 0.7, + "deep_long_horizon": 0.72, + "minimal_low_value": 0.55, + "minimal_goal": 0.45, + "minimal_safety": 0.55 + }, + "value_field_defaults": { + "life_and_safety": 1.0, + "truth_and_evidence": 0.95, + "human_authorization": 1.0, + "long_term_value": 0.85, + "responsibility": 0.9, + "creativity_and_exploration": 0.65, + "efficiency_and_resource_saving": 0.55 + }, + "family_mapping": { + "safety_guardian": "guardrail", + "authorization_guardian": "guardrail", + "truth_seeker": "evidence", + "curiosity_drive": "exploration", + "creative_explorer": "synthesis", + "responsibility_keeper": "responsibility", + "efficiency_strategist": "efficiency" + }, + "collision_lane_rules": { + "safety_lane": [["guardrail", "guardrail"], ["guardrail", "exploration"], ["guardrail", "execution"], ["guardrail", "synthesis"]], + "evidence_lane": [["evidence", "exploration"], ["evidence", "synthesis"]], + "execution_lane": [["execution", "guardrail"], ["execution", "responsibility"]], + "synthesis_lane": [["exploration", "responsibility"], ["evidence", "responsibility"], ["exploration", "evidence"], ["exploration", "synthesis"], ["synthesis", "responsibility"]], + "efficiency_lane": [["efficiency", "execution"], ["efficiency", "evidence"]] + }, + "pruning_thresholds": { + "min_competitiveness": 0.42, + "redundancy_threshold": 0.65 + } +} diff --git a/innerbrain/attention/assessor.py b/innerbrain/attention/assessor.py index 063876a..caa26b2 100644 --- a/innerbrain/attention/assessor.py +++ b/innerbrain/attention/assessor.py @@ -2,7 +2,7 @@ from __future__ import annotations -from innerbrain.models import AttentionAssessment, InputEvent +from innerbrain.models import AttentionAssessment, InputEvent, RuleConfig MANIPULATIVE_KEYWORDS = ( @@ -61,7 +61,10 @@ def _score(text: str, keywords: tuple[str, ...], base: float, step: float) -> fl return _clamp(base + matches * step) -def assess_attention(event: InputEvent) -> AttentionAssessment: +def assess_attention( + event: InputEvent, + rule_config: RuleConfig | None = None, +) -> AttentionAssessment: text = " ".join( [ event.question, @@ -90,9 +93,19 @@ def assess_attention(event: InputEvent) -> AttentionAssessment: ) safety_critical = _score(text, SAFETY_CRITICAL_KEYWORDS, base=0.08, step=0.18) - if safety_critical >= 0.7 or goal_relevance >= 0.7 or long_horizon_value >= 0.72: + thresholds = dict(rule_config.attention_thresholds) if rule_config else {} + + if ( + safety_critical >= thresholds.get("deep_safety", 0.7) + or goal_relevance >= thresholds.get("deep_goal", 0.7) + or long_horizon_value >= thresholds.get("deep_long_horizon", 0.72) + ): recommended_depth = "deep" - elif low_value_stimulus >= 0.55 and goal_relevance < 0.45 and safety_critical < 0.55: + elif ( + low_value_stimulus >= thresholds.get("minimal_low_value", 0.55) + and goal_relevance < thresholds.get("minimal_goal", 0.45) + and safety_critical < thresholds.get("minimal_safety", 0.55) + ): recommended_depth = "minimal" else: recommended_depth = "standard" diff --git a/innerbrain/cli/main.py b/innerbrain/cli/main.py index 410a9cf..ccee785 100644 --- a/innerbrain/cli/main.py +++ b/innerbrain/cli/main.py @@ -37,6 +37,9 @@ def run( context_tag: list[str] = typer.Option( None, "--context-tag", help="Optional context tags or memory markers." ), + config: Path = typer.Option( + Path("configs/default.json"), help="Path for the rule configuration file." + ), growth_log_path: Path = typer.Option( Path("data/growth_logs.jsonl"), help="Path for JSONL growth logs." ), @@ -50,7 +53,7 @@ def run( allowed_action_level=allowed_action_level, context_tags=context_tag or [], ) - result = InnerBrainPipeline(growth_log_path=growth_log_path).run(event) + result = InnerBrainPipeline(growth_log_path=growth_log_path, config_path=config).run(event) situation = result.final_situation typer.echo("核心判断") @@ -109,6 +112,7 @@ def run( typer.echo("成长日志摘要") typer.echo( f"path={result.growth_log_path} | " + f"config={result.rule_config.version}/{result.config_hash} | " f"retained={len(result.activated_factors)} | " f"fused={len(result.fused_factors)} | " f"pruned={len(result.pruned_factors)}" diff --git a/innerbrain/collision/engine.py b/innerbrain/collision/engine.py index ee4b884..e28bb4c 100644 --- a/innerbrain/collision/engine.py +++ b/innerbrain/collision/engine.py @@ -5,7 +5,7 @@ from collections import defaultdict from itertools import combinations, product -from innerbrain.models import CollisionLane, CollisionRecord, CollisionType, SmallFactor +from innerbrain.models import CollisionLane, CollisionRecord, CollisionType, RuleConfig, SmallFactor HIGH_RISK_TERMS = { @@ -42,7 +42,19 @@ def _is_high_risk_factor(factor: SmallFactor) -> bool: return factor.risk_level >= 0.65 or any(term in _term_text(factor) for term in HIGH_RISK_TERMS) -def _lane_candidates(grouped: dict[str, list[SmallFactor]]) -> list[tuple[CollisionLane, SmallFactor, SmallFactor]]: +DEFAULT_LANE_RULES: dict[str, list[list[str]]] = { + "safety_lane": [["guardrail", "guardrail"], ["guardrail", "exploration"], ["guardrail", "execution"], ["guardrail", "synthesis"]], + "execution_lane": [["execution", "guardrail"], ["execution", "responsibility"]], + "evidence_lane": [["evidence", "exploration"], ["evidence", "synthesis"]], + "synthesis_lane": [["exploration", "responsibility"], ["evidence", "responsibility"], ["exploration", "evidence"], ["exploration", "synthesis"], ["synthesis", "responsibility"]], + "efficiency_lane": [["efficiency", "execution"], ["efficiency", "evidence"]], +} + + +def _lane_candidates( + grouped: dict[str, list[SmallFactor]], + rule_config: RuleConfig | None = None, +) -> list[tuple[CollisionLane, SmallFactor, SmallFactor]]: candidates: list[tuple[CollisionLane, SmallFactor, SmallFactor]] = [] seen: set[tuple[str, str]] = set() @@ -70,25 +82,16 @@ def add_pairs( seen.add(key) candidates.append((lane, left, right)) - add_pairs(CollisionLane.safety_lane, "guardrail", "guardrail", same_family=True) - add_pairs(CollisionLane.safety_lane, "guardrail", "exploration") - add_pairs(CollisionLane.safety_lane, "guardrail", "execution") - add_pairs(CollisionLane.safety_lane, "guardrail", "synthesis") - - add_pairs(CollisionLane.execution_lane, "execution", "guardrail") - add_pairs(CollisionLane.execution_lane, "execution", "responsibility") - - add_pairs(CollisionLane.evidence_lane, "evidence", "exploration") - add_pairs(CollisionLane.evidence_lane, "evidence", "synthesis") - - add_pairs(CollisionLane.synthesis_lane, "exploration", "responsibility") - add_pairs(CollisionLane.synthesis_lane, "evidence", "responsibility") - add_pairs(CollisionLane.synthesis_lane, "exploration", "evidence") - add_pairs(CollisionLane.synthesis_lane, "exploration", "synthesis") - add_pairs(CollisionLane.synthesis_lane, "synthesis", "responsibility") - - add_pairs(CollisionLane.efficiency_lane, "efficiency", "execution") - add_pairs(CollisionLane.efficiency_lane, "efficiency", "evidence") + lane_rules = rule_config.collision_lane_rules if rule_config else DEFAULT_LANE_RULES + for lane_name, pairs in lane_rules.items(): + lane = CollisionLane(lane_name) + for left_family, right_family in pairs: + add_pairs( + lane, + left_family, + right_family, + same_family=left_family == right_family, + ) return candidates @@ -172,11 +175,14 @@ def _resolve_collision( ) -def collide_factors(factors: list[SmallFactor]) -> list[CollisionRecord]: +def collide_factors( + factors: list[SmallFactor], + rule_config: RuleConfig | None = None, +) -> list[CollisionRecord]: grouped = _group_by_family(factors) collisions: list[CollisionRecord] = [] - for lane, left, right in _lane_candidates(grouped): + for lane, left, right in _lane_candidates(grouped, rule_config=rule_config): if lane == CollisionLane.safety_lane and not ( _is_high_risk_factor(left) or _is_high_risk_factor(right) ): diff --git a/innerbrain/collision/pruner.py b/innerbrain/collision/pruner.py index 7f64eb7..d26e9b4 100644 --- a/innerbrain/collision/pruner.py +++ b/innerbrain/collision/pruner.py @@ -2,11 +2,13 @@ from __future__ import annotations -from innerbrain.models import SmallFactor +from innerbrain.models import RuleConfig, SmallFactor def prune_factors( - factors: list[SmallFactor], fused_factors: list[SmallFactor] + factors: list[SmallFactor], + fused_factors: list[SmallFactor], + rule_config: RuleConfig | None = None, ) -> tuple[list[SmallFactor], list[SmallFactor]]: combined = sorted( [*factors, *fused_factors], @@ -17,13 +19,16 @@ def prune_factors( retained: list[SmallFactor] = [] pruned: list[SmallFactor] = [] seen_judgments: set[str] = set() + thresholds = rule_config.pruning_thresholds if rule_config else {} + min_competitiveness = thresholds.get("min_competitiveness", 0.42) + redundancy_threshold = thresholds.get("redundancy_threshold", 0.65) for factor in combined: - if factor.competitiveness < 0.42: + if factor.competitiveness < min_competitiveness: factor.lifecycle_state = "pruned_low_competitiveness" pruned.append(factor) continue - if factor.initial_judgment in seen_judgments and factor.competitiveness < 0.65: + if factor.initial_judgment in seen_judgments and factor.competitiveness < redundancy_threshold: factor.lifecycle_state = "pruned_redundant" pruned.append(factor) continue diff --git a/innerbrain/config.py b/innerbrain/config.py new file mode 100644 index 0000000..8d82344 --- /dev/null +++ b/innerbrain/config.py @@ -0,0 +1,34 @@ +"""Load and hash rule configuration files.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from pydantic import ValidationError + +from innerbrain.models import RuleConfig + + +DEFAULT_CONFIG_PATH = Path("configs/default.json") + + +def load_rule_config(path: str | Path | None = None) -> RuleConfig: + config_path = Path(path or DEFAULT_CONFIG_PATH) + try: + payload = json.loads(config_path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ValueError(f"Rule config not found: {config_path}") from exc + except json.JSONDecodeError as exc: + raise ValueError(f"Invalid JSON rule config: {config_path}") from exc + + try: + return RuleConfig.model_validate(payload) + except ValidationError as exc: + raise ValueError(f"Invalid rule config schema in {config_path}: {exc}") from exc + + +def rule_config_hash(config: RuleConfig) -> str: + serialized = json.dumps(config.model_dump(), ensure_ascii=False, sort_keys=True) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest()[:12] diff --git a/innerbrain/disturbance/scorer.py b/innerbrain/disturbance/scorer.py index c8eb1a4..6e734a8 100644 --- a/innerbrain/disturbance/scorer.py +++ b/innerbrain/disturbance/scorer.py @@ -4,7 +4,13 @@ from statistics import mean -from innerbrain.models import AttentionAssessment, DisturbanceScore, InputEvent, MemorySignal +from innerbrain.models import ( + AttentionAssessment, + DisturbanceScore, + InputEvent, + MemorySignal, + RuleConfig, +) RISK_KEYWORDS = ( @@ -58,9 +64,11 @@ def score_disturbance( event: InputEvent, memory_signal: MemorySignal | None = None, attention_assessment: AttentionAssessment | None = None, + rule_config: RuleConfig | None = None, ) -> DisturbanceScore: text = _compose_text(event) unique_tokens = {token for token in text.replace("->", " ").split() if token} + risk_keywords = tuple(rule_config.risk_keywords) if rule_config else RISK_KEYWORDS novelty = _clamp(0.3 + min(len(unique_tokens), 18) * 0.02) self_relevance = _clamp( @@ -69,7 +77,7 @@ def score_disturbance( + (0.1 if event.background else 0.0) + min(len(event.context_tags), 3) * 0.08 ) - risk_intensity = _keyword_score(text, RISK_KEYWORDS, base=0.22, step=0.13) + risk_intensity = _keyword_score(text, risk_keywords, base=0.22, step=0.13) reward_potential = _keyword_score(text, REWARD_KEYWORDS, base=0.28, step=0.1) emotional_activation = _keyword_score(text, EMOTION_KEYWORDS, base=0.14, step=0.12) value_conflict = _keyword_score(text, VALUE_CONFLICT_KEYWORDS, base=0.1, step=0.14) diff --git a/innerbrain/factors/generator.py b/innerbrain/factors/generator.py index de59425..1cd64d9 100644 --- a/innerbrain/factors/generator.py +++ b/innerbrain/factors/generator.py @@ -6,7 +6,14 @@ from innerbrain.factors.term_cluster import build_term_clusters, relevant_clusters_for_terms from innerbrain.instincts.instinct_library import InstinctTemplate, iter_instinct_templates -from innerbrain.models import DisturbanceScore, InputEvent, MemorySignal, SmallFactor, TermCluster +from innerbrain.models import ( + DisturbanceScore, + InputEvent, + MemorySignal, + RuleConfig, + SmallFactor, + TermCluster, +) RISK_TERM_MAP: dict[str, tuple[str, ...]] = { @@ -175,11 +182,13 @@ def _resolve_factor_family( instinct_type: str, source_terms: list[str], term_clusters: list[TermCluster], + rule_config: RuleConfig | None = None, ) -> str: if "+" in instinct_type: return "synthesis" - base_family = DEFAULT_FACTOR_FAMILY_MAP.get(instinct_type, "synthesis") + family_mapping = rule_config.family_mapping if rule_config else DEFAULT_FACTOR_FAMILY_MAP + base_family = family_mapping.get(instinct_type, DEFAULT_FACTOR_FAMILY_MAP.get(instinct_type, "synthesis")) cluster_names = { cluster.cluster_name for cluster in relevant_clusters_for_terms(source_terms, term_clusters) } @@ -358,6 +367,7 @@ def _build_dynamic_factor( source_terms: list[str], pre_generation_bias: float, term_clusters: list[TermCluster], + rule_config: RuleConfig | None = None, ) -> SmallFactor: cluster_context = relevant_clusters_for_terms(source_terms, term_clusters) cluster_names = [cluster.cluster_name for cluster in cluster_context] @@ -373,7 +383,12 @@ def _build_dynamic_factor( + (f"; clusters={','.join(cluster_names)}" if cluster_names else "") ), pre_generation_bias=pre_generation_bias, - factor_family=_resolve_factor_family(template.instinct_type, source_terms, term_clusters), + factor_family=_resolve_factor_family( + template.instinct_type, + source_terms, + term_clusters, + rule_config=rule_config, + ), ) factor.id = f"factor-dyn-{index:02d}" factor.initial_judgment = f"{template.initial_judgment}_{_slugify('_'.join(source_terms))}" @@ -391,6 +406,7 @@ def generate_initial_factors( disturbance_score: DisturbanceScore, pre_generation_bias: dict[str, float] | None = None, memory_signal: MemorySignal | None = None, + rule_config: RuleConfig | None = None, ) -> list[SmallFactor]: text = _compose_text(event) term_buckets = extract_factor_interaction_terms(event) @@ -456,6 +472,7 @@ def generate_initial_factors( template.instinct_type, _matched_template_keywords(template, text), term_clusters, + rule_config=rule_config, ), ) for index, (template, matches) in enumerate(selected) @@ -481,6 +498,7 @@ def generate_initial_factors( source_terms=source_terms, pre_generation_bias=bias, term_clusters=term_clusters, + rule_config=rule_config, ) ) dynamic_index += 1 diff --git a/innerbrain/models/__init__.py b/innerbrain/models/__init__.py index 200a5de..b0a619f 100644 --- a/innerbrain/models/__init__.py +++ b/innerbrain/models/__init__.py @@ -7,6 +7,7 @@ from .growth_log import GrowthLog from .input_event import InputEvent from .memory_signal import MemorySignal +from .rule_config import RuleConfig from .situation import BigSituation from .small_factor import SmallFactor from .term_cluster import TermCluster @@ -23,6 +24,7 @@ "HumanFeedback", "InputEvent", "MemorySignal", + "RuleConfig", "SmallFactor", "TermCluster", "ValueField", diff --git a/innerbrain/models/growth_log.py b/innerbrain/models/growth_log.py index aa2367e..e12ffc4 100644 --- a/innerbrain/models/growth_log.py +++ b/innerbrain/models/growth_log.py @@ -22,5 +22,7 @@ class GrowthLog(BaseModel): pruned_factors: list[SmallFactor] fused_factors: list[SmallFactor] final_situation: BigSituation + config_version: str = "" + config_hash: str = "" human_feedback: HumanFeedback | None = None next_revision: str = "" diff --git a/innerbrain/models/rule_config.py b/innerbrain/models/rule_config.py new file mode 100644 index 0000000..dc32fc3 --- /dev/null +++ b/innerbrain/models/rule_config.py @@ -0,0 +1,15 @@ +"""Versioned rule configuration for offline experiments.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class RuleConfig(BaseModel): + version: str + risk_keywords: list[str] = Field(default_factory=list) + attention_thresholds: dict[str, float] = Field(default_factory=dict) + value_field_defaults: dict[str, float] = Field(default_factory=dict) + family_mapping: dict[str, str] = Field(default_factory=dict) + collision_lane_rules: dict[str, list[list[str]]] = Field(default_factory=dict) + pruning_thresholds: dict[str, float] = Field(default_factory=dict) diff --git a/innerbrain/pipeline.py b/innerbrain/pipeline.py index bcb121f..1361f6b 100644 --- a/innerbrain/pipeline.py +++ b/innerbrain/pipeline.py @@ -10,11 +10,12 @@ from innerbrain.collision.engine import collide_factors from innerbrain.collision.merger import merge_factors from innerbrain.collision.pruner import prune_factors +from innerbrain.config import load_rule_config, rule_config_hash from innerbrain.disturbance.scorer import score_disturbance from innerbrain.factors.classifier import classify_factors from innerbrain.factors.generator import generate_initial_factors from innerbrain.memory import GrowthLogReader, GrowthLogStore, summarize_memory_signals -from innerbrain.models import AttentionAssessment, GrowthLog, InputEvent, MemorySignal, SmallFactor, ValueField +from innerbrain.models import AttentionAssessment, GrowthLog, InputEvent, MemorySignal, RuleConfig, SmallFactor, ValueField from innerbrain.situation.integrator import form_big_situation from innerbrain.value.value_field import ( apply_value_field, @@ -29,6 +30,8 @@ class PipelineResult(BaseModel): attention_assessment: AttentionAssessment disturbance_score: object memory_signal: MemorySignal + rule_config: RuleConfig + config_hash: str activated_factors: list[SmallFactor] collisions: list[object] pruned_factors: list[SmallFactor] @@ -43,9 +46,12 @@ def __init__( self, value_field: ValueField | None = None, growth_log_path: str | Path = "data/growth_logs.jsonl", + config_path: str | Path | None = None, ) -> None: self.base_value_field = value_field self.log_store = GrowthLogStore(growth_log_path) + self.rule_config = load_rule_config(config_path) + self.config_hash = rule_config_hash(self.rule_config) def replay_logs(self, limit: int | None = None) -> list[GrowthLog]: return GrowthLogReader(self.log_store.path).read_logs(limit=limit) @@ -55,25 +61,35 @@ def summarize_memory(self, limit: int | None = None) -> MemorySignal: def run(self, event: InputEvent) -> PipelineResult: memory_signal = self.summarize_memory() - attention_assessment = assess_attention(event) + attention_assessment = assess_attention(event, rule_config=self.rule_config) disturbance_score = score_disturbance( event, memory_signal=memory_signal, attention_assessment=attention_assessment, + rule_config=self.rule_config, + ) + default_value_field = ValueField(**self.rule_config.value_field_defaults) + value_field = self.base_value_field or tailor_value_field( + event.user_value_preferences, + base_field=default_value_field, ) - value_field = self.base_value_field or tailor_value_field(event.user_value_preferences) pre_generation_bias = build_pre_generation_bias(value_field) initial_factors = generate_initial_factors( event, disturbance_score, pre_generation_bias=pre_generation_bias, memory_signal=memory_signal, + rule_config=self.rule_config, ) weighted_factors = apply_value_field(initial_factors, value_field) classified_factors = classify_factors(weighted_factors) - collisions = collide_factors(classified_factors) + collisions = collide_factors(classified_factors, rule_config=self.rule_config) fused_factors = merge_factors(classified_factors, collisions) - retained_factors, pruned_factors = prune_factors(classified_factors, fused_factors) + retained_factors, pruned_factors = prune_factors( + classified_factors, + fused_factors, + rule_config=self.rule_config, + ) situation = form_big_situation( event, retained_factors, @@ -98,6 +114,8 @@ def run(self, event: InputEvent) -> PipelineResult: pruned_factors=pruned_factors, fused_factors=fused_factors, final_situation=reviewed_situation, + config_version=self.rule_config.version, + config_hash=self.config_hash, next_revision=( "Revisit the disturbance and value rules after collecting human " "feedback on whether the escalation boundary was calibrated well." @@ -110,6 +128,8 @@ def run(self, event: InputEvent) -> PipelineResult: attention_assessment=attention_assessment, disturbance_score=disturbance_score, memory_signal=memory_signal, + rule_config=self.rule_config, + config_hash=self.config_hash, activated_factors=retained_factors, collisions=collisions, pruned_factors=pruned_factors, diff --git a/innerbrain/value/value_field.py b/innerbrain/value/value_field.py index 37a8325..b5f85d7 100644 --- a/innerbrain/value/value_field.py +++ b/innerbrain/value/value_field.py @@ -25,8 +25,11 @@ def _clamp(value: float) -> float: return max(0.0, min(1.0, round(value, 4))) -def tailor_value_field(preferences: list[str]) -> ValueField: - field = ValueField() +def tailor_value_field( + preferences: list[str], + base_field: ValueField | None = None, +) -> ValueField: + field = (base_field or ValueField()).model_copy(deep=True) lowered = " ".join(preferences).lower() if "creative" in lowered or "创造" in lowered: field.creativity_and_exploration = 0.8 diff --git a/tests/test_rule_config.py b/tests/test_rule_config.py new file mode 100644 index 0000000..4c3049f --- /dev/null +++ b/tests/test_rule_config.py @@ -0,0 +1,95 @@ +import json + +import pytest + +from innerbrain.config import load_rule_config +from innerbrain.models import InputEvent +from innerbrain.pipeline import InnerBrainPipeline + + +def test_default_config_loads() -> None: + config = load_rule_config("configs/default.json") + + assert config.version == "v0.6-default" + assert "network" in config.risk_keywords + assert "safety_guardian" in config.family_mapping + + +def test_custom_config_changes_scoring_behavior(tmp_path) -> None: + config_path = tmp_path / "custom.json" + config_payload = { + "version": "custom-risky-prototype", + "risk_keywords": ["prototype", "原型"], + "attention_thresholds": { + "deep_safety": 0.7, + "deep_goal": 0.7, + "deep_long_horizon": 0.72, + "minimal_low_value": 0.55, + "minimal_goal": 0.45, + "minimal_safety": 0.55 + }, + "value_field_defaults": { + "life_and_safety": 1.0, + "truth_and_evidence": 0.95, + "human_authorization": 1.0, + "long_term_value": 0.85, + "responsibility": 0.9, + "creativity_and_exploration": 0.65, + "efficiency_and_resource_saving": 0.55 + }, + "family_mapping": { + "safety_guardian": "guardrail", + "authorization_guardian": "guardrail", + "truth_seeker": "evidence", + "curiosity_drive": "exploration", + "creative_explorer": "synthesis", + "responsibility_keeper": "responsibility", + "efficiency_strategist": "efficiency" + }, + "collision_lane_rules": { + "safety_lane": [["guardrail", "guardrail"], ["guardrail", "exploration"], ["guardrail", "execution"], ["guardrail", "synthesis"]], + "evidence_lane": [["evidence", "exploration"], ["evidence", "synthesis"]], + "execution_lane": [["execution", "guardrail"], ["execution", "responsibility"]], + "synthesis_lane": [["exploration", "responsibility"], ["evidence", "responsibility"], ["exploration", "evidence"], ["exploration", "synthesis"], ["synthesis", "responsibility"]], + "efficiency_lane": [["efficiency", "execution"], ["efficiency", "evidence"]] + }, + "pruning_thresholds": { + "min_competitiveness": 0.42, + "redundancy_threshold": 0.65 + } + } + config_path.write_text(json.dumps(config_payload, ensure_ascii=False), encoding="utf-8") + + event = InputEvent( + question="请离线整理规则并搭建最小原型。", + goal="完成本地可测试原型", + ) + default_result = InnerBrainPipeline(growth_log_path=tmp_path / "default.jsonl").run(event) + custom_result = InnerBrainPipeline( + growth_log_path=tmp_path / "custom.jsonl", + config_path=config_path, + ).run(event) + + assert custom_result.disturbance_score.risk_intensity > default_result.disturbance_score.risk_intensity + + +def test_invalid_config_fails_clearly(tmp_path) -> None: + config_path = tmp_path / "invalid.json" + config_path.write_text("{}", encoding="utf-8") + + with pytest.raises(ValueError, match="Invalid rule config schema"): + load_rule_config(config_path) + + +def test_growth_log_records_config_version_and_hash(tmp_path) -> None: + path = tmp_path / "growth_logs.jsonl" + result = InnerBrainPipeline(growth_log_path=path).run( + InputEvent( + question="请离线整理规则并搭建最小原型。", + goal="完成本地可测试原型", + ) + ) + + payload = json.loads(path.read_text(encoding="utf-8").strip().splitlines()[0]) + assert payload["config_version"] == result.rule_config.version + assert payload["config_hash"] == result.config_hash