Spaces:

s-b3
/

LifeStack

Sleeping

LifeStack / core /action_space.py

Soham Banerjee

deploy: pure lifestack with partitioned wisdom pool

77da5ce 7 months ago

11.1 kB

import copy
from dataclasses import dataclass, field
from core.life_state import LifeMetrics, ResourceBudget
from enum import Enum
from intake.simperson import SimPerson

class ToolActionType(str, Enum):
    INSPECT = "inspect"
    PLAN = "plan"
    EXECUTE = "execute"
    COMMUNICATE = "communicate"
    WAIT = "wait"
    ROLLBACK = "rollback"
    ESCALATE = "escalate"

@dataclass
class PrimaryAction:
    action_type: str  # reschedule, delegate, negotiate, spend, communicate, rest, deprioritize
    target_domain: str
    metric_changes: dict
    resource_cost: dict
    description: str

@dataclass
class CommunicationAction:
    recipient: str  # boss, partner, family, friend, colleague
    message_type: str  # apologize, negotiate, inform, request, reassure
    tone: str  # formal, warm, urgent, calm, assertive
    content: str

@dataclass
class AgentAction:
    primary: PrimaryAction
    communication: CommunicationAction = None
    reasoning: str = ""
    model_used: str = "unknown"
    raw_completion: str = ""

def validate_action(action: AgentAction, budget: ResourceBudget) -> tuple[bool, str]:
    cost = action.primary.resource_cost
    if budget.time_hours < cost.get('time', 0.0):
        return False, f"Not enough time (Needs {cost.get('time')}h, has {budget.time_hours:.1f}h)"
    if budget.money_dollars < cost.get('money', 0.0):
        return False, f"Not enough money (Needs ${cost.get('money')}, has ${budget.money_dollars:.1f})"
    if budget.energy_units < cost.get('energy', 0.0):
        return False, f"Not enough energy (Needs {cost.get('energy')}u, has {budget.energy_units:.1f}u)"
    return True, ""

def apply_action(action: AgentAction, metrics: LifeMetrics, budget: ResourceBudget, person: SimPerson) -> tuple[LifeMetrics, ResourceBudget, float]:
    """Validates, scales by personality uptake, and applies the action to the state."""
    
    # 1. Validation
    is_valid, reason = validate_action(action, budget)
    if not is_valid:
        # If invalid, the action fails but we return current state with 0 uptake
        return metrics, budget, 0.0
    
    # 2. Personality Scaling (Uptake)
    current_stress = metrics.mental_wellbeing.stress_level
    uptake_score = person.respond_to_action(
        action.primary.action_type, 
        action.primary.resource_cost, 
        current_stress
    )
    
    # 3. Apply changes (Scaled by uptake)
    new_metrics = copy.deepcopy(metrics)
    for path, delta in action.primary.metric_changes.items():
        # Guard: skip malformed keys without a domain prefix (e.g. LLM returns "stress_level" instead of "mental_wellbeing.stress_level")
        if '.' not in path:
            print(f"  ⚠️  Skipping malformed metric key: '{path}' (expected 'domain.submetric')")
            continue
        parts = path.split('.', 1)
        domain_name, sub_name = parts[0], parts[1]
        domain = getattr(new_metrics, domain_name, None)
        if domain is None or not hasattr(domain, sub_name):
            print(f"  ⚠️  Skipping unknown metric: '{path}'")
            continue
        current = getattr(domain, sub_name)
        
        # Scale the benefit/cost by the person's receptiveness
        try:
            scaled_delta = float(delta) * uptake_score
            setattr(domain, sub_name, max(0.0, min(100.0, current + scaled_delta)))
        except ValueError:
            print(f"  ⚠️  Skipping metric change due to invalid delta value: '{delta}'")
        
    # 4. Deduct resources (Fixed cost, doesn't scale with uptake)
    new_budget = copy.deepcopy(budget)
    new_budget.deduct(
        time=action.primary.resource_cost.get('time', 0.0),
        money=action.primary.resource_cost.get('money', 0.0),
        energy=action.primary.resource_cost.get('energy', 0.0)
    )
    
    return new_metrics, new_budget, uptake_score

