Work Loop¶

Work Loop -- mvp.work_loop

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview¶

Token-aware work loop for pursuing business goals. Decomposes goals into task proposals, ranks them by expected yield (value * likelihood / effort), and routes the next proposed task to a backend (Claude Code or Ollama) based on configurable routing strategy.

Supports goal management (set_goal, list_goals, remove_goal), loop control (start_loop, pause_loop, resume_loop, get_loop_status), task selection (propose_next, score_tasks), execution preparation (prepare_next), execution (execute_next), manual outcome recording (complete_task, fail_task), configuration (set_routing_strategy), and read-only introspection (list_patterns, health, describe_routing).

Production executor boundary

work_loop is production-ready as an orchestration layer, not as a standalone guarantee that arbitrary business tasks can run unattended. Before enabling it for customer-facing or high-impact workflows, configure and test the concrete WorkLoopExecutor, model/provider credentials, task_tracker persistence, token_budget ledger, email approval path, monitoring, and rollback process in the same environment where it will run. The component records approvals, budget preflight, claims, retries, and audit events; the safety and utility of real side effects still depend on the executor and tools it is allowed to call.

When to use:

Running continuous improvement loops that decompose goals and propose the next highest-value task
Prioritising work by token-efficiency when operating under budget constraints
Routing tasks to the optimal backend based on complexity and available budget
Creating task_tracker records for work-loop tasks, recording budget usage, and gating large token spends through approval
Checking dependency health and routing/cost posture before starting a loop or spending tokens

Example:

from mvp.work_loop import WorkLoopBlock, WorkLoopInput

block = WorkLoopBlock()
result = block.infer(WorkLoopInput(
    op="set_goal",
    objective="Improve solver accuracy on ARC-AGI to 80%",
    target_metric="accuracy",
    target_value=0.80,
))
# result.value -> WorkLoopOutput with goal_id, proposals
goal_id = result.value.goal_id
next_task = block.infer(WorkLoopInput(op="propose_next", goal_id=goal_id))
# next_task.value -> WorkLoopOutput with suggested_backend, proposals ranked by yield

health = block.infer(WorkLoopInput(op="health"))
# health.value.completion_state is one of: verified, qualified-draft, blocked-escalated
routing = block.infer(WorkLoopInput(op="describe_routing"))
# routing.value.evidence -> allowed backends/cost tiers and routing strategy descriptions

prepared = block.infer(WorkLoopInput(op="prepare_next", goal_id=goal_id))
# prepared.value.task -> task_tracker task; prepared.value.budget -> token_budget route

executed = block.infer(WorkLoopInput(op="execute_next", goal_id=goal_id))
# executed.value.execution -> executor output, with task_tracker and token_budget updated

Caveats and known limitations:

Goal decomposition delegates to GoalPlanner - decomposition depth and strategy are not user-configurable
Task scoring and ranking logic (rank_tasks, score_task) is internal with no tunable parameters
Routing strategies (PAID_FIRST, LOCAL_FIRST, INTELLIGENT) are predefined - custom strategies not supported
Loop continuation uses simple token budget arithmetic - no actual prompt size estimation
Programmatic WorkLoopBlock() storage defaults to in-memory WorkLoopStore - pass a database path or use the MCP server for persistence
Actual execution depends on a configured WorkLoopExecutor; the default LLM executor requires the LLM router stack to be available
Large estimated token spends require a linked task_tracker review approval before execution; approved=True only asks the loop to verify that stored approval
Production deployments must configure the executor and surrounding services (task_tracker, token_budget, email approvals, credentials, monitoring, rollback) explicitly; passing component tests does not validate those external operational dependencies
All outputs include the canonical envelope fields completion_state, warning_card, evidence, request_id, task_id, and run_id; degraded-but-useful outputs use qualified-draft, hard blocks use blocked-escalated, and deterministic local success uses verified

Works well with: token_budget, task_tracker, email_colleague, payments_x402

Public API¶

`WorkLoopDecisionError(ValueError)`¶

The LLM did not produce a usable, validated backend-routing recommendation.

