LowPriority: Letting a local LLM decide when your background jobs run

A technical deep-dive into an on-device, policy-gated AI scheduler.

Target audience: developers. We're going to walk through the actual
architecture, the decision pipeline, the safety model, and the trade-offs
we made — with real code from the repo.

Desktop software runs a lot of hidden background work: indexers, syncs,
test runners, build pipelines, ML jobs, backups. Almost all of it fires at
the worst time — right when the user is compiling something or on a call —
because nobody asked "is now actually a good moment?"

cron schedules a wall-clock time and never looks at the machine again.
nice/ionice deprioritize a process but never make the scheduling
decision.

LowPriority takes a different angle: it watches your machine, learns your
usage patterns, and uses a local LLM to judge — while a deterministic
policy engine decides. The LLM is an advisor. The policy is the law.

This post covers: the component design, the telemetry contract, the
inference protocol, the deterministic safety layer, the data layer, the
safe execution model, and the test strategy. Real code, real trade-offs.

---

1. The core idea in one picture

Every scheduler is a function of three inputs:

f(machine state, job queue, learned behavior) -> {run, defer, pause, resume} per job

In LowPriority each of those inputs is a first-class, testable module:

┌────────────────────────────┐
│  Typer CLI + Rich dashboard│
└─────────────┬──────────────┘
              │
              ▼
┌────────────────────────────┐
│      Daemon loop           │  every `interval` seconds
└─────────────┬──────────────┘
              │
   ┌──────────┼──────────────┐
   ▼          ▼              ▼
┌────────┐ ┌───────────┐ ┌──────────────┐
│psutil  │ │ SQLite    │ │ JobQueue     │
│snapshot│ │ history   │ │ (SQLite)     │
└───┬────┘ └─────┬─────┘ └──────┬───────┘
    │            │              │
    └────────────┼──────────────┘
                 ▼
        ┌──────────────────────┐   advisor (optional)
        │  Planner             │──▶ Ollama / LM Studio
        │  (one eval cycle)    │◀── structured JSON verdicts
        └───────┬──────┬───────┘
                │      │
        ┌───────▼──┐ ┌─▼──────────────┐
        │ Policy   │ │ Process        │
        │ engine   │ │ Executor       │
        │ (law)    │ │ (process group)│
        └──────────┘ └────────────────┘

Reasons each piece is shaped the way it is:

Component	Why it exists
TelemetrySnapshot	One data contract shared by policy, agent, history, UI
Planner	One function that orchestrates a complete evaluation cycle
policy.py	Deterministic, auditable, authoritative — never overridden
agent.py	Optional LLM advice; zero effect on safety if it fails
executor.py	Isolated process groups, graceful kill, no shell by default
SQLite	Jobs, telemetry, runs, and decisions in one file, WAL mode

---

2. The telemetry contract

Everything downstream speaks one dataclass. That's the single most
important design decision: the planner, the policy, the LLM context, and
the history aggregator all consume a TelemetrySnapshot and nothing
else.

# src/lowpriority/telemetry/collector.py (abridged)
@dataclass
class TelemetrySnapshot:
    timestamp: datetime = field(default_factory=datetime.now)
    cpu_percent: float = 0.0
    cpu_load: float | None = None
    memory_percent: float = 0.0
    available_memory_mb: float = 0.0
    disk_utilization: float | None = None
    disk_read_mbps: float | None = None
    disk_write_mbps: float | None = None
    network_mbps: float | None = None
    user_active: bool = False
    foreground_application: str | None = None
    running_processes: list[str] = field(default_factory=list)
    on_battery: bool = False

Derived values become properties, so every consumer computes them the same
way:

@property
def cpu_headroom(self) -> float:
    return max(0.0, min(1.0, 1.0 - self.cpu_percent / 100.0))

@property
def disk_io_level(self) -> str:
    peak = max(self.disk_read_mbps or 0.0, self.disk_write_mbps or 0.0)
    return "high" if peak >= 50 else "medium" if peak >= 10 else "low"

And context_summary() is the exact view handed to the LLM — capped to
six processes so prompt tokens stay tiny and inference stays cheap:

