src.core.memory module#

Memory System for Marcus.

Multi-tier memory system that enables learning from past experiences and predictive task assignment. Inspired by cognitive memory models with working, episodic, semantic, and procedural memory layers.

class src.core.memory.TaskOutcome[source]#

Bases: object

Record of a task execution outcome.

task_id: str#

agent_id: str#

task_name: str#

estimated_hours: float#

actual_hours: float#

success: bool#

blockers: List[str]#

started_at: datetime | None = None#

completed_at: datetime | None = None#

property estimation_accuracy: float#: Calculate how accurate the time estimate was.

to_dict()[source]#

Convert to dictionary for storage.

Return type:: Dict[str, Any]

__init__(task_id, agent_id, task_name, estimated_hours, actual_hours, success, blockers=<factory>, started_at=None, completed_at=None)#

Parameters:

task_id (str)
agent_id (str)
task_name (str)
estimated_hours (float)
actual_hours (float)
success (bool)
blockers (List[str])
started_at (datetime | None)
completed_at (datetime | None)

Return type:

None

class src.core.memory.AgentProfile[source]#

Bases: object

Learned profile of an agent’s capabilities.

agent_id: str#

total_tasks: int = 0#

successful_tasks: int = 0#

failed_tasks: int = 0#

blocked_tasks: int = 0#

skill_success_rates: Dict[str, float]#

average_estimation_accuracy: float = 0.0#

common_blockers: Dict[str, int]#

peak_performance_hours: List[int]#

property success_rate: float#: Overall success rate.

property blockage_rate: float#: Rate of encountering blockers.

__init__(agent_id, total_tasks=0, successful_tasks=0, failed_tasks=0, blocked_tasks=0, skill_success_rates=<factory>, average_estimation_accuracy=0.0, common_blockers=<factory>, peak_performance_hours=<factory>)#

Parameters:

agent_id (str)
total_tasks (int)
successful_tasks (int)
failed_tasks (int)
blocked_tasks (int)
skill_success_rates (Dict[str, float])
average_estimation_accuracy (float)
common_blockers (Dict[str, int])
peak_performance_hours (List[int])

Return type:

None

class src.core.memory.TaskPattern[source]#

Bases: object

Learned pattern about task types.

pattern_type: str#

task_labels: List[str]#

recent_durations: List[float]#

success_rate: float#

common_blockers: List[str]#

prerequisites: List[str]#

best_agents: List[str]#

max_samples: int = 100#

property median_duration: float#: Calculate median duration from recent samples.

property average_duration: float#: Calculate average duration (for backward compatibility).

__init__(pattern_type, task_labels, recent_durations, success_rate, common_blockers, prerequisites, best_agents, max_samples=100)#

Parameters:

pattern_type (str)
task_labels (List[str])
recent_durations (List[float])
success_rate (float)
common_blockers (List[str])
prerequisites (List[str])
best_agents (List[str])
max_samples (int)

Return type:

None

class src.core.memory.Memory[source]#

Bases: object

Multi-tier memory system for Marcus.

Tiers: - Working Memory: Current state and active tasks - Episodic Memory: Specific task execution records - Semantic Memory: Extracted facts and patterns - Procedural Memory: Learned workflows and strategies

__init__(events=None, persistence=None)[source]#

Initialize the Memory system.

Parameters:

events (Optional[Events]) – Optional Events system for integration.
persistence (Optional[Persistence]) – Optional Persistence for long-term storage.

working: Dict[str, Any]#

episodic: Dict[str, Any]#

semantic: Dict[str, Any]#

procedural: Dict[str, Any]#

async record_task_start(agent_id, task)[source]#

Record that an agent started a task.

Return type:

None

Parameters:

agent_id (str)
task (Task)

async record_task_completion(agent_id, task_id, success, actual_hours, blockers=None)[source]#

Record task completion and learn from it.

Return type:

Optional[TaskOutcome]

Parameters:

agent_id (str)
task_id (str)
success (bool)
actual_hours (float)
blockers (List[str] | None)

async predict_task_outcome(agent_id, task)[source]#

Predict likely outcome for agent-task combination.

Return type:

Dict[str, Any]

Returns:

Dictionary with predictions: - success_probability: 0-1 - estimated_duration: hours - blockage_risk: 0-1 - risk_factors: list of potential issues

Parameters:

agent_id (str)
task (Task)

async predict_completion_time(agent_id, task)[source]#

Predict task completion time with confidence intervals.

Return type:

Dict[str, Any]

Returns:

Dictionary with: - expected_hours: Most likely duration - confidence_interval: {lower, upper} bounds - factors: What influences the prediction - confidence: Overall confidence in prediction (0-1)

Parameters:

agent_id (str)
task (Task)

get_median_duration_by_type(task_type)[source]#

Get median task duration for a given task type.

Parameters:: task_type (str) – Task type label (e.g., “design”, “implement”, “test”)
Returns:: Median duration in hours, or None if no data available
Return type:: Optional[float]

Examples

>>> memory.get_median_duration_by_type("design")
0.1  # 6 minutes

Notes

Uses median instead of average to be robust to outliers (tasks that sat for hours waiting for user input, etc.)

async predict_blockage_probability(agent_id, task)[source]#

Predict likelihood and types of blockages.

Return type:

Dict[str, Any]

Returns:

Dictionary with: - overall_risk: 0-1 probability - risk_breakdown: Dict of blockage type to probability - preventive_measures: Suggested actions to reduce risk - historical_blockers: Common blockers for similar tasks

Parameters:

agent_id (str)
task (Task)

async predict_cascade_effects(task_id, delay_hours)[source]#

Predict impact of task delay on dependent tasks.

Return type:

Dict[str, Any]

Returns:

Dictionary with: - affected_tasks: List of tasks that will be impacted - total_delay: Cumulative delay across all tasks - critical_path_impact: Whether this affects project completion - mitigation_options: Ways to reduce impact

Parameters:

task_id (str)
delay_hours (float)

async calculate_agent_performance_trajectory(agent_id)[source]#

Calculate agent’s skill development trajectory.

Return type:: Dict[str, Any]
Returns:: Dictionary with: - current_skills: Current skill levels - improving_skills: Skills showing improvement - struggling_skills: Skills needing attention - projected_skills: Predicted skill levels in 30 days - recommendations: Training/task recommendations
Parameters:: agent_id (str)

async find_similar_outcomes(task, limit=5)[source]#

Find similar past task executions.

Return type:

List[TaskOutcome]

Parameters:

task (Task)
limit (int)

async get_global_median_duration()[source]#

Get the global median task duration from all completed tasks.

Returns:: Median task duration in hours. Returns 1.0 if no historical data.
Return type:: float

Notes

Uses SQL-based median calculation for efficiency and scalability. Queries ALL historical data from persistence layer, not just in-memory cache. Uses median instead of mean to be more robust to outliers.

get_working_memory_summary()[source]#

Get current working memory state.

Return type:: Dict[str, Any]

update_project_tasks(tasks)[source]#

Update working memory with current project tasks for cascade analysis.

Return type:: None
Parameters:: tasks (List[Task])

get_memory_stats()[source]#

Get memory system statistics.

Return type:: Dict[str, Any]