`RoutingDecision`¶

Validated work_loop backend-routing decision for one task.

Field	Type	Default
`backend`	`str`	required
`baseline_backend`	`str`	required
`cost_tier`	`str`	required
`baseline_cost_tier`	`str`	required
`requires_review`	`bool`	False
`degraded`	`bool`	False
`llm_used`	`bool`	False
`reasons`	`tuple[str, ...]`	()
`rationale`	`str`	''
`raw_response`	`str`	''

Methods:

`escalated() -> bool`¶

`diverged() -> bool`¶

`agentic_evidence() -> dict[str, Any]`¶

`to_dict() -> dict[str, Any]`¶

`WorkLoopRoutingPlanner`¶

Runtime-first facade with deterministic fallback + one-way cost clamp.

Methods:

`route(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> RoutingDecision`¶

`WorkLoopExecutionResult`¶

Structured result returned by work loop executors.

Field	Type	Default
`status`	`str`	'completed'
`summary`	`str`	''
`output`	`str`	''
`tokens_used`	`int`	0
`cost_usd`	`float`	0.0
`progressed`	`bool \\| None`	None
`cost_source`	`str`	'unpriced'

`WorkLoopExecutor(Protocol)`¶

Protocol for executing a selected work-loop task.

Methods:

`execute(proposal: TaskProposal, backend: str, model: str) -> Result[WorkLoopExecutionResult]`¶

Execute the proposal and return a structured result.

`LLMWorkLoopExecutor`¶

Default executor: asks the configured LLM stack to perform the task.

Methods:

`execute(proposal: TaskProposal, backend: str, model: str) -> Result[WorkLoopExecutionResult]`¶

`GoalPlanner`¶

Decomposes business objectives into scored task proposals.

Methods:

`decompose_goal(objective: str, goal_id: str = '') -> list[TaskProposal]`¶

Break a high-level objective into concrete task proposals.

`WorkLoopPatternRuntime`¶

Stateless executable mechanisms for work_loop's applied patterns.

Methods:

`tighten(backend: Any, baseline_backend: Any = '', cost_tier: Any = '', baseline_cost_tier: Any = '', requires_review: bool = False, degraded: bool = False, high_token_task: bool = False) -> WorkLoopPatternReview`¶

One-way backend-routing guard (hook-based-safety-guard-rails).

`LoopScheduler`¶

Determines whether the work loop should continue executing.

Methods:

`should_continue(loop_state: LoopState, budget_snapshot: dict, mode: str | None = None) -> tuple[bool, str]`¶

Evaluate whether the loop should keep running.

`RoutingStrategy(str, Enum)`¶

Backend selection strategy for task execution.

`GoalStatus(str, Enum)`¶

Lifecycle status of a business goal.

`Goal(BaseModel)`¶

A high-level business objective to pursue.

Field	Type	Default
`goal_id`	`str`	''
`objective`	`str`	''
`target_metric`	`str`	''
`target_value`	`float`	0.0
`priority`	`int`	Field(default=1, ge=1, le=10)
`status`	`GoalStatus`	GoalStatus.ACTIVE
`decomposed_tasks_json`	`str`	'[]'

`TaskProposal(BaseModel)`¶

A concrete task derived from a goal, scored for prioritisation.

Field	Type	Default
`task_id`	`str`	''
`goal_id`	`str`	''
`description`	`str`	''
`estimated_tokens`	`int`	Field(default=0, ge=0)
`estimated_value`	`float`	Field(default=0.0, ge=0.0)
`effort_score`	`float`	Field(default=1.0, ge=0.0)
`likelihood`	`float`	Field(default=0.5, ge=0.0, le=1.0)
`yield_score`	`float`	0.0
`routing_preference`	`RoutingStrategy`	RoutingStrategy.INTELLIGENT
`status`	`str`	'pending'
`confidence_basis`	`Literal['llm', 'llm_fallback']`	'llm'
`degraded`	`bool`	False

`LoopState(BaseModel)`¶

Runtime state of a single work loop execution.

