# [DEF:TaskContextModule:Module] # @SEMANTICS: task, context, plugin, execution, logger # @PURPOSE: Provides execution context passed to plugins during task execution. # @LAYER: Core # @RELATION: DEPENDS_ON -> TaskLogger, USED_BY -> plugins # @TIER: CRITICAL # @INVARIANT: Each TaskContext is bound to a single task execution. # [SECTION: IMPORTS] # [SECTION: IMPORTS] from typing import Dict, Any, Callable from .task_logger import TaskLogger from ..logger import belief_scope # [/SECTION] # [DEF:TaskContext:Class] # @SEMANTICS: context, task, execution, plugin # @PURPOSE: A container passed to plugin.execute() providing the logger and other task-specific utilities. # @TIER: CRITICAL # @INVARIANT: logger is always a valid TaskLogger instance. # @UX_STATE: Idle -> Active -> Complete class TaskContext: """ Execution context provided to plugins during task execution. Usage: def execute(params: dict, context: TaskContext = None): if context: context.logger.info("Starting process") context.logger.progress("Processing items", percent=50) # ... plugin logic """ # [DEF:__init__:Function] # @PURPOSE: Initialize the TaskContext with task-specific resources. # @PRE: task_id is a valid task identifier, add_log_fn is callable. # @POST: TaskContext is ready to be passed to plugin.execute(). # @PARAM: task_id (str) - The ID of the task. # @PARAM: add_log_fn (Callable) - Function to add log to TaskManager. # @PARAM: params (Dict) - Task parameters. # @PARAM: default_source (str) - Default source for logs (default: "plugin"). def __init__( self, task_id: str, add_log_fn: Callable, params: Dict[str, Any], default_source: str = "plugin" ): with belief_scope("__init__"): self._task_id = task_id self._params = params self._logger = TaskLogger( task_id=task_id, add_log_fn=add_log_fn, source=default_source ) # [/DEF:__init__:Function] # [DEF:task_id:Function] # @PURPOSE: Get the task ID. # @PRE: TaskContext must be initialized. # @POST: Returns the task ID string. # @RETURN: str - The task ID. @property def task_id(self) -> str: with belief_scope("task_id"): return self._task_id # [/DEF:task_id:Function] # [DEF:logger:Function] # @PURPOSE: Get the TaskLogger instance for this context. # @PRE: TaskContext must be initialized. # @POST: Returns the TaskLogger instance. # @RETURN: TaskLogger - The logger instance. @property def logger(self) -> TaskLogger: with belief_scope("logger"): return self._logger # [/DEF:logger:Function] # [DEF:params:Function] # @PURPOSE: Get the task parameters. # @PRE: TaskContext must be initialized. # @POST: Returns the parameters dictionary. # @RETURN: Dict[str, Any] - The task parameters. @property def params(self) -> Dict[str, Any]: with belief_scope("params"): return self._params # [/DEF:params:Function] # [DEF:get_param:Function] # @PURPOSE: Get a specific parameter value with optional default. # @PRE: TaskContext must be initialized. # @POST: Returns parameter value or default. # @PARAM: key (str) - Parameter key. # @PARAM: default (Any) - Default value if key not found. # @RETURN: Any - Parameter value or default. def get_param(self, key: str, default: Any = None) -> Any: with belief_scope("get_param"): return self._params.get(key, default) # [/DEF:get_param:Function] # [DEF:create_sub_context:Function] # @PURPOSE: Create a sub-context with a different default source. # @PRE: source is a non-empty string. # @POST: Returns new TaskContext with different logger source. # @PARAM: source (str) - New default source for logging. # @RETURN: TaskContext - New context with different source. def create_sub_context(self, source: str) -> "TaskContext": """Create a sub-context with a different default source for logging.""" with belief_scope("create_sub_context"): return TaskContext( task_id=self._task_id, add_log_fn=self._logger._add_log, params=self._params, default_source=source ) # [/DEF:create_sub_context:Function] # [/DEF:TaskContext:Class] # [/DEF:TaskContextModule:Module]