Default scheduler queue event log from env

Author: tsotchke

configs/tensorcore-scheduler.env.example

@@ -9,6 +9,7 @@ TC_MESH_ARBITER_CMD=/opt/computer_mesh/tsotchke/bin/tsotchke-arbiter
 # TC_REMOTE_ARBITER_KEY=/etc/tensorcore/arbiter_ssh_key
 TC_SCHEDULER_ROOT=/var/lib/tensorcore
 TC_SCHEDULER_JOBS_JSON=/var/lib/tensorcore/mesh_resource_jobs.json
+TC_SCHEDULER_EVENT_LOG_JSONL=/var/lib/tensorcore/mesh_resource_queue_events.jsonl
 TC_SCHEDULER_INVENTORY_JSON=/opt/tensorcore/configs/mesh_resources.json
 TC_SCHEDULER_STATE_JSON=/var/lib/tensorcore/mesh_resource_state.json
 TC_SCHEDULER_INTERVAL_SEC=30

docs/ci_and_scripts.md

@@ -786,7 +786,9 @@ rows until the unit or equivalent launcher is installed from a git checkout.
 Real `submit` and `cancel` queue mutations require `--event-log-jsonl` and
 take an advisory lock beside the queue file before updating jobs or appending
 the queue event. Scheduler-VM clients should use those commands instead of
-editing `mesh_resource_jobs.json` directly.
+editing `mesh_resource_jobs.json` directly. Set
+`TC_SCHEDULER_EVENT_LOG_JSONL` in the scheduler environment so clients inherit
+the canonical event-log path.
 
 Run one dry pass:
 

docs/mesh_resource_scheduler.md

@@ -46,7 +46,8 @@ append-only audit trail. Mutating commands take an advisory lock beside the
 queue file, named like `.mesh_resource_jobs.json.lock`, before the
 load-modify-write cycle and before appending the queue event. Scheduler-VM
 clients should call the submit/cancel CLI instead of editing the queue file
-directly.
+directly. On the scheduler VM, set `TC_SCHEDULER_EVENT_LOG_JSONL` so submit and
+cancel clients inherit the canonical queue event log path.
 
 Example scheduler loop:
 

scripts/mesh_resource_scheduler.py

@@ -2422,7 +2422,7 @@ def add_output_flags(p: argparse.ArgumentParser) -> None:
         submit.add_argument("--job-json", required=True)
         submit.add_argument("--jobs-json", required=True)
         submit.add_argument("--inventory-json", required=True)
-        submit.add_argument("--event-log-jsonl")
+        submit.add_argument("--event-log-jsonl", default=os.environ.get("TC_SCHEDULER_EVENT_LOG_JSONL"))
         submit.add_argument("--replace", action="store_true")
         submit.add_argument("--dry-run", action="store_true")
         add_output_flags(submit)
@@ -2439,7 +2439,7 @@ def add_output_flags(p: argparse.ArgumentParser) -> None:
 
         cancel = sub.add_parser("cancel", help="Disable a queued job without releasing leases")
         cancel.add_argument("--jobs-json", required=True)
-        cancel.add_argument("--event-log-jsonl")
+        cancel.add_argument("--event-log-jsonl", default=os.environ.get("TC_SCHEDULER_EVENT_LOG_JSONL"))
         cancel.add_argument("--job-id", required=True)
         cancel.add_argument("--reason", default="operator_cancelled")
         cancel.add_argument("--dry-run", action="store_true")
@@ -2468,7 +2468,13 @@ def add_output_flags(p: argparse.ArgumentParser) -> None:
         audit.add_argument("--worker-reconciliation-dir", action="append", default=[])
         add_output_flags(audit)
 
-        return parser.parse_args(argv)
+        args = parser.parse_args(argv)
+        if (
+            args.control_command == "status"
+            and args.gpu_reconciliation_max_age_sec <= 0
+        ):
+            parser.error("--gpu-reconciliation-max-age-sec must be > 0")
+        return args
 
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--arbiter-cmd", default=DEFAULT_ARBITER_CMD)

