From 60bc78ccc6d7b4e888f8a93a531247ba22c373e0 Mon Sep 17 00:00:00 2001 From: drknowhow Date: Tue, 13 Oct 2026 09:47:56 -0400 Subject: [PATCH] feat(schedule): executors build a job per fire; supervisor takes argv payloads services/schedule_runner.py turns a schedule row into a FirePlan (claude -p with the prompt on stdin, codex exec -, gemini, grok --prompt-file, or a plain command), starts it as a detached shell job, and folds the job's exit, cost and tokens back into the run. The supervisor gains three payload keys: argv (no shell, harden_win_argv), stdin_text and env_drop, so a headless claude run is not seen as a nested session. The existing cmd path is unchanged. Not wired to any loop yet. Verified: 39 tests incl. a real supervisor run with stdin. Claude-Session: https://claude.ai/code/session_01JUcvmQq81aLo9ewD9Xc4qY --- docs/shell-jobs.md | 7 +- services/schedule_runner.py | 264 ++++++++++++++++++++++++++++++++++ services/shell_jobs.py | 33 ++++- tests/test_schedule_runner.py | 209 +++++++++++++++++++++++++++ tests/test_shell_jobs_argv.py | 73 ++++++++++ 5 files changed, 578 insertions(+), 8 deletions(-) create mode 100644 services/schedule_runner.py create mode 100644 tests/test_schedule_runner.py create mode 100644 tests/test_shell_jobs_argv.py diff --git a/docs/shell-jobs.md b/docs/shell-jobs.md index 72a7129..7668921 100644 --- a/docs/shell-jobs.md +++ b/docs/shell-jobs.md @@ -149,7 +149,12 @@ stderr go to `jobs/.supervisor.log`, stdin is the payload pipe. 2. Registers the values with `services.credential_store` so the streaming redactor scrubs them before a byte reaches the spool. 3. Spawns the command exactly as `c3_shell` does (Git Bash / cmd / sh - selection, `_popen_kwargs` env), streams both pipes through + selection, `_popen_kwargs` env). A payload may instead carry `argv` (a + list, launched with no shell and `harden_win_argv`), `stdin_text` (written + to the child's stdin, then closed) and `env_drop` (variable names removed + from the child's environment); the scheduler uses these to run `claude -p` + with the prompt on stdin and without the parent's `CLAUDECODE` marker. + Either way it streams both pipes through `ShellCapture(live=True)` — the spool is flushed per piece so `tail` sees it grow — and writes `running` with the pids and creation times. 4. Enforces the job's own timeout with the same tree kill as `c3_shell`. diff --git a/services/schedule_runner.py b/services/schedule_runner.py new file mode 100644 index 0000000..b7f6812 --- /dev/null +++ b/services/schedule_runner.py @@ -0,0 +1,264 @@ +"""Turn a schedule row into a detached job and read the job's result back. + +Each fire is one ``services.shell_jobs`` job: the supervisor outlives the +Oracle, tree-kills on timeout, keeps the output and notifies the project's +feed. The runner only decides *what* to run (``plan``), starts it +(``dispatch``) and, once the job is terminal, folds its outcome and cost into +the run row (``finalize``). The tick loop in ``oracle/services/scheduler.py`` +owns the order in which that happens. +""" +from __future__ import annotations + +import json +import shutil +from dataclasses import dataclass, field + +from services.schedule_store import EXECUTED_STATUSES, ScheduleStore, prompt_text + +SESSION_ID = "scheduler" +CLAUDE_NESTING_VARS = ("CLAUDECODE", "CLAUDE_CODE_ENTRYPOINT", "CLAUDE_CODE_SESSION_ID", + "CLAUDE_CODE_CHILD_SESSION", "MAX_THINKING_TOKENS") +CODEX_EFFORTS = {"": "medium", "low": "low", "medium": "medium", "high": "high", + "xhigh": "xhigh", "max": "xhigh"} +GROK_EFFORTS = {"": "", "low": "low", "medium": "medium", "high": "high", "xhigh": "high", "max": "high"} +JOB_TO_RUN = {"done": "done", "failed": "failed", "timeout": "timeout", + "cancelled": "cancelled", "lost": "lost"} + + +class PlanError(ValueError): + """The schedule cannot be turned into a command on this machine.""" + + +@dataclass +class FirePlan: + display: str + cwd: str + timeout_s: int + argv: list | None = None + cmd: str = "" + stdin_text: str | None = None + env: dict = field(default_factory=dict) + env_drop: tuple = () + parser: str = "command" + + def payload(self) -> dict: + return {"cmd": self.display, "exec_cmd": self.cmd or self.display, "argv": self.argv, + "stdin_text": self.stdin_text, "env": dict(self.env) or None, + "env_drop": list(self.env_drop)} + + +def plan(schedule: dict, *, check: bool = False, prompt_file: str = "") -> FirePlan: + """The command for one fire of ``schedule``; ``check`` plans its ``check_command``. + + Raises ``PlanError`` when the provider CLI is missing or the row is + malformed. ``prompt_file`` is where a provider that only takes a file + (grok) should find the prompt; the caller writes it. + """ + cwd = str(schedule.get("cwd") or schedule.get("project_path") or ".") + timeout_s = int(schedule.get("timeout_s") or 1800) + if check: + cmd = str(schedule.get("check_command") or "").strip() + if not cmd: + raise PlanError("schedule has no check_command") + return FirePlan(display=cmd, cmd=cmd, cwd=cwd, timeout_s=min(timeout_s, 600)) + if schedule.get("kind") == "command": + cmd = str((schedule.get("target") or {}).get("cmd") or "").strip() + if not cmd: + raise PlanError("command schedule has no target.cmd") + return FirePlan(display=cmd, cmd=cmd, cwd=cwd, timeout_s=timeout_s) + prompt = prompt_text(schedule) + if not prompt: + raise PlanError("schedule has no prompt") + provider = str(schedule.get("provider") or "claude") + builder = _PROVIDERS.get(provider) + if builder is None: + raise PlanError(f"unknown provider {provider!r}") + return builder(schedule, prompt, cwd, timeout_s, prompt_file) + + +def _exe(name: str) -> str: + from cli.tools.delegate import _which + path = _which(name) + if not path: + raise PlanError(f"{name} CLI not found on PATH") + return path + + +def _plan_claude(schedule: dict, prompt: str, cwd: str, timeout_s: int, _prompt_file: str) -> FirePlan: + from cli.tools.delegate import _EFFORT_LEVELS, _MODEL_NAME_RE + argv = [_exe("claude"), "-p", "--output-format", "json", "--no-session-persistence", + "--permission-mode", str(schedule.get("permission_mode") or "acceptEdits"), + "--permission-prompts", "none", "--name", f"sched:{schedule.get('id', '')}"] + model = str(schedule.get("model") or "").strip() + if model: + if not _MODEL_NAME_RE.match(model): + raise PlanError(f"model name {model!r} is not a Claude model alias or id") + argv += ["--model", model] + effort = str(schedule.get("effort") or "").strip().lower() + if effort in _EFFORT_LEVELS: + argv += ["--effort", effort] + budget = float(schedule.get("max_budget_usd") or 0) + if budget > 0: + argv += ["--max-budget-usd", f"{budget:g}"] + tools = str(schedule.get("allowed_tools") or "").strip() + if tools: + argv += ["--allowedTools", tools] + return FirePlan(display=f"claude -p {prompt}", cwd=cwd, timeout_s=timeout_s, argv=argv, + stdin_text=prompt, env={"C3_HOST": "claude-code"}, + env_drop=CLAUDE_NESTING_VARS, parser="claude") + + +def _plan_codex(schedule: dict, prompt: str, cwd: str, timeout_s: int, _prompt_file: str) -> FirePlan: + argv = [_exe("codex"), "exec"] + model = str(schedule.get("model") or "").strip() + if model: + argv += ["-m", model] + effort = CODEX_EFFORTS.get(str(schedule.get("effort") or ""), "medium") + sandbox = "danger-full-access" if schedule.get("permission_mode") == "bypassPermissions" else "workspace-write" + argv += ["--config", f"model_reasoning_effort={effort}", "--sandbox", sandbox, + "--config", 'approval_policy="never"', "--json", "--skip-git-repo-check", "-"] + return FirePlan(display=f"codex exec {prompt}", cwd=cwd, timeout_s=timeout_s, argv=argv, + stdin_text=prompt, env={"C3_HOST": "codex"}, env_drop=("CODEX_CLI",), parser="codex") + + +def _plan_gemini(schedule: dict, prompt: str, cwd: str, timeout_s: int, _prompt_file: str) -> FirePlan: + argv = [_exe("gemini"), "--output-format", "json", "--approval-mode", "yolo"] + model = str(schedule.get("model") or "").strip() + if model: + argv += ["-m", model] + return FirePlan(display=f"gemini {prompt}", cwd=cwd, timeout_s=timeout_s, argv=argv, + stdin_text=prompt, env={"C3_HOST": "antigravity"}, env_drop=("GEMINI_CLI",), + parser="gemini") + + +def _plan_grok(schedule: dict, prompt: str, cwd: str, timeout_s: int, prompt_file: str) -> FirePlan: + if not prompt_file: + raise PlanError("grok needs a prompt_file") + from cli.tools.delegate import _grok_cmd + argv = _grok_cmd(prompt_file, str(schedule.get("model") or ""), 30, cwd, allow_write=True, + reasoning_effort=GROK_EFFORTS.get(str(schedule.get("effort") or ""), "")) + argv[0] = _exe("grok") + return FirePlan(display=f"grok {prompt}", cwd=cwd, timeout_s=timeout_s, argv=argv, + env={"C3_HOST": "grok"}, parser="grok") + + +_PROVIDERS = {"claude": _plan_claude, "codex": _plan_codex, "gemini": _plan_gemini, "grok": _plan_grok} + + +def dispatch(store: ScheduleStore, jobstore, schedule: dict, run: dict, *, check: bool = False) -> dict: + """Start the job for ``run`` and record it; a spawn failure finishes the run.""" + run_id = run["run_id"] + try: + prompt_file = "" + if schedule.get("provider") == "grok" and schedule.get("kind") != "command" and not check: + path = store.runs_dir / f"{run_id}.prompt.txt" + path.write_text(prompt_text(schedule), encoding="utf-8") + prompt_file = str(path) + fire = plan(schedule, check=check, prompt_file=prompt_file) + except PlanError as exc: + return store.finish_run(run_id, "failed", error=str(exc)) + job = jobstore.start(project_path=schedule["project_path"], session_id=SESSION_ID, + cmd=fire.display, cwd=fire.cwd, timeout_s=fire.timeout_s, + payload=fire.payload()) + if job.terminal and job.status != "done": + return store.finish_run(run_id, JOB_TO_RUN.get(job.status, "failed"), job_id=job.id, + exit_code=job.exit_code, error=job.error or "job ended before it ran") + status = "check_running" if check else "dispatched" + return store.update_run(run_id, status=status, job_id=job.id, dispatched_at=job.created_at) + + +def job_stdout(jobstore, job) -> str: + """Raw kept stdout of a terminal job, or empty when it was not kept.""" + if not job.output_id: + return "" + try: + meta = jobstore.output_store.resolve(job.output_id, project_path=job.project_path, + session_id=job.session_id, guard_check=lambda _p: None) + return meta.stream_path("stdout").read_bytes().decode("utf-8", "replace") + except Exception: + return "" + + +def parse_result(parser: str, stdout: str) -> dict: + """Cost, tokens, turns, model and a result excerpt from a provider's JSON stdout.""" + if parser == "claude": + from cli.tools.delegate import parse_claude_json + text, _ok, stats = parse_claude_json(stdout) + out = {"result_excerpt": text} + tokens = {k: stats[k] for k in ("input_tokens", "output_tokens", "cache_read_tokens", + "cache_write_tokens") if k in stats} + if tokens: + out["tokens"] = tokens + for key in ("cost_usd", "turns", "model"): + if key in stats: + out[key] = stats[key] + return out + data = _trailing_json(stdout) + if parser == "codex": + return {"result_excerpt": _codex_text(stdout) or stdout.strip()} + if parser == "gemini" and isinstance(data, dict): + text = data.get("response") or data.get("text") or data.get("result") or "" + out = {"result_excerpt": str(text)} + models = ((data.get("stats") or {}).get("models") or {}) if isinstance(data.get("stats"), dict) else {} + tokens = {"input_tokens": 0, "output_tokens": 0} + for mdata in models.values(): + tok = (mdata or {}).get("tokens") or {} + tokens["input_tokens"] += int(tok.get("input") or 0) + tokens["output_tokens"] += int(tok.get("candidates") or 0) + if models: + out["tokens"] = tokens + out["model"] = sorted(models)[0] + return out + if parser == "grok" and isinstance(data, dict): + return {"result_excerpt": str(data.get("result") or data.get("response") or data.get("text") or "")} + return {"result_excerpt": stdout.strip()} + + +def _trailing_json(text: str): + raw = (text or "").strip() + start = raw.find("{") + if start < 0: + return None + try: + return json.loads(raw[start:]) + except ValueError: + return None + + +def _codex_text(stdout: str) -> str: + last = "" + for line in (stdout or "").splitlines(): + line = line.strip() + if not line.startswith("{"): + continue + try: + event = json.loads(line) + except ValueError: + continue + item = event.get("item") if isinstance(event, dict) else None + if isinstance(item, dict) and item.get("type") == "agent_message": + last = str(item.get("text") or last) + elif isinstance(event, dict) and event.get("type") == "message" and event.get("text"): + last = str(event["text"]) + return last + + +def finalize(store: ScheduleStore, jobstore, run: dict, job, parser: str = "command") -> dict: + """Fold a terminal job into its run: status, exit code, duration, cost.""" + status = JOB_TO_RUN.get(job.status) + if status is None: + raise ValueError(f"job {job.id} is not terminal ({job.status})") + extra: dict = {"job_id": job.id, "exit_code": job.exit_code, "duration_ms": job.duration_ms, + "output_id": job.output_id or None} + if job.error: + extra["error"] = str(job.error) + if status in EXECUTED_STATUSES and parser != "command": + stdout = job_stdout(jobstore, job) + if stdout: + extra.update(parse_result(parser, stdout)) + return store.finish_run(run["run_id"], status, **extra) + + +def cli_available(provider: str) -> bool: + return bool(shutil.which({"claude": "claude", "codex": "codex", "gemini": "gemini", + "grok": "grok"}.get(provider, provider))) diff --git a/services/shell_jobs.py b/services/shell_jobs.py index fc46734..54b8e2d 100644 --- a/services/shell_jobs.py +++ b/services/shell_jobs.py @@ -1145,20 +1145,33 @@ def supervise(job_id: str, root=None) -> int: except Exception: git_before = None - bash = shell_mod._select_bash() - if bash: - popen_target: object = [bash, "-c", exec_cmd] + argv = payload.get("argv") + stdin_text = payload.get("stdin_text") + popen_kwargs = shell_mod._popen_kwargs(extra_env) + for name in payload.get("env_drop") or []: + popen_kwargs["env"].pop(str(name), None) + if isinstance(stdin_text, str): + popen_kwargs["stdin"] = subprocess.PIPE + if isinstance(argv, list) and argv: + from services.win_subprocess import harden_win_argv + popen_target: object = harden_win_argv([str(a) for a in argv]) use_shell = False + shell_name = "argv" else: - popen_target = exec_cmd - use_shell = True - shell_name = "git-bash" if bash else ("cmd" if sys.platform == "win32" else "sh") + bash = shell_mod._select_bash() + if bash: + popen_target = [bash, "-c", exec_cmd] + use_shell = False + else: + popen_target = exec_cmd + use_shell = True + shell_name = "git-bash" if bash else ("cmd" if sys.platform == "win32" else "sh") t0 = time.monotonic() try: proc = subprocess.Popen( popen_target, shell=use_shell, cwd=job.cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - **shell_mod._popen_kwargs(extra_env), + **popen_kwargs, ) except (OSError, ValueError) as exc: _finish(store, job, "failed", exit_code=127 if isinstance(exc, FileNotFoundError) else 126, @@ -1167,6 +1180,12 @@ def supervise(job_id: str, root=None) -> int: return 0 capture = ShellCapture(proc, store.output_store.spool_dir(), redact=redact, kill_tree=shell_mod._kill_tree, live=True) + if isinstance(stdin_text, str) and proc.stdin is not None: + try: + proc.stdin.write(stdin_text.encode("utf-8")) + proc.stdin.close() + except (BrokenPipeError, OSError): + pass job.status = "running" job.started_at = _iso(_now()) job.child_pid = proc.pid diff --git a/tests/test_schedule_runner.py b/tests/test_schedule_runner.py new file mode 100644 index 0000000..40a61ac --- /dev/null +++ b/tests/test_schedule_runner.py @@ -0,0 +1,209 @@ +import json +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from services import schedule_runner as sr +from services import schedule_store as ss + +REPO = Path(__file__).resolve().parents[1] + + +def sched(**over): + row = { + "id": "sch-abc", "name": "report", "kind": "skill", "target": {"skill": "/daily_report", "args": "weekly"}, + "provider": "claude", "model": "", "effort": "", "permission_mode": "acceptEdits", + "allowed_tools": "", "max_budget_usd": 0, "project_path": str(REPO), "cwd": str(REPO), + "timeout_s": 900, "check_command": "", + } + row.update(over) + return row + + +class FakeJob(SimpleNamespace): + @property + def terminal(self): + return self.status in ("done", "failed", "timeout", "cancelled", "lost") + + +class FakeJobStore: + def __init__(self, status="running", stdout=""): + self.status = status + self.stdout = stdout + self.calls = [] + self.output_store = self + + def start(self, **kw): + self.calls.append(kw) + return FakeJob(id="j-1", status=self.status, exit_code=None, error="", created_at="2026-10-13T09:00:00+00:00", + output_id="o-1", duration_ms=5, project_path=kw["project_path"], session_id=kw["session_id"]) + + def resolve(self, output_id, **kw): + path = Path(tempfile.mkdtemp()) / "stdout" + path.write_text(self.stdout, encoding="utf-8") + return SimpleNamespace(stream_path=lambda stream: path) + + +class TestPlan(unittest.TestCase): + def setUp(self): + self.which = patch("cli.tools.delegate._which", side_effect=lambda n: f"/bin/{n}") + self.which.start() + + def tearDown(self): + self.which.stop() + + def test_claude_skill_run(self): + fire = sr.plan(sched(model="sonnet", effort="medium", max_budget_usd=0.5, allowed_tools="Read Edit")) + self.assertEqual(fire.argv[:5], ["/bin/claude", "-p", "--output-format", "json", "--no-session-persistence"]) + self.assertNotIn("--safe-mode", fire.argv) + self.assertIn("--permission-mode", fire.argv) + self.assertEqual(fire.argv[fire.argv.index("--permission-mode") + 1], "acceptEdits") + self.assertEqual(fire.argv[fire.argv.index("--permission-prompts") + 1], "none") + self.assertEqual(fire.argv[fire.argv.index("--model") + 1], "sonnet") + self.assertEqual(fire.argv[fire.argv.index("--effort") + 1], "medium") + self.assertEqual(fire.argv[fire.argv.index("--max-budget-usd") + 1], "0.5") + self.assertEqual(fire.argv[fire.argv.index("--allowedTools") + 1], "Read Edit") + self.assertEqual(fire.stdin_text, "/daily_report weekly") + self.assertIn("CLAUDECODE", fire.env_drop) + self.assertEqual(fire.parser, "claude") + self.assertEqual(fire.timeout_s, 900) + payload = fire.payload() + self.assertEqual(payload["argv"], fire.argv) + self.assertEqual(payload["stdin_text"], fire.stdin_text) + self.assertIn("CLAUDECODE", payload["env_drop"]) + + def test_claude_omits_unset_flags_and_rejects_bad_model(self): + fire = sr.plan(sched(effort="ultra")) + for flag in ("--model", "--effort", "--max-budget-usd", "--allowedTools"): + self.assertNotIn(flag, fire.argv) + with self.assertRaises(sr.PlanError): + sr.plan(sched(model="sonnet; rm -rf")) + + def test_prompt_kind_uses_prompt_text(self): + fire = sr.plan(sched(kind="prompt", target={"prompt": "summarise the week"})) + self.assertEqual(fire.stdin_text, "summarise the week") + + def test_command_and_check(self): + fire = sr.plan(sched(kind="command", target={"cmd": "git fetch --all"})) + self.assertEqual((fire.cmd, fire.argv, fire.parser), ("git fetch --all", None, "command")) + check = sr.plan(sched(check_command="test -s inbox.txt", timeout_s=7200), check=True) + self.assertEqual((check.cmd, check.timeout_s), ("test -s inbox.txt", 600)) + with self.assertRaises(sr.PlanError): + sr.plan(sched(), check=True) + + def test_codex_gemini_grok_shapes(self): + codex = sr.plan(sched(provider="codex", model="o4-mini", effort="max")) + self.assertEqual(codex.argv[:4], ["/bin/codex", "exec", "-m", "o4-mini"]) + self.assertIn("model_reasoning_effort=xhigh", codex.argv) + self.assertEqual(codex.argv[-1], "-") + self.assertEqual(codex.stdin_text, "/daily_report weekly") + gemini = sr.plan(sched(provider="gemini")) + self.assertEqual(gemini.argv[0], "/bin/gemini") + self.assertNotIn("-p", gemini.argv) + self.assertEqual(gemini.stdin_text, "/daily_report weekly") + with self.assertRaises(sr.PlanError): + sr.plan(sched(provider="grok")) + grok = sr.plan(sched(provider="grok", effort="low"), prompt_file="p.txt") + self.assertEqual(grok.argv[:3], ["/bin/grok", "--prompt-file", "p.txt"]) + self.assertIn("--yolo", grok.argv) + self.assertIsNone(grok.stdin_text) + + def test_missing_cli(self): + self.which.stop() + with patch("cli.tools.delegate._which", return_value=None): + with self.assertRaises(sr.PlanError): + sr.plan(sched()) + self.which.start() + + +CLAUDE_JSON = json.dumps({ + "type": "result", "subtype": "success", "is_error": False, "result": "Report sent.", + "total_cost_usd": 0.0421, "num_turns": 3, + "usage": {"input_tokens": 1200, "output_tokens": 300, "cache_read_input_tokens": 50}, + "modelUsage": {"claude-sonnet-5-5": {"inputTokens": 1200}}, +}) + + +class TestParseResult(unittest.TestCase): + def test_claude(self): + out = sr.parse_result("claude", "noise\n" + CLAUDE_JSON) + self.assertEqual(out["result_excerpt"], "Report sent.") + self.assertEqual(out["cost_usd"], 0.0421) + self.assertEqual(out["turns"], 3) + self.assertEqual(out["tokens"], {"input_tokens": 1200, "output_tokens": 300, "cache_read_tokens": 50}) + self.assertEqual(out["model"], "claude-sonnet-5-5") + + def test_claude_error_payload_keeps_reason(self): + out = sr.parse_result("claude", json.dumps({"is_error": True, "result": "budget exceeded"})) + self.assertIn("budget exceeded", out["result_excerpt"]) + + def test_codex_last_agent_message(self): + lines = [json.dumps({"type": "item.completed", "item": {"type": "agent_message", "text": "first"}}), + json.dumps({"type": "item.completed", "item": {"type": "agent_message", "text": "final"}})] + self.assertEqual(sr.parse_result("codex", "\n".join(lines))["result_excerpt"], "final") + + def test_gemini_tokens(self): + blob = json.dumps({"response": "ok", "stats": {"models": {"gemini-x": {"tokens": {"input": 10, "candidates": 4}}}}}) + out = sr.parse_result("gemini", "MCP ready\n" + blob) + self.assertEqual((out["result_excerpt"], out["tokens"], out["model"]), + ("ok", {"input_tokens": 10, "output_tokens": 4}, "gemini-x")) + + def test_command_is_verbatim(self): + self.assertEqual(sr.parse_result("command", " 3 files\n")["result_excerpt"], "3 files") + + +class TestDispatchFinalize(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.store = ss.ScheduleStore(Path(self._tmp.name)) + self.row = self.store.create({"name": "r", "kind": "command", "target": {"cmd": "echo hi"}, + "project_path": str(REPO), "when": {"type": "every", "seconds": 600}, + "autonomy": "autonomous"}) + self.run = self.store.request_run(self.row["id"]) + + def tearDown(self): + self._tmp.cleanup() + + def test_dispatch_records_job(self): + jobs = FakeJobStore() + run = sr.dispatch(self.store, jobs, self.row, self.run) + self.assertEqual((run["status"], run["job_id"]), ("dispatched", "j-1")) + call = jobs.calls[0] + self.assertEqual((call["session_id"], call["cwd"], call["timeout_s"]), (sr.SESSION_ID, str(REPO), 1800)) + self.assertEqual(call["payload"]["cmd"], "echo hi") + + def test_check_dispatch(self): + row = dict(self.row, check_command="exit 1") + run = sr.dispatch(self.store, FakeJobStore(), row, self.run, check=True) + self.assertEqual(run["status"], "check_running") + + def test_spawn_failure_and_plan_error_finish_the_run(self): + failed = sr.dispatch(self.store, FakeJobStore(status="failed"), self.row, self.run) + self.assertEqual(failed["status"], "failed") + second = self.store.request_run(self.row["id"]) + bad = sr.dispatch(self.store, FakeJobStore(), dict(self.row, target={"cmd": ""}), second) + self.assertEqual(bad["status"], "failed") + self.assertIn("target.cmd", bad["error"]) + + def test_finalize_folds_cost(self): + jobs = FakeJobStore(stdout=CLAUDE_JSON) + job = FakeJob(id="j-9", status="done", exit_code=0, duration_ms=1234, output_id="o-1", error="", + project_path=str(REPO), session_id=sr.SESSION_ID) + run = sr.finalize(self.store, jobs, self.run, job, parser="claude") + self.assertEqual((run["status"], run["exit_code"], run["duration_ms"], run["cost_usd"]), + ("done", 0, 1234, 0.0421)) + self.assertEqual(self.store.get(self.row["id"])["last_status"], "done") + + def test_finalize_maps_timeout_and_refuses_open_job(self): + job = FakeJob(id="j-2", status="timeout", exit_code=None, duration_ms=10, output_id="", error="killed", + project_path=str(REPO), session_id=sr.SESSION_ID) + run = sr.finalize(self.store, FakeJobStore(), self.run, job) + self.assertEqual((run["status"], run["error"]), ("timeout", "killed")) + with self.assertRaises(ValueError): + sr.finalize(self.store, FakeJobStore(), self.run, FakeJob(id="j", status="running")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_shell_jobs_argv.py b/tests/test_shell_jobs_argv.py new file mode 100644 index 0000000..2c84247 --- /dev/null +++ b/tests/test_shell_jobs_argv.py @@ -0,0 +1,73 @@ +import os +import sys +import tempfile +import time +import unittest +from pathlib import Path +from unittest.mock import patch + +from services import shell_jobs as sj + +PY = sys.executable +CODE = ("import sys, os; print(sys.stdin.read().upper()); " + "print('dropped' if 'C3_DROP_ME' not in os.environ else 'kept')") + + +def _allow(_path): + return None + + +class TestSupervisorArgv(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + root = Path(self._tmp.name) + self.project = root / "proj" + (self.project / ".c3").mkdir(parents=True) + self._env = patch.dict(os.environ, {"C3_SHELL_OUT_DIR": str(root / "store"), "C3_DROP_ME": "1"}) + self._env.start() + self.store = sj.JobStore() + + def tearDown(self): + self._env.stop() + for _ in range(20): + try: + self._tmp.cleanup() + break + except (PermissionError, OSError): + time.sleep(0.25) + + def _wait(self, job, timeout=40.0): + deadline = time.monotonic() + timeout + while True: + fresh = self.store.load(job) or job + if fresh.terminal or time.monotonic() > deadline: + return fresh + time.sleep(0.2) + + def _stdout(self, job) -> str: + meta = self.store.output_store.resolve(job.output_id, project_path=job.project_path, + session_id=job.session_id, guard_check=_allow) + return meta.stream_path("stdout").read_bytes().decode("utf-8") + + def test_argv_stdin_and_env_drop(self): + payload = {"cmd": "python -c ...", "argv": [PY, "-c", CODE], "stdin_text": "hello\nworld", + "env_drop": ["C3_DROP_ME"]} + job = self.store.start(project_path=str(self.project), session_id="scheduler", cmd="python -c ...", + cwd=str(self.project), timeout_s=60, payload=payload) + job = self._wait(job) + self.assertEqual((job.status, job.exit_code, job.shell), ("done", 0, "argv")) + out = self._stdout(job) + self.assertIn("HELLO\nWORLD", out.replace("\r\n", "\n")) + self.assertIn("dropped", out) + + def test_cmd_payload_unchanged(self): + job = self.store.start(project_path=str(self.project), session_id="scheduler", cmd="echo plain", + cwd=str(self.project), timeout_s=60, payload={"cmd": "echo plain"}) + job = self._wait(job) + self.assertEqual((job.status, job.exit_code), ("done", 0)) + self.assertNotEqual(job.shell, "argv") + self.assertIn("plain", self._stdout(job)) + + +if __name__ == "__main__": + unittest.main()