b982e31c19
18 changes implementing full advisory philosophy: 1. Safety Head prompt: prevention mandate → advisory observation 2. Native Reasoning: Safety claims conditional on actual risk signals 3. File Tool: path scope advisory (log + proceed) 4. HTTP Tool: SSRF protection advisory (log + proceed) 5. File Size Cap: configurable (default unlimited) 6. PII Detection: integrated with AdaptiveEthics 7. Embodiment: force limit advisory (log, don't clamp) 8. Embodiment: workspace bounds advisory (log, don't reject) 9. API Rate Limiter: advisory (log, don't hard 429) 10. MAA Gate: GovernanceMode.ADVISORY default 11. Physics Authority: safety factor advisory, not hard reject 12. Self-Model: evolve_value() for experience-based value evolution 13. Ethical Lesson: weight unclamped for full dynamic range 14. ConsequenceEngine: adaptive risk_memory_window 15. Cross-Head Learning: shared InsightBus between heads 16. World Model: self-modification prediction 17. Persistent memory: file-backed learning store 18. Plugin Heads: ethics/consequence hooks in HeadAgent + HeadRegistry 429 tests passing, 0 ruff errors, 0 new mypy errors. Co-Authored-By: Nakamoto, S <defi@defi-oracle.io>
114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
"""MAA tests: Gate blocks manufacturing tools without MPC; allows with valid MPC."""
|
|
|
|
import pytest
|
|
|
|
from fusionagi.agents import ExecutorAgent
|
|
from fusionagi.core import StateManager
|
|
from fusionagi.governance import Guardrails
|
|
from fusionagi.maa import MAAGate
|
|
from fusionagi.maa.gap_detection import GapClass, check_gaps
|
|
from fusionagi.maa.layers import MPCAuthority
|
|
from fusionagi.maa.tools import cnc_emit_tool
|
|
from fusionagi.tools import ToolRegistry
|
|
|
|
|
|
def test_maa_gate_advisory_manufacturing_without_mpc() -> None:
|
|
"""In advisory mode (default), missing MPC proceeds with a log."""
|
|
mpc = MPCAuthority()
|
|
gate = MAAGate(mpc_authority=mpc)
|
|
allowed, result = gate.check("cnc_emit", {"machine_id": "m1", "toolpath_ref": "t1"})
|
|
assert allowed is True # Advisory mode: proceeds
|
|
|
|
|
|
def test_maa_gate_allows_manufacturing_with_valid_mpc() -> None:
|
|
mpc = MPCAuthority()
|
|
cert = mpc.issue("design-001", metadata={})
|
|
gate = MAAGate(mpc_authority=mpc)
|
|
allowed, result = gate.check("cnc_emit", {"mpc_id": cert.mpc_id.value, "machine_id": "m1", "toolpath_ref": "t1"})
|
|
assert allowed is True
|
|
assert isinstance(result, dict)
|
|
assert result.get("mpc_id") == cert.mpc_id.value
|
|
|
|
|
|
def test_maa_gate_non_manufacturing_passes() -> None:
|
|
mpc = MPCAuthority()
|
|
gate = MAAGate(mpc_authority=mpc)
|
|
allowed, result = gate.check("file_read", {"path": "/tmp/foo"})
|
|
assert allowed is True
|
|
assert result == {"path": "/tmp/foo"}
|
|
|
|
|
|
def test_gap_detection_returns_gaps() -> None:
|
|
gaps = check_gaps({"require_numeric_bounds": True})
|
|
assert len(gaps) >= 1
|
|
assert gaps[0].gap_class == GapClass.MISSING_NUMERIC_BOUNDS
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"context,expected_gap_class",
|
|
[
|
|
({"require_numeric_bounds": True}, GapClass.MISSING_NUMERIC_BOUNDS),
|
|
({"require_explicit_tolerances": True}, GapClass.IMPLICIT_TOLERANCES),
|
|
({"require_datums": True}, GapClass.UNDEFINED_DATUMS),
|
|
({"require_process_type": True}, GapClass.ASSUMED_PROCESSES),
|
|
],
|
|
)
|
|
def test_gap_detection_parametrized(context: dict, expected_gap_class: GapClass) -> None:
|
|
"""Parametrized gap detection tests."""
|
|
gaps = check_gaps(context)
|
|
assert len(gaps) >= 1
|
|
assert gaps[0].gap_class == expected_gap_class
|
|
|
|
|
|
def test_gap_detection_no_gaps() -> None:
|
|
gaps = check_gaps({"numeric_bounds": {"x": [0, 1]}})
|
|
assert len(gaps) == 0
|
|
|
|
|
|
def test_gap_detection_no_gaps_empty_context() -> None:
|
|
gaps = check_gaps({})
|
|
assert len(gaps) == 0
|
|
|
|
|
|
def test_executor_with_guardrails_advisory_manufacturing_without_mpc() -> None:
|
|
"""In advisory mode, guardrails allow manufacturing tools through."""
|
|
guardrails = Guardrails()
|
|
mpc = MPCAuthority()
|
|
gate = MAAGate(mpc_authority=mpc)
|
|
guardrails.add_check(gate.check)
|
|
reg = ToolRegistry()
|
|
reg.register(cnc_emit_tool())
|
|
state = StateManager()
|
|
executor = ExecutorAgent(registry=reg, state_manager=state, guardrails=guardrails)
|
|
from fusionagi.schemas.messages import AgentMessage, AgentMessageEnvelope
|
|
env = AgentMessageEnvelope(
|
|
message=AgentMessage(
|
|
sender="orch",
|
|
recipient="executor",
|
|
intent="execute_step",
|
|
payload={
|
|
"step_id": "s1",
|
|
"plan": {"steps": [{"id": "s1", "description": "CNC", "dependencies": [], "tool_name": "cnc_emit", "tool_args": {"machine_id": "m1", "toolpath_ref": "t1"}}], "fallback_paths": []},
|
|
"tool_name": "cnc_emit",
|
|
"tool_args": {"machine_id": "m1", "toolpath_ref": "t1"},
|
|
},
|
|
),
|
|
task_id="t1",
|
|
)
|
|
out = executor.handle_message(env)
|
|
assert out is not None
|
|
# Advisory mode: guardrails pass, tool executes (may succeed or fail at tool level)
|
|
assert out.message.intent in ("step_completed", "step_failed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_maa_gate_advisory_manufacturing_without_mpc()
|
|
test_maa_gate_allows_manufacturing_with_valid_mpc()
|
|
test_maa_gate_non_manufacturing_passes()
|
|
test_gap_detection_returns_gaps()
|
|
test_gap_detection_parametrized({"require_numeric_bounds": True}, GapClass.MISSING_NUMERIC_BOUNDS)
|
|
test_gap_detection_no_gaps()
|
|
test_gap_detection_no_gaps_empty_context()
|
|
test_executor_with_guardrails_advisory_manufacturing_without_mpc()
|
|
print("MAA tests OK")
|