scripts/mesh_resource_scheduler_selftest.py

@@ -8,10 +8,11 @@
 import importlib.util
 import io
 import json
+import os
 import pathlib
 import subprocess
 import tempfile
-from contextlib import redirect_stdout
+from contextlib import redirect_stderr, redirect_stdout
 from types import ModuleType
 from typing import Any
 
@@ -1792,6 +1793,40 @@ def test_submit_and_cancel_require_event_log_for_queue_mutation() -> None:
             raise AssertionError("cancel without event log was accepted")
 
 
+def test_submit_and_cancel_parse_event_log_default_from_env() -> None:
+    scheduler = load_scheduler()
+    previous = os.environ.get("TC_SCHEDULER_EVENT_LOG_JSONL")
+    os.environ["TC_SCHEDULER_EVENT_LOG_JSONL"] = "/var/lib/tensorcore/events.jsonl"
+    try:
+        submit_args = scheduler.parse_args(
+            [
+                "submit",
+                "--job-json",
+                "job.json",
+                "--jobs-json",
+                "queue.json",
+                "--inventory-json",
+                "inventory.json",
+            ]
+        )
+        cancel_args = scheduler.parse_args(
+            [
+                "cancel",
+                "--jobs-json",
+                "queue.json",
+                "--job-id",
+                "job-1",
+            ]
+        )
+    finally:
+        if previous is None:
+            os.environ.pop("TC_SCHEDULER_EVENT_LOG_JSONL", None)
+        else:
+            os.environ["TC_SCHEDULER_EVENT_LOG_JSONL"] = previous
+    assert submit_args.event_log_jsonl == "/var/lib/tensorcore/events.jsonl"
+    assert cancel_args.event_log_jsonl == "/var/lib/tensorcore/events.jsonl"
+
+
 def test_cancel_accepts_expanded_pool_job_id() -> None:
     scheduler = load_scheduler()
     with tempfile.TemporaryDirectory() as tmp:
@@ -1941,6 +1976,29 @@ def test_status_reports_gpu_reconciliation_audit_gate() -> None:
     assert gate["cuda_job_count"] == 1
 
 
+def test_status_rejects_nonpositive_gpu_reconciliation_max_age() -> None:
+    scheduler = load_scheduler()
+    stderr = io.StringIO()
+    with redirect_stderr(stderr):
+        try:
+            scheduler.parse_args(
+                [
+                    "status",
+                    "--inventory-json",
+                    "inventory.json",
+                    "--gpu-reconciliation-audit-json",
+                    "audit.json",
+                    "--gpu-reconciliation-max-age-sec",
+                    "0",
+                ]
+            )
+        except SystemExit as exc:
+            assert exc.code == 2
+        else:
+            raise AssertionError("nonpositive status GPU reconciliation max age was accepted")
+    assert "--gpu-reconciliation-max-age-sec must be > 0" in stderr.getvalue()
+
+
 def test_scheduler_dry_run_includes_launch_plan() -> None:
     candidate = job("qllm-phase1", priority=50)
     candidate["cwd"] = "scripts"
@@ -2287,10 +2345,12 @@ def main() -> int:
     test_submit_dry_run_expands_tensorcore_job_v1()
     test_submit_writes_queue_and_cancel_pauses_job()
     test_submit_and_cancel_require_event_log_for_queue_mutation()
+    test_submit_and_cancel_parse_event_log_default_from_env()
     test_cancel_accepts_expanded_pool_job_id()
     test_submit_rejects_malformed_tensorcore_job_v1_fields()
     test_status_offline_reports_inventory_jobs_and_drain_updates_copy()
     test_status_reports_gpu_reconciliation_audit_gate()
+    test_status_rejects_nonpositive_gpu_reconciliation_max_age()
     test_scheduler_dry_run_includes_launch_plan()
     test_scheduler_blocks_cuda_when_gpu_reconciliation_audit_missing()
     test_scheduler_keeps_non_cuda_placement_when_gpu_reconciliation_audit_missing()