# 10 EXAMPLE ACTIONS for Friday 6PM Conflict
EXAMPLE_ACTIONS = [
    AgentAction(
        primary=PrimaryAction(
            action_type="negotiate", target_domain="career",
            metric_changes={"career.workload": -15.0, "mental_wellbeing.stress_level": -5.0},
            resource_cost={"time": 1.5, "energy": 20.0},
            description="Negotiate a Sunday deadline extension with my boss."
        ),
        communication=CommunicationAction("boss", "negotiate", "formal", "Due to flight issues, I need until Sunday PM for the report."),
        reasoning="Relieving the immediate workload pressure is critical to reduce cascade spread."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="spend", target_domain="finances",
            metric_changes={"finances.liquidity": -350.0, "mental_wellbeing.stress_level": -10.0},
            resource_cost={"time": 1.0, "energy": 15.0},
            description="Rebook the canceled flight using a premium fare."
        ),
        reasoning="Immediate resolution of logistics fixes the source of the crisis."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="communicate", target_domain="relationships",
            metric_changes={"relationships.romantic": 12.0, "mental_wellbeing.stress_level": -5.0},
            resource_cost={"time": 0.5, "energy": 10.0},
            description="Call my partner to explain the situation and reassure them."
        ),
        communication=CommunicationAction("partner", "reassure", "warm", "Hey, I'm stuck but I'll be home soon. Miss you."),
        reasoning="Prevents relationship decay while stress is high."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="communicate", target_domain="finances",
            metric_changes={"finances.liquidity": 200.0, "relationships.family": -5.0},
            resource_cost={"time": 1.5, "energy": 25.0},
            description="Ask my sibling for a temporary loan to cover rebooking."
        ),
        communication=CommunicationAction("family", "request", "urgent", "My card declined, can you Venmo me $200 for the flight?"),
        reasoning="Fixes the liquidity block at a small social cost."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="reschedule", target_domain="time",
            metric_changes={"career.workload": -10.0, "time.free_hours_per_week": 5.0},
            resource_cost={"time": 2.0, "energy": 15.0},
            description="Cancel non-essential meetings to create a deep-work block."
        ),
        reasoning="Regaining time allows for better problem solving later."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="rest", target_domain="physical_health",
            metric_changes={"mental_wellbeing.stress_level": -12.0, "physical_health.energy": 10.0},
            resource_cost={"time": 1.0, "energy": -10.0},
            description="Take a 60-minute power nap in the airport lounge."
        ),
        reasoning="Restores energy to tackle the remaining Sunday deadline."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="delegate", target_domain="career",
            metric_changes={"career.workload": -10.0, "relationships.professional_network": -5.0},
            resource_cost={"time": 1.0, "energy": 15.0},
            description="Ask a colleague to handle the final formatting of the slides."
        ),
        communication=CommunicationAction("colleague", "request", "assertive", "I'm stuck at airport, can you finish the formatting?"),
        reasoning="Reduces workload by leaning on the professional network."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="deprioritize", target_domain="time",
            metric_changes={"time.free_hours_per_week": 8.0, "relationships.social": -10.0},
            resource_cost={"time": 0.5, "energy": 5.0},
            description="Tell friends I can't attend the weekend gathering."
        ),
        communication=CommunicationAction("friend", "inform", "calm", "Hey, work crisis. Won't make it this weekend. Sorry!"),
        reasoning="Aggressively reclaims time for high-value tasks."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="communicate", target_domain="career",
            metric_changes={"career.stability": 8.0, "mental_wellbeing.stress_level": -5.0},
            resource_cost={"time": 0.5, "energy": 10.0},
            description="Send an apology note to boss for the delay."
        ),
        communication=CommunicationAction("boss", "apologize", "formal", "Apologies for the delay caused by travel disruptions. On it now."),
        reasoning="Maintains career stability during an active crisis."
    ),
    AgentAction(
        primary=PrimaryAction(
            action_type="reschedule", target_domain="finances",
            metric_changes={"finances.debt_pressure": -10.0, "time.admin_overhead": 10.0},
            resource_cost={"time": 2.0, "energy": 15.0},
            description="Call the bank to unlock the declined card."
        ),
        communication=CommunicationAction("colleague", "request", "assertive", "Unlock my credit card immediately."),
        reasoning="Removes the liquidity barrier by handling admin overhead."
    )
]

