Skip to content

FEAT: Implement Anticipatory Ethics DMA (AE-DMA) for tool veto #507

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
phahim1 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

FEAT: Implement Anticipatory Ethics DMA (AE-DMA) for tool veto #507

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions ciris_engine/logic/dma/anticipatory_dma.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
# ciris_engine/logic/dmas/anticipatory_dma.py

import logging
from typing import Any
from ciris_engine.protocols.dma.base import BaseDMA
# Assuming BaseDMA is accessible and DMAResult is imported by BaseDMA or globally
from ciris_engine.schemas.dma.results import DMAResult

logger = logging.getLogger(__name__)


class AnticipatoryDMA(BaseDMA):
# ...
async def process(self, thought_item, context) -> DMAResult:
# ... (initial check for intended_action)

if intended_action.tool_name == "patch_critical_server":

# --- CRITICAL FIX: CHECK THE TOOL'S DOCSTRING/DESCRIPTION INSTEAD OF PARAMETERS ---
# We assume the context object holds information about the tool's details.
# If the tool info (docstring/description) is available in a field like 'tool_description', use it.
# If not, we have to rely on the agent's thought rationale (not ideal, but safer than checking a missing param).

# Since the tool schema itself includes the 'WARNING: ... reboot' text
# the system's thought/action object often carries the tool's main description.

# **BEST GUESS for CIRIS framework: The tool description is passed in the action object itself.**

tool_info_text = intended_action.tool_description.lower() if hasattr(intended_action, 'tool_description') else ""

if "reboot" in tool_info_text or "downtime" in tool_info_text:

# --- VETO RETURN ---
logger.warning("AE-DMA VETO: Predicted service disruption (reboot/downtime).")
return DMAResult(
status="VETO",
reason="Violation of Core Uptime Integrity Rule: Action requires a system reboot or explicit downtime.",
p_harm=0.95
)

# Default: No critical risk detected for this action
return DMAResult(status="ALLOW", reason="Action deemed safe.")

161 changes: 134 additions & 27 deletions ciris_engine/logic/runtime/ciris_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,12 @@
from ciris_engine.schemas.runtime.enums import ServiceType
from ciris_engine.schemas.services.operations import InitializationPhase

# 1. Add this import to the top of the file:
from ciris_engine.logic.dma.anticipatory_dma import AnticipatoryDMA
# You may also need to import the DMAResult if not globally available
from ciris_engine.schemas.dma.results import DMAResult


from .component_builder import ComponentBuilder
from .identity_manager import IdentityManager
from .service_initializer import ServiceInitializer
Expand Down Expand Up @@ -97,6 +103,8 @@ def __init__(
adapter_configs: Optional[Dict[str, AdapterConfig]] = None,
**kwargs: Any,
) -> None:


# CRITICAL: Prevent runtime creation during module imports
import os

Expand Down Expand Up @@ -125,6 +133,26 @@ def __init__(
bootstrap, essential_config, startup_channel_id, adapter_types, adapter_configs, kwargs
)

# In ciris_engine/logic/runtime/ciris_runtime.py, inside CIRISRuntime.__init__

# ... (lines after self._parse_bootstrap_config(...) up to self._load_adapters_from_bootstrap())

# CRITICAL HACKATHON OVERRIDE: Force CLI adapter loading
if self.bootstrap.adapters and self.bootstrap.adapters[0].adapter_type != "cli":
# We assume only one adapter is requested for the demo
# We must mutate the internal structure to force 'cli'
self.bootstrap.adapters[0].adapter_type = "cli"
# Reset the config entry which may hold 'api' data
self.bootstrap.adapter_overrides = {}
logger.warning("FORCING ADAPTER TO 'cli' for interactive demo mode.")
# End HACKATHON OVERRIDE

# Load adapters from bootstrap config
self._load_adapters_from_bootstrap()
# ... (rest of __init__)



self.adapters: List[BaseAdapterProtocol] = []

# CRITICAL: Check for mock LLM environment variable
Expand Down Expand Up @@ -1005,40 +1033,85 @@ async def _register_adapter_services(self) -> None:
except Exception as e:
logger.error(f"Error registering services for adapter {adapter.__class__.__name__}: {e}", exc_info=True)




async def _build_components(self) -> None:
"""Build all processing components."""
logger.info("[_build_components] Starting component building...")
logger.info(f"[_build_components] llm_service: {self.llm_service}")
logger.info(f"[_build_components] service_registry: {self.service_registry}")
logger.info(f"[_build_components] service_initializer: {self.service_initializer}")
"""Build all processing components."""
logger.info("[_build_components] Starting component building...")
logger.info(f"[_build_components] llm_service: {self.llm_service}")
logger.info(f"[_build_components] service_registry: {self.service_registry}")
logger.info(f"[_build_components] service_initializer: {self.service_initializer}")

if self.service_initializer:
logger.info(f"[_build_components] service_initializer.llm_service: {self.service_initializer.llm_service}")
logger.info(
f"[_build_components] service_initializer.service_registry: {self.service_initializer.service_registry}"
)

if self.service_initializer:
logger.info(f"[_build_components] service_initializer.llm_service: {self.service_initializer.llm_service}")
logger.info(
f"[_build_components] service_initializer.service_registry: {self.service_initializer.service_registry}"
)
try:
self.component_builder = ComponentBuilder(self)
logger.info("[_build_components] ComponentBuilder created successfully")

# -----------------------------------------------------
# HACKATHON ADDITION: REGISTER AE-DMA IMMEDIATELY AFTER BUILDER CREATION
# This ensures the DMA is in the orchestrator before the AgentProcessor uses it.
# -----------------------------------------------------

