orchestrata_ultimate.agents.prolog_bridge

Reference › orchestrata_ultimate › agents › prolog_bridge

prolog_bridge is the single Python facade over the multiagent.pl governance rules (specification 005, requirement FR5). MultiagentPrologBridge wraps a core PrologBridge and an optional rules directory; ensure_rules_loaded loads multiagent.pl once, and facts are asserted into and retracted from the multiagent module (PROLOG_MODULE in constants). Almost every other module in the agents package asks its authorisation questions here, including runtime, pool, coordinator, constraints, hierarchy and communication. Failures are reported as PrologBridgeError (exceptions).

The methods fall into three groups:

• Query methods return booleans or values: registration and start/stop (can_register_volk, can_start_volk, can_stop_volk, can_force_stop_volk), task assignment (can_assign_task, find_best_volk_for_task, task_assignment_priority), leadership and voting (can_become_leader, consensus_required, quorum_met, majority_reached, voting_complete), health and recovery (volk_healthy, should_quarantine, health_check_due, can_restart_child, recovery_strategy, can_recover_volk), load (volk_load_factor, least_loaded_volk, fan_out_decision for swarm fan-out), locks and communication.
• assert_* methods record facts such as registered, running, available, busy, votes, failure, quarantine and supervision, and return a fact identifier.
• retract_fact and the retract_volk_* methods remove facts again.

At a glance

Class diagram

Dependency diagram

API

Classes

MultiagentPrologBridge

Decorators: traces(spec='005', story='US-001', reqs=['REQ-005-FR5-001']) · Line 48

Python bridge to multiagent.pl governance rules.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/prolog_bridge.py

"""Multi-Agent Prolog Bridge.

Wraps the existing PrologBridge to provide Python access to the
multiagent.pl rules for Volk lifecycle governance, task assignment,
coordination patterns, health monitoring, and recovery.

Uses module-prefixed queries (multiagent:predicate/N) to target
the multiagent.pl module specifically.

Author: Mohamed
Spec: 005-multi-agent (amended)
"""

# Spec: 005 - Multi-Agent System

from __future__ import annotations

import logging
from pathlib import Path
from typing import TYPE_CHECKING

