src.core.context module#

Context System for Marcus.

Provides rich context for task assignments including previous implementations, dependency awareness, and relevant patterns. Enhances agent effectiveness by reducing time spent understanding existing code and architectural decisions.

class src.core.context.TaskContext[source]#

Bases: object

Complete context for a task assignment.

task_id: str#
previous_implementations: Dict[str, Any]#
dependent_tasks: List[Dict[str, Any]]#
related_patterns: List[Dict[str, Any]]#
architectural_decisions: List[Dict[str, Any]]#
to_dict()[source]#

Convert to dictionary for serialization.

Return type:

Dict[str, Any]

__init__(task_id, previous_implementations=<factory>, dependent_tasks=<factory>, related_patterns=<factory>, architectural_decisions=<factory>)#
Parameters:
Return type:

None

class src.core.context.DependentTask[source]#

Bases: object

Information about a task that depends on another.

task_id: str#
task_name: str#
expected_interface: str#
dependency_type: str = 'functional'#
__init__(task_id, task_name, expected_interface, dependency_type='functional')#
Parameters:

  • task_id (str)

  • task_name (str)

  • expected_interface (str)

  • dependency_type (str)

Return type:

None

class src.core.context.Decision[source]#

Bases: object

An architectural decision made during development.

decision_id: str#
task_id: str#
agent_id: str#
timestamp: datetime#
what: str#
why: str#
impact: str#
project_id: str | None = None#
to_dict()[source]#

Convert to dictionary for storage.

Return type:

Dict[str, Any]

__init__(decision_id, task_id, agent_id, timestamp, what, why, impact, project_id=None)#
Parameters:
Return type:

None

class src.core.context.Context[source]#

Bases: object

Manages context for task assignments.

Features: - Tracks implementations from completed tasks - Identifies dependent tasks - Stores architectural decisions - Provides rich context for new assignments - Optional persistence for long-term storage

__init__(events=None, persistence=None, use_hybrid_inference=True, ai_engine=None, project_id=None)[source]#

Initialize the Context system.

Parameters:

  • events (Optional[Events]) – Optional Events system for integration.

  • persistence (Optional[Any]) – Optional Persistence instance for storing context.

  • use_hybrid_inference (bool) – Whether to use hybrid dependency inference if available.

  • ai_engine (Optional[Any]) – Optional AI engine for hybrid inference.

  • project_id (Optional[str]) – Project identifier for tracking decisions and artifacts.

implementations: Dict[str, Dict[str, Any]]#
dependencies: Dict[str, List[DependentTask]]#
decisions: List[Decision]#
patterns: Dict[str, List[Dict[str, Any]]]#
async add_implementation(task_id, implementation)[source]#

Add implementation details from a completed task.

Parameters:

  • task_id (str) – ID of the completed task.

  • implementation (Dict[str, Any]) – Details about the implementation (APIs, models, patterns).

Return type:

None

add_dependency(task_id, dependent_task)[source]#

Record that one task depends on another.

Parameters:

  • task_id (str) – The task being depended upon.

  • dependent_task (DependentTask) – Information about the dependent task.

Return type:

None

async log_decision(agent_id, task_id, what, why, impact)[source]#

Log an architectural decision made by an agent.

Parameters:

  • agent_id (str) – ID of the agent making the decision.

  • task_id (str) – Current task ID.

  • what (str) – What was decided.

  • why (str) – Reasoning behind the decision.

  • impact (str) – Expected impact on other components.

Returns:

The logged Decision object.

Return type:

Decision

async get_context(task_id, task_dependencies)[source]#

Get complete context for a task assignment.

Parameters:

  • task_id (str) – The task being assigned.

  • task_dependencies (List[str]) – IDs of tasks this task depends on.

Returns:

Complete context for the task.

Return type:

TaskContext

async analyze_dependencies(tasks, infer_implicit=True)[source]#

Analyze task list to identify dependencies (both explicit and implicit).

Parameters:

  • tasks (List[Task]) – List of all tasks.

  • infer_implicit (bool) – Whether to infer implicit dependencies (default: True).

Return type:

Dict[str, List[str]]

Returns:

Mapping of task_id to list of dependent task IDs

infer_needed_interface(dependent_task, dependency_task_id)[source]#

Infer what interface a dependent task needs from its dependency.

Parameters:

  • dependent_task (Task) – The task that depends on another.

  • dependency_task_id (str) – The ID of the task it depends on.

Return type:

str

Returns:

String describing the expected interface/functionality

async suggest_task_order(tasks)[source]#

Suggest an optimal order for tasks based on dependencies.

Uses topological sort with priority consideration.

Parameters:

tasks (List[Task]) – List of tasks to order.

Return type:

List[Task]

Returns:

Ordered list of tasks

async get_decisions_for_task(task_id)[source]#

Get all decisions related to a specific task.

Parameters:

task_id (str) – The task ID.

Return type:

List[Decision]

Returns:

List of related decisions

async get_implementation_summary()[source]#

Get a summary of all tracked implementations.

Return type:

Dict[str, Any]

Returns:

Summary statistics and recent implementations

async clear_old_data(days=30)[source]#

Clear context data older than specified days.

Parameters:

days (int) – Number of days to retain.

Return type:

None