canvas_engineering.scheduling

Region scheduling: clock-driven firing decisions, learned top-k selection from residual summaries, and a hybrid combination of both.

canvas_engineering.scheduling.RegionScheduler

Evaluates clock rules to decide which regions fire each step.

Each region's ClockSpec determines when it fires: - periodic: fires when external_t % period == 0 - on_event: fires when a residual summary exceeds a threshold - boundary: fires on a named lifecycle event

Regions without a ClockSpec always fire (backward compatible).

Cooldown suppresses re-firing for N steps after a fire. max_silence forces firing after N steps of silence regardless.

Internal microsteps

Regions with clock.domain == "internal" and clock.max_inner_steps produce additional inner steps beyond the main external step. Use step_plan() to get a list of active-region sets: the first set is the normal external step, subsequent sets are internal microsteps for regions with internal clocks.

reset()

Reset scheduler state (e.g., between episodes).

step(external_t, summaries=None, boundary=None)

Determine which regions should fire this step.

Parameters:

Name	Type	Description	Default
external_t	int	Current external timestep.	required
summaries	Optional[Dict[str, Dict[str, float]]]	Residual summaries from ResidualAccumulator.summaries(). Format: {region_name: {kind_name: float_value}}.	None
boundary	Optional[str]	Optional named boundary event (e.g., "episode_end").	None

Returns:

Type	Description
Set[str]	Set of region names that should update this step.

step_plan(external_t, summaries=None, boundary=None)

Returns a list of active-region sets, one per inner step.

The first set is the external step (always present, same as step()). Subsequent sets are internal microsteps for regions with clock.domain == "internal" and clock.max_inner_steps.

Usage

plan = scheduler.step_plan(t, summaries) for active in plan: canvas = dispatcher(canvas, active_regions=active)

canvas_engineering.scheduling.LearnedScheduler

Bases: Module

Learns which regions to activate based on residual summaries.

Uses a small MLP to score regions and selects top-k via straight-through Gumbel top-k for differentiable discrete selection.

Parameters:

Name	Type	Description	Default
n_regions	int	Number of regions to score.	required
summary_dim	int	Dimension of the flattened summary input.	required
max_active	int	Maximum number of regions to activate per step.	required
temperature	float	Gumbel softmax temperature. Lower = more discrete.	1.0

Usage

scheduler = LearnedScheduler(n_regions=5, summary_dim=10, max_active=3) summaries_flat = torch.randn(1, 10) # flattened residual summaries active_indices, log_probs = scheduler(summaries_flat)

forward(summaries)

Score regions, select top-k.

Parameters:

Name	Type	Description	Default
summaries	Tensor	(summary_dim,) or (B, summary_dim) flattened residual summaries. If batched, uses the first item for selection (scheduling is per-step, not per-sample).	required

Returns:

Type	Description
Set[int]	Tuple of:
Tensor	active_indices: Set of int region indices to activate.
Tuple[Set[int], Tensor]	log_probs: (n_regions,) tensor of log-probabilities for each region being selected (for REINFORCE-style training).

canvas_engineering.scheduling.HybridScheduler

Bases: Module

Compose RegionScheduler (rule-based) with LearnedScheduler.

Some regions fire according to declarative ClockSpec rules; others are gated by a small learned MLP that scores them from residual summaries. HybridScheduler produces a unified Set[str] of region names to fire per step.

Parameters:

Name	Type	Description	Default
program	CanvasProgram	The CanvasProgram providing rule-based clocks.	required
learned_regions	Optional[List[str]]	Names of regions whose firing decisions are taken by LearnedScheduler instead of their declared ClockSpec. If empty, behaves like RegionScheduler.	None
summary_dim	int	Dimension of the flattened summary input that the learned scheduler consumes.	0
max_active	int	Maximum number of learned-region activations per step.	1
temperature	float	Gumbel temperature for the learned scheduler.	1.0

Usage

sched = HybridScheduler(program, learned_regions=["belief"], summary_dim=16, max_active=1) active = sched.step(t, summaries={"err": {"prediction": 0.4}})

-> active is a set of region names (rule-based ∪ learned)

reset()

Reset internal scheduling state (e.g., between episodes).

step(external_t, summaries=None, boundary=None, summary_tensor=None)

Return the set of region names that fire this step.

Parameters:

Name	Type	Description	Default
external_t	int	Current external timestep.	required
summaries	Optional[Dict[str, Dict[str, float]]]	Residual summaries (forwarded to RegionScheduler).	None
boundary	Optional[str]	Optional boundary event name.	None
summary_tensor	Optional[Tensor]	Flat (summary_dim,) tensor consumed by the learned scheduler. Required when learned_regions is non-empty. If omitted while learned regions exist, the learned regions simply don't fire this step.	None