def context_summary(self) -> dict[str, Any]:
    return {
        "cpu_percent": round(self.cpu_percent, 1),
        "memory_percent": round(self.memory_percent, 1),
        "disk_io": self.disk_io_level,
        "user_active": self.user_active,
        "active_processes": self.running_processes[:6],
        "on_battery": self.on_battery,
    }

Trade-off we made: "user active" is a heuristic, not a ground truth. We
consider a process interactive if it has a terminal attached and burns
CPU, fall back to "no logged-in user → idle", and finally use a CPU
threshold. It's deliberately cheap and best-effort — perfection here costs
more than it earns.

---

3. History: statistics, not ML

The model needs context, and context means history. UsageHistory
aggregates the last 7 days of telemetry into per-hour buckets. It's plain,
inspectable statistics — you can query it yourself:

SELECT hour, samples, activity_probability, mean_cpu, mean_memory
FROM ...

# src/lowpriority/history/usage.py
def hourly_stats(self, days: int = 7) -> dict[int, dict]:
    rows = self.db.query("SELECT * FROM telemetry WHERE timestamp >= ? ...", ...)
    buckets: dict[int, list[dict]] = {h: [] for h in range(24)}
    for row in rows:
        ts = datetime.fromisoformat(row["timestamp"])
        buckets[ts.hour].append(row)

    for hour, rows in buckets.items():
        active = sum(1 for r in rows if r.get("user_active"))
        cpus  = [r["cpu_percent"] for r in rows if r.get("cpu_percent") is not None]
        stats[hour] = {
            "activity_probability": round(active / len(rows), 2),
            "mean_cpu": round(sum(cpus) / len(cpus), 1) if cpus else 0.0,
            ...
        }

From that, we derive an idle score used across the planner:

idle_score = 0.5 × (1 − activity_probability)
           + 0.3 × cpu_headroom
           + 0.2 × memory_headroom

Why 0.5/0.3/0.2 and not a fitted model? Because they're interpretable
and auditable. A user (or a reviewer) can read the formula and understand
why a 9am job defers. We can revisit the weights in a config file later —
we chose explainability over a black box on purpose.

typical_idle_windows() is a nice byproduct: runs of consecutive hours
where learned activity probability is low, e.g. (22, 6, 0.94) → "your
machine is idle from 22:00 to 06:00, on average 94%."

---

4. The agent protocol: a tiny, validated vocabulary

The LLM is an advisor. To keep it harmless we constrained the protocol
before any model touches it:

class AgentDecision(BaseModel):
    decision: Literal["run", "defer", "pause", "resume"] = "defer"
    job_id: str | None = None
    reason: str = ""
    confidence: float = Field(default=0.5, ge=0.0, le=1.0)
    recommended_delay_seconds: int | None = Field(default=None, ge=0)

class DecisionBatch(BaseModel):
    decisions: list[AgentDecision] = Field(default_factory=list)

Note what is not in the vocabulary: no command, no shell string, no
job registration. The model cannot express "run rm -rf /". It can only
pick among job IDs already in the database.

The full InferenceProvider protocol is three methods — small on purpose:

class InferenceProvider(Protocol):
    name: str
    async def health_check(self) -> bool: ...
    async def list_models(self) -> list[str]: ...
    async def generate(self, messages, response_schema) -> T: ...

Ollama and LM Studio both expose OpenAI-compatible APIs, so the two
providers share the same HTTP shape (httpx):

payload = {
    "model": model,
    "messages": messages,
    "temperature": 0.1,          # greedy: we want determinism, not creativity
    "stream": False,
    "response_format": {"type": "json_object"},
}

The response is parsed defensively — local models wrap JSON in prose and
markdown fences all the time:

def extract_json_object(text: str) -> dict:
    fenced = re.search(r"```

(?:json)?\s*(.*?)

```", text, re.DOTALL)
    if fenced:
        text = fenced.group(1).strip()
    start, end = text.find("{"), text.rfind("}")
    if start == -1 or end <= start:
        raise ProviderError("no JSON object found in model output ...")
    return json.loads(text[start:end + 1])

Everything is run through Pydantic. Malformed model output is never
executed — it's an error, kept out of the pipeline.

The context handed to the model is deliberately minimal (one call covers
all pending jobs; batches keep inference cheap):

{
  "system": { "cpu_percent": 67.0, "memory_percent": 59.0,
              "disk_io": "low", "user_active": false,
              "active_processes": ["Code"], "on_battery": false },
  "jobs": [
    { "id": "a83f4c2e", "name": "Generate embeddings", "priority": "low",
      "status": "queued", "estimated_cpu": 25,
      "estimated_memory_mb": 500, "deadline": null }
  ]
}

---

5. The deterministic safety layer (the law)

This is the file to read first if you're security-conscious:
scheduler/policy.py. The agent can suggest; this engine decides.

def system_gate(snapshot: TelemetrySnapshot, cfg: Config) -> PolicyResult:
    limits, behavior = cfg.limits, cfg.behavior

    # strength order: battery > free memory > CPU > memory > user activity
    if snapshot.on_battery and behavior.pause_on_battery:
        return PolicyResult.pause("on battery; pausing background work")

    if snapshot.available_memory_mb < limits.min_free_memory_mb:
        return PolicyResult.defer("insufficient free memory (...)")

    if snapshot.cpu_percent > limits.max_background_cpu:
        return PolicyResult.defer("CPU usage elevated (...)")

    if snapshot.user_active and behavior.pause_on_high_activity:
        return PolicyResult.defer("user activity detected")

    return PolicyResult.allow("resources available", weight=1)

Per-job rules layer on top (job_check): job-specific hard limits
(max_cpu_percent, max_memory_mb), plus a deadline-urgency override —
a high priority job close to its deadline may push past a moderate
gate (e.g. CPU at 60% vs limit 50%), and that override is recorded in
plain text so it's auditable.

urgency = deadline_elapsed_ratio(job)      # 0.0 far, 1.0 overdue
weight  = int(urgency * 10)

if job.priority == "high" and urgency > 0.7:
    return PolicyResult.allow(
        f"high priority + deadline urgency ({urgency:.0%}); overriding: {gate.reason}",
        weight=weight + 10,
    )

The planner makes the authority explicit — the agent's recommendation is
folded into the reason, but the verdict always comes from policy:

# src/lowpriority/scheduler/planner.py (abridged)
reason, decision, source = self._combine(job, gate, agent_by_job.get(job.id))
# policy is authoritative; source records whether policy or agent decided

Why a deterministic layer at all, when we have an LLM? Three reasons:
(1) an LLM is non-deterministic — a throttled, auditable engine guarantees
repeatable behavior; (2) latency and cost — small models on a laptop are
slow, and we can gate before ever paying for inference; (3) safety —
no matter what the model does, the machine has hard invariants.

---

6. The planner: one evaluation cycle

Planner.evaluate_once() is the heart. It orchestrates nine steps in one
async function, which makes the whole scheduler unit-testable with a fake
runner and a fake provider:

async def evaluate_once(self, snapshot=None) -> PlanResult:
    snapshot = snapshot or collect()          # 1. telemetry
    self.history.record(snapshot)             # 2. history
    result.completed = self.runner.poll()     # 3. reap finished processes
    self.runner.enforce_limits(self.config)   # 4. kill runaway jobs
    pending = self.queue.pending()            # 5. gather work
    gates = [(j, policy_mod.job_check(snapshot, j, self.config))
             for j in pending]                # 6. policy gate
    agent_by_job = {}
    if self.agent.available:
        for rec in await self.agent.recommend(snapshot, pending, hourly):
            if rec.job_id and self.agent.validate_decision(rec):
                agent_by_job[rec.job_id] = rec   # 7. agent advice
    for job, gate in gates:
        reason, decision, _source = self._combine(job, gate,
                                                  agent_by_job.get(job.id))
        # 8. apply: start / pause / resume (dry-run aware)
        # 9. persist the decision to SQLite for explainability

Because the LLM call is async and the rest is pure logic, the whole cycle
is testable without a model, without processes, without hardware. That's
deliberate — see §9.

---

7. Execution: the parts most apps get wrong

jobs/model.py defines the Job (user-registered only) and
scheduler/executor.py owns process lifetimes with three rules:

1. Isolated process groups.

subprocess.Popen(
    command,                     # argument list — shell=False by default
    start_new_session=True,      # own process group → can kill the whole tree
    stdout=log_file, stderr=log_file,
    cwd=job.working_directory,
)

2. Graceful termination — SIGTERM, wait, then SIGKILL.

# executor.stop()
proc.send_signal(signal.SIGTERM)
try:
    proc.wait(timeout=TERM_GRACE_SECONDS)   # 8s of good citizenship
except subprocess.TimeoutExpired:
    proc.kill()                              # then force

3. No shell unless the user opts in per-job. By default commands are
shlex.split into argv arrays (shell=False). --shell is an explicit,
warned-about opt-in for workflows that genuinely need &&, pipes, or
redirection.

JobRunner keeps SQLite state and live processes in sync, and self-heals:
if the DB says running but no process exists (a previous runtime died),
it adopts or abandons the job gracefully — state never gets stuck.

---

8. Persistence and explainability

SQLite, WAL mode, four tables:

jobs       -- id, name, command, priority, status, limits, deadline, shell
telemetry  -- one row per snapshot (the 7-day history)
job_runs   -- per process: pid, exit_code, cpu/mem peaks, log path
decisions  -- every scheduling decision + who decided + why + confidence

The decisions table is the part we wrote first, because it's the reason
you can trust the app: every run/defer/pause/resume is recorded
with a human-readable reason, the confidence, and its source
(policy or agent).

The DB wrapper is thread-safe by design: WAL lets readers and writer
coexist, a threading.Lock serializes writes, and each thread gets its own
connection via threading.local().

# src/lowpriority/database.py
class Database:
    def execute(self, sql, params=()):
        with self._lock:                       # writes serialized
            cur = self._conn.execute(sql, params)
            self._conn.commit()
            return cur

    def query(self, sql, params=()):
        return [dict(r) for r in self._conn.execute(sql, params).fetchall()]

---

9. Testing strategy: no hardware, no network, no model

The three scary dependencies — processes, AI backends, system metrics —
are all test doubles behind narrow seams. That's why each module exists.

• Virtual telemetry. Tests build TelemetrySnapshot directly (cpu_percent=85, user_active=True, ...) and drive the planner with it. No OS metrics, no flakiness.
• Virtual inference. httpx.MockTransport simulates Ollama / LM Studio responses — including malformed and hostile ones — so the whole validation path is provable without a server.
• Virtual execution. A FakeRunner records started/stopped and touches only the in-memory Database(":memory:").

# tests/conftest.py — the eternal sandbox
@pytest.fixture()
def db() -> Database:
    return Database(":memory:")   # tests never touch a real file

The suite asserts the safety invariants, not just behavior:

async def test_policy_overrides_agent_run_when_busy(...):
    """Agent says run, CPU is busy -> defer regardless."""
    # decision.source == "policy", nothing was started
async def test_unknown_decision_job_id_ignored(...):
    """The LLM can't invent job IDs."""
    # only the real job's process could ever start

Run it:

pytest                 # 70 tests
ruff check src tests   # lint

---

10. What we learned (the honest part)

1. A tiny, validated LLM protocol beats a clever one. Giving the model a four-word vocabulary (run|defer|pause|resume) plus JSON validation removed an entire class of bugs before they could exist.
2. "Local" is a feature, not a limitation. No API keys, no billing, no data leaving the laptop — it changes how defensively you write the compatibility layer and how you test it (MockTransport, not a sandboxed cloud).
3. Explainability is a product requirement. The decisions table and the --dry-run flag aren't features for us — they're what let a user trust an AI-based scheduler with their machine.
4. Deterministic rules + probabilistic advice is a great pairing. Rules give hard guarantees and a floor; the model gives nuance ("your machine looks idle now, and historically it stays idle for the next hour").
5. Plausible next hard problem: learned resource estimation. Right now job_runs already records real CPU/memory peaks — feeding those back into Job.estimated_* is the natural way to make the agent's context honest.

---

11. Dive in

• Repository: github.com/harishkotra/lowpriority
• Source layout: src/lowpriority/{telemetry,inference,jobs,scheduler,history,ui}
• Start reading: scheduler/planner.py then scheduler/policy.py
• Claims verified by: pytest (70 tests) and ruff check src tests

Suggested first contributions: quiet hours / cooldown enforcement in
policy.py; launchd/systemd user units; GPU-aware gating; learned
resource estimation from job_runs.

Code & more: https://www.dailybuild.xyz/project/223-low-priority