Field	Type	Default
`loop_id`	`str`	''
`routing_strategy`	`RoutingStrategy`	RoutingStrategy.INTELLIGENT
`tokens_consumed_this_run`	`int`	0
`tokens_budget_remaining`	`int`	0
`tasks_completed`	`int`	0
`tasks_failed`	`int`	0
`status`	`str`	'idle'
`reserve_tokens`	`int`	5000000

`WorkLoopInput(BaseModel)`¶

Input envelope for the WorkLoopBlock.

Field	Type	Default
`op`	`Literal['set_goal', 'list_goals', 'remove_goal', 'start_loop', 'pause_loop', 'resume_loop', 'resume_goal', 'cancel_loop', 'get_loop_status', 'propose_next', 'prepare_next', 'execute_next', 'complete_task', 'fail_task', 'score_tasks', 'set_routing_strategy', 'list_patterns', 'health', 'describe_routing']`	required
`goal_id`	`str`	''
`objective`	`str`	''
`target_metric`	`str`	''
`target_value`	`float`	0.0
`priority`	`int`	1
`routing_strategy`	`str`	''
`loop_id`	`str`	''
`task_id`	`str`	''
`tokens_budget`	`int`	0
`approved`	`bool`	False
`reviewer`	`str`	''
`execution_result`	`str`	''
`error_message`	`str`	''
`cost_usd`	`float`	0.0
`tokens_used`	`int`	0
`request_id`	`str`	''
`run_id`	`str`	''
`proposals`	`list[dict[str, Any]]`	Field(default_factory=list)

`WorkLoopOutput(BaseModel)`¶

Output envelope from the WorkLoopBlock.

