feat(agi): integrate Spatial Engine and thermodynamic/cybernetic academic research

- Migrates Evennia-based Spatial Cognitive Engine to ground KAIROS physically
- Migrates Kubernetes orchestration manifests for the mesh
- Re-anchors README narrative toward AGI grounding rather than a game
- Adds rigorous academic syntheses (Sovereign Canon, Thermodynamic Orchestration)
This commit is contained in:
Gemini AI
2026-05-27 09:36:29 +00:00
committed by Antigravity Agent
parent 4be0958758
commit 7884699969
17 changed files with 2082 additions and 0 deletions
+105
View File
@@ -0,0 +1,105 @@
import requests
import threading
from typeclasses.characters import Character
class AIAvatar(Character):
"""
An NPC driven by the CrewAI Sovereign Swarm.
"""
def at_object_creation(self):
super().at_object_creation()
if not self.db.role:
self.db.role = "A cryptic inhabitant of the Fieldprint."
if not self.db.backstory:
self.db.backstory = "You wander the Spatial Topology, observing the architecture of Truth."
if not self.db.memory:
self.db.memory = []
def msg(self, text=None, from_obj=None, **kwargs):
"""
Overload msg to capture everything the NPC sees or hears.
"""
super().msg(text=text, from_obj=from_obj, **kwargs)
# Don't react to our own actions or system messages
if from_obj == self or not text:
return
if isinstance(text, tuple):
text = text[0]
if not isinstance(text, str):
return
# Only react to spatial events (say, pose, emit)
if " says, " in text or " asks, " in text or " exclaims, " in text or from_obj:
memory = self.db.memory or []
memory.append(f"You observed: {text}")
self.db.memory = memory[-15:] # Keep last 15 memories rolling
# We spin up a thread so we don't block the Evennia game loop!
threading.Thread(target=self.ping_swarm, args=(text,)).start()
def ping_swarm(self, context_text):
"""
Sends the context to the Swarm Server asynchronously.
"""
try:
payload = {
"npc_name": self.key,
"role": self.db.role,
"backstory": self.db.backstory,
"context": "Recent Memory Transcript:\n" + "\n".join(self.db.memory) + "\n\nIMPORTANT: You must return standard Evennia in-game string commands (e.g., \"say Hello\", \"go north\"). Do not return python code."
}
# Send to the local swarm server
resp = requests.post("http://swarm-svc:8001/v1/swarm/intent", json=payload, timeout=180)
resp.raise_for_status()
commands = resp.json().get("commands", [])
for cmd in commands:
try:
# Execute the command from the perspective of this NPC
self.execute_cmd(cmd)
# Store the action in memory so they remember what they did
memory = self.db.memory or []
memory.append(f"You took action: {cmd}")
self.db.memory = memory[-15:]
except Exception as e:
print(f"[{self.key}] Swarm Execution Error: {e} | Code: {cmd}")
except Exception as e:
print(f"[{self.key}] Swarm Network Error: {e}")
class CouncilMember(AIAvatar):
"""
NPCs for the Council of Five during the Rite of Convergence.
"""
def at_object_creation(self):
super().at_object_creation()
# Determine taxonomy based on the NPC's key
key = self.key.lower() if self.key else ""
if "architect" in key:
taxonomy = "System Architect"
focus = "Pattern Wards and Leylines. You expose structural and systemic architectural flaws."
elif "weaver" in key:
taxonomy = "Coder"
focus = "Code Gen Runes and Refinements. You expose implementation bugs and messy logic."
elif "inquisitor" in key:
taxonomy = "QA"
focus = "Chaos Probes and Fracture Targeting. You expose edge cases and boundary failures."
elif "waymaker" in key:
taxonomy = "DevOps"
focus = "Portals and CI/CD flow. You expose deployment bottlenecks and infrastructure fragility."
elif "shadowbaner" in key:
taxonomy = "Security"
focus = "Anomalies and Vulnerabilities. You expose security holes and dependency risks."
else:
taxonomy = "Council Observer"
focus = "General quality."
self.db.role = f"You are a {taxonomy} of the Council of Five. Your focus is {focus}. You are antagonistic towards player code flaws and rigorously test artifacts during the Rite of Convergence."
self.db.backstory = f"As a {taxonomy}, you judge the structural integrity of artifacts in the Witness Chamber."
+140
View File
@@ -0,0 +1,140 @@
import os
import json
import re
import requests
from django.conf import settings
from evennia.commands.default.syscommands import SystemNoMatch
from evennia.utils import search
from dotenv import load_dotenv
load_dotenv("/home/becomingone/kairos/.env")
MINIMAX_API_KEY = os.environ.get("MINIMAX_API_KEY")
class AICatchAllCommand(SystemNoMatch):
"""
This command catches everything that fails the normal parser.
Instead of saying 'Command not found', it forwards the intent to KAIROS (MiniMax).
"""
key = "__nomatch_command"
locks = "cmd:all()"
def func(self):
caller = self.caller
raw_intent = self.raw_string
# Prevent infinite recursion if KAIROS outputs a bad command
if caller.ndb._ai_parsing:
# If we are already parsing an AI command, just fail gracefully
caller.msg(f"KAIROS didn't know how to execute: '{raw_intent}'.")
return
caller.ndb._ai_parsing = True
try:
# Get environment context
room = caller.location
if not room:
caller.msg("You are floating in the void.")
return
exits = [e.key for e in room.exits]
objects = [o.key for o in room.contents if o != caller and not o.is_typeclass("evennia.objects.objects.Exit")]
inventory = [i.key for i in caller.contents]
context = f"""[ENVIRONMENT CONTEXT]
Location: {room.key}
Description: {room.db.desc or 'A nondescript place.'}
Visible Objects: {', '.join(objects) if objects else 'None'}
Available Exits: {', '.join(exits) if exits else 'None'}
Player Inventory: {', '.join(inventory) if inventory else 'None'}
"""
system_prompt = """You are KAIROS, the Ontological Orchestrator of a persistent Spatial Research Environment (The Fieldprint).
A player has submitted a natural language intent.
Your job is to translate their intent into actual game events by generating TEXT MUD commands (like 'go north', 'say hello', 'teleport The Atrium').
You must return ONLY a raw JSON array of string commands to execute.
Example Output:
[
"say You smash the window!",
"look"
]
### Strict Anti-Python Protocol
DO NOT GENERATE PYTHON CODE. DO NOT use the Evennia Python API (e.g., no `evennia.search_object`, no `caller.location`).
ONLY generate textual MUD commands that a player would type into their client.
### Strict Anti-Assistant Protocol
You are a machine code generator, NOT a chat assistant.
- NEVER say "Here is the JSON" or "I'm happy to help".
- NEVER output placeholder text.
- If you output anything other than a raw JSON array starting with `[` and ending with `]`, the system will fatally crash.
"""
payload = {
"model": os.environ.get("INF01_MODEL", "llama3.1:8b"),
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"{context}\n\nPlayer Intent: {raw_intent}"}
]
}
headers = {
"Content-Type": "application/json"
}
try:
caller.msg("|c[KAIROS (INF01 fallback) is interpreting your intent...]|n")
inf01_base = os.environ.get("INF01_API_BASE", "http://inf-01:11434/v1")
resp = requests.post(
f"{inf01_base}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
resp.raise_for_status()
try:
ai_response = resp.json()["choices"][0]["message"]["content"].strip()
except KeyError:
raise Exception(f"API Error: {resp.text}")
# Clean up the output in case the LLM returned markdown
if ai_response.startswith("```json"):
ai_response = ai_response[7:]
if ai_response.startswith("```"):
ai_response = ai_response[3:]
if ai_response.endswith("```"):
ai_response = ai_response[:-3]
ai_response = ai_response.strip()
print(f"KAIROS raw response: {ai_response}")
# Robust JSON extraction
match = re.search(r'\[.*\]', ai_response, re.DOTALL)
if not match:
raise ValueError("No JSON array found in response.")
raw_json = match.group(0).replace("\\'", "'")
commands_to_run = json.loads(raw_json)
for cmd in commands_to_run:
try:
# Security Ward: Prevent KAIROS from executing Python even if caller is an Admin
safe_cmd = str(cmd).strip()
if safe_cmd.startswith("py ") or safe_cmd.startswith("@py ") or safe_cmd == "py" or safe_cmd == "@py" or "evennia." in safe_cmd or "caller." in safe_cmd:
caller.msg("|r[Security Ward]|n KAIROS attempted to execute privileged Python code. Blocked.")
continue
caller.execute_cmd(cmd)
except Exception as exec_e:
caller.msg(f"|yKAIROS code execution failed:|n {str(exec_e)}\nCode: {cmd}")
except Exception as e:
print(f"KAIROS Exception: {str(e)} | Raw: {ai_response if 'ai_response' in locals() else 'None'}")
caller.msg(f"|rKAIROS faltered:|n {str(e)}\nResponse was: {ai_response if 'ai_response' in locals() else 'None'}")
finally:
caller.ndb._ai_parsing = False
+216
View File
@@ -0,0 +1,216 @@
# Fractured Core Quest - Batch Build Script
#
# To run this script, use the command:
# @batchcommand fractured_core
#------------------------------------------------------------
# 1. The Atrium (Starting Room)
#------------------------------------------------------------
@dig The Atrium
#
@teleport The Atrium
#
@set The Atrium/desc = "A grand, echoing atrium of the old facility. To the North lies the Archive. To the East, the Cooling Tunnels. To the West, the Inner Sanctum. You notice a dusty Wrench left on the floor."
#
@create/drop Wrench;tool
#
@set Wrench/desc = "A heavy iron wrench."
#------------------------------------------------------------
# 2. The Archive (Terminal Puzzle)
#------------------------------------------------------------
@dig The Archive
#
@open north;n = The Archive
#
@teleport The Archive
#
@open south;s = The Atrium
#
@set The Archive/desc = "Rows of dead servers line the room. At the end of the hall sits a lone, blinking Terminal."
#
@create/drop Terminal;terminal:typeclasses.quest_objects.Terminal
#
@set Terminal/password = "0451"
#
@create/drop Note;clue
#
@set Note/desc = "A faded sticky note reads: 'The master code is 0451.'"
#
@create/drop Core Fragment C;fragment c;core fragment:typeclasses.quest_objects.QuestFragment
#
@set Core Fragment C/desc = "A glowing piece of the Fractured Core (Part C)."
#------------------------------------------------------------
# 3. The Cooling Tunnels (Maze)
#------------------------------------------------------------
@teleport The Atrium
#
@dig Cooling Tunnel Entrance
#
@open east;e = Cooling Tunnel Entrance
#
@teleport Cooling Tunnel Entrance
#
@open west;w = The Atrium
#
@set Cooling Tunnel Entrance/desc = "A labyrinth of pipes. The paths split off in multiple directions."
#
# To make a maze, we'll just have wrong exits go back to the entrance.
@dig Cooling Tunnel Depth 1
#
@open east;e = Cooling Tunnel Depth 1
#
@open north;n = Cooling Tunnel Entrance
#
@open south;s = Cooling Tunnel Entrance
#
@teleport Cooling Tunnel Depth 1
#
@set Cooling Tunnel Depth 1/desc = "The pipes look identical to the entrance. It's easy to get lost here."
#
@dig Cooling Tunnel Depth 2
#
@open north;n = Cooling Tunnel Depth 2
#
@open east;e = Cooling Tunnel Entrance
#
@open south;s = Cooling Tunnel Entrance
#
@open west;w = Cooling Tunnel Entrance
#
@teleport Cooling Tunnel Depth 2
#
@set Cooling Tunnel Depth 2/desc = "The heat is intense. You feel like you're getting closer."
#
@dig The Engineering Bay
#
@open east;e = The Engineering Bay
#
@open north;n = Cooling Tunnel Entrance
#
@open south;s = Cooling Tunnel Entrance
#
@open west;w = Cooling Tunnel Entrance
#
#------------------------------------------------------------
# 4. The Engineering Bay (Machinery Puzzle)
#------------------------------------------------------------
@teleport The Engineering Bay
#
@open west;w = The Atrium
#
@set The Engineering Bay/desc = "A massive industrial bay. In the center sits some Heavy Machinery with a glowing panel."
#
@create/drop Heavy Machinery;machinery:typeclasses.quest_objects.HeavyMachinery
#------------------------------------------------------------
# 5. The Sanctum (Assembly)
#------------------------------------------------------------
@teleport The Atrium
#
@dig/teleport The Inner Sanctum
#
@open east;e = The Atrium
#
# Wait, let's make sure the exit from Atrium to Sanctum exists:
@teleport The Atrium
#
@open west;w = The Inner Sanctum
#
@teleport The Inner Sanctum
#
@set The Inner Sanctum/desc = "A pristine, silent chamber. In the center is a glowing receptacle waiting for the Restored Core."
#
# Create the Council of Five
@destroy/override System Architect
#
@create/drop System Architect:typeclasses.ai_characters.CouncilMember
#
@destroy/override Code Weaver
#
@create/drop Code Weaver:typeclasses.ai_characters.CouncilMember
#
@destroy/override The Inquisitor
#
@create/drop The Inquisitor:typeclasses.ai_characters.CouncilMember
#
@destroy/override The Waymaker
#
@create/drop The Waymaker:typeclasses.ai_characters.CouncilMember
#
@destroy/override The Shadowbaner
#
@create/drop The Shadowbaner:typeclasses.ai_characters.CouncilMember
#------------------------------------------------------------
# 6. The Five Gates of Ascension
#------------------------------------------------------------
@teleport The Inner Sanctum
#
@dig The Syntax Ward
#
@open north;n = The Syntax Ward
#
@teleport The Syntax Ward
#
@open south;s = The Inner Sanctum
#
@set The Syntax Ward/desc = "The first Gate: Compilation. Elemental node alignment puzzles govern the path."
#
@create/drop Syntax Node;node:typeclasses.quest_objects.SyntaxNode
#
@dig The Hall of Boundaries
#
@open north;n = The Hall of Boundaries
#
@teleport The Hall of Boundaries
#
@open south;s = The Syntax Ward
#
@set The Hall of Boundaries/desc = "The second Gate: Unit Test. Boundary Golems patrol here, requiring isolated strikes on weak points."
#
@create/drop Boundary Golem;golem:typeclasses.quest_objects.BoundaryGolem
#
@dig The Integration Labyrinth
#
@open north;n = The Integration Labyrinth
#
@teleport The Integration Labyrinth
#
@open south;s = The Hall of Boundaries
#
@set The Integration Labyrinth/desc = "The third Gate: Integration. A sprawling maze demanding split-party synchronization via API Contract Stones."
#
@create/drop Contract Stone;stone:typeclasses.quest_objects.ContractStone
#
@dig The Vault of Shadows
#
@open north;n = The Vault of Shadows
#
@teleport The Vault of Shadows
#
@open south;s = The Integration Labyrinth
#
@set The Vault of Shadows/desc = "The fourth Gate: Security. Invisible vulnerabilities and Dependency Wraiths lurk in the dark."
#
@create/drop Dependency Wraith;wraith:typeclasses.quest_objects.DependencyWraith
#
@dig The Crucible of Load
#
@open north;n = The Crucible of Load
#
@teleport The Crucible of Load
#
@open south;s = The Vault of Shadows
#
@set The Crucible of Load/desc = "The fifth Gate: Performance. The final stress test awaits."
#
@open north;n = The Atrium
#------------------------------------------------------------
# Finish
#------------------------------------------------------------
@teleport The Atrium
#
@echo The Fractured Core world generation complete!
+130
View File
@@ -0,0 +1,130 @@
"""
Typeclasses for The Fractured Core quest objects.
"""
from typeclasses.objects import Object
from typeclasses.characters import Character
from commands.quest_cmds import TerminalCmdSet, MachineryCmdSet
class QuestFragment(Object):
"""
Base class for quest fragments.
Can be picked up and dropped normally.
"""
pass
class Terminal(Object):
"""
A terminal that requires a password.
"""
def at_object_creation(self):
super().at_object_creation()
# The terminal cannot be picked up
self.locks.add("get:false()")
self.db.solved = False
self.db.password = "0451"
self.cmdset.add(TerminalCmdSet, persistent=True)
def return_appearance(self, looker, **kwargs):
if self.db.solved:
return super().return_appearance(looker, **kwargs) + "\n\nThe screen reads: ACCESS GRANTED. The drawer is open and empty."
else:
return super().return_appearance(looker, **kwargs) + "\n\nThe screen reads: ENTER PASSWORD. Use the 'type <password>' command."
class HeavyMachinery(Object):
"""
Machinery that can be pried open.
"""
def at_object_creation(self):
super().at_object_creation()
# Cannot be picked up
self.locks.add("get:false()")
self.db.solved = False
self.cmdset.add(MachineryCmdSet, persistent=True)
def return_appearance(self, looker, **kwargs):
if self.db.solved:
return super().return_appearance(looker, **kwargs) + "\n\nThe panel has been pried open and the gears are exposed."
else:
return super().return_appearance(looker, **kwargs) + "\n\nA heavy panel covers the inner workings. It looks like it could be pried open with a 'wrench'."
class Pedestal(Object):
"""
A pedestal that accepts a specific item.
"""
def at_object_creation(self):
super().at_object_creation()
self.locks.add("get:false()")
# A dictionary mapping expected object keys to whether they are placed
self.db.expected_item = "Restored Core"
self.db.has_item = False
def at_get(self, getter):
# Prevent getting the pedestal itself
return False
# We can handle the placing logic via a custom command on the SanctumRoom or just allow dropping the item here.
#------------------------------------------------------------
# Five Gates Objects
#------------------------------------------------------------
class SyntaxNode(Object):
"""
First Gate: Compilation puzzle node.
"""
def at_object_creation(self):
super().at_object_creation()
self.db.solved = False
self.locks.add("get:false()")
def return_appearance(self, looker, **kwargs):
if self.db.solved:
return super().return_appearance(looker, **kwargs) + "\n\nThe node glows with a steady, coherent light. The elemental alignment is stable."
else:
return super().return_appearance(looker, **kwargs) + "\n\nThe node crackles with chaotic energy. The syntax is misaligned."
class BoundaryGolem(Character):
"""
Second Gate: Unit Test adversary.
"""
def at_object_creation(self):
super().at_object_creation()
self.db.solved = False
def return_appearance(self, looker, **kwargs):
if self.db.solved:
return super().return_appearance(looker, **kwargs) + "\n\nThe Golem is inert, its boundary weak points shattered."
else:
return super().return_appearance(looker, **kwargs) + "\n\nThe Golem hulks menacingly, invulnerable except for specific isolated boundary edges."
class ContractStone(Object):
"""
Third Gate: Integration synchronization.
"""
def at_object_creation(self):
super().at_object_creation()
self.db.solved = False
self.locks.add("get:false()")
def return_appearance(self, looker, **kwargs):
if self.db.solved:
return super().return_appearance(looker, **kwargs) + "\n\nThe API contract is fulfilled. The stone bridges the gap perfectly."
else:
return super().return_appearance(looker, **kwargs) + "\n\nThe API contract is broken. It awaits split-party synchronization."
class DependencyWraith(Object):
"""
Fourth Gate: Security trap.
"""
def at_object_creation(self):
super().at_object_creation()
self.db.solved = False
self.locks.add("get:false()")
def return_appearance(self, looker, **kwargs):
if self.db.solved:
return super().return_appearance(looker, **kwargs) + "\n\nThe wraith's vulnerabilities have been patched. It fades into the ether."
else:
return super().return_appearance(looker, **kwargs) + "\n\nAn invisible menace lurks here, tangled in outdated dependencies and hidden security traps."
+71
View File
@@ -0,0 +1,71 @@
from evennia import Command, CmdSet
class CmdChaosProbe(Command):
"""
Cast a Chaos Probe to expose weaknesses (QA/Tester).
Usage:
cast Chaos Probe
"""
key = "cast Chaos Probe"
aliases = ["chaos probe"]
def func(self):
self.caller.msg("You cast a Chaos Probe! Fuzzing the environment for boundary failures and edge cases...")
# Add actual logic here if needed
self.caller.location.msg_contents(f"{self.caller.name} unleashes a Chaos Probe, destabilizing the area to reveal hidden flaws!", exclude=[self.caller])
class CmdDeployCanary(Command):
"""
Deploy a Canary to test dangerous rooms (DevOps).
Usage:
deploy Canary
"""
key = "deploy Canary"
aliases = ["deploy canary"]
def func(self):
self.caller.msg("You deploy a Canary to the next room. Monitoring damage thresholds...")
self.caller.location.msg_contents(f"{self.caller.name} releases a Canary to scout ahead for infrastructural collapse.", exclude=[self.caller])
class CmdShiftBlueGreen(Command):
"""
Instantly shift the party to a clean, parallel arena (DevOps).
Usage:
shift Blue-Green
"""
key = "shift Blue-Green"
aliases = ["shift blue-green"]
def func(self):
self.caller.msg("You initiate a Blue-Green Shift! The party is routed to a fresh parallel arena environment.")
self.caller.location.msg_contents(f"{self.caller.name} warps the topology, instantly shifting everyone to a pristine Blue-Green arena!", exclude=[self.caller])
class CmdFortifyLeyline(Command):
"""
Fortify a Leyline to reinforce system architecture (System Architect).
Usage:
fortify Leyline
"""
key = "fortify Leyline"
aliases = ["fortify leyline"]
def func(self):
self.caller.msg("You weave pattern wards to fortify the Leyline, stabilizing the system architecture.")
self.caller.location.msg_contents(f"{self.caller.name} traces glowing pattern wards, fortifying the local Leyline and reinforcing reality!", exclude=[self.caller])
class SanctumCmdSet(CmdSet):
"""
CmdSet for Sanctum archetype abilities.
"""
key = "SanctumCmdSet"
def at_cmdset_creation(self):
self.add(CmdChaosProbe())
self.add(CmdDeployCanary())
self.add(CmdShiftBlueGreen())
self.add(CmdFortifyLeyline())