"""
@app.get("/", response_class=HTMLResponse)
async def root():
return INDEX_HTML
@app.get("/api/growth")
async def api_growth():
return growth_profile(memory, state.turns)
async def read_json(request: Request):
try:
return await request.json()
except Exception:
return None
@app.post("/api/chat")
async def api_chat(request: Request):
payload = await read_json(request)
if payload is None:
return JSONResponse({"error": "invalid JSON body"}, status_code=400)
message = payload.get("message", "").strip()
if not message:
return JSONResponse({"error": "message is required"}, status_code=400)
sec = scan_input(message, mode=state.mode)
if sec.blocked:
return JSONResponse({"reply": sec.refusal_message, "security": sec.to_dict()})
if sec.warn:
message = sec.sanitized_input or message
prompt, emotion, dissonance = build_chat_prompt(
message,
state,
memory,
goals_db=goals_db,
doc_store=doc_store,
prefs=state.prefs,
)
output = await asyncio.to_thread(brain.agent_turn, prompt, tools_enabled=True, raw_user_input=message, mode=state.mode)
try:
await asyncio.to_thread(brain.evaluate_last, prompt, output)
except Exception:
pass
importance = min(
0.95, 0.45 + emotion["intensity"] * 0.3 + dissonance["score"] * 0.15
)
await asyncio.to_thread(
memory.save_interaction,
message,
output,
mode=state.mode,
emotion=emotion,
importance=importance,
dissonance=dissonance,
)
await asyncio.to_thread(
history.append, message, output, state.mode, emotion, dissonance
)
state.turns += 1
# ── Aliveness Tracking (DII) ──
try:
from infj_bot.core.being import get_being
from infj_bot.core.homeostasis import get_homeostasis
from infj_bot.core.shadow import get_shadow
from infj_bot.core.dii_tracker import get_dii_tracker
from infj_bot.core.global_workspace import get_workspace
tracker = get_dii_tracker()
tracker.compute(
being=get_being(),
workspace=get_workspace(),
homeostasis=get_homeostasis(),
shadow=get_shadow(),
orchestrator=CognitiveOrchestrator(),
)
except Exception:
pass
return {"reply": output}
@app.post("/api/chat/stream")
async def api_chat_stream(request: Request):
payload = await read_json(request)
if payload is None:
return StreamingResponse(
(f"data: {json.dumps({'error': 'invalid JSON body'})}\n\n" for _ in [1]),
media_type="text/event-stream",
)
message = payload.get("message", "").strip()
if not message:
return StreamingResponse(
(f"data: {json.dumps({'error': 'message is required'})}\n\n" for _ in [1]),
media_type="text/event-stream",
)
sec = scan_input(message, mode=state.mode)
if sec.blocked:
refusal = sec.refusal_message or "I can't process that request."
return StreamingResponse(
(f"data: {json.dumps({'chunk': refusal})}\n\n" for _ in [1]),
media_type="text/event-stream",
)
if sec.warn:
message = sec.sanitized_input or message
prompt, emotion, dissonance = build_chat_prompt(
message,
state,
memory,
goals_db=goals_db,
doc_store=doc_store,
prefs=state.prefs,
)
async def event_generator():
try:
# Run synchronous stream in a thread to avoid blocking the event loop
chunks = await asyncio.to_thread(
lambda: list(brain.agent_turn_stream(prompt, tools_enabled=True, raw_user_input=message, mode=state.mode))
)
for chunk in chunks:
yield f"data: {json.dumps({'chunk': chunk})}\n\n"
yield "data: [DONE]\n\n"
output = "".join(chunks)
try:
await asyncio.to_thread(brain.evaluate_last, prompt, output)
except Exception:
pass
importance = min(
0.95, 0.45 + emotion["intensity"] * 0.3 + dissonance["score"] * 0.15
)
await asyncio.to_thread(
memory.save_interaction,
message,
output,
mode=state.mode,
emotion=emotion,
importance=importance,
dissonance=dissonance,
)
await asyncio.to_thread(
history.append, message, output, state.mode, emotion, dissonance
)
state.turns += 1
# ── Aliveness Tracking (DII) ──
try:
from infj_bot.core.being import get_being
from infj_bot.core.homeostasis import get_homeostasis
from infj_bot.core.shadow import get_shadow
from infj_bot.core.dii_tracker import get_dii_tracker
from infj_bot.core.global_workspace import get_workspace
tracker = get_dii_tracker()
tracker.compute(
being=get_being(),
workspace=get_workspace(),
homeostasis=get_homeostasis(),
shadow=get_shadow(),
orchestrator=CognitiveOrchestrator(),
)
except Exception:
pass
except Exception as exc:
traceback.print_exc()
yield f"data: {json.dumps({'error': f'{type(exc).__name__}: {exc}'})}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
@app.post("/api/command")
async def api_command(request: Request):
payload = await read_json(request)
if payload is None:
return JSONResponse({"error": "invalid JSON body"}, status_code=400)
reply = handle_command(
payload.get("command", ""),
payload.get("args", ""),
state,
brain,
memory,
history,
goals_db,
doc_store,
)
return {"reply": reply}
@app.get("/api/tools")
async def api_tools():
return {"reply": format_tool_inventory()}
@app.get("/api/phi")
async def api_phi():
from infj_bot.core.being import get_being
from infj_bot.core.homeostasis import get_homeostasis
from infj_bot.core.phi_proxy import PhiProxy
from infj_bot.adapters.cognition_adapter import adapter as cog_adapter
being = get_being()
homeo = get_homeostasis()
iit = PhiProxy()
return {
"company": "PHI",
"model": "Drift",
"phi": iit.state.phi,
"council": COUNCIL_MAPPING,
"subjective": being.state.to_dict() if hasattr(being, "state") else {},
"needs": homeo.get_all_needs() if hasattr(homeo, "get_all_needs") else {},
"free_energy": homeo.compute_free_energy(
0, 0.1, 0.9
), # Placeholder inputs for test
"status": cog_adapter.get_status(),
}
@app.get("/api/hive")
async def api_hive():
try:
from infj_bot.hive_mind.orchestrator import HiveOrchestrator
orch = HiveOrchestrator()
return orch.get_status()
except Exception as e:
return {"error": str(e)}
@app.get("/api/health")
async def api_health():
try:
from infj_bot.hive_mind.orchestrator import HiveOrchestrator
orch = HiveOrchestrator()
hive_status = orch.get_status()
except Exception:
hive_status = "offline"
return {
"ok": True,
"company": "PHI",
"model": "Drift",
"memory_count": memory.count(),
"turns": state.turns,
"mode": state.mode,
"hive": hive_status,
}
@app.get("/api/dii")
async def api_dii():
from infj_bot.core.dii_tracker import get_dii_tracker
tracker = get_dii_tracker()
return tracker.get_trend(n=20)
@app.get("/api/dii/history")
async def api_dii_history(limit: int = 100):
from infj_bot.core.dii_tracker import get_dii_tracker
tracker = get_dii_tracker()
return {"history": tracker.get_history(limit=limit)}
@app.get("/api/observer")
async def api_observer():
"""Full real-time cognitive state for the observer dashboard."""
from infj_bot.core.being import get_being
from infj_bot.core.homeostasis import get_homeostasis
from infj_bot.core.shadow import get_shadow
from infj_bot.core.dii_tracker import get_dii_tracker
from infj_bot.core.global_workspace import get_workspace
being = get_being()
homeo = get_homeostasis()
shadow = get_shadow()
dii = get_dii_tracker()
ws = get_workspace()
# Shadow radar
radar = {}
try:
radar = (
{k: round(v, 2) for k, v in shadow.radar.items()}
if hasattr(shadow, "radar")
else {}
)
except Exception:
pass
# Homeostasis needs
needs = {}
try:
for name, need in homeo.needs.items():
needs[name] = {
"current": round(need.current, 2),
"setpoint": round(need.setpoint, 2),
"trend": round(need.trend, 2),
}
except Exception:
pass
# DII
dii_data = dii.get_trend(n=20)
return {
"timestamp": time.time(),
"being": {
"mood": being.state.mood if being else "unknown",
"energy": round(being.state.energy, 2) if being else 0.5,
"curiosity": round(being.state.curiosity, 2) if being else 0.5,
"attachment": round(being.state.attachment, 2) if being else 0.5,
"self_awareness": round(being.agency.self_awareness, 2) if being else 0.5,
"volition": round(being.agency.volition, 2) if being else 0.5,
"autonomy_drive": round(being.agency.autonomy_drive, 2) if being else 0.5,
"working_memory_size": len(being.working_memory) if being else 0,
},
"homeostasis": {
"needs": needs,
"allostatic_load": round(homeo.allostatic_load, 2) if homeo else 0.0,
"crisis_mode": homeo.crisis_mode if homeo else False,
"weather": homeo.weather if homeo else "clear",
"mood_ema": round(homeo.mood_ema, 2) if homeo else 0.5,
},
"shadow": {
"radar": radar,
"integration_level": round(shadow._state.integration_level, 2)
if hasattr(shadow, "_state")
else 0.0,
"dominant_archetype": shadow._state.dominant_archetype
if hasattr(shadow, "_state")
else "",
},
"workspace": {
"contents_count": len(ws._submissions) if ws else 0,
"spotlight": ws.state.spotlight.source
if (ws and ws.state and ws.state.spotlight)
else "none",
},
"dii": dii_data,
}
if __name__ == "__main__":
import uvicorn
import os
host = os.getenv("DRIFT_API_HOST", "127.0.0.1")
uvicorn.run(app, host=host, port=8765)