diff --git a/docs/a38.md b/docs/a38.md index 0cd86f8..7d19a18 100644 --- a/docs/a38.md +++ b/docs/a38.md @@ -22,7 +22,7 @@ The [example manifest](../examples/a38.json) is illustrative: replace its job an Separately from that optional policy object, when every changed path ends with `.md` (case-sensitive) and the same fail-closed inventory rules hold, A38 omits **all** required jobs as `not_applicable` with log text `omitted: markdown-only change set` and report flag `markdown_only: true`. No author report is required for Ready when the guard independently confirms that markdown-only inventory. A pure `README.md` change set is both README-only and markdown-only; the full markdown skip applies and `readme_only: true` is set only when configured `omit_jobs` also matched. -Each required job contains `id`, `name`, `timeout_s`, `workflow`, and `job`, plus exactly one input: either `command` or `executor`. The legacy `command` form remains an exact local shell command and is preserved unchanged. An `executor` contains exactly `adapter` and `config`; `adapter` is one of `commands`, `compose`, `http-smoke`, or `immutable`, and its object-valued `config` is validated by that adapter's pure, strict parser. For example: +Each required job contains `id`, `name`, `timeout_s`, `workflow`, and `job`, plus exactly one input: either `command` or `executor`. An optional `lock` may appear on the `command` form (`null` or a nonempty name of ASCII letters, digits, `.`, `_`, or `-`). Executor jobs set `lock` in `executor.config`, not as a sibling of `executor`. The legacy `command` form remains an exact local shell command and is preserved unchanged. An `executor` contains exactly `adapter` and `config`; `adapter` is one of `commands`, `compose`, `http-smoke`, or `immutable`, and its object-valued `config` is validated by that adapter's pure, strict parser. For example: ```json { @@ -33,7 +33,7 @@ Each required job contains `id`, `name`, `timeout_s`, `workflow`, and `job`, plu } ``` -Central policy loading normalizes this structured form to a command-only internal job: `agent a38 job ADAPTER --config ` followed by shell-quoted, compact, sorted-key, ASCII JSON. Non-finite values are rejected. Execution, the frozen command-only report schema, report verification, and Guard comparison all use that same deterministic command string. The [A38 job adapters](a38-job-adapters.md) document each adapter's configuration contract. +Central policy loading normalizes this structured form to an internal job with that same command string plus the resolved `lock` (`null` when none): `agent a38 job ADAPTER --config ` followed by shell-quoted, compact, sorted-key, ASCII JSON. Non-finite values are rejected. Execution, the frozen command-only report schema, report verification, and Guard comparison all use that same deterministic command string. The [A38 job adapters](a38-job-adapters.md) document each adapter's configuration contract. IDs are unique, and timeouts are finite positive seconds up to 86400. Workflow paths identify `.github/workflows/*.yml` or `*.yaml`; `job` is the workflow's job identifier. An exclusion contains only `workflow`, `job`, and a meaningful `reason`. A workflow/job pair appears exactly once across required jobs and exclusions. @@ -90,7 +90,7 @@ agent a38 run --repo . --policy /tmp/a38-run/policy.json \ --output /tmp/a38-run/report.md --logs-dir /tmp/a38-run/logs --private ``` -The runner identifies the GitHub repository from `origin` by default; for a fork, pass `--repository OWNER/NAME` with the PR target repository. Without `--private` or `--public`, visibility lookup requires `--github-session SESSION` bound explicitly in `$AGENT_HOME/github-accounts.json`. The executor verifies the selected account and refuses an absent binding or mismatched login; it never uses ambient host authentication. Explicit `--private` or `--public` supplies the report visibility without a GitHub lookup. A38 checks that visibility against the actual repository. The runner walks the complete required job list sequentially, records each command's result, exit code, elapsed time and timeout, and continues independent jobs after failures. When the change set is markdown-only, every required job is recorded as `not_applicable` without executing the command. Otherwise, configured `readme_only.omit_jobs` are recorded as `not_applicable` without executing the command when the change set is README-only; other jobs still run. It exposes `A38_HEAD_SHA` and `A38_BASE_SHA` to commands, removes GitHub token variables from job environments, and retains per-job logs locally. It terminates timed-out process groups. An interrupted run or checkout drift must never produce a successful report. +The runner identifies the GitHub repository from `origin` by default; for a fork, pass `--repository OWNER/NAME` with the PR target repository. Without `--private` or `--public`, visibility lookup requires `--github-session SESSION` bound explicitly in `$AGENT_HOME/github-accounts.json`. The executor verifies the selected account and refuses an absent binding or mismatched login; it never uses ambient host authentication. Explicit `--private` or `--public` supplies the report visibility without a GitHub lookup. A38 checks that visibility against the actual repository. The runner records each command's result, exit code, elapsed time and timeout. Jobs whose lock domains do not conflict may run at the same time: non-docker jobs share a CPU domain so lint, type-check and Jest do not overlap each other; `docker-heavy` may overlap that CPU domain. A `lock` on the command form coordinates only inside that `agent a38 run`. Cross-process exclusivity is the adapter lock (`executor.config.lock`, including the `docker-heavy` default for compose and http-smoke). `A38_MAX_IN_FLIGHT` caps concurrency (default 2, integer ≥ 1). After a job failure, independent jobs still run; a dirty tree or HEAD change still aborts the rest. When the change set is markdown-only, every required job is recorded as `not_applicable` without executing the command. Otherwise, configured `readme_only.omit_jobs` are recorded as `not_applicable` without executing the command when the change set is README-only; other jobs still run. It exposes `A38_HEAD_SHA` and `A38_BASE_SHA` to commands, removes GitHub token variables from job environments, and retains per-job logs locally. It terminates timed-out process groups. An interrupted run or checkout drift must never produce a successful report. Job adapter behavior is defined separately in [A38 job adapters](a38-job-adapters.md). Use `agent a38 job --config ''`; this standard intentionally does not duplicate adapter schemas. @@ -106,7 +106,7 @@ agent a38 verify --policy /tmp/a38-run/policy.json \ Only a complete `run` plus successful local `verify` for the signed, clean final SHA may be recorded as `local_check_pass`. That final SHA must be on the open draft with no intervening commit after the verified measurement. Early draft publication may precede this final measurement ([pull request lifecycle](pull-request-lifecycle.md)). Any fix, amend, rebase, or other new SHA requires a new signed clean commit and a complete run and verification from the beginning. -The generated `report.md` is ready to publish: one short sentence under `EN:`, one under `DE:`, and a closed `
` section containing a mandatory table followed by the unchanged machine-readable block in its own nested closed details section. Labels appear on their own lines; blank lines separate the languages and follow `` so GitHub renders the enclosed Markdown. The summary describes the evidence without declaring failed or interrupted runs successful. The table lists **every recorded A38 job**, in execution order, with its ID/name, duration, final result and exit code, including failed, errored or timed-out jobs. Display durations in seconds, always rounded **up to whole seconds** (84.467 → 85 s; 84 → 84 s; 0 → 0 s). Preserve the exact fractional measurements in the original machine block. Rows represent A38 jobs, not every underlying test case in a suite. Job names are escaped so they cannot alter the table or HTML structure. Commands and exact evidence remain in the nested original report. Adopters and plugins use this central output rather than maintaining another format template. +The generated `report.md` is ready to publish: one short sentence under `EN:`, one under `DE:`, and a closed `
` section containing a mandatory table followed by the unchanged machine-readable block in its own nested closed details section. Labels appear on their own lines; blank lines separate the languages and follow `` so GitHub renders the enclosed Markdown. The summary describes the evidence without declaring failed or interrupted runs successful. The table lists **every recorded A38 job**, in policy order (manifest `jobs` order), with its ID/name, duration, final result and exit code, including failed, errored or timed-out jobs. Display durations in seconds, always rounded **up to whole seconds** (84.467 → 85 s; 84 → 84 s; 0 → 0 s). Preserve the exact fractional measurements in the original machine block. Rows represent A38 jobs, not every underlying test case in a suite. Job names are escaped so they cannot alter the table or HTML structure. Commands and exact evidence remain in the nested original report. Adopters and plugins use this central output rather than maintaining another format template. Post the generated `report.md` unchanged as a PR comment **using the PR author's GitHub account**, preserving its markers and fenced JSON block. For legacy reports, adding this presentation around the original marked block is allowed only if that block remains byte-for-byte unchanged; preserve all measured data and timestamps, then verify and reassess the edited comment. Publishing is separate from running and verification. Do not paste raw logs containing credentials into the comment and do not hand-edit or reconstruct a passing payload. diff --git a/src/agent_cli/a38.py b/src/agent_cli/a38.py index fac843f..19cc25c 100644 --- a/src/agent_cli/a38.py +++ b/src/agent_cli/a38.py @@ -8,6 +8,7 @@ from __future__ import annotations import argparse +import concurrent.futures import html import json import math @@ -29,7 +30,7 @@ from .a38_job_adapters import ADAPTERS from .readme_only import git_is_markdown_only, git_is_readme_only from .a38_job_adapters.commands import parse_commands_config -from .a38_job_adapters.common import BUILTIN_UNSET, JobError +from .a38_job_adapters.common import BUILTIN_UNSET, DOCKER_HEAVY_LOCK, LOCK_NAME_RE, JobError from .a38_job_adapters.compose import companion_env_missing, parse_compose_config from .a38_job_adapters.http_smoke import parse_http_smoke_config from .a38_job_adapters.immutable import parse_immutable_config @@ -54,7 +55,8 @@ {"schema", "standard", "documentation", "mode", "jobs", "exclusions"} ) JOB_COMMON_KEYS = frozenset({"id", "name", "timeout_s", "workflow", "job"}) -JOB_INPUT_KEYS = JOB_COMMON_KEYS | frozenset({"command", "executor"}) +JOB_INPUT_KEYS = JOB_COMMON_KEYS | frozenset({"command", "executor", "lock"}) +_DOCKER_DEFAULT_LOCK_ADAPTERS = frozenset({"compose", "http-smoke"}) EXECUTOR_KEYS = frozenset({"adapter", "config"}) EXCLUSION_KEYS = frozenset({"workflow", "job", "reason"}) @@ -224,10 +226,21 @@ def _require_job_keys(obj: Mapping[str, Any], label: str) -> str: inputs = keys & {"command", "executor"} if len(inputs) != 1: raise A38Error(f"{label} must contain exactly one of command or executor") - return inputs.pop() + input_key = inputs.pop() + if input_key == "executor" and "lock" in keys: + raise A38Error(f"{label} executor jobs must set lock in executor.config, not as a sibling") + return input_key -def _executor_command(value: Any, label: str) -> str: +def _optional_job_lock(value: Any, label: str) -> str | None: + if value is None: + return None + if not isinstance(value, str) or LOCK_NAME_RE.fullmatch(value) is None: + raise A38Error(f"{label} must be a lock name or null") + return value + + +def _executor_command_and_lock(value: Any, label: str) -> tuple[str, str | None]: if not isinstance(value, dict): raise A38Error(f"{label} must be an object") _require_keys(value, EXECUTOR_KEYS, label) @@ -249,11 +262,15 @@ def _executor_command(value: Any, label: str) -> str: raise A38Error(f"{label}.config cannot be serialized as JSON: {exc}") from exc parser = _ADAPTER_CONFIG_PARSERS[adapter] try: - parser(config_text) + common, _parsed = parser(config_text) except (JobError, OverflowError, RecursionError) as exc: raise A38Error(f"{label}.config is invalid: {exc}") from exc command = f"agent a38 job {adapter} --config {shlex.quote(config_text)}" - return _validate_command(command, f"{label} command") + command = _validate_command(command, f"{label} command") + lock = common.lock + if lock is None and adapter in _DOCKER_DEFAULT_LOCK_ADAPTERS: + lock = DOCKER_HEAVY_LOCK + return command, lock def load_policy(text: str) -> dict: @@ -330,8 +347,14 @@ def load_policy(text: str) -> dict: name = _validate_name(item["name"], f"jobs[{index}].name") if input_key == "command": command = _validate_command(item["command"], f"jobs[{index}].command") + if "lock" in item: + lock = _optional_job_lock(item["lock"], f"jobs[{index}].lock") + else: + lock = None else: - command = _executor_command(item["executor"], f"jobs[{index}].executor") + command, lock = _executor_command_and_lock( + item["executor"], f"jobs[{index}].executor" + ) jobs.append( { "id": ident, @@ -340,6 +363,7 @@ def load_policy(text: str) -> dict: "timeout_s": timeout_out, "workflow": workflow, "job": gh_job, + "lock": lock, } ) @@ -840,6 +864,18 @@ def _write_report(output: Path, payload: Mapping[str, Any]) -> None: _write_bytes_atomic(output, text.encode("utf-8")) +_active_job_procs: list[subprocess.Popen[Any]] = [] +_active_job_procs_lock = threading.Lock() + + +def _terminate_active_job_procs() -> None: + # SIGTERM is delivered to the main thread; workers will not see KeyboardInterrupt. + with _active_job_procs_lock: + procs = list(_active_job_procs) + for proc in procs: + _terminate_process_group(proc) + + def _terminate_process_group(proc: subprocess.Popen[Any]) -> None: """Allow the leader up to 30 seconds for owned-resource cleanup on TERM. @@ -896,6 +932,8 @@ def _run_one_job( if sys.platform != "win32": popen_kwargs["start_new_session"] = True proc = subprocess.Popen(**popen_kwargs) # noqa: S603 + with _active_job_procs_lock: + _active_job_procs.append(proc) timed_out = False try: try: @@ -909,6 +947,11 @@ def _run_one_job( finally: # The session leader may have exited while descendants still run. _terminate_process_group(proc) + with _active_job_procs_lock: + try: + _active_job_procs.remove(proc) + except ValueError: + pass duration = time.monotonic() - start if timed_out or duration > timeout_s: return "timeout", proc.returncode if proc.returncode is not None else -1, duration @@ -929,6 +972,56 @@ def _force_fail_runs(runs: list[dict[str, Any]]) -> None: run["exit_code"] = 1 +def _job_conflict_keys(job: Mapping[str, Any]) -> frozenset[str]: + lock = job["lock"] + if lock == DOCKER_HEAVY_LOCK: + return frozenset({DOCKER_HEAVY_LOCK}) + keys: set[str] = {"#cpu"} + if isinstance(lock, str) and lock: + keys.add(lock) + return frozenset(keys) + + +def _max_in_flight(env: Mapping[str, str]) -> int: + raw = env.get("A38_MAX_IN_FLIGHT", "2") + try: + value = int(raw) + except (TypeError, ValueError): + raise A38Error("A38_MAX_IN_FLIGHT must be an integer >= 1") from None + if value < 1: + raise A38Error("A38_MAX_IN_FLIGHT must be an integer >= 1") + return value + + +def _await_in_flight_as_interrupted( + in_flight: dict[ + concurrent.futures.Future[Any], tuple[Mapping[str, Any], float] + ], + run_by_id: dict[str, dict[str, Any]], + reasons: list[str], +) -> None: + if in_flight: + concurrent.futures.wait(tuple(in_flight)) + for fut, (job, started) in list(in_flight.items()): + ident = str(job["id"]) + try: + fut.result() + except Exception: + pass + if ident not in run_by_id: + run_by_id[ident] = _run_entry( + ident=ident, + name=str(job["name"]), + command=str(job["command"]), + result="error", + exit_code=-1, + duration_s=time.monotonic() - started, + timeout_s=float(job["timeout_s"]), + ) + reasons.append(f"{ident}: interrupted") + in_flight.clear() + + def _prepare_run_output( repo_path: Path, output: Path, @@ -1033,8 +1126,8 @@ def run_policy( omit = set(required) env = _job_env(head, base) + max_in_flight = _max_in_flight(env) _preflight_jobs([job for job in jobs if str(job["id"]) not in omit], env) - runs: list[dict[str, Any]] = [] reasons: list[str] = [] interrupted = False drift = False @@ -1043,97 +1136,177 @@ def run_policy( if markdown_only else "omitted: README-only change set\n" ) + run_by_id: dict[str, dict[str, Any]] = {} + stop_starting = False + + def execute_job(job: Mapping[str, Any]) -> dict[str, Any]: + ident = str(job["id"]) + name = str(job["name"]) + command = str(job["command"]) + timeout_s = float(job["timeout_s"]) + log_path = logs_dir / f"{ident}.log" + started = time.monotonic() + result, exit_code, duration_s = _run_one_job( + repo_path=root, + command=command, + timeout_s=timeout_s, + log_path=log_path, + env=env, + ) + if result == "pass" and duration_s > timeout_s: + result = "timeout" + return _run_entry( + ident=ident, + name=name, + command=command, + result=result, + exit_code=exit_code, + duration_s=duration_s, + timeout_s=timeout_s, + ) previous_sigterm = None if threading.current_thread() is threading.main_thread(): previous_sigterm = signal.getsignal(signal.SIGTERM) signal.signal(signal.SIGTERM, _interrupt_run) + in_flight: dict[ + concurrent.futures.Future[dict[str, Any]], tuple[Mapping[str, Any], float] + ] = {} + pending: list[Mapping[str, Any]] = [] try: for job in jobs: ident = str(job["id"]) - name = str(job["name"]) - command = str(job["command"]) - timeout_s = float(job["timeout_s"]) - log_path = logs_dir / f"{ident}.log" - started = time.monotonic() if ident in omit: - log_path.write_text(omit_log, encoding="utf-8") - runs.append( - _run_entry( - ident=ident, - name=name, - command=command, - result="not_applicable", - exit_code=0, - duration_s=0.0, - timeout_s=timeout_s, - ) - ) - continue - try: - result, exit_code, duration_s = _run_one_job( - repo_path=root, - command=command, - timeout_s=timeout_s, - log_path=log_path, - env=env, - ) - except KeyboardInterrupt: - interrupted = True - runs.append( - _run_entry( - ident=ident, - name=name, - command=command, - result="error", - exit_code=-1, - duration_s=time.monotonic() - started, - timeout_s=timeout_s, - ) - ) - reasons.append(f"{ident}: interrupted") - break - except OSError as exc: - result, exit_code, duration_s = "error", -1, time.monotonic() - started - reasons.append(f"{ident}: execution failed ({type(exc).__name__})") - if result == "pass" and duration_s > timeout_s: - result = "timeout" - runs.append( - _run_entry( + (logs_dir / f"{ident}.log").write_text(omit_log, encoding="utf-8") + run_by_id[ident] = _run_entry( ident=ident, - name=name, - command=command, - result=result, - exit_code=exit_code, - duration_s=duration_s, - timeout_s=timeout_s, + name=str(job["name"]), + command=str(job["command"]), + result="not_applicable", + exit_code=0, + duration_s=0.0, + timeout_s=float(job["timeout_s"]), ) - ) - if result != "pass" or exit_code != 0: - reasons.append(f"{ident}: result is {result}") - - try: - _require_clean_tree(root, run=runner) - after = _head_sha(root, run=runner) - except A38Error as exc: - drift = True - reasons.append(f"working tree or HEAD drifted: {exc}") - break - if after != head: - drift = True - reasons.append("HEAD drifted during run") - break + else: + pending.append(job) + + with concurrent.futures.ThreadPoolExecutor(max_workers=max_in_flight) as pool: + try: + while pending or in_flight: + if not stop_starting: + inflight_keys: set[str] = set() + for running_job, _started in in_flight.values(): + inflight_keys.update(_job_conflict_keys(running_job)) + still_pending: list[Mapping[str, Any]] = [] + for job in pending: + keys = _job_conflict_keys(job) + if ( + len(in_flight) >= max_in_flight + or keys & inflight_keys + ): + still_pending.append(job) + continue + started = time.monotonic() + future = pool.submit(execute_job, job) + in_flight[future] = (job, started) + inflight_keys.update(keys) + pending = still_pending + + if not in_flight: + break + + try: + done, _pending_futs = concurrent.futures.wait( + tuple(in_flight), + return_when=concurrent.futures.FIRST_COMPLETED, + ) + except KeyboardInterrupt: + interrupted = True + stop_starting = True + reasons.append("run interrupted") + _terminate_active_job_procs() + _await_in_flight_as_interrupted(in_flight, run_by_id, reasons) + break + + for future in done: + job, started = in_flight.pop(future) + ident = str(job["id"]) + try: + entry = future.result() + except KeyboardInterrupt: + interrupted = True + stop_starting = True + entry = _run_entry( + ident=ident, + name=str(job["name"]), + command=str(job["command"]), + result="error", + exit_code=-1, + duration_s=time.monotonic() - started, + timeout_s=float(job["timeout_s"]), + ) + reasons.append(f"{ident}: interrupted") + _terminate_active_job_procs() + _await_in_flight_as_interrupted(in_flight, run_by_id, reasons) + run_by_id[ident] = entry + pending.clear() + break + except Exception as exc: + entry = _run_entry( + ident=ident, + name=str(job["name"]), + command=str(job["command"]), + result="error", + exit_code=-1, + duration_s=time.monotonic() - started, + timeout_s=float(job["timeout_s"]), + ) + reasons.append(f"{ident}: execution failed ({type(exc).__name__})") + run_by_id[ident] = entry + if entry["result"] != "pass" or entry["exit_code"] != 0: + if f"{ident}: result is {entry['result']}" not in reasons: + reasons.append(f"{ident}: result is {entry['result']}") + if interrupted: + break + try: + _require_clean_tree(root, run=runner) + after = _head_sha(root, run=runner) + except A38Error as exc: + drift = True + stop_starting = True + reasons.append(f"working tree or HEAD drifted: {exc}") + pending.clear() + _terminate_active_job_procs() + _await_in_flight_as_interrupted(in_flight, run_by_id, reasons) + break + if after != head: + drift = True + stop_starting = True + reasons.append("HEAD drifted during run") + pending.clear() + _terminate_active_job_procs() + _await_in_flight_as_interrupted(in_flight, run_by_id, reasons) + break + except KeyboardInterrupt: + interrupted = True + stop_starting = True + reasons.append("run interrupted") + _terminate_active_job_procs() + _await_in_flight_as_interrupted(in_flight, run_by_id, reasons) except KeyboardInterrupt: interrupted = True reasons.append("run interrupted") + _terminate_active_job_procs() + _await_in_flight_as_interrupted(in_flight, run_by_id, reasons) finally: if previous_sigterm is not None: signal.signal(signal.SIGTERM, previous_sigterm) - finished_ids = {run_item["id"] for run_item in runs} + runs: list[dict[str, Any]] = [] for job in jobs: ident = str(job["id"]) - if ident in finished_ids: + if ident in run_by_id: + runs.append(run_by_id[ident]) continue runs.append( _run_entry( diff --git a/tests/test_a38.py b/tests/test_a38.py index db0106c..980612d 100644 --- a/tests/test_a38.py +++ b/tests/test_a38.py @@ -8,6 +8,7 @@ import sys import subprocess import tempfile +import threading import time import unittest from datetime import datetime, timedelta, timezone @@ -238,6 +239,32 @@ def test_rejects_unknown_key(self) -> None: with self.assertRaisesRegex(A38Error, "unknown keys"): load_policy(json.dumps(raw)) + def test_command_job_lock_round_trip(self) -> None: + raw = _policy_dict() + raw["jobs"][0]["lock"] = "jest-full" + policy = load_policy(json.dumps(raw)) + self.assertEqual(policy["jobs"][0]["lock"], "jest-full") + + def test_command_job_rejects_empty_lock(self) -> None: + raw = _policy_dict() + raw["jobs"][0]["lock"] = "" + with self.assertRaisesRegex(A38Error, "lock name"): + load_policy(json.dumps(raw)) + + def test_command_job_rejects_illegal_lock_name(self) -> None: + raw = _policy_dict() + raw["jobs"][0]["lock"] = "jest full" + with self.assertRaisesRegex(A38Error, "lock name"): + load_policy(json.dumps(raw)) + + def test_executor_job_rejects_sibling_lock(self) -> None: + raw = _policy_dict( + jobs=[_commands_lock_job(ident="jest", gh_job="jest", lock="jest-full")] + ) + raw["jobs"][0]["lock"] = "jest-full" + with self.assertRaisesRegex(A38Error, "executor.config"): + load_policy(json.dumps(raw)) + def test_rejects_empty_jobs(self) -> None: with self.assertRaisesRegex(A38Error, "non-empty"): load_policy(_policy_text(jobs=[])) @@ -1810,6 +1837,256 @@ def test_base_sha_required(self) -> None: ) +def _commands_lock_job( + *, + ident: str, + gh_job: str, + lock: str | None, + timeout_s: float = 30, +) -> dict: + config: dict = {"steps": [["true", ident]]} + if lock is not None: + config["lock"] = lock + return { + "id": ident, + "name": ident, + "executor": {"adapter": "commands", "config": config}, + "timeout_s": timeout_s, + "workflow": ".github/workflows/ci.yml", + "job": gh_job, + } + + +class SchedulerTests(unittest.TestCase): + def _run_with_sleeper( + self, + jobs: list[dict], + *, + sleep_s: float = 0.25, + env: dict[str, str] | None = None, + outcomes: dict[str, tuple[str, int, float]] | None = None, + ) -> tuple[dict, float, list[tuple[str, str, float]]]: + events: list[tuple[str, str, float]] = [] + lock = threading.Lock() + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + head = _init_repo(repo) + policy = load_policy(_policy_text(jobs=jobs)) + ident_by_command = {str(job["command"]): str(job["id"]) for job in policy["jobs"]} + + def fake_run_one_job_mapped(**kwargs: object) -> tuple[str, int, float]: + command = str(kwargs["command"]) + ident = ident_by_command.get(command, "unknown") + t0 = time.monotonic() + with lock: + events.append(("start", ident, t0)) + time.sleep(sleep_s) + t1 = time.monotonic() + with lock: + events.append(("end", ident, t1)) + if outcomes and ident in outcomes: + return outcomes[ident] + return ("pass", 0, t1 - t0) + + extra_env = env or {} + with mock.patch.dict(os.environ, extra_env, clear=False): + with mock.patch("agent_cli.a38._run_one_job", side_effect=fake_run_one_job_mapped): + started = time.monotonic() + verdict = run_policy( + repo, + policy, + output=root / "report.md", + logs_dir=root / "logs", + base_sha=head, + private=True, + ) + wall = time.monotonic() - started + return verdict, wall, events + + def test_disjoint_lock_domains_overlap(self) -> None: + jobs = [ + _commands_lock_job(ident="jest", gh_job="jest", lock="jest-full"), + _commands_lock_job(ident="docker", gh_job="compose", lock="docker-heavy"), + ] + verdict, wall, events = self._run_with_sleeper(jobs, sleep_s=0.3) + self.assertTrue(verdict["ok"], msg=verdict) + starts = [item for item in events if item[0] == "start"] + ends = [item for item in events if item[0] == "end"] + self.assertEqual({item[1] for item in starts}, {"jest", "docker"}) + first_end = min(item[2] for item in ends) + second_start = max(item[2] for item in starts) + self.assertLess(second_start, first_end) + self.assertLess(wall, 0.55) + + def test_same_lock_does_not_overlap(self) -> None: + jobs = [ + _commands_lock_job(ident="a", gh_job="a", lock="jest-full"), + _commands_lock_job(ident="b", gh_job="b", lock="jest-full"), + ] + verdict, wall, events = self._run_with_sleeper(jobs, sleep_s=0.25) + self.assertTrue(verdict["ok"], msg=verdict) + starts = sorted((item[2], item[1]) for item in events if item[0] == "start") + ends = {item[1]: item[2] for item in events if item[0] == "end"} + first_id = starts[0][1] + self.assertGreaterEqual(starts[1][0], ends[first_id]) + self.assertGreaterEqual(wall, 0.45) + + def test_command_and_jest_share_cpu_domain(self) -> None: + jobs = [ + { + "id": "lint", + "name": "lint", + "command": "true", + "timeout_s": 30, + "workflow": ".github/workflows/ci.yml", + "job": "lint", + }, + _commands_lock_job(ident="jest", gh_job="jest", lock="jest-full"), + ] + verdict, wall, events = self._run_with_sleeper(jobs, sleep_s=0.25) + self.assertTrue(verdict["ok"], msg=verdict) + starts = sorted((item[2], item[1]) for item in events if item[0] == "start") + ends = {item[1]: item[2] for item in events if item[0] == "end"} + first_id = starts[0][1] + self.assertGreaterEqual(starts[1][0], ends[first_id]) + self.assertGreaterEqual(wall, 0.45) + + def test_max_in_flight_cap_prevents_overlap(self) -> None: + jobs = [ + _commands_lock_job(ident="jest", gh_job="jest", lock="jest-full"), + _commands_lock_job(ident="docker", gh_job="compose", lock="docker-heavy"), + ] + verdict, wall, events = self._run_with_sleeper( + jobs, sleep_s=0.25, env={"A38_MAX_IN_FLIGHT": "1"} + ) + self.assertTrue(verdict["ok"], msg=verdict) + starts = sorted((item[2], item[1]) for item in events if item[0] == "start") + ends = {item[1]: item[2] for item in events if item[0] == "end"} + first_id = starts[0][1] + self.assertGreaterEqual(starts[1][0], ends[first_id]) + self.assertGreaterEqual(wall, 0.45) + + def test_failure_still_runs_independent_domain(self) -> None: + jobs = [ + _commands_lock_job(ident="jest", gh_job="jest", lock="jest-full"), + _commands_lock_job(ident="docker", gh_job="compose", lock="docker-heavy"), + ] + verdict, _wall, events = self._run_with_sleeper( + jobs, + sleep_s=0.05, + outcomes={"jest": ("fail", 1, 0.05)}, + ) + self.assertFalse(verdict["ok"]) + ran = {item[1] for item in events if item[0] == "start"} + self.assertEqual(ran, {"jest", "docker"}) + + def test_report_runs_follow_policy_order(self) -> None: + jobs = [ + _commands_lock_job(ident="docker", gh_job="compose", lock="docker-heavy"), + _commands_lock_job(ident="jest", gh_job="jest", lock="jest-full"), + ] + hold_docker = threading.Event() + + def fake_run_one_job(**kwargs: object) -> tuple[str, int, float]: + command = str(kwargs["command"]) + if "docker-heavy" in command: + hold_docker.wait(timeout=2) + time.sleep(0.05) + return ("pass", 0, 0.05) + time.sleep(0.02) + hold_docker.set() + return ("pass", 0, 0.02) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + head = _init_repo(repo) + policy = load_policy(_policy_text(jobs=jobs)) + with mock.patch("agent_cli.a38._run_one_job", side_effect=fake_run_one_job): + verdict = run_policy( + repo, + policy, + output=root / "report.md", + logs_dir=root / "logs", + base_sha=head, + private=True, + ) + self.assertTrue(verdict["ok"], msg=verdict) + report = parse_comment((root / "report.md").read_text(encoding="utf-8")) + self.assertEqual([run.id for run in report.runs], ["docker", "jest"]) + + def test_head_drift_terminates_in_flight_peer(self) -> None: + jobs = [ + _commands_lock_job(ident="jest", gh_job="jest", lock="jest-full"), + _commands_lock_job(ident="docker", gh_job="compose", lock="docker-heavy"), + ] + terminated = {"n": 0} + tree_calls = {"n": 0} + + def fake_run_one_job(**kwargs: object) -> tuple[str, int, float]: + time.sleep(0.15) + return ("pass", 0, 0.15) + + def fake_clean_tree(*args: object, **kwargs: object) -> None: + tree_calls["n"] += 1 + if tree_calls["n"] >= 2: + raise A38Error("dirty") + + def fake_terminate() -> None: + terminated["n"] += 1 + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + head = _init_repo(repo) + policy = load_policy(_policy_text(jobs=jobs)) + with mock.patch("agent_cli.a38._run_one_job", side_effect=fake_run_one_job): + with mock.patch("agent_cli.a38._require_clean_tree", side_effect=fake_clean_tree): + with mock.patch( + "agent_cli.a38._terminate_active_job_procs", + side_effect=fake_terminate, + ): + verdict = run_policy( + repo, + policy, + output=root / "report.md", + logs_dir=root / "logs", + base_sha=head, + private=True, + ) + self.assertFalse(verdict["ok"]) + self.assertGreaterEqual(terminated["n"], 1) + + def test_invalid_max_in_flight_raises_before_jobs(self) -> None: + jobs = [_commands_lock_job(ident="jest", gh_job="jest", lock="jest-full")] + called = {"n": 0} + + def fake_run_one_job(**kwargs: object) -> tuple[str, int, float]: + called["n"] += 1 + return ("pass", 0, 0.01) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + head = _init_repo(repo) + policy = load_policy(_policy_text(jobs=jobs)) + with mock.patch.dict(os.environ, {"A38_MAX_IN_FLIGHT": "0"}): + with mock.patch("agent_cli.a38._run_one_job", side_effect=fake_run_one_job): + with self.assertRaises(A38Error) as ctx: + run_policy( + repo, + policy, + output=root / "report.md", + logs_dir=root / "logs", + base_sha=head, + private=True, + ) + self.assertIn("A38_MAX_IN_FLIGHT", str(ctx.exception)) + self.assertEqual(called["n"], 0) + + class CliTests(unittest.TestCase): def test_policy_subcommand(self) -> None: with tempfile.TemporaryDirectory() as tmp: diff --git a/tests/test_a38_executor.py b/tests/test_a38_executor.py index f4226fb..826a71d 100644 --- a/tests/test_a38_executor.py +++ b/tests/test_a38_executor.py @@ -134,8 +134,12 @@ def test_all_adapter_ids_normalize_to_command_only_jobs(self) -> None: self.assertNotIn("executor", normalized) self.assertEqual( set(normalized), - {"id", "name", "command", "timeout_s", "workflow", "job"}, + {"id", "name", "command", "timeout_s", "workflow", "job", "lock"}, ) + if adapter in ("compose", "http-smoke"): + self.assertEqual(normalized["lock"], "docker-heavy") + else: + self.assertIsNone(normalized["lock"]) argv = shlex.split(normalized["command"]) self.assertEqual(argv[:5], ["agent", "a38", "job", adapter, "--config"]) self.assertEqual(len(argv), 6) @@ -153,6 +157,12 @@ def test_all_adapter_ids_normalize_to_command_only_jobs(self) -> None: f"agent a38 job {adapter} --config {shlex.quote(canonical)}", ) + def test_explicit_compose_lock_overrides_docker_heavy_default(self) -> None: + config = dict(_minimal_configs()["compose"]) + config["lock"] = "custom-lock" + policy = load_policy(_policy_text(_executor_job("compose", config))) + self.assertEqual(policy["jobs"][0]["lock"], "custom-lock") + def test_reordered_nested_keys_have_identical_normalization(self) -> None: first = { "steps": [{"argv": ["printf", "%s", "{repo}"], "stdout": "result.txt"}],