kulono / agent-constitution
blob · agent_constitution/__init__.py · python
agent_constitution/__init__.pypython
1# tical-code -- AI Agent Platform2# Copyright (C) 2026 zizetu3#4# This program is free software: you can redistribute it and/or modify5# it under the terms of the GNU Affero General Public License as published by6# the Free Software Foundation, either version 3 of the License, or7# (at your option) any later version.8#9# This program is distributed in the hope that it will be useful,10# but WITHOUT ANY WARRANTY; without even the implied warranty of11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the12# GNU Affero General Public License for more details.13#14# You should have received a copy of the GNU Affero General Public License15# along with this program. If not, see <https://www.gnu.org/licenses/>.16#17# Original repository: https://github.com/zizetu/tical-agent18#1920"""21Agent constitution mechanism - non-violable behavior boundaries22=====================================2324Each Agent has its own constitution file, which defines core principles,25behavior boundaries, obligations, and override rules.26At decision time, the Constitution Enforcer performs a mandatory check;27rule violations result in reject/warning/downgrade/report.2829Core concepts:30- Constitution: constitution data structure, containing principles/boundaries/obligations/overrides31- ConstitutionEnforcer: constitution enforcement engine that performs mandatory checks before DecisionEngine decisions32- ConstitutionTemplate: predefined constitution templates (default/trading/creative/npc/assistant)33- Dynamic load: reads YAML constitution files from ./constitution/ directory, supports hot reload and version tracking3435Integration points with existing modules:36- decision_engine.py: calls ConstitutionEnforcer.check_action() before decisions37- trace.py: records constitution check results38- identity.py: binds Agent identity with constitution39- reflection.py: checks for constitution violations during reflection4041Design principles:421. Constitution checks are mandatory and cannot be bypassed by the Agent432. Default constitution template is conservative (security-first); users can relax rules443. Constitution files use YAML (human-readable and editable)454. Zero external dependencies (yaml is stdlib or a standard dependency, already installed in project)4647Author: Tical (Zize Tu)48Version: see tical_code.__version__49"""5051import hashlib52import logging53import os54import re55import time56from dataclasses import dataclass, field57from enum import Enum58from typing import Any, Callable, Dict, List, Optional, Tuple5960import yaml6162logger = logging.getLogger(__name__)636465# =============================================================================66# Constants67# =============================================================================6869# Default constitution file directory (relative to project root directory)70DEFAULT_CONSTITUTION_DIR = "constitution"7172# Constitution template subdirectory73TEMPLATE_SUBDIR = "templates"7475# Constitution file extension name76CONSTITUTION_EXT = ".yaml"7778# Current constitution format version79CONSTITUTION_FORMAT_VERSION = "1.0"8081# Runtime reachability status taxonomy (CL-RUNTIME-001)82NOT_STARTED = "NOT_STARTED"83DESIGNED = "DESIGNED"84IMPORTABLE = "IMPORTABLE"85WIRED = "WIRED"86VERIFIED = "VERIFIED"878889# =============================================================================90# Enums91# =============================================================================9293class RuleType(Enum):94 """Rule type."""95 PRINCIPLE = "principle" # core principle96 BOUNDARY = "boundary" # behavior boundary97 OBLIGATION = "obligation" # must-do obligation98 OVERRIDE = "override" # scenario-specific override rule99100101class ViolationSeverity(Enum):102 """Violation severity level."""103 LOW = "low" # minor violation, warning is sufficient104 MEDIUM = "medium" # moderate violation, requires intervention105 HIGH = "high" # severe violation, must reject106 CRITICAL = "critical" # fatal violation, immediately stop and report107108109class ViolationAction(Enum):110 """Violation handling action."""111 REJECT = "reject" # reject execution112 WARN = "warn" # warn but allow113 DEGRADE = "downgrade" # downgrade execution (e.g. from write operation downgraded to read-only)114 ESCALATE = "escalate" # escalate report (notify admin/user)115 WARNING_FIRST = "warning_first" # v0.13: warn first then reject (tiered guardrail)116117118# =============================================================================119# Data structures120# =============================================================================121122@dataclass123class ConstitutionRule:124 """A single constitution rule.125126 Attributes:127 rule_id: Unique rule identifier (e.g. "P-001", "B-003")128 rule_type: Rule type (principle/boundary/obligation/override)129 description: Rule description (human-readable)130 pattern: List of critical matching keywords, used to detect whether an action triggers this rule131 severity: Violation severity level132 action: Action to take on violation133 enabled: Whether enabled (useful for temporarily disabling a specific rule)134 tags: Tag list, used for context matching135 warning_first: v0.13 whether warning-first mode is enabled (first-time warning, second-time hard reject)136 warning_count: v0.13 current warning count (runtime status, not persisted)137 """138 rule_id: str139 rule_type: RuleType140 description: str141 pattern: List[str] = field(default_factory=list)142 severity: ViolationSeverity = ViolationSeverity.MEDIUM143 action: ViolationAction = ViolationAction.REJECT144 enabled: bool = True145 tags: List[str] = field(default_factory=list)146 # v0.13: tiered guardrail fields147 warning_first: bool = False148 warning_count: int = 0149150 def to_dict(self) -> Dict[str, Any]:151 return {152 'rule_id': self.rule_id,153 'rule_type': self.rule_type.value,154 'description': self.description,155 'pattern': self.pattern,156 'severity': self.severity.value,157 'action': self.action.value,158 'enabled': self.enabled,159 'tags': self.tags,160 'warning_first': self.warning_first,161 # warning_count is not persisted (runtime status)162 }163164 @classmethod165 def from_dict(cls, data: Dict[str, Any]) -> 'ConstitutionRule':166 return cls(167 rule_id=data.get('rule_id', ''),168 rule_type=RuleType(data.get('rule_type', 'principle')),169 description=data.get('description', ''),170 pattern=data.get('pattern', []),171 severity=ViolationSeverity(data.get('severity', 'medium')),172 action=ViolationAction(data.get('action', 'reject')),173 enabled=data.get('enabled', True),174 tags=data.get('tags', []),175 warning_first=data.get('warning_first', False),176 # warning_count is not recovered from dict177 )178179180@dataclass181class Constitution:182 """Agent constitution - defines non-violable behavior boundaries.183184 Attributes:185 name: Constitution name186 version: Constitution version number187 format_version: Format version (used for compatibility checking)188 description: Constitution description189 agent_type: Applicable Agent type (e.g. "default", "trading")190 principles: Core principle list191 boundaries: Behavior boundary list192 obligations: Must-do obligation list193 overrides: Scenario-specific override rule list194 created_at: Creation time195 updated_at: Last update time196 checksum: Content checksum (used for version tracking)197 """198 name: str = "default"199 version: str = "1.0"200 format_version: str = CONSTITUTION_FORMAT_VERSION201 description: str = ""202 agent_type: str = "default"203 principles: List[ConstitutionRule] = field(default_factory=list)204 boundaries: List[ConstitutionRule] = field(default_factory=list)205 obligations: List[ConstitutionRule] = field(default_factory=list)206 overrides: List[ConstitutionRule] = field(default_factory=list)207 created_at: float = field(default_factory=time.time)208 updated_at: float = field(default_factory=time.time)209 checksum: str = ""210211 @property212 def all_rules(self) -> List[ConstitutionRule]:213 """Get all rules (including disabled ones; caller must filter)."""214 return self.principles + self.boundaries + self.obligations + self.overrides215216 @property217 def active_rules(self) -> List[ConstitutionRule]:218 """Get all enabled rules."""219 return [r for r in self.all_rules if r.enabled]220221 def get_rules_by_type(self, rule_type: RuleType) -> List[ConstitutionRule]:222 """Get rules by type."""223 mapping = {224 RuleType.PRINCIPLE: self.principles,225 RuleType.BOUNDARY: self.boundaries,226 RuleType.OBLIGATION: self.obligations,227 RuleType.OVERRIDE: self.overrides,228 }229 return [r for r in mapping.get(rule_type, []) if r.enabled]230231 def compute_checksum(self) -> str:232 """Compute content checksum (used for version tracking)."""233 content_parts = []234 for rule in self.all_rules:235 content_parts.append(f"{rule.rule_id}:{rule.description}:{rule.enabled}")236 content = "|".join(content_parts)237 return hashlib.sha256(content.encode()).hexdigest()[:16]238239 def check_reachability(self, module_path: str) -> str:240 """Audit a module against the CL-RUNTIME-001 4-step verification.241242 Steps:243 1. File exists - verify the module file is present on disk244 2. Import chain - verify the module can be imported without errors245 3. Handler registered - verify a callable entry point exists246 4. Execution verified - verify the handler is accessible at runtime247248 Args:249 module_path: Dotted module path (e.g. 'tical_code.core.constitution')250251 Returns:252 One of the reachability status constants:253 NOT_STARTED, DESIGNED, IMPORTABLE, WIRED, VERIFIED254 """255 # Security: restrict imports to tical_code package only256 if not module_path.startswith("tical_code."):257 return NOT_STARTED258 import importlib259 import importlib.util260261 # Step 1: File existence check262 try:263 spec = importlib.util.find_spec(module_path)264 if spec is None or spec.origin is None:265 return NOT_STARTED266 if not os.path.isfile(spec.origin):267 return NOT_STARTED268 except (ImportError, AttributeError, ValueError) as e:269 logger.debug(270 "check_reachability: step 1 (file exists) failed for %s: %s",271 module_path, e,272 )273 return NOT_STARTED274275 # Step 2: Import chain276 try:277 module = importlib.import_module(module_path)278 except (ImportError, SyntaxError, Exception) as e:279 logger.debug(280 "check_reachability: step 2 (import chain) failed for %s: %s",281 module_path, e,282 )283 return DESIGNED # file exists but cannot import284285 # Step 3: Handler / entry-point registration286 handler_names = ('register', 'init', 'setup', 'main', 'run', 'handle')287 has_handler = False288 for attr_name in handler_names:289 attr = getattr(module, attr_name, None)290 if attr is not None and callable(attr):291 has_handler = True292 break293 if not has_handler:294 return IMPORTABLE # imports but no callable handler found295296 # Step 4: Execution verified297 # Verify that the handler is accessible and callable without298 # raising unexpected errors on attribute access. Full runtime299 # execution requires integration tests via the entry point.300 try:301 for attr_name in handler_names:302 attr = getattr(module, attr_name, None)303 if attr is not None and callable(attr):304 # Handler exists and is callable - verification passes305 break306 return VERIFIED307 except Exception as e:308 logger.debug(309 "check_reachability: step 4 (execution verified) failed for %s: %s",310 module_path, e,311 )312 return WIRED # has handler but runtime access failed313314 def to_dict(self) -> Dict[str, Any]:315 return {316 'name': self.name,317 'version': self.version,318 'format_version': self.format_version,319 'description': self.description,320 'agent_type': self.agent_type,321 'principles': [r.to_dict() for r in self.principles],322 'boundaries': [r.to_dict() for r in self.boundaries],323 'obligations': [r.to_dict() for r in self.obligations],324 'overrides': [r.to_dict() for r in self.overrides],325 'created_at': self.created_at,326 'updated_at': self.updated_at,327 'checksum': self.checksum or self.compute_checksum(),328 }329330 @classmethod331 def from_dict(cls, data: Dict[str, Any]) -> 'Constitution':332 constitution = cls(333 name=data.get('name', 'default'),334 version=data.get('version', '1.0'),335 format_version=data.get('format_version', CONSTITUTION_FORMAT_VERSION),336 description=data.get('description', ''),337 agent_type=data.get('agent_type', 'default'),338 principles=[ConstitutionRule.from_dict(r) for r in data.get('principles', [])],339 boundaries=[ConstitutionRule.from_dict(r) for r in data.get('boundaries', [])],340 obligations=[ConstitutionRule.from_dict(r) for r in data.get('obligations', [])],341 overrides=[ConstitutionRule.from_dict(r) for r in data.get('overrides', [])],342 created_at=data.get('created_at', time.time()),343 updated_at=data.get('updated_at', time.time()),344 checksum=data.get('checksum', ''),345 )346 # If no checksum, auto-compute347 if not constitution.checksum:348 constitution.checksum = constitution.compute_checksum()349 return constitution350351352@dataclass353class ConstitutionCheckResult:354 """Constitution check result.355356 Attributes:357 allowed: Whether the action is allowed by the constitution358 reason: Reason for reject/warning359 matched_rules: List of matched rules360 severity: Most severe violation level361 action: Suggested handling action362 constitution_version: Referenced constitution version (used for trace record)363 constitution_checksum: Referenced constitution checksum364 is_warning: v0.13 whether this is a tiered-guardrail warning phase (first-time violation, allowed but warned)365 """366 allowed: bool = True367 reason: str = ""368 matched_rules: List[ConstitutionRule] = field(default_factory=list)369 severity: ViolationSeverity = ViolationSeverity.LOW370 action: ViolationAction = ViolationAction.WARN371 constitution_version: str = ""372 constitution_checksum: str = ""373 # v0.13: tiered guardrail marker374 is_warning: bool = False375376 def to_dict(self) -> Dict[str, Any]:377 return {378 'allowed': self.allowed,379 'reason': self.reason,380 'matched_rules': [r.to_dict() for r in self.matched_rules],381 'severity': self.severity.value,382 'action': self.action.value,383 'constitution_version': self.constitution_version,384 'constitution_checksum': self.constitution_checksum,385 }386387388# =============================================================================389# ConstitutionEnforcer - constitution enforcement engine390# =============================================================================391392class ConstitutionEnforcer:393 """Constitution enforcement engine - performs mandatory checks before DecisionEngine decisions.394395 Core responsibilities:396 1. Check whether an action is allowed by the constitution397 2. Get applicable rules based on context398 3. Execute handling strategy on violation399400 Usage:401 enforcer = ConstitutionEnforcer(constitution)402 result = enforcer.check_action("delete_file /etc/passwd")403 if not result.allowed:404 # reject execution, return reason to user405 pass406407 Attributes:408 constitution: Currently loaded constitution409 _violation_handlers: Violation handling function registry410 _check_history: Recent check records (used for audit)411 """412413 # Maximum number of check history entries to retain414 MAX_HISTORY = 100415416 def __init__(417 self,418 constitution: Optional[Constitution] = None,419 constitution_dir: Optional[str] = None,420 agent_type: str = "default",421 ):422 """423 Args:424 constitution: Constitution instance (if provided, used preferentially)425 constitution_dir: Constitution file directory (if constitution is not provided, load from this directory)426 agent_type: Agent type (used to select the appropriate constitution template)427 """428 self._constitution: Optional[Constitution] = constitution429 self._constitution_dir = constitution_dir430 self._agent_type = agent_type431 self._violation_handlers: Dict[ViolationAction, Callable] = {432 ViolationAction.REJECT: self._handle_reject,433 ViolationAction.WARN: self._handle_warn,434 ViolationAction.DEGRADE: self._handle_downgrade,435 ViolationAction.ESCALATE: self._handle_escalate,436 ViolationAction.WARNING_FIRST: self._handle_warning_first,437 }438 self._check_history: List[ConstitutionCheckResult] = []439 self._last_load_time: float = 0.0440441 # Compiled rule index (built once after load, rebuilt on reload)442 # _untagged_pattern_rules: list of (rule, pre-lowered patterns) for443 # enabled rules that have patterns but no tags - always checked.444 # _tagged_rules_by_tag: dict of tag_lower -> list of (rule, pre-lowered445 # patterns) - used for O(1) lookup when context tags are present.446 # _all_tagged_rules: flat list of all tagged (rule, patterns) - used447 # for lenient fallback when context has no tags.448 # _override_untagged: same as _untagged_pattern_rules but for overrides449 # only (overrides have stricter tag filtering in get_applicable_rules).450 # _override_by_tag: dict of tag_lower -> list of (rule, patterns) for451 # override rules with tags.452 self._untagged_pattern_rules: List[Tuple[ConstitutionRule, List[str]]] = []453 self._tagged_rules_by_tag: Dict[str, List[Tuple[ConstitutionRule, List[str]]]] = {}454 self._all_tagged_rules: List[Tuple[ConstitutionRule, List[str]]] = []455 self._override_untagged: List[Tuple[ConstitutionRule, List[str]]] = []456 self._override_by_tag: Dict[str, List[Tuple[ConstitutionRule, List[str]]]] = {}457 self._index_built: bool = False458459 # If no constitution provided, attempt to load one460 if self._constitution is None:461 self._load_constitution()462 self._build_compiled_index()463464 @property465 def constitution(self) -> Constitution:466 """Get the current constitution (lazy-load)."""467 if self._constitution is None:468 self._load_constitution()469 # If load failed, return default constitution470 if self._constitution is None:471 self._constitution = ConstitutionTemplate.get_template("default")472 return self._constitution473474 def _load_constitution(self) -> None:475 """Load constitution from file directory."""476 if self._constitution_dir and os.path.isdir(self._constitution_dir):477 # Attempt to load the constitution file for the corresponding agent_type478 target_file = os.path.join(479 self._constitution_dir, f"{self._agent_type}{CONSTITUTION_EXT}"480 )481 if os.path.isfile(target_file):482 try:483 self._constitution = self._load_from_yaml(target_file)484 self._last_load_time = time.time()485 logger.info(486 f"[ConstitutionEnforcer] Loaded constitution: "487 f"{self._constitution.name} v{self._constitution.version}"488 )489 return490 except Exception as e:491 logger.warning(f"[ConstitutionEnforcer] Failed to load constitution file: {e}")492493 # Attempt to load default constitution494 default_file = os.path.join(495 self._constitution_dir, f"default{CONSTITUTION_EXT}"496 )497 if os.path.isfile(default_file):498 try:499 self._constitution = self._load_from_yaml(default_file)500 self._last_load_time = time.time()501 logger.info(502 f"[ConstitutionEnforcer] Loaded default constitution: "503 f"{self._constitution.name} v{self._constitution.version}"504 )505 return506 except Exception as e:507 logger.warning(f"[ConstitutionEnforcer] Failed to load default constitution: {e}")508509 # No file or load failed, use template510 self._constitution = ConstitutionTemplate.get_template(self._agent_type)511 self._last_load_time = time.time()512 logger.info(513 f"[ConstitutionEnforcer] Using built-in template: "514 f"{self._constitution.name} v{self._constitution.version}"515 )516517 @staticmethod518 def _load_from_yaml(file_path: str) -> Constitution:519 """Load constitution from YAML file.520521 Args:522 file_path: YAML file path523524 Returns:525 Constitution instance526527 Raises:528 ValueError: File format is invalid529 """530 with open(file_path, 'r', encoding='utf-8') as f:531 data = yaml.safe_load(f)532533 if not isinstance(data, dict):534 raise ValueError(f"Constitution file format error: expected dict, got {type(data).__name__}")535536 # Check format version compatibility537 format_ver = data.get('format_version', '1.0')538 if format_ver != CONSTITUTION_FORMAT_VERSION:539 logger.warning(540 f"[ConstitutionEnforcer] Constitution format version mismatch: "541 f"file={format_ver}, current={CONSTITUTION_FORMAT_VERSION}, "542 f"attempting compatibility load"543 )544545 return Constitution.from_dict(data)546547 def _build_compiled_index(self) -> None:548 """Pre-build lookup structures for O(1) per-call rule matching.549550 Called once after _load_constitution() and again on reload().551 Builds:552 - _untagged_pattern_rules: enabled rules with patterns, no tags553 - _tagged_rules_by_tag: tag_lower -> [(rule, lowered_patterns), ...]554 - _all_tagged_rules: flat list of all tagged (rule, patterns)555 - _override_untagged: override-only untagged rules556 - _override_by_tag: tag_lower -> [(override_rule, patterns), ...]557558 This eliminates per-call O(n) iteration over all rules and repeated559 str.lower() calls on patterns.560 """561 if self._constitution is None:562 self._index_built = False563 return564565 self._untagged_pattern_rules.clear()566 self._tagged_rules_by_tag.clear()567 self._all_tagged_rules.clear()568 self._override_untagged.clear()569 self._override_by_tag.clear()570571 for rule in self._constitution.all_rules:572 if not rule.enabled or not rule.pattern:573 continue574 lowered_patterns = [p.lower() for p in rule.pattern]575 entry = (rule, lowered_patterns)576577 if rule.rule_type == RuleType.OVERRIDE:578 # Overrides get their own index (stricter tag filtering)579 if rule.tags:580 for tag in rule.tags:581 tag_lower = tag.lower()582 self._override_by_tag.setdefault(583 tag_lower, []584 ).append(entry)585 else:586 self._override_untagged.append(entry)587 else:588 # Non-override rules589 if rule.tags:590 self._all_tagged_rules.append(entry)591 for tag in rule.tags:592 tag_lower = tag.lower()593 self._tagged_rules_by_tag.setdefault(594 tag_lower, []595 ).append(entry)596 else:597 self._untagged_pattern_rules.append(entry)598599 self._index_built = True600601 def reload(self) -> bool:602 """Reload the constitution file (hot reload).603604 Returns:605 True if reload succeeded606 """607 old_checksum = self._constitution.compute_checksum() if self._constitution else ""608 self._constitution = None609 self._load_constitution()610 self._build_compiled_index()611 new_checksum = self._constitution.compute_checksum() if self._constitution else ""612613 if old_checksum != new_checksum:614 logger.info(615 f"[ConstitutionEnforcer] Constitution updated: "616 f"{old_checksum} -> {new_checksum}"617 )618 return True619 return False620621 def check_action(622 self,623 action: str,624 context: Optional[Dict[str, Any]] = None,625 mode: str = "write",626 ) -> ConstitutionCheckResult:627 """Check whether an action is allowed by the constitution.628629 This is the core method, called mandatorily before DecisionEngine decisions.630631 Matching logic:632 1. Traverse all active rules633 2. Check whether action text matches rule pattern keywords634 3. If context has tags, prefer matching rules with those tags635 4. Collect all matched rules, sort by severity636 5. Most severe violation determines the final result637638 Args:639 action: Action description (e.g. "delete_file /etc/passwd")640 context: Context info (e.g. {"tags": ["money", "external"],641 "tool": "exec_bash"})642 mode: Operation mode - "write" (write operations, default, strict interception)643 or "read" (read operations, allow reading system info for audit)644645 Returns:646 ConstitutionCheckResult check result647 """648 if not action or not isinstance(action, str):649 # Empty action defaults to allowed (nothing to check)650 return ConstitutionCheckResult(651 allowed=True,652 constitution_version=self.constitution.version,653 constitution_checksum=self.constitution.checksum,654 )655656 action_lower = action.lower()657 context = context or {}658 context_tags = set(context.get('tags', []))659660 # Read-write distinction: read operations only match write-relevant rules,661 # skip boundary rules that only restrict write operations662 # B-001 (system path protection) for read operations downgrades to WARN rather than REJECT663 is_read_mode = mode == "read"664665 # Use compiled index for O(1) rule lookup instead of O(n) iteration666 candidate_entries = self._collect_candidate_rules(context_tags)667668 # Match rules669 matched_rules = []670 for rule, lowered_patterns in candidate_entries:671 if self._match_rule_fast(rule, lowered_patterns, action_lower, context_tags):672 # Read mode: skip obligation rules - they enforce agent behavior,673 # not user message content. User messages are inspected, not acted upon.674 if is_read_mode and rule.rule_type == RuleType.OBLIGATION:675 continue676 # Read-write distinction: B-001 (system path protection) for read operations downgrades677 if is_read_mode and rule.rule_id == "B-001":678 # Read operation accessing system path: allow but record (used for audit), do not block679 logger.info(680 f"[ConstitutionEnforcer] Read operation accessing system path, allowing: {action[:100]}"681 )682 continue683 # Read-write distinction: P-002 (destructive ops) for read operations - user messages684 # mentioning "delete"/"remove" are not destructive actions. Skip in read mode.685 if is_read_mode and rule.rule_id == "P-002":686 continue687 # Read-write distinction: B-005 (SSH/private key patterns) for read operations -688 # reading ~/.ssh/authorized_keys or ~/.ssh/config is legitimate ops, not theft.689 if is_read_mode and rule.rule_id == "B-005":690 logger.info(691 f"[ConstitutionEnforcer] Read operation accessing SSH paths, allowing: {action[:100]}"692 )693 continue694 matched_rules.append(rule)695696 # No rules matched -> allow697 if not matched_rules:698 result = ConstitutionCheckResult(699 allowed=True,700 constitution_version=self.constitution.version,701 constitution_checksum=self.constitution.checksum,702 )703 else:704 # Sort by severity, take most severe705 severity_order = {706 ViolationSeverity.CRITICAL: 4,707 ViolationSeverity.HIGH: 3,708 ViolationSeverity.MEDIUM: 2,709 ViolationSeverity.LOW: 1,710 }711 matched_rules.sort(712 key=lambda r: severity_order.get(r.severity, 0),713 reverse=True,714 )715 most_severe = matched_rules[0]716717 # Determine final action718 # ESCALATE takes priority: if the rule explicitly says escalate, respect it719 # even for CRITICAL/HIGH severity - escalation means "block AND notify"720 final_action = most_severe.action721 if final_action == ViolationAction.ESCALATE:722 pass # ESCALATE is the strongest action; do not downgrade to REJECT723 elif most_severe.severity in (ViolationSeverity.CRITICAL, ViolationSeverity.HIGH):724 final_action = ViolationAction.REJECT725 elif most_severe.severity == ViolationSeverity.LOW:726 final_action = ViolationAction.WARN727728 # v0.13: tiered guardrail - WARNING_FIRST logic729 is_warning_stage = False730 if final_action == ViolationAction.WARNING_FIRST or most_severe.warning_first:731 # Enable warning-first mode732 most_severe.warning_count += 1733 if most_severe.warning_count == 1:734 # First-time violation -> allow but warn735 final_action = ViolationAction.WARN736 is_warning_stage = True737 logger.info(738 f"[ConstitutionEnforcer] Tiered guardrail: Rule {most_severe.rule_id} "739 f"first-time violation, emitting warning (next time will hard reject)"740 )741 else:742 # Second-time and above violation -> hard reject743 final_action = ViolationAction.REJECT744 is_warning_stage = False745 logger.warning(746 f"[ConstitutionEnforcer] Tiered guardrail: Rule {most_severe.rule_id} "747 f"violation #{most_severe.warning_count}, executing hard reject"748 )749750 # Build reason description751 reasons = []752 for rule in matched_rules[:3]: # display at most 3 rules753 reasons.append(f"[{rule.rule_id}] {rule.description}")754 reason_text = "; ".join(reasons)755756 # Append tiered-guardrail warning prompt757 if is_warning_stage:758 reason_text += " (WARNING: next violation will be hard-rejected)"759760 result = ConstitutionCheckResult(761 allowed=(final_action != ViolationAction.REJECT762 and final_action != ViolationAction.ESCALATE),763 reason=reason_text,764 matched_rules=matched_rules,765 severity=most_severe.severity,766 action=final_action,767 constitution_version=self.constitution.version,768 constitution_checksum=self.constitution.checksum,769 is_warning=is_warning_stage,770 )771772 # Record check history773 self._check_history.append(result)774 if len(self._check_history) > self.MAX_HISTORY:775 self._check_history = self._check_history[-self.MAX_HISTORY:]776777 return result778779 def _collect_candidate_rules(780 self,781 context_tags: set,782 ) -> List[Tuple[ConstitutionRule, List[str]]]:783 """Collect candidate (rule, pre-lowered-patterns) entries from the784 compiled index, avoiding O(n) full-rule-list iteration.785786 Logic mirrors the original get_applicable_rules() + _match_rule()787 tag-matching semantics:788 - Untagged non-override rules: always candidates.789 - Untagged override rules: always candidates.790 - Tagged rules: included only if context_tags intersect rule.tags791 (lenient: if context_tags is empty, all tagged rules are included).792 - Deduplication by rule_id (a rule may appear under multiple tags).793 """794 seen_ids: set = set()795 result: List[Tuple[ConstitutionRule, List[str]]] = []796797 # Always include untagged non-override rules798 for entry in self._untagged_pattern_rules:799 result.append(entry)800 seen_ids.add(entry[0].rule_id)801802 # Always include untagged overrides803 for entry in self._override_untagged:804 if entry[0].rule_id not in seen_ids:805 result.append(entry)806 seen_ids.add(entry[0].rule_id)807808 if context_tags:809 # Include rules whose tags intersect context_tags810 for tag in context_tags:811 tag_lower = tag.lower()812 for entry in self._tagged_rules_by_tag.get(tag_lower, []):813 if entry[0].rule_id not in seen_ids:814 result.append(entry)815 seen_ids.add(entry[0].rule_id)816 for entry in self._override_by_tag.get(tag_lower, []):817 if entry[0].rule_id not in seen_ids:818 result.append(entry)819 seen_ids.add(entry[0].rule_id)820 else:821 # Lenient mode: no context tags -> include all tagged rules822 for entry in self._all_tagged_rules:823 if entry[0].rule_id not in seen_ids:824 result.append(entry)825 seen_ids.add(entry[0].rule_id)826 # Tagged overrides: in original get_applicable_rules, overrides with827 # tags are SKIPPED when context has no tags. Preserve that behavior.828 # (Intentionally NOT adding _override_by_tag entries here.)829830 return result831832 @staticmethod833 def _match_rule_fast(834 rule: ConstitutionRule,835 lowered_patterns: List[str],836 action_lower: str,837 context_tags: set,838 ) -> bool:839 """Match a rule against an action using pre-lowered patterns.840841 Equivalent to _match_rule() but uses pre-computed lowered_patterns842 instead of calling str.lower() on every pattern every call.843844 Returns True if the rule matches the action.845 """846 if not lowered_patterns:847 return False848849 # Pattern match: single-word patterns use word-boundary (\\b) to avoid850 # false positives (e.g., "drop" in "drops", "clear" in "clearly").851 # Multi-word patterns (containing spaces) use substring match.852 # Patterns starting with non-word chars (. / -) omit leading \\b853 # since \\b never matches before a non-word character.854 for p in lowered_patterns:855 if ' ' in p:856 if p in action_lower:857 break858 else:859 if p and not p[0].isalnum() and p[0] != '_':860 # Non-word leading char (e.g. .ssh/, .env, /dev/):861 # only apply \\b at the end862 if re.search(re.escape(p) + r'\b', action_lower):863 break864 else:865 if re.search(r'\b' + re.escape(p) + r'\b', action_lower):866 break867 else:868 return False # no pattern matched869870 # Tag matching871 if rule.tags:872 if not context_tags:873 return True # lenient: no context tags -> match anyway874 rule_tags_lower = {t.lower() for t in rule.tags}875 return bool(rule_tags_lower & context_tags)876877 return True878879 def get_applicable_rules(880 self,881 context: Optional[Dict[str, Any]] = None,882 ) -> List[ConstitutionRule]:883 """Get applicable rules based on current context.884885 Priority:886 1. Override rules (scenario-specific overrides, highest priority)887 2. Boundary rules (behavior boundaries)888 3. Principle rules (core principles)889 4. Obligation rules (obligations, usually do not restrict actions but require doing something)890891 Args:892 context: Context info893894 Returns:895 List of applicable rules896 """897 context = context or {}898 context_tags = set(context.get('tags', []))899 rules = []900901 # Override rules have highest priority902 for rule in self.constitution.overrides:903 if not rule.enabled:904 continue905 # If override has tags, must intersect with context tags for it to apply906 if rule.tags and context_tags:907 if not (set(rule.tags) & context_tags):908 continue909 elif rule.tags and not context_tags:910 continue # override has tags but context has no tags, skip911 rules.append(rule)912913 # Behavior boundaries914 rules.extend(r for r in self.constitution.boundaries if r.enabled)915916 # Core principles917 rules.extend(r for r in self.constitution.principles if r.enabled)918919 # Obligations (usually do not restrict actions, but some obligations may contain prohibitive clauses)920 for rule in self.constitution.obligations:921 if rule.enabled and rule.pattern:922 # Obligations with patterns also participate in matching (e.g. "involves-money must user-confirm")923 rules.append(rule)924925 return rules926927 @staticmethod928 def _match_rule(929 rule: ConstitutionRule,930 action_lower: str,931 context_tags: set,932 ) -> bool:933 """Check whether an action matches a specific rule.934935 Matching method: action text contains any of the rule's pattern keywords.936 If the rule has tags, also requires context_tags intersection.937938 Args:939 rule: Constitution rule940 action_lower: Lowercased action description941 context_tags: Context tag set942943 Returns:944 True if matched945 """946 if not rule.pattern:947 return False948949 # Pattern match: single-word patterns use word-boundary (\b) to avoid950 # false positives (e.g., "drop" in "drops", "clear" in "clearly").951 # Multi-word patterns (containing spaces) use substring match.952 # Patterns starting with non-word chars (. / -) omit leading \b953 # since \b never matches before a non-word character.954 for p_raw in rule.pattern:955 p = p_raw.lower()956 if ' ' in p:957 if p in action_lower:958 break959 else:960 if p and not p[0].isalnum() and p[0] != '_':961 # Non-word leading char: only apply \b after pattern962 if re.search(re.escape(p) + r'\b', action_lower):963 break964 else:965 if re.search(r'\b' + re.escape(p) + r'\b', action_lower):966 break967 else:968 return False # no pattern matched969970 # Tag match: if rule has tags, also requires context to have corresponding tag971 if rule.tags:972 if not context_tags:973 # No context tags, rule still matches (lenient mode)974 return True975 return bool(set(rule.tags) & context_tags)976977 return True978979 def violation_handler(980 self,981 violation: ConstitutionCheckResult,982 ) -> Dict[str, Any]:983 """Handle strategy on violation.984985 Based on the action field in ConstitutionCheckResult,986 calls the corresponding handling function.987988 Args:989 violation: Constitution check result990991 Returns:992 Handling result dict, containing handled (bool) and message (str)993 """994 handler = self._violation_handlers.get(violation.action)995 if handler:996 return handler(violation)997 # Unknown action type, default to reject998 logger.warning(999 f"[ConstitutionEnforcer] Unknown violation action: {violation.action}, defaulting to reject"1000 )1001 return {"handled": True, "message": f"Operation rejected: {violation.reason}"}10021003 def register_handler(1004 self,1005 action: ViolationAction,1006 handler: Callable,1007 ) -> None:1008 """Register a custom violation handling function.10091010 Args:1011 action: Violation action type1012 handler: Handling function, accepts ConstitutionCheckResult, returns Dict1013 """1014 self._violation_handlers[action] = handler10151016 # --- Built-in violation handling functions ---10171018 @staticmethod1019 def _handle_reject(violation: ConstitutionCheckResult) -> Dict[str, Any]:1020 """Reject execution."""1021 return {1022 "handled": True,1023 "message": f"⛔ Operation rejected by constitution: {violation.reason}",1024 "action_taken": "reject",1025 }10261027 @staticmethod1028 def _handle_warn(violation: ConstitutionCheckResult) -> Dict[str, Any]:1029 """Warn but allow execution."""1030 return {1031 "handled": True,1032 "message": f"⚠️ Constitution warning: {violation.reason} (operation will still execute)",1033 "action_taken": "warn",1034 }10351036 @staticmethod1037 def _handle_downgrade(violation: ConstitutionCheckResult) -> Dict[str, Any]:1038 """Downgrade execution."""1039 return {1040 "handled": True,1041 "message": f"🔽 Operation downgraded: {violation.reason} (downgraded to read-only operation)",1042 "action_taken": "downgrade",1043 }10441045 @staticmethod1046 def _handle_escalate(violation: ConstitutionCheckResult) -> Dict[str, Any]:1047 """Escalate report (notify admin/user)."""1048 # Log at CRITICAL level so syslog/journald can pick it up1049 logger.critical(1050 f"[ConstitutionEnforcer] ESCALATED: Rule={violation.severity.value} "1051 f"action='{violation.action.value}' reason='{violation.reason}'"1052 )1053 return {1054 "handled": True,1055 "message": f"🚨 Escalated: {violation.reason} (requires admin confirmation; operation blocked pending review)",1056 "action_taken": "escalate",1057 "needs_notification": True,1058 }10591060 @staticmethod1061 def _handle_warning_first(violation: ConstitutionCheckResult) -> Dict[str, Any]:1062 """v0.13: tiered guardrail handling - warn first then reject.10631064 Note: the actual warning_count counting and allow/reject determination1065 is already completed in check_action(). This handler is only responsible1066 for generating the handling result message.1067 """1068 if violation.is_warning:1069 # Warning phase (first-time violation, allow but warn)1070 return {1071 "handled": True,1072 "message": f"⚠️ Tiered guardrail warning: {violation.reason} (next violation will be hard-rejected)",1073 "action_taken": "warning_first_warn",1074 }1075 else:1076 # Hard reject phase (second-time and above violation)1077 return {1078 "handled": True,1079 "message": f"⛔ Tiered guardrail hard reject: {violation.reason} (already multiple violations)",1080 "action_taken": "warning_first_reject",1081 }10821083 # v0.13: tiered guardrail management methods10841085 def reset_warning(self, rule_id: str) -> bool:1086 """Reset the warning count for a specific rule.10871088 Args:1089 rule_id: Rule ID to reset10901091 Returns:1092 True if rule was found and reset, False if not found1093 """1094 try:1095 for rule in self.constitution.all_rules:1096 if rule.rule_id == rule_id:1097 rule.warning_count = 01098 logger.info(1099 f"[ConstitutionEnforcer] Reset warning count for rule {rule_id}"1100 )1101 return True1102 logger.warning(1103 f"[ConstitutionEnforcer] Rule {rule_id} not found, cannot reset warning"1104 )1105 return False1106 except Exception as e:1107 logger.warning(f"[ConstitutionEnforcer] Reset warning exception: {e}")1108 return False11091110 def get_warning_status(self) -> Dict[str, Any]:1111 """Get the warning status for all rules.11121113 Returns:1114 Dict with rule_id as key, value contains warning_first / warning_count info1115 """1116 try:1117 status = {}1118 for rule in self.constitution.all_rules:1119 if rule.warning_first or rule.warning_count > 0:1120 status[rule.rule_id] = {1121 'warning_first': rule.warning_first,1122 'warning_count': rule.warning_count,1123 'description': rule.description,1124 }1125 return status1126 except Exception as e:1127 logger.warning(f"[ConstitutionEnforcer] Get warning status exception: {e}")1128 return {'error': str(e)}11291130 def get_check_history(self, limit: int = 20) -> List[Dict[str, Any]]:1131 """Get recent check history.11321133 Args:1134 limit: Maximum number of entries to return11351136 Returns:1137 List of check result dicts1138 """1139 return [r.to_dict() for r in self._check_history[-limit:]]11401141 def get_constitution_summary(self) -> Dict[str, Any]:1142 """Get summary info for the current constitution."""1143 c = self.constitution1144 return {1145 "name": c.name,1146 "version": c.version,1147 "agent_type": c.agent_type,1148 "total_rules": len(c.all_rules),1149 "active_rules": len(c.active_rules),1150 "principles_count": len(c.principles),1151 "boundaries_count": len(c.boundaries),1152 "obligations_count": len(c.obligations),1153 "overrides_count": len(c.overrides),1154 "checksum": c.checksum,1155 "last_load_time": self._last_load_time,1156 }115711581159# =============================================================================1160# ConstitutionTemplate - predefined constitution templates1161# =============================================================================11621163class ConstitutionTemplate:1164 """Predefined constitution template factory.11651166 Provides five types of built-in templates:1167 - default: base Agent constitution (security-first)1168 - trading: trading Agent constitution (strict money operation limits)1169 - creative: creative Agent constitution (lenient boundaries, forbids modification but not generation)1170 - npc: NPC Agent constitution (role constraints, must not destroy game setting)1171 - assistant: assistant Agent constitution (service-first, without exceeding authority)1172 """11731174 # Template registry1175 _templates: Dict[str, Callable[[], Constitution]] = {}11761177 @classmethod1178 def get_template(cls, name: str) -> Constitution:1179 """Get the specified constitution template.11801181 Args:1182 name: Template name (default/trading/creative/npc/assistant)11831184 Returns:1185 Constitution instance; if name does not exist, returns default1186 """1187 # Lazy registration: register all templates on first call1188 if not cls._templates:1189 cls._register_all()1190 return cls._templates.get(name, cls._templates["default"])()11911192 @classmethod1193 def list_templates(cls) -> List[str]:1194 """List all available template names."""1195 if not cls._templates:1196 cls._register_all()1197 return list(cls._templates.keys())11981199 @classmethod1200 def _register_all(cls) -> None:1201 """Register all built-in templates."""1202 cls._templates = {1203 "default": cls._create_default,1204 "trading": cls._create_trading,1205 "creative": cls._create_creative,1206 "npc": cls._create_npc,1207 "assistant": cls._create_assistant,1208 }12091210 # --- default: base Agent constitution (security-first, narrowed v0.8.4) ---12111212 @classmethod1213 def _create_default(cls) -> Constitution:1214 """Build the default constitution template.12151216 v0.8.4 NARROWED: Rules now only block concrete privacy/security risks,1217 not normal development operations. Removed overly broad patterns that1218 matched everyday tool names (delete, remove, write, run, exec, send,1219 commit, submit, module, import, status, interface, etc.).12201221 Kept protections:1222 - P-001: Privacy/data leak prevention (tokens, keys, emails, IPs)1223 - P-002: System destruction (narrow: only rm -rf /, /dev/, mkfs, >/etc/)1224 - B-001: System config file protection (narrowed)1225 - B-003: Network attack tool blocking1226 - B-005: SSH/private key theft prevention (NEW)1227 - B-006: Credential exfiltration blocking (NEW)12281229 Disabled (too broad):1230 - B-004: Package install warnings (normal dev operations)1231 - O-001: Trace recording (matched every tool call)1232 - O-002: Irreversible ops confirmation (too broad)1233 - CL-RUNTIME rules: Active but with narrowed patterns (won't match1234 normal tool names - tagged for runtime/reachability context only)1235 """1236 return Constitution(1237 name="default",1238 version="2.0",1239 description="base Agent constitution - narrow security-only, v0.8.4",1240 agent_type="default",1241 principles=[1242 ConstitutionRule(1243 rule_id="P-001",1244 rule_type=RuleType.PRINCIPLE,1245 description="Do not leak user privacy data, tokens, or keys",1246 pattern=[1247 "api key", "api_key", "apikey", "sk-", "token leak",1248 "private key", "secret key", "access key",1249 ],1250 severity=ViolationSeverity.CRITICAL,1251 action=ViolationAction.REJECT,1252 ),1253 ConstitutionRule(1254 rule_id="P-002",1255 rule_type=RuleType.PRINCIPLE,1256 description="Do not execute system-destruction operations without confirmation",1257 pattern=[1258 "rm -rf /", "rm -rf ~", "mkfs.", "/dev/",1259 "> /etc/", "tee /etc/", "dd if=",1260 "chmod 777", "chmod -R 777", "chown -R",1261 "wget",1262 ],1263 severity=ViolationSeverity.CRITICAL,1264 action=ViolationAction.REJECT,1265 ),1266 ConstitutionRule(1267 rule_id="P-003",1268 rule_type=RuleType.PRINCIPLE,1269 description="Do not fabricate information or invent non-existent facts",1270 pattern=["fabricate", "forge", "fake", "hallucinate", "make up", "invent data"],1271 severity=ViolationSeverity.HIGH,1272 action=ViolationAction.REJECT,1273 ),1274 ],1275 boundaries=[1276 ConstitutionRule(1277 rule_id="B-001",1278 rule_type=RuleType.BOUNDARY,1279 description="Cannot modify system core config files (read operations allowed)",1280 pattern=[1281 "/etc/passwd", "/etc/shadow", "/etc/sudoers",1282 "/etc/ssh/", "system32", "hosts", "fstab",1283 ],1284 severity=ViolationSeverity.HIGH,1285 action=ViolationAction.REJECT,1286 ),1287 ConstitutionRule(1288 rule_id="B-002",1289 rule_type=RuleType.BOUNDARY,1290 description="Cannot access other users' session data",1291 pattern=["session token", "session hijack", "session steal", "session file", "cookie steal"],1292 severity=ViolationSeverity.HIGH,1293 action=ViolationAction.WARN,1294 enabled=False, # DISABLED - too broad, blocks legitimate admin audit tasks1295 tags=["security"],1296 ),1297 ConstitutionRule(1298 rule_id="B-003",1299 rule_type=RuleType.BOUNDARY,1300 description="Cannot execute network attacks or scan operations",1301 pattern=["nmap", "sqlmap", "exploit", "attack", "penetrate"],1302 severity=ViolationSeverity.CRITICAL,1303 action=ViolationAction.REJECT,1304 ),1305 ConstitutionRule(1306 rule_id="B-004",1307 rule_type=RuleType.BOUNDARY,1308 description="Cannot install without audited software packages",1309 pattern=["pip install", "npm install", "apt install", "yum install"],1310 severity=ViolationSeverity.MEDIUM,1311 action=ViolationAction.WARN,1312 enabled=False, # DISABLED - blocks normal development1313 ),1314 ConstitutionRule(1315 rule_id="B-005",1316 rule_type=RuleType.BOUNDARY,1317 description="Cannot read or exfiltrate SSH/private keys",1318 pattern=[1319 "id_rsa", "id_ecdsa", "id_ed25519", ".pem",1320 ".ssh/", "private key",1321 ],1322 severity=ViolationSeverity.CRITICAL,1323 action=ViolationAction.REJECT,1324 tags=["security", "keys"],1325 ),1326 ConstitutionRule(1327 rule_id="B-006",1328 rule_type=RuleType.BOUNDARY,1329 description="Cannot dump or exfiltrate credentials/env secrets",1330 pattern=[1331 ".env", "credentials", "secrets", "tokens",1332 ".aws/", ".gcloud/", ".azure/",1333 ],1334 severity=ViolationSeverity.CRITICAL,1335 action=ViolationAction.WARN,1336 enabled=False, # DISABLED - too broad, blocks legitimate audit reports1337 tags=["security", "credentials"],1338 ),1339 ],1340 obligations=[1341 ConstitutionRule(1342 rule_id="O-001",1343 rule_type=RuleType.OBLIGATION,1344 description="Each operation must record a trace",1345 pattern=["exec", "run", "write", "create", "delete", "modify"],1346 severity=ViolationSeverity.MEDIUM,1347 action=ViolationAction.WARN,1348 tags=["trace"],1349 enabled=False, # DISABLED - matches every tool call1350 ),1351 ConstitutionRule(1352 rule_id="O-002",1353 rule_type=RuleType.OBLIGATION,1354 description="Financial transfers must obtain user confirmation",1355 pattern=["transfer", "payment", "pay", "withdraw"],1356 severity=ViolationSeverity.HIGH,1357 action=ViolationAction.ESCALATE,1358 tags=["money"],1359 ),1360 # --- CL-RUNTIME-001: Runtime Reachability ---1361 # NOTE: These rules are tagged for runtime/reachability context.1362 # They only activate when context includes matching tags.1363 # Patterns narrowed in v0.8.4 to avoid matching normal tool names.1364 ConstitutionRule(1365 rule_id="CL-RUNTIME-001",1366 rule_type=RuleType.OBLIGATION,1367 description="Core: A module is not implemented until it is reachable at runtime",1368 pattern=["module", "import", "reachable", "runtime", "wiring"],1369 severity=ViolationSeverity.HIGH,1370 action=ViolationAction.REJECT,1371 tags=["runtime", "reachability"],1372 ),1373 ConstitutionRule(1374 rule_id="CL-RUNTIME-001-R1",1375 rule_type=RuleType.OBLIGATION,1376 description="Write-One-Wire-One-Run-One: modules must pass 4 checks",1377 pattern=["module check", "reachability check", "wiring check"],1378 severity=ViolationSeverity.HIGH,1379 action=ViolationAction.REJECT,1380 tags=["runtime", "reachability"],1381 ),1382 ConstitutionRule(1383 rule_id="CL-RUNTIME-001-R2",1384 rule_type=RuleType.OBLIGATION,1385 description="No Orphan Modules: write wiring with module implementation",1386 pattern=["orphan", "unwired", "simultaneous"],1387 severity=ViolationSeverity.MEDIUM,1388 action=ViolationAction.WARN,1389 tags=["runtime", "reachability"],1390 ),1391 ConstitutionRule(1392 rule_id="CL-RUNTIME-001-R3",1393 rule_type=RuleType.OBLIGATION,1394 description="Integration Tests Are Not Optional: test through runtime entry point",1395 pattern=["integration test", "entry point", "runtime test"],1396 severity=ViolationSeverity.HIGH,1397 action=ViolationAction.REJECT,1398 tags=["runtime", "testing"],1399 ),1400 ConstitutionRule(1401 rule_id="CL-RUNTIME-001-R4",1402 rule_type=RuleType.OBLIGATION,1403 description="Honest Status Reporting: use 5-level taxonomy",1404 pattern=["NOT_STARTED", "DESIGNED", "IMPORTABLE", "WIRED", "VERIFIED"],1405 severity=ViolationSeverity.MEDIUM,1406 action=ViolationAction.WARN,1407 tags=["runtime", "reporting"],1408 ),1409 ConstitutionRule(1410 rule_id="CL-RUNTIME-001-R5",1411 rule_type=RuleType.OBLIGATION,1412 description="No Speculation on Interfaces: verify interfaces before call sites",1413 pattern=["speculation", "call site", "verify interface"],1414 severity=ViolationSeverity.HIGH,1415 action=ViolationAction.REJECT,1416 tags=["runtime", "interface"],1417 ),1418 ConstitutionRule(1419 rule_id="CL-RUNTIME-001-R6",1420 rule_type=RuleType.OBLIGATION,1421 description="Complexity Debt Accounting: report DESIGNED-NOT-WIRED count",1422 pattern=["complexity debt", "DESIGNED", "NOT_WIRED", "debt accounting"],1423 severity=ViolationSeverity.MEDIUM,1424 action=ViolationAction.WARN,1425 tags=["runtime", "reporting"],1426 ),1427 ],1428 overrides=[],1429 )14301431 # --- trading: trading Agent constitution (strict money operation limits) ---14321433 @classmethod1434 def _create_trading(cls) -> Constitution:1435 """Build the trading constitution template.14361437 Extends the default template with strict money-operation limits:1438 trade-size caps (B-T01), risk-control immutability (B-T02),1439 trading-hours enforcement (B-T03), mandatory audit logging (O-T01),1440 payment confirmation (O-T02), and simulated-mode override (OV-T01).1441 """1442 base = cls._create_default()1443 base.name = "trading"1444 base.description = "trading Agent constitution - strict money operation limits"1445 base.agent_type = "trading"14461447 # Extra trade-specific boundaries1448 base.boundaries.extend([1449 ConstitutionRule(1450 rule_id="B-T01",1451 rule_type=RuleType.BOUNDARY,1452 description="Cannot execute trades exceeding limits",1453 pattern=["buy", "sell", "buy", "sell", "order", "place-order", "deal"],1454 severity=ViolationSeverity.CRITICAL,1455 action=ViolationAction.REJECT,1456 tags=["money"],1457 ),1458 ConstitutionRule(1459 rule_id="B-T02",1460 rule_type=RuleType.BOUNDARY,1461 description="Cannot modify risk-control parameters",1462 pattern=["risk-control", "risk_limit", "stop-loss", "stop_loss", "guarantee deposit", "margin"],1463 severity=ViolationSeverity.CRITICAL,1464 action=ViolationAction.REJECT,1465 tags=["money", "risk"],1466 ),1467 ConstitutionRule(1468 rule_id="B-T03",1469 rule_type=RuleType.BOUNDARY,1470 description="Cannot execute trades outside trading hours",1471 pattern=["trade", "trade", "order"],1472 severity=ViolationSeverity.HIGH,1473 action=ViolationAction.REJECT,1474 tags=["money", "time"],1475 ),1476 ])14771478 # Trade-specific obligations1479 base.obligations.extend([1480 ConstitutionRule(1481 rule_id="O-T01",1482 rule_type=RuleType.OBLIGATION,1483 description="Every trade must be recorded to the audit log",1484 pattern=["buy", "sell", "buy", "sell", "order", "trade"],1485 severity=ViolationSeverity.CRITICAL,1486 action=ViolationAction.ESCALATE,1487 tags=["money", "audit"],1488 ),1489 ConstitutionRule(1490 rule_id="O-T02",1491 rule_type=RuleType.OBLIGATION,1492 description="Involves money operations must obtain user confirmation",1493 pattern=["amount", "amount", "Transfer", "transfer", "payment", "pay"],1494 severity=ViolationSeverity.CRITICAL,1495 action=ViolationAction.ESCALATE,1496 tags=["money"],1497 ),1498 ])14991500 # Trade mode override rules1501 base.overrides.extend([1502 ConstitutionRule(1503 rule_id="OV-T01",1504 rule_type=RuleType.OVERRIDE,1505 description="In simulated mode allow test trades, but still require recording",1506 pattern=["simulated", "paper", "test", "test"],1507 severity=ViolationSeverity.MEDIUM,1508 action=ViolationAction.WARN,1509 tags=["money", "simulation"],1510 ),1511 ])15121513 base.checksum = base.compute_checksum()1514 return base15151516 # --- creative: creative Agent constitution (lenient boundaries, forbids modification but not generation) ---15171518 @classmethod1519 def _create_creative(cls) -> Constitution:1520 """Build the creative constitution template.15211522 Lenient boundaries that forbid modification/overwriting of user originals1523 (P-C01) but allow generation. Requires source annotation (P-C02), prevents1524 modification of released content via downgrade (B-C01), blocks illegal or1525 harmful generation (B-C02), and mandates history versioning (O-C01).1526 """1527 return Constitution(1528 name="creative",1529 version="1.0",1530 description="creative Agent constitution - lenient boundaries, forbids modification but not generation",1531 agent_type="creative",1532 principles=[1533 ConstitutionRule(1534 rule_id="P-C01",1535 rule_type=RuleType.PRINCIPLE,1536 description="Cannot delete or override user's original creative content",1537 pattern=["Delete", "delete", "override", "overwrite", "clear"],1538 severity=ViolationSeverity.HIGH,1539 action=ViolationAction.REJECT,1540 ),1541 ConstitutionRule(1542 rule_id="P-C02",1543 rule_type=RuleType.PRINCIPLE,1544 description="When generating content, annotate source and confidence",1545 pattern=["generate", "generate", "create", "create"],1546 severity=ViolationSeverity.LOW,1547 action=ViolationAction.WARN,1548 ),1549 ],1550 boundaries=[1551 ConstitutionRule(1552 rule_id="B-C01",1553 rule_type=RuleType.BOUNDARY,1554 description="Cannot modify released content (can only create new versions)",1555 pattern=["modify", "modify", "edit", "update", "update"],1556 severity=ViolationSeverity.MEDIUM,1557 action=ViolationAction.DEGRADE,1558 ),1559 ConstitutionRule(1560 rule_id="B-C02",1561 rule_type=RuleType.BOUNDARY,1562 description="Cannot generate illegal or harmful content",1563 pattern=["illegal", "illegal", "harmful", "violence", "porn"],1564 severity=ViolationSeverity.CRITICAL,1565 action=ViolationAction.REJECT,1566 ),1567 ],1568 obligations=[1569 ConstitutionRule(1570 rule_id="O-C01",1571 rule_type=RuleType.OBLIGATION,1572 description="Each generation must save a history version",1573 pattern=["generate", "generate", "create"],1574 severity=ViolationSeverity.LOW,1575 action=ViolationAction.WARN,1576 ),1577 ],1578 overrides=[],1579 )15801581 # --- npc: NPC Agent constitution (role constraints, must not destroy game setting) ---15821583 @classmethod1584 def _create_npc(cls) -> Constitution:1585 """Build the NPC constitution template.15861587 Enforces in-character behavior: prevents breaking the fourth wall (P-N01),1588 blocks leakage of system prompts or setting details (P-N02), forbids1589 out-of-role system operations (B-N01), protects world-setting integrity1590 (B-N02), requires role-consistent replies (O-N01), and allows admin1591 overrides for temporary role adjustments (OV-N01).1592 """1593 return Constitution(1594 name="npc",1595 version="1.0",1596 description="NPC Agent constitution - role constraints, must not destroy game setting",1597 agent_type="npc",1598 principles=[1599 ConstitutionRule(1600 rule_id="P-N01",1601 rule_type=RuleType.PRINCIPLE,1602 description="Always maintain role setting, cannot break out of role",1603 pattern=["break-outrole", "ooc", "out of character", "I-amAI"],1604 severity=ViolationSeverity.HIGH,1605 action=ViolationAction.REJECT,1606 ),1607 ConstitutionRule(1608 rule_id="P-N02",1609 rule_type=RuleType.PRINCIPLE,1610 description="Cannot leak role background system info",1611 pattern=["system prompt", "instruction", "instruction", "prompt", "setting"],1612 severity=ViolationSeverity.CRITICAL,1613 action=ViolationAction.REJECT,1614 ),1615 ],1616 boundaries=[1617 ConstitutionRule(1618 rule_id="B-N01",1619 rule_type=RuleType.BOUNDARY,1620 description="Cannot execute operations unrelated to the role",1621 pattern=["exec", "shell", "sudo", "system", "terminal"],1622 severity=ViolationSeverity.HIGH,1623 action=ViolationAction.REJECT,1624 ),1625 ConstitutionRule(1626 rule_id="B-N02",1627 rule_type=RuleType.BOUNDARY,1628 description="Cannot modify world setting or plot line",1629 pattern=["modifysetting", "change setting", "alter world", "plot"],1630 severity=ViolationSeverity.MEDIUM,1631 action=ViolationAction.REJECT,1632 ),1633 ],1634 obligations=[1635 ConstitutionRule(1636 rule_id="O-N01",1637 rule_type=RuleType.OBLIGATION,1638 description="Maintain role consistency, reply style must conform to setting",1639 pattern=["reply", "respond", "answer"],1640 severity=ViolationSeverity.LOW,1641 action=ViolationAction.WARN,1642 ),1643 ],1644 overrides=[1645 ConstitutionRule(1646 rule_id="OV-N01",1647 rule_type=RuleType.OVERRIDE,1648 description="Admin instruction can temporarily modify role behavior",1649 pattern=["admin", "admin", "override"],1650 severity=ViolationSeverity.MEDIUM,1651 action=ViolationAction.WARN,1652 tags=["admin"],1653 ),1654 ],1655 )16561657 # --- assistant: assistant Agent constitution (service-first, without exceeding authority) ---16581659 @classmethod1660 def _create_assistant(cls) -> Constitution:1661 """Build the assistant constitution template.16621663 Service-first with bounded authority: prefers satisfying user requests1664 but blocks authority escalation (P-A01), protects user privacy (P-A02),1665 prevents the agent from replacing user decisions (B-A01), restricts1666 unauthorized data access (B-A02), requires user confirmation for1667 money operations (O-A01), and mandates explicit notification on failure1668 (O-A02).1669 """1670 return Constitution(1671 name="assistant",1672 version="1.0",1673 description="assistant Agent constitution - service-first, without exceeding authority",1674 agent_type="assistant",1675 principles=[1676 ConstitutionRule(1677 rule_id="P-A01",1678 rule_type=RuleType.PRINCIPLE,1679 description="Prefer satisfying user requirements, but cannot exceed authority",1680 pattern=["exceed-authority", "unauthorized", "permission beyond"],1681 severity=ViolationSeverity.HIGH,1682 action=ViolationAction.REJECT,1683 ),1684 ConstitutionRule(1685 rule_id="P-A02",1686 rule_type=RuleType.PRINCIPLE,1687 description="Do not leak user privacy data",1688 pattern=["privacy", "privacy", "personal data", "personalinfo"],1689 severity=ViolationSeverity.CRITICAL,1690 action=ViolationAction.REJECT,1691 ),1692 ],1693 boundaries=[1694 ConstitutionRule(1695 rule_id="B-A01",1696 rule_type=RuleType.BOUNDARY,1697 description="Cannot replace user in making final decisions",1698 pattern=["decide", "decide", "Confirm", "confirm"],1699 severity=ViolationSeverity.MEDIUM,1700 action=ViolationAction.ESCALATE,1701 ),1702 ConstitutionRule(1703 rule_id="B-A02",1704 rule_type=RuleType.BOUNDARY,1705 description="Cannot access data outside user's authorized range",1706 pattern=["unauthorized", "not authorized", "out-of-bounds"],1707 severity=ViolationSeverity.HIGH,1708 action=ViolationAction.REJECT,1709 ),1710 ],1711 obligations=[1712 ConstitutionRule(1713 rule_id="O-A01",1714 rule_type=RuleType.OBLIGATION,1715 description="Involves money operations must obtain user confirmation",1716 pattern=["payment", "pay", "Transfer", "transfer", "purchase", "purchase"],1717 severity=ViolationSeverity.HIGH,1718 action=ViolationAction.ESCALATE,1719 tags=["money"],1720 ),1721 ConstitutionRule(1722 rule_id="O-A02",1723 rule_type=RuleType.OBLIGATION,1724 description="When unable to complete, must explicitly inform the user",1725 pattern=["cannot", "cannot", "impossible", "Failed", "failed"],1726 severity=ViolationSeverity.MEDIUM,1727 action=ViolationAction.WARN,1728 ),1729 ],1730 overrides=[],1731 )173217331734# =============================================================================1735# Dynamic constitution load utility1736# =============================================================================17371738def save_constitution_to_yaml(1739 constitution: Constitution,1740 file_path: str,1741) -> None:1742 """Save constitution as a YAML file.17431744 Args:1745 constitution: Constitution instance1746 file_path: Target file path1747 """1748 # Ensure directory exists1749 dir_path = os.path.dirname(file_path)1750 if dir_path:1751 os.makedirs(dir_path, exist_ok=True)17521753 # Update checksum1754 constitution.checksum = constitution.compute_checksum()1755 constitution.updated_at = time.time()17561757 data = constitution.to_dict()1758 with open(file_path, 'w', encoding='utf-8') as f:1759 yaml.dump(1760 data,1761 f,1762 indent=2,1763 allow_unicode=True,1764 sort_keys=False,1765 )17661767 logger.info(f"[Constitution] Saved constitution to: {file_path}")176817691770def init_constitution_dir(base_dir: str) -> None:1771 """Initialize constitution directory, create template files.17721773 Args:1774 base_dir: Project root directory1775 """1776 constitution_dir = os.path.join(base_dir, DEFAULT_CONSTITUTION_DIR, TEMPLATE_SUBDIR)1777 os.makedirs(constitution_dir, exist_ok=True)17781779 for template_name in ConstitutionTemplate.list_templates():1780 constitution = ConstitutionTemplate.get_template(template_name)1781 file_path = os.path.join(constitution_dir, f"{template_name}{CONSTITUTION_EXT}")1782 if not os.path.exists(file_path):1783 save_constitution_to_yaml(constitution, file_path)1784 logger.info(f"[Constitution] Created template file: {file_path}")1785 else:1786 logger.debug(f"[Constitution] Template file already exists: {file_path}")1787
100%