diff --git a/Cargo.lock b/Cargo.lock index d8ec7aa5..c552e886 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2649,7 +2649,7 @@ dependencies = [ [[package]] name = "ptask-bot" -version = "3.50.6" +version = "3.50.7" dependencies = [ "anyhow", "jiff", @@ -2665,7 +2665,7 @@ dependencies = [ [[package]] name = "ptask-cli" -version = "3.50.6" +version = "3.50.7" dependencies = [ "anyhow", "axum", @@ -2693,7 +2693,7 @@ dependencies = [ [[package]] name = "ptask-core" -version = "3.50.6" +version = "3.50.7" dependencies = [ "hex", "interim", @@ -2714,7 +2714,7 @@ dependencies = [ [[package]] name = "ptask-distill" -version = "3.50.6" +version = "3.50.7" dependencies = [ "anyhow", "candle-core", @@ -2738,7 +2738,7 @@ dependencies = [ [[package]] name = "ptask-notify" -version = "3.50.6" +version = "3.50.7" dependencies = [ "lettre", "ptask-core", @@ -2751,7 +2751,7 @@ dependencies = [ [[package]] name = "ptask-server" -version = "3.50.6" +version = "3.50.7" dependencies = [ "anyhow", "axum", @@ -2783,7 +2783,7 @@ dependencies = [ [[package]] name = "ptask-tui" -version = "3.50.6" +version = "3.50.7" dependencies = [ "anyhow", "crossterm", diff --git a/Cargo.toml b/Cargo.toml index 0458a215..1499b63a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ members = [ resolver = "2" [workspace.package] -version = "3.50.6" +version = "3.50.7" edition = "2024" license-file = "LICENSE" authors = ["PureTensor, Inc. "] @@ -52,12 +52,12 @@ rmcp = { version = "3.3.0", features = ["server", "macros", "schemars", "transpo futures-util = { version = "0.3", default-features = false } # Internal crates -ptask-core = { path = "crates/ptask-core", version = "3.50.6" } -ptask-tui = { path = "crates/ptask-tui", version = "3.50.6" } -ptask-server = { path = "crates/ptask-server", version = "3.50.6" } -ptask-bot = { path = "crates/ptask-bot", version = "3.50.6" } -ptask-distill = { path = "crates/ptask-distill", version = "3.50.6" } -ptask-notify = { path = "crates/ptask-notify", version = "3.50.6" } +ptask-core = { path = "crates/ptask-core", version = "3.50.7" } +ptask-tui = { path = "crates/ptask-tui", version = "3.50.7" } +ptask-server = { path = "crates/ptask-server", version = "3.50.7" } +ptask-bot = { path = "crates/ptask-bot", version = "3.50.7" } +ptask-distill = { path = "crates/ptask-distill", version = "3.50.7" } +ptask-notify = { path = "crates/ptask-notify", version = "3.50.7" } # HMAC for outbound webhook signing (v0.3.5). sha2 = "0.10" diff --git a/docs/gen/pt.1 b/docs/gen/pt.1 index 71d0f9b0..7080583b 100644 --- a/docs/gen/pt.1 +++ b/docs/gen/pt.1 @@ -1,6 +1,6 @@ .ie \n(.g .ds Aq \(aq .el .ds Aq ' -.TH pt 1 "pt 3.50.6" +.TH pt 1 "pt 3.50.7" .SH NAME pt \- Sovereign task manager for PureTensor .SH SYNOPSIS @@ -208,4 +208,4 @@ Generate shell completions (bash/zsh/fish) to stdout pt\-help(1) Print this message or the help of the given subcommand(s) .SH VERSION -v3.50.6 +v3.50.7 diff --git a/scripts/systemd/ptask-distill.service b/scripts/systemd/ptask-distill.service index 80200349..d812c8d4 100644 --- a/scripts/systemd/ptask-distill.service +++ b/scripts/systemd/ptask-distill.service @@ -12,11 +12,12 @@ OnFailure=ptask-failure-alert@%n.service Type=oneshot EnvironmentFile=%h/puretensor-tasks/.env # Wait up to ~15 min for the local provider (the seat takes ~10 min to load -# after a reboot or a GPU lease). If it is still down and a pt-gpu lease is -# active or ended <1 h ago, the broker deliberately stopped it: skip this run -# (exit 1 = condition failed, unit not failed). Any other outage proceeds and -# fails closed in pt, so a real provider outage still alerts. -ExecCondition=/bin/sh -c 'case "$${LOCAL_LLM_URL:-}" in http*) ;; *) exit 0;; esac; for i in $$(seq 1 90); do curl -sf -m3 "$${LOCAL_LLM_URL%%/}/models" >/dev/null && exit 0; sleep 10; done; if python3 -c "import json,os,sys,time; L=open(os.path.expanduser(\\\"~/.local/state/ptcompute/audit.jsonl\\\")).read().splitlines(); e=[json.loads(x) for x in L[-50:] if x.strip()]; e=[x for x in e if x.get(\\\"event\\\") in (\\\"grant\\\",\\\"release\\\")]; l=e[-1] if e else {}; sys.exit(0 if l.get(\\\"event\\\")==\\\"grant\\\" or (l.get(\\\"event\\\")==\\\"release\\\" and time.time()-l.get(\\\"ts\\\",0)<3600) else 1)" 2>/dev/null; then echo "local llm down under a pt-gpu lease; skipping this run" >&2; exit 1; fi; echo "local llm not ready after wait; proceeding" >&2; exit 0' +# after a reboot or a GPU lease). If it is still down and the last pt-gpu +# lease event is a grant <12 h old or a release <1 h old, the broker +# deliberately stopped it: skip this run (exit 1 = condition failed, unit not +# failed). A stale grant (leaked by a crashed broker/holder) is ignored, so +# any other outage proceeds and fails closed in pt and still alerts. +ExecCondition=/bin/sh -c 'case "$${LOCAL_LLM_URL:-}" in http*) ;; *) exit 0;; esac; for i in $$(seq 1 90); do curl -sf -m3 "$${LOCAL_LLM_URL%%/}/models" >/dev/null && exit 0; sleep 10; done; if python3 -c "import json,os,sys,time; L=open(os.path.expanduser(\\\"~/.local/state/ptcompute/audit.jsonl\\\")).read().splitlines(); e=[json.loads(x) for x in L[-50:] if x.strip()]; e=[x for x in e if x.get(\\\"event\\\") in (\\\"grant\\\",\\\"release\\\")]; l=e[-1] if e else {}; a=time.time()-l.get(\\\"ts\\\",0); sys.exit(0 if (l.get(\\\"event\\\")==\\\"grant\\\" and -300/dev/null; then echo "local llm down under a pt-gpu lease; skipping this run" >&2; exit 1; fi; echo "local llm not ready after wait; proceeding" >&2; exit 0' ExecStart=%h/.cargo/bin/pt distill --batch 300 Nice=10 IOSchedulingClass=best-effort diff --git a/tests/test_distill_service_condition.py b/tests/test_distill_service_condition.py new file mode 100644 index 00000000..52457c18 --- /dev/null +++ b/tests/test_distill_service_condition.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +"""Regression test for the ptask-distill.service ExecCondition lease check. + +Runs the real ExecCondition shell (after systemd's $$/%%/backslash +unescaping) with fake curl/sleep on PATH so the provider looks down and the +wait loop is instant, against a temporary $HOME audit log. +""" +import json +import os +import re +import subprocess +import tempfile +import time +import unittest + +UNIT = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", "scripts", "systemd", "ptask-distill.service", +) + + +def condition_script(): + with open(UNIT) as f: + for line in f: + if line.startswith("ExecCondition="): + value = line.rstrip("\r\n")[len("ExecCondition="):] + break + else: + raise AssertionError("no ExecCondition= in " + UNIT) + # systemd: "$$" -> "$", "%%" -> "%", then C-style backslash unescaping. + value = value.replace("$$", "$").replace("%%", "%") + value = re.sub(r"\\(.)", lambda m: m.group(1), value) + prefix = "/bin/sh -c '" + assert value.startswith(prefix) and value.endswith("'"), value + return value[len(prefix):-1] + + +class DistillLeaseConditionTest(unittest.TestCase): + def run_condition(self, events): + with tempfile.TemporaryDirectory() as home: + state = os.path.join(home, ".local", "state", "ptcompute") + os.makedirs(state) + with open(os.path.join(state, "audit.jsonl"), "w") as f: + for e in events: + f.write(json.dumps(e) + "\n") + bindir = os.path.join(home, "bin") + os.makedirs(bindir) + for name, body in (("curl", "exit 7"), ("sleep", "exit 0")): + path = os.path.join(bindir, name) + with open(path, "w") as f: + f.write("#!/bin/sh\n" + body + "\n") + os.chmod(path, 0o755) + env = { + "HOME": home, + "PATH": bindir + os.pathsep + os.environ.get("PATH", "/usr/bin:/bin"), + "LOCAL_LLM_URL": "http://127.0.0.1:9/v1", + } + r = subprocess.run( + ["/bin/sh", "-c", condition_script()], + env=env, capture_output=True, text=True, timeout=120, + ) + return r.returncode, r.stderr + + def test_stale_grant_proceeds_and_fails_closed(self): + now = time.time() + code, err = self.run_condition([ + {"event": "grant", "ts": now - 48 * 3600}, + ]) + self.assertEqual(code, 0, err) + self.assertIn("proceeding", err) + + def test_recent_grant_skips(self): + code, err = self.run_condition([ + {"event": "grant", "ts": time.time() - 60}, + ]) + self.assertEqual(code, 1, err) + self.assertIn("skipping", err) + + def test_recent_release_skips(self): + now = time.time() + code, err = self.run_condition([ + {"event": "grant", "ts": now - 1800}, + {"event": "release", "ts": now - 600}, + ]) + self.assertEqual(code, 1, err) + + def test_old_release_proceeds(self): + now = time.time() + code, err = self.run_condition([ + {"event": "grant", "ts": now - 4 * 3600}, + {"event": "release", "ts": now - 2 * 3600}, + ]) + self.assertEqual(code, 0, err) + + def test_no_lease_events_proceeds(self): + code, err = self.run_condition([]) + self.assertEqual(code, 0, err) + + +if __name__ == "__main__": + unittest.main()