jev_serial_scheduler.py
原始文件
1#!/usr/bin/env python3
2"""Durable event-driven Jev queue for the 2026-11-07 five-App experiment.
3
4Only one Jev run is ever submitted by this process. It waits for the prior
5run to reach a terminal state before creating the next request, and records
6every decision in scheduler-state.json. It never retries a run and never
7changes a run after submission.
8"""
9from __future__ import annotations
10
11import fcntl
12import json
13import os
14import time
15import urllib.request
16from pathlib import Path
17
18ROOT = Path(__file__).resolve().parents[1]
19STATE_PATH = ROOT / "admission" / "scheduler-state.json"
20LOCK_PATH = ROOT / "admission" / "scheduler.lock"
21API = "http://127.0.0.1:8000"
22PROJECT = "guitaboo-5app-1rep-baseline-comparison-20261107"
23TERMINAL = {"completed", "invalid", "interrupted", "startup_failed", "cancelled", "reused"}
24
25# ODK is already queued by the initial submission. The remaining entries are
26# templates copied from the cancelled, pre-created Jev requests; no old
27# idempotency key is reused.
28FIRST = "run_b4871490b9e84bbcb1a9b3464ca596fd"
29TEMPLATES = [
30 ("wikipedia", "run_e2ec41a9bbeb4392be74bbc647c70fca"),
31 ("scarletnotes", "run_c6b3ac0504354073bb8b664b60a4ddbd",),
32 ("omninotes", "run_03a3021d78324925aff96071a9ef4f93"),
33 ("activitydiary", "run_95ba6d15213c4b8e94673fe9af556c91"),
34]
35
36
37def api(path, payload=None):
38 data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode()
39 req = urllib.request.Request(API + path, data=data, headers={"Content-Type": "application/json"})
40 with urllib.request.urlopen(req, timeout=30) as r:
41 return json.loads(r.read())
42
43
44def save(state):
45 tmp = STATE_PATH.with_suffix(".tmp")
46 tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n")
47 os.replace(tmp, STATE_PATH)
48
49
50def template_payload(config, app, ordinal):
51 fields = [
52 "project", "app", "rep", "tool", "condition", "apk_path", "apk_sha256",
53 "device_id", "seed", "budget_seconds", "provider_id", "code", "parameters",
54 "prompt_config", "initial_state", "evaluation_protocol", "command",
55 ]
56 payload = {k: config[k] for k in fields if k in config}
57 payload["app"] = app
58 payload["idempotency_key"] = (
59 f"guitaboo-5app-fixab-serial-jev-{app}-rep2-20261107-{ordinal}"
60 )
61 payload["rerun_of"] = None
62 payload["reuse_from"] = None
63 return payload
64
65
66def main():
67 LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
68 with LOCK_PATH.open("a+") as lock:
69 fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
70 state = json.loads(STATE_PATH.read_text()) if STATE_PATH.exists() else {
71 "project": PROJECT, "first_run": FIRST, "next_index": 0,
72 "submitted": [], "events": [], "halted": False,
73 }
74 while not state.get("halted") and state["next_index"] < len(TEMPLATES):
75 prior = state["submitted"][-1]["run_id"] if state["submitted"] else FIRST
76 detail = api("/api/runs/" + prior)
77 state["events"].append({"at": time.time(), "observed": prior, "state": detail.get("state")})
78
79 # Once ODK Jev has actually acquired the device, release the
80 # deferred ODK StartLux run into the same device queue. It then
81 # follows Jev after cleanup without delaying Jev admission.
82 if (prior == FIRST and detail.get("state") in
83 {"admission", "environment_setup", "starting", "exploring", "collecting", "validating"}
84 and not state.get("deferred_startlux_run_id")):
85 deferred = json.loads((ROOT / "admission" / "odk-startlux-deferred-config.json").read_text())
86 payload = template_payload(deferred["config"], "odkcollect", 1)
87 payload["idempotency_key"] = "guitaboo-5app-fixab-serial-startlux-odkcollect-rep2-20261107"
88 created = api("/api/runs", payload)
89 state["deferred_startlux_run_id"] = created["id"]
90 state["events"].append({"at": time.time(), "deferred_startlux_submitted": created["id"],
91 "after_jev": prior})
92 save(state)
93 if detail.get("state") not in TERMINAL:
94 save(state)
95 time.sleep(15)
96 continue
97
98 app, template_id = TEMPLATES[state["next_index"]]
99 existing = api("/api/runs/" + template_id)
100 payload = template_payload(existing["config"], app, state["next_index"] + 2)
101 created = api("/api/runs", payload)
102 entry = {"app": app, "run_id": created["id"], "template": template_id,
103 "created_at": time.time(), "prior_run": prior}
104 state["submitted"].append(entry)
105 state["next_index"] += 1
106 state["events"].append({"at": time.time(), "submitted": entry})
107 save(state)
108 state["complete"] = state["next_index"] >= len(TEMPLATES)
109 save(state)
110
111
112if __name__ == "__main__":
113 main()