Field	Type	Default
`success`	`bool`	True
`op`	`str`	''
`goal_id`	`str`	''
`loop_id`	`str`	''
`goals`	`list[dict[str, Any]]`	Field(default_factory=list)
`proposals`	`list[dict[str, Any]]`	Field(default_factory=list)
`task`	`dict[str, Any]`	Field(default_factory=dict)
`budget`	`dict[str, Any]`	Field(default_factory=dict)
`execution`	`dict[str, Any]`	Field(default_factory=dict)
`approval`	`dict[str, Any]`	Field(default_factory=dict)
`loop_state`	`dict[str, Any]`	Field(default_factory=dict)
`suggested_backend`	`str`	''
`message`	`str`	''
`error`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`list[Any] \\| dict[str, Any]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`routing_cost_tier`	`str`	''
`baseline_backend`	`str`	''
`requires_review`	`bool`	False
`llm_used`	`bool`	False
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`WorkLoopSkillCatalog`¶

Maps each applied pattern slug to a work_loop backend-routing skill record.

Methods:

`list_skills() -> list[WorkLoopSkill]`¶

`executable_skills() -> list[WorkLoopSkill]`¶

`get(slug: str) -> WorkLoopSkill | None`¶

`WorkLoopStore`¶

SQLite-backed persistence for work loop state.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`close() -> None`¶

`ping() -> bool`¶

Read-only store reachability check.

`create_goal(objective: str, target_metric: str = '', target_value: float = 0.0, priority: int = 1) -> str`¶

`get_goal(goal_id: str) -> dict | None`¶

`list_goals(status: str = '', limit: int = 50) -> list[dict]`¶

`update_goal(goal_id: str, **kwargs: object) -> bool`¶

`remove_goal(goal_id: str) -> bool`¶

`create_proposal(goal_id: str, description: str, estimated_tokens: int = 0, estimated_value: float = 0.0, effort_score: float = 1.0, likelihood: float = 0.5, yield_score: float = 0.0, routing_preference: str = 'intelligent') -> str`¶

`list_proposals(goal_id: str = '', status: str = '', limit: int = 50) -> list[dict]`¶

`get_proposal(task_id: str) -> dict | None`¶

`update_proposal(task_id: str, **kwargs: object) -> bool`¶

`claim_proposal(task_id: str, claim_id: str, expected_statuses: tuple[str, ...] = ('ready', 'awaiting_approval', 'pending')) -> dict | None`¶

`recover_stuck_claims(older_than_seconds: int = 3600) -> int`¶

`release_in_progress_claims(goal_id: str = '') -> int`¶

Release every in_progress proposal claim back to 'ready'.

`record_audit(task_id: str, event_type: str, detail: object | None = None, tracker_task_id: str = '') -> str`¶

`list_audit(task_id: str, limit: int = 50) -> list[dict]`¶

`create_loop_run(routing_strategy: str = 'intelligent', tokens_budget: int = 0, reserve_tokens: int = 5000000) -> str`¶

`get_loop_run(loop_id: str) -> dict | None`¶

`list_loop_runs(status: str = '', limit: int = 100) -> list[dict]`¶

`update_loop_run(loop_id: str, **kwargs: object) -> bool`¶

`proposal_counts_by_status(goal_id: str = '') -> dict[str, int]`¶

`record_milestone(goal_id: str, description: str) -> str`¶

`list_milestones(goal_id: str, limit: int = 50) -> list[dict]`¶

`WorkLoopBlock(AIBlock[WorkLoopInput, WorkLoopOutput, None])`¶

Token-aware continuous work loop for pursuing business goals 24/7.

Field	Type	Default
`name`	`str`	'work_loop'
`approval_threshold_tokens`	`int`	1000000
`max_execution_attempts`	`int`	2
`stuck_claim_seconds`	`int`	3600
`no_improvement_limit`	`int`	2

Methods:

`infer(data: WorkLoopInput) -> Result[WorkLoopOutput]`¶

Dispatch to the appropriate operation handler.

Functions¶

`summarize_work_loop_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]`¶

Summarise runtime-vs-fallback routing decisions with path redaction.

`deterministic_routing_decision(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> RoutingDecision`¶

Demoted-real backend-routing floor (the zero-LLM baseline).

`build_default_planner() -> 'WorkLoopRoutingPlanner | None'`¶

Build the on-by-default agentic planner, or None when suppressed.

`grounded_route_backend(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict, planner: WorkLoopRoutingPlanner | None = None, approval_threshold: int = 1000000) -> RoutingDecision`¶

The ONE shared grounded chokepoint for EVERY work_loop backend-routing decision.

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

Return compact metadata for work_loop-applied vendored patterns.

`propose_optimization(completed_goals: list[str]) -> str`¶

Suggest the next strategic move after goals are completed.

`select_backend(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> str`¶

Select LLM backend for a task given the routing strategy.

`get_skill_catalog() -> WorkLoopSkillCatalog`¶

`score_task(proposal: TaskProposal) -> float`¶

Compute yield score: value * likelihood / max(effort, 0.1).

`rank_tasks(proposals: list[TaskProposal]) -> list[TaskProposal]`¶

Return proposals sorted by yield_score descending.

Work Loop¶

Overview¶

Public API¶

WorkLoopDecisionError(ValueError)¶

RoutingDecision¶

escalated() -> bool¶

diverged() -> bool¶

agentic_evidence() -> dict[str, Any]¶

to_dict() -> dict[str, Any]¶

WorkLoopRoutingPlanner¶

route(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> RoutingDecision¶

WorkLoopExecutionResult¶

WorkLoopExecutor(Protocol)¶

execute(proposal: TaskProposal, backend: str, model: str) -> Result[WorkLoopExecutionResult]¶

LLMWorkLoopExecutor¶

execute(proposal: TaskProposal, backend: str, model: str) -> Result[WorkLoopExecutionResult]¶

GoalPlanner¶

decompose_goal(objective: str, goal_id: str = '') -> list[TaskProposal]¶

WorkLoopPatternRuntime¶

tighten(backend: Any, baseline_backend: Any = '', cost_tier: Any = '', baseline_cost_tier: Any = '', requires_review: bool = False, degraded: bool = False, high_token_task: bool = False) -> WorkLoopPatternReview¶

LoopScheduler¶

should_continue(loop_state: LoopState, budget_snapshot: dict, mode: str | None = None) -> tuple[bool, str]¶

RoutingStrategy(str, Enum)¶

GoalStatus(str, Enum)¶

Goal(BaseModel)¶

TaskProposal(BaseModel)¶

LoopState(BaseModel)¶

WorkLoopInput(BaseModel)¶

WorkLoopOutput(BaseModel)¶

WorkLoopSkillCatalog¶

list_skills() -> list[WorkLoopSkill]¶

executable_skills() -> list[WorkLoopSkill]¶

get(slug: str) -> WorkLoopSkill | None¶

WorkLoopStore¶

close() -> None¶

ping() -> bool¶

create_goal(objective: str, target_metric: str = '', target_value: float = 0.0, priority: int = 1) -> str¶

get_goal(goal_id: str) -> dict | None¶

list_goals(status: str = '', limit: int = 50) -> list[dict]¶

update_goal(goal_id: str, **kwargs: object) -> bool¶

remove_goal(goal_id: str) -> bool¶

create_proposal(goal_id: str, description: str, estimated_tokens: int = 0, estimated_value: float = 0.0, effort_score: float = 1.0, likelihood: float = 0.5, yield_score: float = 0.0, routing_preference: str = 'intelligent') -> str¶

list_proposals(goal_id: str = '', status: str = '', limit: int = 50) -> list[dict]¶

get_proposal(task_id: str) -> dict | None¶

update_proposal(task_id: str, **kwargs: object) -> bool¶

claim_proposal(task_id: str, claim_id: str, expected_statuses: tuple[str, ...] = ('ready', 'awaiting_approval', 'pending')) -> dict | None¶

recover_stuck_claims(older_than_seconds: int = 3600) -> int¶

release_in_progress_claims(goal_id: str = '') -> int¶

record_audit(task_id: str, event_type: str, detail: object | None = None, tracker_task_id: str = '') -> str¶

list_audit(task_id: str, limit: int = 50) -> list[dict]¶

create_loop_run(routing_strategy: str = 'intelligent', tokens_budget: int = 0, reserve_tokens: int = 5000000) -> str¶

get_loop_run(loop_id: str) -> dict | None¶

list_loop_runs(status: str = '', limit: int = 100) -> list[dict]¶

update_loop_run(loop_id: str, **kwargs: object) -> bool¶

proposal_counts_by_status(goal_id: str = '') -> dict[str, int]¶

record_milestone(goal_id: str, description: str) -> str¶

list_milestones(goal_id: str, limit: int = 50) -> list[dict]¶

WorkLoopBlock(AIBlock[WorkLoopInput, WorkLoopOutput, None])¶

infer(data: WorkLoopInput) -> Result[WorkLoopOutput]¶

Functions¶

summarize_work_loop_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]¶

deterministic_routing_decision(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> RoutingDecision¶

build_default_planner() -> 'WorkLoopRoutingPlanner | None'¶

grounded_route_backend(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict, planner: WorkLoopRoutingPlanner | None = None, approval_threshold: int = 1000000) -> RoutingDecision¶

applied_agentic_patterns() -> list[dict[str, Any]]¶

propose_optimization(completed_goals: list[str]) -> str¶

select_backend(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> str¶

get_skill_catalog() -> WorkLoopSkillCatalog¶

score_task(proposal: TaskProposal) -> float¶

rank_tasks(proposals: list[TaskProposal]) -> list[TaskProposal]¶

`WorkLoopDecisionError(ValueError)`¶

`RoutingDecision`¶

`escalated() -> bool`¶

`diverged() -> bool`¶

`agentic_evidence() -> dict[str, Any]`¶

`to_dict() -> dict[str, Any]`¶

`WorkLoopRoutingPlanner`¶

`route(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> RoutingDecision`¶

`WorkLoopExecutionResult`¶

`WorkLoopExecutor(Protocol)`¶

`execute(proposal: TaskProposal, backend: str, model: str) -> Result[WorkLoopExecutionResult]`¶

`LLMWorkLoopExecutor`¶

`execute(proposal: TaskProposal, backend: str, model: str) -> Result[WorkLoopExecutionResult]`¶

`GoalPlanner`¶

`decompose_goal(objective: str, goal_id: str = '') -> list[TaskProposal]`¶

`WorkLoopPatternRuntime`¶

`tighten(backend: Any, baseline_backend: Any = '', cost_tier: Any = '', baseline_cost_tier: Any = '', requires_review: bool = False, degraded: bool = False, high_token_task: bool = False) -> WorkLoopPatternReview`¶

`LoopScheduler`¶

`should_continue(loop_state: LoopState, budget_snapshot: dict, mode: str | None = None) -> tuple[bool, str]`¶

`RoutingStrategy(str, Enum)`¶

`GoalStatus(str, Enum)`¶

`Goal(BaseModel)`¶

`TaskProposal(BaseModel)`¶

`LoopState(BaseModel)`¶

`WorkLoopInput(BaseModel)`¶

`WorkLoopOutput(BaseModel)`¶

`WorkLoopSkillCatalog`¶

`list_skills() -> list[WorkLoopSkill]`¶

`executable_skills() -> list[WorkLoopSkill]`¶

`get(slug: str) -> WorkLoopSkill | None`¶

`WorkLoopStore`¶

`close() -> None`¶

`ping() -> bool`¶

`create_goal(objective: str, target_metric: str = '', target_value: float = 0.0, priority: int = 1) -> str`¶

`get_goal(goal_id: str) -> dict | None`¶

`list_goals(status: str = '', limit: int = 50) -> list[dict]`¶

`update_goal(goal_id: str, **kwargs: object) -> bool`¶

`remove_goal(goal_id: str) -> bool`¶

`create_proposal(goal_id: str, description: str, estimated_tokens: int = 0, estimated_value: float = 0.0, effort_score: float = 1.0, likelihood: float = 0.5, yield_score: float = 0.0, routing_preference: str = 'intelligent') -> str`¶

`list_proposals(goal_id: str = '', status: str = '', limit: int = 50) -> list[dict]`¶

`get_proposal(task_id: str) -> dict | None`¶

`update_proposal(task_id: str, **kwargs: object) -> bool`¶

`claim_proposal(task_id: str, claim_id: str, expected_statuses: tuple[str, ...] = ('ready', 'awaiting_approval', 'pending')) -> dict | None`¶

`recover_stuck_claims(older_than_seconds: int = 3600) -> int`¶

`release_in_progress_claims(goal_id: str = '') -> int`¶

`record_audit(task_id: str, event_type: str, detail: object | None = None, tracker_task_id: str = '') -> str`¶

`list_audit(task_id: str, limit: int = 50) -> list[dict]`¶

`create_loop_run(routing_strategy: str = 'intelligent', tokens_budget: int = 0, reserve_tokens: int = 5000000) -> str`¶

`get_loop_run(loop_id: str) -> dict | None`¶

`list_loop_runs(status: str = '', limit: int = 100) -> list[dict]`¶

`update_loop_run(loop_id: str, **kwargs: object) -> bool`¶

`proposal_counts_by_status(goal_id: str = '') -> dict[str, int]`¶

`record_milestone(goal_id: str, description: str) -> str`¶

`list_milestones(goal_id: str, limit: int = 50) -> list[dict]`¶

`WorkLoopBlock(AIBlock[WorkLoopInput, WorkLoopOutput, None])`¶

`infer(data: WorkLoopInput) -> Result[WorkLoopOutput]`¶

`summarize_work_loop_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]`¶

`deterministic_routing_decision(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> RoutingDecision`¶

`build_default_planner() -> 'WorkLoopRoutingPlanner | None'`¶

`grounded_route_backend(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict, planner: WorkLoopRoutingPlanner | None = None, approval_threshold: int = 1000000) -> RoutingDecision`¶

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

`propose_optimization(completed_goals: list[str]) -> str`¶

`select_backend(task: TaskProposal, strategy: RoutingStrategy, budget_snapshot: dict) -> str`¶

`get_skill_catalog() -> WorkLoopSkillCatalog`¶

`score_task(proposal: TaskProposal) -> float`¶

`rank_tasks(proposals: list[TaskProposal]) -> list[TaskProposal]`¶