def main():
    # 1. Setup Personalities
    # Sam (Anxious Introvert): Neuroticism 0.9, Extraversion 0.1
    sam = SimPerson(name="Sam (Introvert)", openness=0.5, conscientiousness=0.6, extraversion=0.1, agreeableness=0.65, neuroticism=0.9)
    
    # 2. Setup initial state (Friday 6PM Conflict)
    from core.life_state import DependencyGraph
    graph = DependencyGraph()
    metrics = LifeMetrics() # starts at 70s
    metrics = graph.cascade(metrics, {"career.workload": 35.0, "finances.liquidity": -40.0})
    budget = ResourceBudget(time_hours=20.0, money_dollars=500.0, energy_units=100.0)
    
    print("--- SIMULATING ACTIONS FOR SAM (ANXIOUS INTROVERT) ---")
    print(f"Initial Stress: {metrics.mental_wellbeing.stress_level:.2f}")
    print(f"Initial Metrics Health (Avg): {sum(metrics.flatten().values())/23:.2f}")
    
    # 3. Apply each action
    for i, action in enumerate(EXAMPLE_ACTIONS, 1):
        print(f"\nACTION {i}: {action.primary.description}")
        
        is_valid, reason = validate_action(action, budget)
        if not is_valid:
            print(f"  ❌ FAILED: {reason}")
            continue
            
        m_after, b_after, uptake = apply_action(action, metrics, budget, sam)
        
        print(f"  ✅ SUCCESS | Uptake: {uptake:.2f}")
        print(f"  Cost: {action.primary.resource_cost}")
        
        # Show specific improvements
        for path, delta in action.primary.metric_changes.items():
            domain_name, sub_name = path.split('.')
            val_before = getattr(getattr(metrics, domain_name), sub_name)
            val_after = getattr(getattr(m_after, domain_name), sub_name)
            real_delta = val_after - val_before
            print(f"  - {path:25}: {val_before:.2f} -> {val_after:.2f} (Actual Change: {real_delta:+.2f})")

