Intercept memory writes for a CrewAI agent's shared memory. CrewAI agents may have a ``memory`` or ``shared_memory`` attribute. This

Wrap the function_map on an AutoGen AssistantAgent. AutoGen agents store callable functions in a ``function_map`` dict. This method w

Hook into GroupChat's select_speaker and message routing. If the agent is a GroupChat manager (has a ``groupchat`` attribute), this m

Intercept memory writes on the wrapped agent. If the underlying object exposes a ``memory`` attribute with a ``save_context`` method,

Track agent state changes for governance audit. Wraps ``update_system_message`` and ``reset`` (if present) to monitor and validate st

Hook into individual step execution within a task. If the agent exposes a ``step`` or ``_execute_step`` method, it is wrapped so that

Intercept the agent's tool registry to apply per-tool governance. After the agent is wrapped this method inspects its ``tools`` attri

Run *callback* in a daemon thread bounded by ``callback_timeout``. Returns ``True`` if the callback completed cleanly within the time

* Structural validation for an unknown payload claiming to be a Workflow. * * Loaded workflow files arrive from disk and may have been hand-

Return ``True`` if *tool_name* should be blocked (v4 path).

Check if a similar incident was recently created.

Return True if a repo has enough traction to be considered established.

Check if current date is within range

Return True when the HTTP request originated from a loopback peer.

Check if domain resolves to private IP (basic check).

Check replay set membership under lock.

Check if line is safe to skip (comments, tests, etc.).

Check if detected text is likely test/example data.

Check if content is Tier 1 (High Value).

Check if content is Tier 3 (Low Value).

Issue a CRITICAL alert for detected fraud.

Walk ``root`` recursively, yielding manifest files outside any excluded directory. Uses os.walk so we can prune subtrees in-place — r

Yield every string-like leaf reachable from *value*. Walks dicts (keys and values), lists/tuples/sets, and decodes ``bytes`` via UTF-8 with `

Yield every string contained in a nested params structure. Used by the no_pii policy check so the PII detector runs against every string valu

Safe JSON serialization for parameter scanning.

Best-effort kill of a timed-out exec process by PID.

KL(P || Q) with Laplace smoothing.

Calculate length of Longest Common Subsequence. Pure function using dynamic programming.

Compute Levenshtein edit distance between two strings.

Simple Levenshtein distance.

Calculate Levenshtein edit distance between two sequences. Pure function using dynamic programming.

Calculate lexical drift between two texts. Pure function - no side effects.

Validate individual rules and detect conflicts.

List repositories for a GitHub organization.

Background task that reads messages from the WebSocket.

Listen for messages on a topic.

Main loop for liveness checks

Load scores from disk.

Load federation data from the directory.

Load inventory from disk.

Return a map of action key to a `uses` string with an inline comment.

Load embeddings of known adversarial prompts. In production, would use real embeddings from a threat database.

Load bias detection patterns. In production, would load from comprehensive bias datasets.

Load built-in experiment templates.

Load the last processing checkpoint.

Load risk classification config from YAML or JSON. Falls back to the bundled ``eu_ai_act_defaults.yaml``.

Load default compliance controls for enabled frameworks.

Load default policies based on policy level.

Load default safety rules. .. deprecated:: Uses built-in sample rules. For production use, load an explicit config wi

Load design proposals from file.

Load security exemptions from plugin config. Validates the exemptions file against a strict schema to prevent malicious or malformed

Load a single JSON or YAML file.

Load fixtures from a file or directory. Supports JSON and YAML. A single file may contain one fixture (object) or many (array / YAML list). A

Load documents from disk.

Load the file system state from disk.

Restore registry state from the storage file.

Load entries from persist file.

Restore registry state from the storage file.

Load HIPAA-specific policies.

Load message index from existing files.

Load known jailbreak patterns. In production, would load from comprehensive adversarial dataset.

Load the bundled policy JSON schema, returning None if unavailable.

Load PII detection patterns. In production, would use more sophisticated NER models.

Load policy files.

Load policy rules from a YAML file. Expected format:: rules: require_signature: true allowed_types: [integration, ag

Load governance policy from YAML config.

Load pending review items from file.

Load safety corrections from disk.

Load sample knowledge base articles.

Load circuit breaker state from file.

Read and validate the trusted override token from the environment. Rejects empty, short, or blacklisted (well-known/weak) values so a misconf

Load the model upgrade log from disk. Returns: Dict containing: - upgrades: List of upgrade reco

Load user preferences from disk.

Load a single YAML file, returning an empty dict on failure.

Load a YAML file. Uses PyYAML when available, else a minimal parser.

Log an access attempt.