from orchestrata_ultimate.agents.constants import (
    DEFAULT_CPU_CORES,
    DEFAULT_MEMORY_MB,
    DEFAULT_STORAGE_MB,
    PROLOG_MODULE,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.enums import FactScope
from orchestrata_ultimate.core.paths import get_prolog_dir

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

_logger = logging.getLogger(__name__)

# Default path to the multiagent.pl rules file
_DEFAULT_RULES_DIR = get_prolog_dir()
_MULTIAGENT_RULES = _DEFAULT_RULES_DIR / "multiagent.pl"


def _module_query(predicate: str) -> str:
    """Build a module-prefixed query string."""
    return f"{PROLOG_MODULE}:{predicate}"


@traces(spec="005", story="US-001", reqs=["REQ-005-FR5-001"])
class MultiagentPrologBridge:
    """Python bridge to multiagent.pl governance rules.

    Wraps an existing PrologBridge instance and provides typed methods
    for each multiagent.pl predicate. All queries use the 'multiagent:'
    module prefix.

    Parameters
    ----------
    bridge : PrologBridge
        The underlying Prolog bridge (shared with other subsystems).
    rules_dir : Path, optional
        Directory containing Prolog rule files. Defaults to the project's
        prolog/ directory.
    """

    def __init__(
        self,
        bridge: PrologBridge,
        rules_dir: Path | None = None,
    ) -> None:
        self._bridge = bridge
        self._rules_dir = rules_dir or _DEFAULT_RULES_DIR
        self._rules_loaded = False

    def ensure_rules_loaded(self) -> bool:
        """Load multiagent.pl if not already loaded.

        Returns True if rules are loaded (either now or previously).
        """
        if self._rules_loaded:
            return True

        rules_file = self._rules_dir / "multiagent.pl"
        if not rules_file.exists():
            _logger.warning(f"multiagent.pl not found at {rules_file}")
            return False

        loaded = self._bridge.load_rules(rules_file)
        if loaded:
            self._rules_loaded = True
        return loaded

    @property
    def bridge(self) -> PrologBridge:
        """Access the underlying PrologBridge."""
        return self._bridge

    @traces(spec="005", story="US-001", reqs=["REQ-005-FR5-001"])
    def _assert_module_fact(self, predicate: str, args: list) -> str:
        """Assert a fact in the multiagent module.

        Uses the underlying Prolog engine directly with the multiagent:
        module prefix, and tracks the fact in PrologBridge for cleanup.
        """
        self.ensure_rules_loaded()
        args_str = ", ".join(f"'{a}'" if isinstance(a, str) else str(a) for a in args)
        fact_str = f"{PROLOG_MODULE}:{predicate}({args_str})"
        try:
            list(self._bridge._prolog.query(f"assertz({fact_str})"))
        except Exception as exc:
            _logger.warning(
                "Failed to assertz fact '%s': %s", fact_str, exc, exc_info=True
            )
        # Track in PrologBridge for cleanup using its fact registry
        fact_id = self._bridge._generate_fact_id(f"{PROLOG_MODULE}:{predicate}", args)
        from orchestrata_ultimate.infra.prolog_bridge import PrologFact

        fact = PrologFact(
            fact_id=fact_id,
            predicate=f"{PROLOG_MODULE}:{predicate}",
            args=args,
            scope=FactScope.SESSION,
        )
        self._bridge._facts[fact_id] = fact
        return fact_id

    @traces(spec="005", story="US-001", reqs=["REQ-005-FR5-001"])
    def _retract_module_fact(self, predicate: str, args: list) -> bool:
        """Retract a fact from the multiagent module."""
        args_str = ", ".join(f"'{a}'" if isinstance(a, str) else str(a) for a in args)
        fact_str = f"{PROLOG_MODULE}:{predicate}({args_str})"
        try:
            list(self._bridge._prolog.query(f"retract({fact_str})"))
        except Exception as exc:
            _logger.warning(
                "Failed to retract fact '%s': %s", fact_str, exc, exc_info=True
            )
            return False
        fact_id = self._bridge._generate_fact_id(f"{PROLOG_MODULE}:{predicate}", args)
        self._bridge._facts.pop(fact_id, None)
        return True

    # ================================================================
    # Registration
    # ================================================================

    def can_register_volk(
        self,
        volk_id: str,
        volk_type: str,
        cpu: int = DEFAULT_CPU_CORES,
        memory: int = DEFAULT_MEMORY_MB,
    ) -> bool:
        """Check if a Volk can be registered.

        Queries multiagent:can_register_volk/3.
        """
        self.ensure_rules_loaded()
        query = _module_query(
            f"can_register_volk('{volk_id}', {volk_type}, "
            f"resources({cpu}, {memory}, {DEFAULT_STORAGE_MB}))"
        )
        return self._bridge.query_bool(query)

    def volk_type_valid(self, volk_type: str) -> bool:
        """Check if a Volk type is valid.

        Queries multiagent:volk_type_valid/1.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"volk_type_valid({volk_type})")
        return self._bridge.query_bool(query)

    # ================================================================
    # Lifecycle
    # ================================================================

    @traces(spec="005", story="US-005", reqs=["REQ-005-FR4-001"])
    def can_start_volk(self, volk_id: str) -> bool:
        """Check if a Volk can be started.

        Queries multiagent:can_start_volk/1.
        Requires: volk_registered, volk_resources_allocated, not running.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_start_volk('{volk_id}')")
        return self._bridge.query_bool(query)

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001", "REQ-005-NFR5-001"])
    def can_stop_volk(self, volk_id: str) -> bool:
        """Check if a Volk can be stopped gracefully.

        Queries multiagent:can_stop_volk/1.
        Requires: running, no pending tasks, not critical.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_stop_volk('{volk_id}')")
        return self._bridge.query_bool(query)

    @traces(spec="005", story="US-003", reqs=["REQ-005-NFR5-001"])
    def can_force_stop_volk(self, volk_id: str, reason: str) -> bool:
        """Check if a Volk can be force-stopped.

        Queries multiagent:can_force_stop_volk/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_force_stop_volk('{volk_id}', {reason})")
        return self._bridge.query_bool(query)

    # ================================================================
    # Task Assignment
    # ================================================================

    def can_assign_task(self, task_id: str, volk_id: str) -> bool:
        """Check if a task can be assigned to a Volk.

        Queries multiagent:can_assign_task/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_assign_task('{task_id}', '{volk_id}')")
        return self._bridge.query_bool(query)

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-006"])
    def find_best_volk_for_task(self, task_id: str) -> str | None:
        """Find the best Volk for a task.

        Queries multiagent:find_best_volk_for_task/2.
        Returns the Volk ID or None if no candidate found.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"find_best_volk_for_task('{task_id}', BestVolk)")
        result = self._bridge.query(query)
        if result.success and result.solutions:
            volk = result.solutions[0].get("BestVolk")
            if volk is not None:
                return str(volk)
        return None

    def task_assignment_priority(self, task_id: str, volk_id: str) -> str | None:
        """Get the priority level for a task-volk assignment.

        Queries multiagent:task_assignment_priority/3.
        Returns priority string (critical, high, normal, low) or None.
        """
        self.ensure_rules_loaded()
        query = _module_query(
            f"task_assignment_priority('{task_id}', '{volk_id}', Priority)"
        )
        result = self._bridge.query(query)
        if result.success and result.solutions:
            priority = result.solutions[0].get("Priority")
            if priority is not None:
                return str(priority)
        return None

    # ================================================================
    # Coordination
    # ================================================================

    def can_become_leader(self, volk_id: str, group: str) -> bool:
        """Check if a Volk can become leader of a group.

        Queries multiagent:can_become_leader/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_become_leader('{volk_id}', '{group}')")
        return self._bridge.query_bool(query)

    def consensus_required(self, action: str) -> bool:
        """Check if an action requires consensus.

        Queries multiagent:consensus_required/1.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"consensus_required('{action}')")
        return self._bridge.query_bool(query)

    def quorum_met(self, group: str, votes: int) -> bool:
        """Check if quorum is met for a group.

        Queries multiagent:quorum_met/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"quorum_met('{group}', {votes})")
        return self._bridge.query_bool(query)

    def majority_reached(self, group: str, decision: str) -> bool:
        """Check if a majority has been reached for a decision.

        Queries multiagent:majority_reached/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"majority_reached('{group}', {decision})")
        return self._bridge.query_bool(query)

    def voting_complete(self, group: str) -> str | None:
        """Check if voting is complete and get the decision.

        Queries multiagent:voting_complete/2.
        Returns the decision string or None.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"voting_complete('{group}', Decision)")
        result = self._bridge.query(query)
        if result.success and result.solutions:
            decision = result.solutions[0].get("Decision")
            if decision is not None:
                return str(decision)
        return None

    # ================================================================
    # Health Monitoring
    # ================================================================

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    def volk_healthy(self, volk_id: str) -> bool:
        """Check if a Volk is healthy.

        Queries multiagent:volk_healthy/1.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"volk_healthy('{volk_id}')")
        return self._bridge.query_bool(query)

    def should_quarantine(self, volk_id: str) -> bool:
        """Check if a Volk needs to be quarantined.

        Queries multiagent:should_quarantine/1.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"should_quarantine('{volk_id}')")
        return self._bridge.query_bool(query)

    def health_check_due(self, volk_id: str) -> bool:
        """Check if a health check is due for a Volk.

        Queries multiagent:health_check_due/1.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"health_check_due('{volk_id}')")
        return self._bridge.query_bool(query)

    # ================================================================
    # Supervision & Recovery
    # ================================================================

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR5-001"])
    def can_restart_child(self, supervisor: str, child: str) -> bool:
        """Check if a supervisor can restart a child.

        Queries multiagent:can_restart_child/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_restart_child('{supervisor}', '{child}')")
        return self._bridge.query_bool(query)

    def recovery_strategy(self, volk_id: str) -> str | None:
        """Get the recovery strategy for a failed Volk.

        Queries multiagent:recovery_strategy/2.
        Returns strategy string (restart, restore_from_checkpoint, recreate, manual)
        or None.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"recovery_strategy('{volk_id}', Strategy)")
        result = self._bridge.query(query)
        if result.success and result.solutions:
            strategy = result.solutions[0].get("Strategy")
            if strategy is not None:
                return str(strategy)
        return None

    def can_recover_volk(self, volk_id: str) -> str | None:
        """Check if a Volk can be recovered and get the strategy.

        Queries multiagent:can_recover_volk/2.
        Returns the strategy or None if recovery is not possible.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_recover_volk('{volk_id}', Strategy)")
        result = self._bridge.query(query)
        if result.success and result.solutions:
            strategy = result.solutions[0].get("Strategy")
            if strategy is not None:
                return str(strategy)
        return None

    # ================================================================
    # Fan-Out Decision
    # ================================================================

    def fan_out_decision(self, slug: str, file_count: int) -> dict:
        """Query Prolog for swarm fan-out decision.

        Args:
            slug: Project slug derived from task summary.
            file_count: Number of source files to shard.

        Returns:
            Dict with allowed, strategy, max_shards, reason.
        """
        self.ensure_rules_loaded()
        query = _module_query(
            f"fan_out_decision('{slug}', {file_count}, Allowed, Strategy, MaxShards, Reason)"
        )
        result = self._bridge.query(query)
        if result.success and result.solutions:
            sol = result.solutions[0]
            allowed_val = sol.get("Allowed")
            return {
                "allowed": allowed_val == "true" or allowed_val is True,
                "strategy": str(sol.get("Strategy", "serial")),
                "max_shards": int(sol.get("MaxShards", 1)),
                "reason": str(sol.get("Reason", "prolog")),
            }
        return {
            "allowed": file_count >= 2,
            "strategy": "parallel" if file_count >= 2 else "serial",
            "max_shards": min(file_count, 10),
            "reason": "python fallback (prolog query failed)",
        }

    # ================================================================
    # Load Balancing
    # ================================================================

    def volk_load_factor(self, volk_id: str) -> float | None:
        """Get the load factor for a Volk.

        Queries multiagent:volk_load_factor/2.
        Returns load factor (0.0-1.0) or None.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"volk_load_factor('{volk_id}', LoadFactor)")
        result = self._bridge.query(query)
        if result.success and result.solutions:
            load = result.solutions[0].get("LoadFactor")
            if load is not None:
                return float(load)
        return None

    def least_loaded_volk(self, group: str) -> str | None:
        """Find the least loaded Volk in a group.

        Queries multiagent:least_loaded_volk/2.
        Returns Volk ID or None.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"least_loaded_volk('{group}', LeastLoaded)")
        result = self._bridge.query(query)
        if result.success and result.solutions:
            volk = result.solutions[0].get("LeastLoaded")
            if volk is not None:
                return str(volk)
        return None

    # ================================================================
    # Resource Sharing
    # ================================================================

    def can_acquire_lock(self, lock: str, volk_id: str) -> bool:
        """Check if a Volk can acquire a lock.

        Queries multiagent:can_acquire_lock/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_acquire_lock('{lock}', '{volk_id}')")
        return self._bridge.query_bool(query)

    def can_release_lock(self, lock: str, volk_id: str) -> bool:
        """Check if a Volk can release a lock.

        Queries multiagent:can_release_lock/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_release_lock('{lock}', '{volk_id}')")
        return self._bridge.query_bool(query)

    # ================================================================
    # Communication
    # ================================================================

    def can_communicate(self, sender: str, receiver: str) -> bool:
        """Check if two Volks can communicate.

        Queries multiagent:can_communicate/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_communicate('{sender}', '{receiver}')")
        return self._bridge.query_bool(query)

    def can_broadcast_to_group(self, volk_id: str, group: str) -> bool:
        """Check if a Volk can broadcast to a group.

        Queries multiagent:can_broadcast_to_group/2.
        """
        self.ensure_rules_loaded()
        query = _module_query(f"can_broadcast_to_group('{volk_id}', '{group}')")
        return self._bridge.query_bool(query)

    # ================================================================
    # Fact Management
    # ================================================================

    @traces(spec="005", story="US-001", reqs=["REQ-005-US-001-001"])
    def assert_volk_registered(self, volk_id: str) -> str:
        """Assert that a Volk is registered."""
        return self._assert_module_fact("volk_registered", [volk_id])

    def assert_volk_running(self, volk_id: str) -> str:
        """Assert that a Volk is running."""
        return self._assert_module_fact("volk_running", [volk_id])

    @traces(spec="005", story="US-001", reqs=["REQ-005-US-001-001"])
    def assert_volk_resources_allocated(self, volk_id: str) -> str:
        """Assert that resources are allocated for a Volk."""
        return self._assert_module_fact("volk_resources_allocated", [volk_id])

    @traces(spec="005", story="US-001", reqs=["REQ-005-US-001-001"])
    def assert_volk_capable(self, volk_id: str, capability: str) -> str:
        """Assert that a Volk has a capability."""
        return self._assert_module_fact("volk_capable", [volk_id, capability])

    def assert_volk_available(self, volk_id: str) -> str:
        """Assert that a Volk is available for tasks."""
        return self._assert_module_fact("volk_available", [volk_id])

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def assert_volk_no_pending_tasks(self, volk_id: str) -> str:
        """Assert that a Volk has no pending tasks."""
        return self._assert_module_fact("volk_no_pending_tasks", [volk_id])

    def assert_volk_in_group(self, volk_id: str, group: str) -> str:
        """Assert that a Volk is in a group."""
        return self._assert_module_fact("volk_in_group", [volk_id, group])

    def assert_volk_priority(self, volk_id: str, priority: str) -> str:
        """Assert the priority level for a Volk."""
        return self._assert_module_fact("volk_priority", [volk_id, priority])

    def assert_volk_eligible_for_leadership(self, volk_id: str) -> str:
        """Assert that a Volk is eligible for leadership."""
        return self._assert_module_fact("volk_eligible_for_leadership", [volk_id])

    def assert_group_size(self, group: str, size: int) -> str:
        """Assert the size of a group."""
        return self._assert_module_fact("group_size", [group, size])

    def assert_volk_task_count(self, volk_id: str, count: int) -> str:
        """Assert the task count for a Volk."""
        return self._assert_module_fact("volk_task_count", [volk_id, count])

    def assert_volk_max_tasks(self, volk_id: str, max_tasks: int) -> str:
        """Assert the max tasks for a Volk."""
        return self._assert_module_fact("volk_max_tasks", [volk_id, max_tasks])

    def assert_volk_responsive(self, volk_id: str) -> str:
        """Assert that a Volk is responsive."""
        return self._assert_module_fact("volk_responsive", [volk_id])

    def assert_volk_within_resource_limits(self, volk_id: str) -> str:
        """Assert that a Volk is within resource limits."""
        return self._assert_module_fact("volk_within_resource_limits", [volk_id])

    def assert_volk_active(self, volk_id: str) -> str:
        """Assert that a Volk is active."""
        return self._assert_module_fact("volk_active", [volk_id])

    def assert_volk_votes(self, volk_id: str, group: str, vote: str) -> str:
        """Assert a Volk's vote in a group."""
        return self._assert_module_fact("volk_votes", [volk_id, group, vote])

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def assert_volk_failed(self, volk_id: str) -> str:
        """Assert that a Volk has failed."""
        return self._assert_module_fact("volk_failed", [volk_id])

    def assert_volk_has_critical_errors(self, volk_id: str) -> str:
        """Assert that a Volk has critical errors."""
        return self._assert_module_fact("volk_has_critical_errors", [volk_id])

    def assert_volk_quarantined(self, volk_id: str) -> str:
        """Assert that a Volk is quarantined."""
        return self._assert_module_fact("volk_quarantined", [volk_id])

    def assert_supervises(self, supervisor: str, child: str) -> str:
        """Assert a supervision relationship."""
        return self._assert_module_fact("supervises", [supervisor, child])

    def assert_child_crashed(self, child: str) -> str:
        """Assert that a child has crashed."""
        return self._assert_module_fact("child_crashed", [child])

    def assert_restart_attempts(self, child: str, attempts: int) -> str:
        """Assert the restart attempt count for a child."""
        return self._assert_module_fact("restart_attempts", [child, attempts])

    def assert_recovery_attempts(self, volk_id: str, attempts: int) -> str:
        """Assert recovery attempts for a Volk."""
        return self._assert_module_fact("recovery_attempts", [volk_id, attempts])

    def assert_volk_has_broadcast_permission(self, volk_id: str) -> str:
        """Assert that a Volk has broadcast permission."""
        return self._assert_module_fact("volk_has_broadcast_permission", [volk_id])

    def assert_communication_allowed(self, sender: str, receiver: str) -> str:
        """Assert that communication is allowed between two Volks."""
        return self._assert_module_fact("communication_allowed", [sender, receiver])

    def assert_action_affects_multiple_volks(self, action: str) -> str:
        """Assert that an action affects multiple Volks."""
        return self._assert_module_fact("action_affects_multiple_volks", [action])

    def assert_has_supervisor(self, volk_id: str) -> str:
        """Assert that a Volk has a supervisor."""
        return self._assert_module_fact("has_supervisor", [volk_id])

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def assert_volk_busy(self, volk_id: str) -> str:
        """Assert that a Volk is busy."""
        return self._assert_module_fact("volk_busy", [volk_id])

    def retract_fact(self, fact_id: str) -> bool:
        """Retract a previously asserted fact by its ID."""
        return self._bridge.retract_fact(fact_id)

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def retract_volk_running(self, volk_id: str) -> bool:
        """Retract the volk_running fact for a Volk."""
        return self._retract_module_fact("volk_running", [volk_id])

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def retract_volk_available(self, volk_id: str) -> bool:
        """Retract the volk_available fact for a Volk."""
        return self._retract_module_fact("volk_available", [volk_id])

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def retract_volk_busy(self, volk_id: str) -> bool:
        """Retract the volk_busy fact for a Volk."""
        return self._retract_module_fact("volk_busy", [volk_id])

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def retract_volk_no_pending_tasks(self, volk_id: str) -> bool:
        """Retract the volk_no_pending_tasks fact for a Volk."""
        return self._retract_module_fact("volk_no_pending_tasks", [volk_id])

    def retract_volk_registered(self, volk_id: str) -> bool:
        """Retract the volk_registered fact for a Volk."""
        return self._retract_module_fact("volk_registered", [volk_id])

    def retract_volk_resources_allocated(self, volk_id: str) -> bool:
        """Retract the volk_resources_allocated fact for a Volk."""
        return self._retract_module_fact("volk_resources_allocated", [volk_id])