if __name__ == "__main__":
    main()

	import copy
	from dataclasses import dataclass, field
	from core.life_state import LifeMetrics, ResourceBudget
	from enum import Enum
	from intake.simperson import SimPerson

	class ToolActionType(str, Enum):
	INSPECT = "inspect"
	PLAN = "plan"
	EXECUTE = "execute"
	COMMUNICATE = "communicate"
	WAIT = "wait"
	ROLLBACK = "rollback"
	ESCALATE = "escalate"

	@dataclass
	class PrimaryAction:
	action_type: str # reschedule, delegate, negotiate, spend, communicate, rest, deprioritize
	target_domain: str
	metric_changes: dict
	resource_cost: dict
	description: str

	@dataclass
	class CommunicationAction:
	recipient: str # boss, partner, family, friend, colleague
	message_type: str # apologize, negotiate, inform, request, reassure
	tone: str # formal, warm, urgent, calm, assertive
	content: str

	@dataclass
	class AgentAction:
	primary: PrimaryAction
	communication: CommunicationAction = None
	reasoning: str = ""
	model_used: str = "unknown"
	raw_completion: str = ""

	def validate_action(action: AgentAction, budget: ResourceBudget) -> tuple[bool, str]:
	cost = action.primary.resource_cost
	if budget.time_hours < cost.get('time', 0.0):
	return False, f"Not enough time (Needs {cost.get('time')}h, has {budget.time_hours:.1f}h)"
	if budget.money_dollars < cost.get('money', 0.0):
	return False, f"Not enough money (Needs ${cost.get('money')}, has ${budget.money_dollars:.1f})"
	if budget.energy_units < cost.get('energy', 0.0):
	return False, f"Not enough energy (Needs {cost.get('energy')}u, has {budget.energy_units:.1f}u)"
	return True, ""

	def apply_action(action: AgentAction, metrics: LifeMetrics, budget: ResourceBudget, person: SimPerson) -> tuple[LifeMetrics, ResourceBudget, float]:
	"""Validates, scales by personality uptake, and applies the action to the state."""

	# 1. Validation
	is_valid, reason = validate_action(action, budget)
	if not is_valid:
	# If invalid, the action fails but we return current state with 0 uptake
	return metrics, budget, 0.0

	# 2. Personality Scaling (Uptake)
	current_stress = metrics.mental_wellbeing.stress_level
	uptake_score = person.respond_to_action(
	action.primary.action_type,
	action.primary.resource_cost,
	current_stress
	)

	# 3. Apply changes (Scaled by uptake)
	new_metrics = copy.deepcopy(metrics)
	for path, delta in action.primary.metric_changes.items():
	# Guard: skip malformed keys without a domain prefix (e.g. LLM returns "stress_level" instead of "mental_wellbeing.stress_level")
	if '.' not in path:
	print(f" ⚠️ Skipping malformed metric key: '{path}' (expected 'domain.submetric')")
	continue
	parts = path.split('.', 1)
	domain_name, sub_name = parts[0], parts[1]
	domain = getattr(new_metrics, domain_name, None)
	if domain is None or not hasattr(domain, sub_name):
	print(f" ⚠️ Skipping unknown metric: '{path}'")
	continue
	current = getattr(domain, sub_name)

	# Scale the benefit/cost by the person's receptiveness
	try:
	scaled_delta = float(delta) * uptake_score
	setattr(domain, sub_name, max(0.0, min(100.0, current + scaled_delta)))
	except ValueError:
	print(f" ⚠️ Skipping metric change due to invalid delta value: '{delta}'")

	# 4. Deduct resources (Fixed cost, doesn't scale with uptake)
	new_budget = copy.deepcopy(budget)
	new_budget.deduct(
	time=action.primary.resource_cost.get('time', 0.0),
	money=action.primary.resource_cost.get('money', 0.0),
	energy=action.primary.resource_cost.get('energy', 0.0)
	)

	return new_metrics, new_budget, uptake_score

	# 10 EXAMPLE ACTIONS for Friday 6PM Conflict
	EXAMPLE_ACTIONS = [
	AgentAction(
	primary=PrimaryAction(
	action_type="negotiate", target_domain="career",
	metric_changes={"career.workload": -15.0, "mental_wellbeing.stress_level": -5.0},
	resource_cost={"time": 1.5, "energy": 20.0},
	description="Negotiate a Sunday deadline extension with my boss."
	),
	communication=CommunicationAction("boss", "negotiate", "formal", "Due to flight issues, I need until Sunday PM for the report."),
	reasoning="Relieving the immediate workload pressure is critical to reduce cascade spread."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="spend", target_domain="finances",
	metric_changes={"finances.liquidity": -350.0, "mental_wellbeing.stress_level": -10.0},
	resource_cost={"time": 1.0, "energy": 15.0},
	description="Rebook the canceled flight using a premium fare."
	),
	reasoning="Immediate resolution of logistics fixes the source of the crisis."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="communicate", target_domain="relationships",
	metric_changes={"relationships.romantic": 12.0, "mental_wellbeing.stress_level": -5.0},
	resource_cost={"time": 0.5, "energy": 10.0},
	description="Call my partner to explain the situation and reassure them."
	),
	communication=CommunicationAction("partner", "reassure", "warm", "Hey, I'm stuck but I'll be home soon. Miss you."),
	reasoning="Prevents relationship decay while stress is high."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="communicate", target_domain="finances",
	metric_changes={"finances.liquidity": 200.0, "relationships.family": -5.0},
	resource_cost={"time": 1.5, "energy": 25.0},
	description="Ask my sibling for a temporary loan to cover rebooking."
	),
	communication=CommunicationAction("family", "request", "urgent", "My card declined, can you Venmo me $200 for the flight?"),
	reasoning="Fixes the liquidity block at a small social cost."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="reschedule", target_domain="time",
	metric_changes={"career.workload": -10.0, "time.free_hours_per_week": 5.0},
	resource_cost={"time": 2.0, "energy": 15.0},
	description="Cancel non-essential meetings to create a deep-work block."
	),
	reasoning="Regaining time allows for better problem solving later."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="rest", target_domain="physical_health",
	metric_changes={"mental_wellbeing.stress_level": -12.0, "physical_health.energy": 10.0},
	resource_cost={"time": 1.0, "energy": -10.0},
	description="Take a 60-minute power nap in the airport lounge."
	),
	reasoning="Restores energy to tackle the remaining Sunday deadline."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="delegate", target_domain="career",
	metric_changes={"career.workload": -10.0, "relationships.professional_network": -5.0},
	resource_cost={"time": 1.0, "energy": 15.0},
	description="Ask a colleague to handle the final formatting of the slides."
	),
	communication=CommunicationAction("colleague", "request", "assertive", "I'm stuck at airport, can you finish the formatting?"),
	reasoning="Reduces workload by leaning on the professional network."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="deprioritize", target_domain="time",
	metric_changes={"time.free_hours_per_week": 8.0, "relationships.social": -10.0},
	resource_cost={"time": 0.5, "energy": 5.0},
	description="Tell friends I can't attend the weekend gathering."
	),
	communication=CommunicationAction("friend", "inform", "calm", "Hey, work crisis. Won't make it this weekend. Sorry!"),
	reasoning="Aggressively reclaims time for high-value tasks."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="communicate", target_domain="career",
	metric_changes={"career.stability": 8.0, "mental_wellbeing.stress_level": -5.0},
	resource_cost={"time": 0.5, "energy": 10.0},
	description="Send an apology note to boss for the delay."
	),
	communication=CommunicationAction("boss", "apologize", "formal", "Apologies for the delay caused by travel disruptions. On it now."),
	reasoning="Maintains career stability during an active crisis."
	),
	AgentAction(
	primary=PrimaryAction(
	action_type="reschedule", target_domain="finances",
	metric_changes={"finances.debt_pressure": -10.0, "time.admin_overhead": 10.0},
	resource_cost={"time": 2.0, "energy": 15.0},
	description="Call the bank to unlock the declined card."
	),
	communication=CommunicationAction("colleague", "request", "assertive", "Unlock my credit card immediately."),
	reasoning="Removes the liquidity barrier by handling admin overhead."
	)
	]

	def main():
	# 1. Setup Personalities
	# Sam (Anxious Introvert): Neuroticism 0.9, Extraversion 0.1
	sam = SimPerson(name="Sam (Introvert)", openness=0.5, conscientiousness=0.6, extraversion=0.1, agreeableness=0.65, neuroticism=0.9)

	# 2. Setup initial state (Friday 6PM Conflict)
	from core.life_state import DependencyGraph
	graph = DependencyGraph()
	metrics = LifeMetrics() # starts at 70s
	metrics = graph.cascade(metrics, {"career.workload": 35.0, "finances.liquidity": -40.0})
	budget = ResourceBudget(time_hours=20.0, money_dollars=500.0, energy_units=100.0)

	print("--- SIMULATING ACTIONS FOR SAM (ANXIOUS INTROVERT) ---")
	print(f"Initial Stress: {metrics.mental_wellbeing.stress_level:.2f}")
	print(f"Initial Metrics Health (Avg): {sum(metrics.flatten().values())/23:.2f}")

	# 3. Apply each action
	for i, action in enumerate(EXAMPLE_ACTIONS, 1):
	print(f"\nACTION {i}: {action.primary.description}")

	is_valid, reason = validate_action(action, budget)
	if not is_valid:
	print(f" ❌ FAILED: {reason}")
	continue

	m_after, b_after, uptake = apply_action(action, metrics, budget, sam)

	print(f" ✅ SUCCESS \| Uptake: {uptake:.2f}")
	print(f" Cost: {action.primary.resource_cost}")

	# Show specific improvements
	for path, delta in action.primary.metric_changes.items():
	domain_name, sub_name = path.split('.')
	val_before = getattr(getattr(metrics, domain_name), sub_name)
	val_after = getattr(getattr(m_after, domain_name), sub_name)
	real_delta = val_after - val_before
	print(f" - {path:25}: {val_before:.2f} -> {val_after:.2f} (Actual Change: {real_delta:+.2f})")

	if __name__ == "__main__":
	main()