tiers.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/timing/tiers.py
 
"""
Purpose
-------
Declare each timed stage's cost-of-quality tier and that tier's budget.
 
Responsibilities
----------------
- Keep local timing data bounded, private and advisory.
 
Diagnostics
-----------
Domain: GENERAL
Levels:
    L1 -- instrumentation failures
    L2 -- lifecycle
    L3 -- semantic details
    L4 -- deep tracing
 
Contracts
---------
- Timing instrumentation never changes the wrapped operation's outcome.
- A project may lower a shipped tier budget and never raise one.
 
Dependencies
------------
- oc_diagnostics (structured timing diagnostics)
"""
from __future__ import annotations
 
from collections.abc import Mapping
from types import MappingProxyType
from oc_diagnostics import _dbg
 
INNER_LOOP = "inner_loop"
PER_COMMIT = "per_commit"
PER_LANDING = "per_landing"
PERIODIC = "periodic"
 
#: The Tooling spec's "Cost of quality: tier budgets" table. Aims, never
#: limits: the unattended tier ships no budget, so it has none to lower.
TIER_BUDGETS_MS: Mapping[str, int | None] = MappingProxyType({
    INNER_LOOP: 10_000,
    PER_COMMIT: 30_000,
    PER_LANDING: 900_000,
    PERIODIC: None,
})
 
#: Every stage the ledger can record: each command path `command_stage`
#: builds from the CLI group, plus the three engine operations. A stage
#: whose moment is unclear takes the slower tier rather than a tight one.
_STAGES_BY_TIER: Mapping[str, tuple[str, ...]] = {
    INNER_LOOP: ("format", "health", "lint", "new", "oct"),
    PER_COMMIT: ("git", "git.add", "git.branch", "git.check", "git.commit",
                 "git.commit-msg", "git.ignore", "git.restore", "git.status"),
    PER_LANDING: ("agent", "audit", "clean", "complexity", "deps", "docs",
                  "engine.dispatch", "engine.prepare", "engine.teardown",
                  "export-skeleton", "export-source", "git.changelog", "test",
                  "typecheck"),
    PERIODIC: ("diag", "diag.list-domains", "diag.migrate-config",
               "diag.restore", "diag.set-level", "diag.validate-config",
               "git.hooks", "git.hooks.install", "git.hooks.remove",
               "git.hooks.status", "git.init", "install", "install-hooks",
               "install-mcp", "migrate", "migrate.option-c", "mutate",
               "scaffold"),
}
 
STAGE_TIERS: Mapping[str, str] = MappingProxyType(
    {stage: tier for tier, stages in _STAGES_BY_TIER.items() for stage in stages})
 
def tier_for(stage: object) -> str | None:
    """Return the declared tier, never inventing one for an unknown stage."""
    return STAGE_TIERS.get(stage) if isinstance(stage, str) else None
 
def budgets_ms(overrides: object = None) -> dict[str, int | None]:
    """Apply project overrides, which may only lower a shipped budget."""
    func = "budgets_ms"
    budgets = dict(TIER_BUDGETS_MS)
    if not isinstance(overrides, Mapping):
        return budgets
    for tier, value in overrides.items():
        shipped = TIER_BUDGETS_MS.get(tier) if isinstance(tier, str) else None
        if shipped is None or type(value) is not int or value <= 0 or value > shipped:
            _dbg("GENERAL", func, "VALIDATION_ERROR: invalid timing budget", 3)
            continue
        budgets[tier] = value
    return budgets
 
def budget_for(stage: object, budgets: Mapping[str, int | None] | None = None) -> int | None:
    """Return the effective budget, or None when the stage declares no tier."""
    tier = tier_for(stage)
    return None if tier is None else (TIER_BUDGETS_MS if budgets is None else budgets).get(tier)