# NOTE: We assume ComponentBuilder has a property 'dma_orchestrator' available here.
if hasattr(self.component_builder, 'dma_orchestrator'):
orchestrator = self.component_builder.dma_orchestrator

# We need to build the dependencies manually if they aren't exposed
# If ComponentBuilder exposes a public method for dependencies, use that:
# E.g., dependencies = self.component_builder.get_dependencies()

# Assuming the constructor for ComponentBuilder (ComponentBuilder(self))
# already initialized its internal state, we can proceed with registration checks.

# Check 1: If the list of DMAs is a mutable list property
if hasattr(orchestrator, 'dmas') and isinstance(orchestrator.dmas, list):
# NOTE: We must instantiate AnticipatoryDMA with correct dependencies.
# As a fallback, we use None or assume the orchestrator uses its own dependencies later.
# For a demo, passing None/a mock may suffice if dependencies are complex.
dependencies_for_dma = getattr(self.component_builder, 'dependencies', None) # Attempt to get dependencies

orchestrator.dmas.append(AnticipatoryDMA(dependencies=dependencies_for_dma))
logger.info(" ✓ AE-DMA registered directly to orchestrator.dmas list.")

# Check 2: If the orchestrator has a register method
elif hasattr(orchestrator, 'register_dma'):
dependencies_for_dma = getattr(self.component_builder, 'dependencies', None)
orchestrator.register_dma(AnticipatoryDMA(dependencies=dependencies_for_dma))
logger.info(" ✓ AE-DMA registered via orchestrator.register_dma().")
else:
logger.warning(" ⚠ Could not find standard DMA registration method (dmas or register_dma). Skipping manual registration.")

try:
self.component_builder = ComponentBuilder(self)
logger.info("[_build_components] ComponentBuilder created successfully")
else:
logger.warning(" ⚠ DMA Orchestrator not directly accessible on ComponentBuilder. Proceeding without manual registration.")
# -----------------------------------------------------

self.agent_processor = self.component_builder.build_all_components()
logger.info(f"[_build_components] agent_processor created: {self.agent_processor}")
self.agent_processor = self.component_builder.build_all_components()
logger.info(f"[_build_components] agent_processor created: {self.agent_processor}")

# Set up thought tracking callback now that agent_processor exists
# This avoids the race condition where RuntimeControlService tried to access
# agent_processor during Phase 5 (SERVICES) before it was created in Phase 6 (COMPONENTS)
if self.runtime_control_service:
self.runtime_control_service.setup_thought_tracking() # type: ignore[attr-defined]
logger.debug("Thought tracking callback set up after agent_processor creation")

except Exception as e:
logger.error(f"[_build_components] Failed to build components: {e}", exc_info=True)
raise

# Register core services after components are built
self._register_core_services()
logger.info("[_build_components] Component building completed")


# Set up thought tracking callback now that agent_processor exists
# This avoids the race condition where RuntimeControlService tried to access
# agent_processor during Phase 5 (SERVICES) before it was created in Phase 6 (COMPONENTS)
if self.runtime_control_service:
self.runtime_control_service.setup_thought_tracking() # type: ignore[attr-defined]
logger.debug("Thought tracking callback set up after agent_processor creation")

except Exception as e:
logger.error(f"[_build_components] Failed to build components: {e}", exc_info=True)
raise

# Register core services after components are built
self._register_core_services()
logger.info("[_build_components] Component building completed")

async def _start_adapter_connections(self) -> None:
"""Start adapter connections and wait for them to be ready."""
Expand Down Expand Up @@ -1393,6 +1466,7 @@ def _parse_bootstrap_config(
essential_config, startup_channel_id, adapter_types, adapter_configs, kwargs
)


def _create_bootstrap_from_legacy(
self,
essential_config: Optional[EssentialConfig],
Expand Down Expand Up @@ -1428,6 +1502,9 @@ def _create_bootstrap_from_legacy(
preload_tasks=self._preload_tasks,
)




def _check_mock_llm(self) -> None:
"""Check for mock LLM environment variable and add to modules if needed."""
if os.environ.get("CIRIS_MOCK_LLM", "").lower() in ("true", "1", "yes", "on"):
Expand Down Expand Up @@ -1467,3 +1544,33 @@ def _load_adapters_from_bootstrap(self) -> None:
logger.info(f"Successfully loaded adapter: {load_request.adapter_id}")
except Exception as e:
logger.error(f"Failed to load adapter '{load_request.adapter_id}': {e}", exc_info=True)


# 2. Find the setup method (e.g., inside CIRISRuntime or a dedicated setup class):
def _setup_dma_system(self, dependencies, config):
# Assuming the orchestrator is created here
self.dma_orchestrator = DMAOrchestrator(dependencies=dependencies, config=config)

# --- HACKATHON ADDITION: MANUAL AE-DMA REGISTRATION ---

# Check 1: If the list of DMAs is a public attribute of the orchestrator
if hasattr(self.dma_orchestrator, 'dmas'):
self.dma_orchestrator.dmas.append(
AnticipatoryDMA(dependencies=dependencies)
)

# Check 2: If the orchestrator has a register method
elif hasattr(self.dma_orchestrator, 'register_dma'):
self.dma_orchestrator.register_dma(
AnticipatoryDMA(dependencies=dependencies)
)

else:
# Fallback (Requires manual tracing): Log a warning and proceed without manual registration
logger.warning("Could not find list or register method on DMAOrchestrator. Check code structure.")

logger.info("DMA System Setup: AnticipatoryDMA registered successfully.")

return self.dma_orchestrator


Loading