diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f6fa75..8f1ad6d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,71 @@ All notable changes to Code Context Control (C3) are documented here. The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.130.0] - 2026-09-12 + +### Added — AgentCI owns the DAG: job outputs, run-time `needs.*`, one job per act + +A workflow shaped like most real ones — a `changes` job that decides whether +the suite must run, shards gated on `if: needs.changes.outputs.code == 'true'`, +and a trailing aggregator whose `run:` reads `${{ needs..result }}` — +could not run under `c3_ci` at all. Every job came back `UNSUP` with +"unresolved expression(s): github.event.before, needs.changes.result". + +Three things were missing, each measured with a two-job probe before the +fix: + +- **Job outputs.** The runner recorded each job's `result` for `if:` and + nothing else. Now every native step gets its own `GITHUB_OUTPUT` file + (`k=v` and `k<.outputs.`, and dependents read `needs..outputs.` + in `if:` and in `run:`. Under act the same values come from the + `::set-output:: k=v` lines act logs. An output that was declared but never + written is refused, not blanked — CI would hand a dependent the real + value, and a blank standing in for it is the exact bug this engine exists + to catch. Jobs that publish outputs are never served from the verdict + cache, because a cache hit replays a verdict, not the values. +- **`needs.*` in `run:` text.** Deferred at DAG-build time instead of + blocking, then filled in just before each step executes. `steps.*` the + same. A leftover at run time still refuses the step. +- **`github.event.*`.** There is no payload locally. With a declared event + the field is `""` — what GitHub yields for a field a sparse payload lacks + and what act does — so a workflow written to handle an empty value runs. + Without a declared event it stays a blocker for the native engine (act + resolves it itself). `secrets.*` and `vars.*` block both engines as before. + +And one thing that was silently wrong: **`act -j ` runs the job's +`needs` first**. C3 orders the DAG and runs each job with `-j`, so every +dependency executed twice — once by C3, again inside the dependent's +container — and an aggregator job replayed the whole workflow. act now +receives a derived copy of the workflow (temp file, `-W` outside the tree, +repository untouched) holding that one job with `needs` and `if:` removed +and `needs.*` spelled out from C3's own results and outputs. A `needs.*` +reference C3 never recorded is reported before act starts. + +### Fixed — two ways an act run died before its first step + +- **A cached image was still "pulled".** Every run passed `--pull=true`, + and act sends the stored Docker Hub credential on a pull where the CLI + sends none — so an expired login turned a public image already on disk + into "authentication required" (issue #173, now reproduced). The runner + checks `docker image inspect` first and pulls only what is missing. +- **Two runs on one box collided on port 34567.** `--artifact-server-path` + makes act bind its artifact server to one fixed port, so a second act run + — another session, another project, the test suite beside a real run — + died with "bind: Only one usage of each socket address" and reported + `unparsed`. Each run now gets a free port. +- **`--job X` dropped X's own dependencies.** Selecting `smokes` alone + deselected `changes`, and then `if: needs.changes.outputs.code == 'true'` + was unjudgeable — the selection had removed the job whose output it + reads. A selected job (and a `rerun`) now brings its transitive `needs` + along; everything else stays DESELECTED and the verdict stays PARTIAL. + +Measured on the workflow above (drknowhow/Yep, 10,683 tests): 6 of 6 jobs +ran — `changes` alone in its container and its `code=true` captured, three +shards gated on it and green, the aggregator handed real results. Verdict +FAIL, honestly: one smoke module needs `java`, which GitHub's runner has and +the act image does not. Before this release every job was `UNSUP`. + ## [2.129.2] - 2026-09-10 ### Added — the hub could decide override requests but not allow them diff --git a/cli/c3.py b/cli/c3.py index fe0a3f5..1522ed6 100644 --- a/cli/c3.py +++ b/cli/c3.py @@ -91,7 +91,7 @@ # Config CONFIG_DIR = ".c3" CONFIG_FILE = ".c3/config.json" -__version__ = "2.129.2" +__version__ = "2.130.0" def _compress_file_cli(compressor, path, mode="smart", **kw): diff --git a/pyproject.toml b/pyproject.toml index 6c9675c..9710a3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "code-context-control" -version = "2.129.2" +version = "2.130.0" description = "Local MCP code-intelligence for AI coding tools: surgical search/read/edit, agent-config version history, path-level access + masking guards, and a multi-project hub." readme = "README.md" requires-python = ">=3.10" diff --git a/services/ci_act.py b/services/ci_act.py index c6ca185..b153bbb 100644 --- a/services/ci_act.py +++ b/services/ci_act.py @@ -44,6 +44,8 @@ import time from pathlib import Path +import yaml + # act's community runner images. These mirror a good deal of the GitHub # runner's preinstalled toolchain; a bare `ubuntu:24.04` does not, and a job # that assumes a preinstalled tool then fails locally while passing in CI. A @@ -132,6 +134,18 @@ def availability() -> dict: return info +def image_present(image: str) -> bool: + """Is *image* already in the local Docker store? + + Decides `--pull` for a run: an image that is here needs no registry + round-trip, and skipping it also skips the issue-#173 trap — act sends + the stored Docker Hub credential on every pull, so an expired login + turns a public image that is already on disk into an auth failure. + """ + rc, _ = _probe(["docker", "image", "inspect", image], timeout=20) + return rc == 0 + + # ── Side-effect gate ──────────────────────────────────────────────────────── # With a real action runner, jobs that were previously unrunnable become # runnable — including ones whose entire purpose is to publish. Capability no @@ -208,14 +222,28 @@ def host_mutations(inst) -> list: def build_command(inst, project, act_path: str, event: str = "", image: str = "", network: str = "", secret_file: str = "", env_file: str = "", - artifact_dir: str = "", pull: bool = True) -> list: - """The exact argv handed to act. Pure, so it is unit-testable.""" + artifact_dir: str = "", pull: bool = True, + workflow_file: str = "", artifact_port: int = 0) -> list: + """The exact argv handed to act. Pure, so it is unit-testable. + + *workflow_file* overrides the job's own workflow path — the derived copy + from derive_workflow(). act accepts `-W` outside the working tree, so it + lives in a temp dir and the repository stays untouched. + + *artifact_port*: act's artifact server binds ONE fixed port (34567) by + default, so two act runs on the same box — two sessions, two projects, + a test suite beside a real run — collide and the second dies with + "bind: Only one usage of each socket address" before any step runs. + run_job() picks a free port per run. + """ cmd: list = [act_path] if event: cmd.append(event) - workflow_path = getattr(inst, "workflow_path", "") or "" - if workflow_path: + workflow_path = workflow_file or getattr(inst, "workflow_path", "") or "" + if workflow_file: + cmd += ["-W", str(workflow_file).replace("\\", "/")] + elif workflow_path: try: rel = Path(workflow_path).resolve().relative_to(Path(project).resolve()) cmd += ["-W", str(rel).replace("\\", "/")] @@ -241,9 +269,22 @@ def build_command(inst, project, act_path: str, event: str = "", cmd += ["--env-file", env_file or os.devnull] if artifact_dir: cmd += ["--artifact-server-path", artifact_dir] + if artifact_port: + cmd += ["--artifact-server-port", str(int(artifact_port))] return cmd +def free_port() -> int: + """A TCP port nothing is listening on right now, for act's artifact server.""" + import socket + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("", 0)) + return int(sock.getsockname()[1]) + except OSError: + return 0 + + # act prefixes every line it emits with `[Workflow/job]`, and marks actual # program output with a `|` gutter: # @@ -316,10 +357,111 @@ def program_output(text: str) -> str: return "\n".join(lines) if lines else (text or "") +# ── Owning the DAG ────────────────────────────────────────────────────────── +# `act -j ` runs the job's `needs` first — measured 2026-09-12 with a +# two-job probe: `-j b` ran `a`, then `b`. C3 already ran `a`, ordered the DAG +# and recorded the result, so under the old per-job invocation every +# dependency executed twice and an aggregator job replayed the whole workflow +# inside its container. The derived copy below holds one job, no `needs`, no +# `if:` (C3 evaluated it before dispatch), and `needs.*` spelled out. + +_NEEDS_REF = re.compile( + r"\$\{\{\s*needs\.([A-Za-z_][\w-]*)\.(result|outputs\.([A-Za-z_][\w-]*))\s*\}\}") +_SET_OUTPUT = re.compile(r"::set-output::\s*([^=\s]+)=(.*)$") + + +def _reads_needs(inst) -> bool: + for step in getattr(inst, "steps", []): + if "needs." in str(getattr(step, "run", "") or ""): + return True + for value in (getattr(step, "env", None) or {}).values(): + if "needs." in str(value): + return True + return False + + +def derive_workflow(inst, needs_ctx: dict, source_text: str = "") -> tuple: + """(yaml_text, unresolved) — this one job, ready for `act -j`. + + *source_text* is the workflow file's content (read from + ``inst.workflow_path`` when omitted). The job keeps everything else — + matrix, env, steps, `outputs:` — so act's behaviour is unchanged. + Returns (None, []) when the source cannot be read or parsed. + + A `needs.*` reference to a result or output C3 does not hold is returned + in *unresolved* rather than left for act, which would resolve it to "" + and run something CI never would. + """ + text = source_text + if not text: + try: + text = Path(getattr(inst, "workflow_path", "")).read_text(encoding="utf-8") + except (OSError, TypeError, ValueError): + return None, [] + try: + raw = yaml.safe_load(text) or {} + except Exception: + return None, [] + if not isinstance(raw, dict): + return None, [] + jobs = raw.get("jobs") + job = jobs.get(inst.job_id) if isinstance(jobs, dict) else None + if not isinstance(job, dict): + return None, [] + + job = dict(job) + job.pop("needs", None) + job.pop("if", None) + derived = {k: v for k, v in raw.items() if k != "jobs"} + derived["jobs"] = {inst.job_id: job} + # `on:` parses as the boolean True under YAML 1.1; put the key back as a + # string so act's parser sees a trigger list. + if True in derived: + derived["on"] = derived.pop(True) + + unresolved: list = [] + + def _one(match: re.Match) -> str: + dep, kind, key = match.group(1), match.group(2), match.group(3) + rec = (needs_ctx or {}).get(dep) + if not isinstance(rec, dict): + unresolved.append(match.group(0).strip()) + return match.group(0) + if kind == "result": + return str(rec.get("result", "")) + outputs = rec.get("outputs") or {} + if key not in outputs: + unresolved.append(match.group(0).strip()) + return match.group(0) + return str(outputs[key]) + + dumped = yaml.safe_dump(derived, sort_keys=False, allow_unicode=True, + width=10_000) + substituted = _NEEDS_REF.sub(_one, dumped) + return substituted, sorted(set(unresolved)) + + +def job_outputs_from_log(text: str) -> dict: + """Every `::set-output:: k=v` act logged, last write wins.""" + found: dict = {} + for line in (text or "").splitlines(): + m = _SET_OUTPUT.search(line) + if m: + found[m.group(1)] = m.group(2).rstrip() + return found + + def run_job(inst, project, timeout: int = DEFAULT_ACT_TIMEOUT, event: str = "", network: str = "", artifact_dir: str = "", - act_path: str = "") -> dict: - """Run one job through act. Returns {exit_code, output, timed_out, command}.""" + act_path: str = "", needs_ctx: dict = None) -> dict: + """Run one job through act. Returns {exit_code, output, timed_out, command}. + + *needs_ctx* is `{job_id: {"result": ..., "outputs": {...}}}` for the + job's dependencies, as C3 recorded them. When the job declares `needs` + or reads `needs.*`, act is handed a derived workflow (derive_workflow) + so it runs this job alone. An unresolvable `needs.*` reference comes + back as {"unresolved": [...]} without act ever starting. + """ act_path = act_path or find_act() if not act_path: return {"exit_code": 127, "output": "act is not installed", @@ -329,9 +471,27 @@ def run_job(inst, project, timeout: int = DEFAULT_ACT_TIMEOUT, empty = Path(tmpdir) / "empty" empty.write_text("", encoding="utf-8") + workflow_file = "" + if getattr(inst, "needs", None) or _reads_needs(inst): + derived, unresolved = derive_workflow(inst, needs_ctx or {}) + if unresolved: + shutil.rmtree(tmpdir, ignore_errors=True) + return {"exit_code": 0, "output": "", "timed_out": False, + "command": "", "unresolved": unresolved} + if derived is None: + shutil.rmtree(tmpdir, ignore_errors=True) + return {"exit_code": 126, "timed_out": False, "command": "", + "output": "could not read the workflow file to derive a " + "single-job copy for act"} + workflow_file = str(Path(tmpdir) / "workflow.yml") + Path(workflow_file).write_text(derived, encoding="utf-8") + + image = image_for(inst.runs_on) cmd = build_command(inst, project, act_path, event=event, network=network, secret_file=str(empty), env_file=str(empty), - artifact_dir=artifact_dir) + artifact_dir=artifact_dir, workflow_file=workflow_file, + pull=not image_present(image), + artifact_port=free_port() if artifact_dir else 0) from cli.tools.shell import _kill_tree, _popen_kwargs diff --git a/services/ci_runner.py b/services/ci_runner.py index 6fc0601..a896d91 100644 --- a/services/ci_runner.py +++ b/services/ci_runner.py @@ -30,7 +30,7 @@ import subprocess import time import uuid -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from datetime import datetime, timezone from pathlib import Path @@ -127,6 +127,9 @@ class JobResult: failures: list = field(default_factory=list) parser: str = "" log_path: str = "" + # The job's declared `outputs:`, resolved from what its steps wrote to + # GITHUB_OUTPUT. Dependents read them as `needs..outputs.`. + outputs: dict = field(default_factory=dict) @property def ran(self) -> bool: @@ -142,7 +145,7 @@ def to_dict(self) -> dict: "cross_os": self.cross_os, "duration_ms": self.duration_ms, "steps": [s.to_dict() for s in self.steps], "failures": self.failures, "parser": self.parser, - "log_path": self.log_path, + "log_path": self.log_path, "outputs": dict(self.outputs), } @@ -367,8 +370,16 @@ def _pick_engine(inst, engine: str, act_state: dict, allow_foreign: bool): def _run_job_act(inst, project: Path, run_dir: Path, event: str, - network: str, fidelity: str) -> JobResult: - """Delegate one job to act, then read its result the same way as any other.""" + network: str, fidelity: str, needs_ctx: dict = None) -> JobResult: + """Delegate one job to act, then read its result the same way as any other. + + C3 owns the DAG. act, handed `-j `, would run the job's `needs` + first — every dependency again, inside the container, on top of the run + C3 already made of them (an aggregator job would replay the whole + workflow). So act receives a derived copy of the workflow holding this + one job with `needs` removed and `needs.*` spelled out from C3's own + results (ci_act.derive_workflow), and runs exactly that job. + """ result = JobResult(key=inst.key, job_id=inst.job_id, name=inst.name, workflow=inst.workflow, runs_on=inst.runs_on, status=PASSED, engine="act", fidelity=fidelity, @@ -377,19 +388,42 @@ def _run_job_act(inst, project: Path, run_dir: Path, event: str, artifacts.mkdir(parents=True, exist_ok=True) outcome = ci_act.run_job(inst, project, event=event, network=network, - artifact_dir=str(artifacts)) + artifact_dir=str(artifacts), + needs_ctx=needs_ctx or {}) header = (f"[c3:ci] job {inst.key} via act " f"(image {ci_act.image_for(inst.runs_on)}, host {host_os()})\n" f"[c3:ci] $ {outcome.get('command', '')}\n\n") log = header + (outcome.get("output") or "") result.duration_ms = outcome.get("duration_ms", 0) - if outcome.get("timed_out"): + if outcome.get("unresolved"): + result.status = UNSUPPORTED + result.reason = ("`needs.*` this job reads was never produced: " + + ", ".join(outcome["unresolved"])) + elif outcome.get("timed_out"): result.status = TIMEOUT result.reason = "act exceeded its timeout; the container tree was killed" elif outcome.get("exit_code"): result.status = FAILED result.reason = f"act exited {outcome['exit_code']}" + elif inst.outputs: + # act logs every GITHUB_OUTPUT write as `::set-output:: k=v` but not + # which step wrote it; the derived workflow holds only this job, so + # every such line is this job's, and each declared output resolves + # against the union. Two steps writing the same key is the one shape + # this cannot tell apart (last wins, as it would on GitHub). + written = ci_act.job_outputs_from_log(outcome.get("output") or "") + steps_ctx = {(s.id or f"__step{s.index}"): {"outputs": dict(written)} + for s in inst.steps} + result.outputs, missing = resolve_job_outputs(inst, steps_ctx) + if missing: + result.status = UNSUPPORTED + result.reason = ("job `outputs:` could not be resolved from act's " + "log: " + "; ".join(missing)) + else: + log += ("\n[c3:ci] job outputs: " + + ", ".join(f"{k}={v}" for k, v in result.outputs.items()) + + "\n") # act reports per-step results in its own log format rather than as data, # so the job is recorded as one unit. The failure parsers read the log the @@ -433,6 +467,23 @@ def _run_job_act(inst, project: Path, run_dir: Path, event: str, return result +def _with_dependencies(chosen: set, ordered: list) -> set: + """*chosen* plus every job it transitively `needs` (all matrix cells).""" + by_job: dict = {} + for inst in ordered: + by_job.setdefault((inst.workflow, inst.job_id), []).append(inst) + out = set(chosen) + frontier = [i for i in ordered if i.key in chosen] + while frontier: + inst = frontier.pop() + for dep in inst.needs: + for dep_inst in by_job.get((inst.workflow, dep), []): + if dep_inst.key not in out: + out.add(dep_inst.key) + frontier.append(dep_inst) + return out + + def _eval_values(inst, needs_results: dict, github: dict) -> dict: """The contexts an `if:` may read, built from facts we actually have.""" return { @@ -447,8 +498,59 @@ def _eval_values(inst, needs_results: dict, github: dict) -> dict: } +def _read_github_output(path: Path) -> dict: + """Parse one step's GITHUB_OUTPUT file: `k=v` lines and `k< tuple: + """(outputs, unresolved) for a job's `outputs:` mapping. + + Each declared output is an expression, almost always + `${{ steps..outputs. }}`. An output whose step never wrote the key + is reported, not blanked: a dependent that reads it would otherwise be + told "" where CI would tell it the real value. + """ + from services.ci_workflow import substitute + + outputs: dict = {} + unresolved: list = [] + for name, expr in (inst.outputs or {}).items(): + res = substitute(str(expr), {"steps": steps_ctx}) + if res.unresolved: + unresolved.append(f"{name} <- {expr}") + continue + outputs[name] = res.text + return outputs, unresolved + + def _run_job(inst, project: Path, run_dir: Path, timeout: int, - eval_values: dict = None) -> JobResult: + eval_values: dict = None, event_declared: bool = False) -> JobResult: + from services.ci_workflow import late_substitute + eval_values = eval_values or _eval_values(inst, {}, {}) result = JobResult(key=inst.key, job_id=inst.job_id, name=inst.name, workflow=inst.workflow, runs_on=inst.runs_on, @@ -464,6 +566,13 @@ def _run_job(inst, project: Path, run_dir: Path, timeout: int, started = time.time() env = _job_env(inst, project) + # Run-time contexts. `needs` arrived from the runner (what earlier jobs + # produced); `steps` fills in here as each step writes GITHUB_OUTPUT. + steps_ctx: dict = dict(eval_values.get("steps") or {}) + runtime = {"needs": dict(eval_values.get("needs") or {}), "steps": steps_ctx} + eval_values["steps"] = steps_ctx + output_dir = run_dir / "outputs" / _safe(inst.key) + # A failed step no longer ends the job outright: `if: always()` exists # precisely so cleanup and reporting steps run afterwards. The first # failure fixes the job's verdict; later steps run only if their own @@ -501,14 +610,62 @@ def _run_job(inst, project: Path, run_dir: Path, timeout: int, f"\n[c3:ci] step {step.index} skipped by `if: {step.if_}`\n") continue - sres, output = _run_step(step, project, env, timeout) + # Second substitution pass: `needs.*` / `steps.*` exist now, and + # `github.event.*` is "" under a declared event. A leftover here is a + # value CI would have and we do not — refuse, never run a literal. + run_step = step + if step.run and "${{" in step.run: + late = late_substitute(step.run, runtime, event_declared) + if late.unresolved: + result.status = UNSUPPORTED + result.reason = (f"step {step.index} still has unresolved " + f"expression(s) at run time: " + + ", ".join(sorted(set(late.unresolved)))) + chunks.append(f"\n[c3:ci] {result.reason}\n") + break + run_step = replace(step, run=late.text) + step_env = dict(env) + for key, value in (step.env or {}).items(): + late = late_substitute(str(value), runtime, event_declared) + step_env[str(key)] = late.text + output_file = output_dir / f"step-{step.index}.txt" + try: + output_dir.mkdir(parents=True, exist_ok=True) + output_file.write_text("", encoding="utf-8") + step_env["GITHUB_OUTPUT"] = str(output_file) + except OSError: + output_file = None + + sres, output = _run_step(run_step, project, step_env, timeout) result.steps.append(sres) chunks.append(output) + if output_file is not None: + written = _read_github_output(output_file) + if written: + sid = step.id or f"__step{step.index}" + steps_ctx.setdefault(sid, {}).setdefault("outputs", {}).update(written) + chunks.append(f"[c3:ci] step {step.index} outputs: " + + ", ".join(f"{k}={v}" for k, v in written.items()) + + "\n") if sres.status in (FAILED, TIMEOUT) and not job_failed: job_failed = True result.status = sres.status result.reason = f"step {sres.index} ({sres.name}) exited {sres.exit_code}" + if result.status == PASSED and inst.outputs: + result.outputs, missing = resolve_job_outputs(inst, steps_ctx) + if missing: + # Declared, never produced. CI would hand dependents "" here; we + # say so instead of letting a blank masquerade as a value. + result.status = UNSUPPORTED + result.reason = ("job `outputs:` could not be resolved: " + + "; ".join(missing)) + chunks.append(f"\n[c3:ci] {result.reason}\n") + else: + chunks.append("[c3:ci] job outputs: " + + ", ".join(f"{k}={v}" for k, v in result.outputs.items()) + + "\n") + result.duration_ms = round((time.time() - started) * 1000) log = "".join(chunks) if len(log) > MAX_LOG_CHARS: @@ -683,6 +840,14 @@ def run_ci(project_path, selector: str = "", allow_foreign: bool = False, else: chosen = {i.key for i in ordered} + # A selected job runs WITH its `needs`. Selecting `smokes` alone used to + # deselect `changes`, and then `if: needs.changes.outputs.code == 'true'` + # was unjudgeable — the selection had removed the very job whose output + # it reads (hit 2026-09-12 on the first `--job` run of a gated workflow). + # GitHub has no "one job" run; the closure is the smallest honest one. + if selector or only: + chosen = _with_dependencies(chosen, ordered) + failed_jobs: set = set() # job_ids whose failure blocks dependents results: list = [] # (workflow, job_id) -> {"result": "success"|"failure"|"skipped"} so a @@ -691,14 +856,15 @@ def run_ci(project_path, selector: str = "", allow_foreign: bool = False, def _needs_ctx(inst) -> dict: return {jid: needs_results.get((inst.workflow, jid), - {"result": "skipped"}) + {"result": "skipped", "outputs": {}}) for jid in inst.needs} - def _record(inst, status: str) -> None: + def _record(inst, status: str, outputs: dict = None) -> None: outcome = {PASSED: "success", SKIPPED_IF: "skipped"}.get( status, "failure" if status in (FAILED, TIMEOUT, UNSUPPORTED) else "skipped") - needs_results[(inst.workflow, inst.job_id)] = {"result": outcome} + needs_results[(inst.workflow, inst.job_id)] = { + "result": outcome, "outputs": dict(outputs or {})} for inst in ordered: if inst.key not in chosen: @@ -839,6 +1005,11 @@ def _record(inst, status: str) -> None: scope=ci_cache.scope_for(inst, cache_rules)) except Exception: job_fp = "" # never let caching break a run + if inst.outputs: + # A cache hit replays a verdict, not the job's `outputs:` — and a + # dependent reading `needs..outputs.` would then find + # nothing and be refused. Jobs that publish outputs always run. + job_fp = "" if job_fp and not no_cache: hit = ci_cache.lookup(project, job_fp) @@ -855,10 +1026,11 @@ def _record(inst, status: str) -> None: if chosen_engine == "act": job_result = _run_job_act(inst, project, run_dir, event, network, - fidelity) + fidelity, needs_ctx=_needs_ctx(inst)) else: job_result = _run_job(inst, project, run_dir, timeout, - eval_values=eval_values) + eval_values=eval_values, + event_declared=bool(event)) job_result.fidelity = fidelity job_result.fingerprint = job_fp if job_fp and job_result.status == PASSED: @@ -867,7 +1039,7 @@ def _record(inst, status: str) -> None: results.append(job_result) if job_result.status in (FAILED, TIMEOUT, UNSUPPORTED): failed_jobs.add((inst.workflow, inst.job_id)) - _record(inst, job_result.status) + _record(inst, job_result.status, job_result.outputs) result.jobs = results result.finished_at = _now() diff --git a/services/ci_workflow.py b/services/ci_workflow.py index 614dc3b..c5582de 100644 --- a/services/ci_workflow.py +++ b/services/ci_workflow.py @@ -133,6 +133,65 @@ def _one(match: re.Match) -> str: return ExprResult(_EXPR_RE.sub(_one, text), unresolved) +# Contexts that do not exist until the run is under way. `needs.*` is what the +# jobs before this one produced; `steps.*` is what the steps before this one +# produced. Neither is a blocker at DAG-build time: the native runner fills +# them in just before each step executes, and the act engine receives a copy +# of the workflow in which `needs.*` is already spelled out (see +# ci_act.derive_workflow), so act evaluates `steps.*` itself. +DEFERRED_ROOTS = ("needs", "steps") + + +def classify_unresolved(unresolved: list) -> tuple: + """Split substitute()'s leftovers into (hard, deferred, event). + + hard — nothing local will ever supply it (`secrets.*`, `vars.*`, an + operator expression). Blocks the job on every engine. + deferred — `needs.*` / `steps.*`: resolved at run time. + event — `github.event.*`: the payload. There is none locally; with a + declared event both GitHub (sparse payload) and act resolve a + missing field to "", so the runner does the same — but only + when the caller declared the event, never by guessing one. + """ + hard: list = [] + deferred: list = [] + event: list = [] + for expr in unresolved: + root = expr.split(".", 1)[0] + if root in DEFERRED_ROOTS: + deferred.append(expr) + elif expr.startswith("github.event."): + event.append(expr) + else: + hard.append(expr) + return hard, deferred, event + + +def late_substitute(text: str, runtime: dict, event_declared: bool) -> ExprResult: + """Second pass, at run time, over what instantiate() left in place. + + *runtime* carries the contexts that now exist (`needs`, `steps`). With a + declared event, `github.event.*` becomes "" — the value a sparse payload + yields on GitHub and the value act uses. Anything still unresolved after + this is a genuine gap and the caller must refuse the step. + """ + if not text or "${{" not in text: + return ExprResult(text or "", []) + first = substitute(text, runtime) + if not event_declared: + return first + leftover: list = [] + + def _blank_event(match: re.Match) -> str: + expr = match.group(1).strip() + if expr.startswith("github.event."): + return "" + leftover.append(expr) + return match.group(0) + + return ExprResult(_EXPR_RE.sub(_blank_event, first.text), leftover) + + # ── Normalized IR ─────────────────────────────────────────────────────────── @dataclass @@ -148,6 +207,9 @@ class Step: with_: dict = field(default_factory=dict) # Set during instantiation when the step is a `uses:` we stand in for. shim: str = "" + # The YAML `id:` — what `steps..outputs.` and a job's `outputs:` + # mapping refer to. Empty for steps that declared none. + id: str = "" @property def is_run(self) -> bool: @@ -182,6 +244,11 @@ class JobInstance: act_blockers: list = field(default_factory=list) # why ACT cannot either workflow: str = "" workflow_path: str = "" # act needs -W to disambiguate job names + # The raw `outputs:` mapping (name -> expression, usually + # `${{ steps..outputs. }}`). Resolved after the job runs, from the + # GITHUB_OUTPUT lines its steps wrote, and handed to dependents as + # `needs..outputs.`. + outputs: dict = field(default_factory=dict) @property def key(self) -> str: @@ -489,6 +556,9 @@ def instantiate(workflow: Workflow, event: str = "", git: dict = None) -> list: act_blockers=act_blockers, workflow=workflow.name, workflow_path=workflow.path, + outputs={str(k): str(v) for k, v in + (raw.get("outputs") or {}).items()} + if isinstance(raw.get("outputs"), dict) else {}, ) instances.append(inst) return instances @@ -531,23 +601,35 @@ def _build_step(index: int, raw: dict, ctx: dict, blockers.append(f"step {index} has neither `run` nor `uses`") act_blockers.append(f"step {index} has neither `run` nor `uses`") - if run_res.unresolved: - # Executing a command with a literal ${{ }} in it would run something - # other than what CI runs. Refuse the job instead. act substitutes the - # same expressions, but a missing secret is missing on either engine. - msg = (f"step {index} has unresolved expression(s): " - + ", ".join(sorted(set(run_res.unresolved)))) - blockers.append(msg) - act_blockers.append(msg) + # Executing a command with a literal ${{ }} in it would run something + # other than what CI runs. What cannot be resolved now falls into three + # classes (classify_unresolved): a genuine gap blocks the job on every + # engine — act substitutes the same expressions, and a missing secret is + # missing on either; `needs.*`/`steps.*` are filled in at run time; and + # `github.event.*` is "" once the caller has declared the event (act + # does the same), and a blocker for the native engine until they do. + event_declared = "event_name" in (github_fields or set()) + + def _unresolved(label: str, unresolved: list) -> None: + hard, _deferred, event = classify_unresolved(unresolved) + if hard: + msg = (f"step {index}{label} has unresolved expression(s): " + + ", ".join(sorted(set(hard)))) + blockers.append(msg) + act_blockers.append(msg) + if event and not event_declared: + blockers.append( + f"step {index}{label} reads {', '.join(sorted(set(event)))} — " + "there is no event payload locally; pass the event you are " + "simulating and it resolves to \"\" as it would on a sparse " + "payload (the act engine already does this)") + + _unresolved("", run_res.unresolved) env_res = {} for key, value in (raw.get("env") or {}).items(): sub = substitute(str(value), ctx) - if sub.unresolved: - msg = (f"step {index} env {key} has unresolved expression(s): " - + ", ".join(sorted(set(sub.unresolved)))) - blockers.append(msg) - act_blockers.append(msg) + _unresolved(f" env {key}", sub.unresolved) env_res[str(key)] = sub.text step = Step( @@ -561,6 +643,7 @@ def _build_step(index: int, raw: dict, ctx: dict, if_=str(raw.get("if") or ""), with_=dict(raw.get("with") or {}), shim=shim, + id=str(raw.get("id") or ""), ) return step, blockers, act_blockers diff --git a/tests/test_ci_dag_ownership.py b/tests/test_ci_dag_ownership.py new file mode 100644 index 0000000..bf6ced7 --- /dev/null +++ b/tests/test_ci_dag_ownership.py @@ -0,0 +1,398 @@ +"""AgentCI — C3 owns the DAG: job outputs, run-time `needs.*`, one job per act. + +Three facts, each measured on 2026-09-12 with a two-job probe workflow +before a line of this was written: + + * `act -j b` runs `a` first when `b` needs it. Under C3's per-job act + invocation that meant every dependency ran twice and an aggregator job + replayed the whole workflow inside its container. + * act logs each GITHUB_OUTPUT write as `::set-output:: k=v`, without the + step id. + * With a declared event and no payload, `github.event.*` is "" — the same + thing GitHub yields for a field a sparse payload lacks. + +The tests below hold C3 to those facts: a derived single-job workflow for +act, real `needs..outputs.` for both engines, and `github.event.*` +that is "" only when the caller declared the event. +""" +from __future__ import annotations + +import shutil +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +from services import ci_act # noqa: E402 +from services import ci_runner as cr # noqa: E402 +from services import ci_workflow as cw # noqa: E402 +from services.ci_workflow import ( # noqa: E402 + build_dag, + discover_workflows, + host_os, + parse_workflow, +) + +LOCAL_RUNNER = {"Linux": "ubuntu-latest", "Darwin": "macos-latest", + "Windows": "windows-latest"}[host_os()] + + +def workflow(root: Path, body: str, name: str = "ci.yml") -> Path: + wf = root / ".github" / "workflows" + wf.mkdir(parents=True, exist_ok=True) + path = wf / name + path.write_text(body, encoding="utf-8") + return path + + +def instances(root: Path, event: str = "") -> dict: + dag = build_dag([parse_workflow(p) for p in discover_workflows(root)], + event=event) + return {i.job_id: i for i in dag.instances} + + +class TempProject(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + +# ── Classification at DAG-build time ──────────────────────────────────────── + +class TestClassification(unittest.TestCase): + def test_three_classes(self): + hard, deferred, event = cw.classify_unresolved([ + "secrets.TOKEN", "needs.build.result", "steps.x.outputs.y", + "github.event.before", "vars.REGION", + ]) + self.assertEqual(hard, ["secrets.TOKEN", "vars.REGION"]) + self.assertEqual(deferred, ["needs.build.result", "steps.x.outputs.y"]) + self.assertEqual(event, ["github.event.before"]) + + def test_late_substitute_fills_needs_and_blanks_event_only_when_declared(self): + runtime = {"needs": {"a": {"result": "success", "outputs": {"v": "42"}}}, + "steps": {}} + text = "r=${{ needs.a.result }} v=${{ needs.a.outputs.v }} e=[${{ github.event.before }}]" + undeclared = cw.late_substitute(text, runtime, event_declared=False) + self.assertEqual(undeclared.unresolved, ["github.event.before"]) + declared = cw.late_substitute(text, runtime, event_declared=True) + self.assertEqual(declared.text, "r=success v=42 e=[]") + self.assertEqual(declared.unresolved, []) + + def test_late_substitute_reports_a_genuine_gap(self): + res = cw.late_substitute("${{ needs.a.outputs.missing }}", + {"needs": {"a": {"result": "success", "outputs": {}}}, + "steps": {}}, True) + self.assertEqual(res.unresolved, ["needs.a.outputs.missing"]) + + +class TestBuildTimeBlockers(TempProject): + BODY = """ +name: CI +on: [push, pull_request] +jobs: + changes: + runs-on: ubuntu-latest + outputs: + code: ${{ steps.decide.outputs.code }} + steps: + - id: decide + run: echo "base=${{ github.event.pull_request.base.sha }}"; echo "code=true" >> "$GITHUB_OUTPUT" + test: + needs: changes + if: needs.changes.outputs.code == 'true' + runs-on: ubuntu-latest + steps: + - run: echo "changes said ${{ needs.changes.result }}" + secretive: + runs-on: ubuntu-latest + steps: + - run: echo ${{ secrets.TOKEN }} +""" + + def test_needs_refs_are_not_blockers_on_either_engine(self): + by = instances(self.tmp_with(self.BODY)) + self.assertEqual(by["test"].blockers, []) + self.assertEqual(by["test"].act_blockers, []) + self.assertEqual(by["test"].needs, ["changes"]) + + def test_event_refs_block_native_until_the_event_is_declared_never_act(self): + by = instances(self.tmp_with(self.BODY)) + self.assertTrue(any("github.event.pull_request.base.sha" in b + for b in by["changes"].blockers)) + self.assertEqual(by["changes"].act_blockers, []) + declared = instances(self.tmp, event="pull_request") + self.assertEqual(declared["changes"].blockers, []) + + def test_secrets_still_block_both_engines(self): + by = instances(self.tmp_with(self.BODY)) + self.assertTrue(any("secrets.TOKEN" in b for b in by["secretive"].blockers)) + self.assertTrue(any("secrets.TOKEN" in b for b in by["secretive"].act_blockers)) + + def test_outputs_and_step_ids_survive_instantiation(self): + by = instances(self.tmp_with(self.BODY)) + self.assertEqual(by["changes"].outputs, + {"code": "${{ steps.decide.outputs.code }}"}) + self.assertEqual(by["changes"].steps[0].id, "decide") + + def tmp_with(self, body: str) -> Path: + workflow(self.tmp, body) + return self.tmp + + +# ── The derived single-job workflow for act ───────────────────────────────── + +class TestDeriveWorkflow(TempProject): + BODY = """ +name: probe +on: [push] +jobs: + a: + runs-on: ubuntu-latest + outputs: + val: ${{ steps.s.outputs.val }} + steps: + - id: s + run: echo "val=42" >> "$GITHUB_OUTPUT" + b: + needs: a + if: needs.a.outputs.val == '42' + runs-on: ubuntu-latest + strategy: + matrix: + n: [1, 2] + steps: + - run: echo "r=${{ needs.a.result }} v=${{ needs.a.outputs.val }} n=${{ matrix.n }}" +""" + + def test_holds_one_job_without_needs_or_if_and_spells_needs_out(self): + workflow(self.tmp, self.BODY) + b = next(i for i in instances(self.tmp).values() if i.job_id == "b") + text, unresolved = ci_act.derive_workflow( + b, {"a": {"result": "success", "outputs": {"val": "42"}}}) + self.assertEqual(unresolved, []) + derived = parse_workflow_text(text) + self.assertEqual(list(derived["jobs"]), ["b"]) + self.assertNotIn("needs", derived["jobs"]["b"]) + self.assertNotIn("if", derived["jobs"]["b"]) + run = derived["jobs"]["b"]["steps"][0]["run"] + self.assertIn("r=success v=42", run) + self.assertIn("${{ matrix.n }}", run) # act still expands matrix + self.assertIn("on", derived) # not the YAML-1.1 `True` + self.assertEqual(derived["name"], "probe") + + def test_an_output_c3_never_recorded_is_reported_not_blanked(self): + workflow(self.tmp, self.BODY) + b = next(i for i in instances(self.tmp).values() if i.job_id == "b") + text, unresolved = ci_act.derive_workflow( + b, {"a": {"result": "success", "outputs": {}}}) + self.assertEqual(unresolved, ["${{ needs.a.outputs.val }}"]) + + def test_build_command_prefers_the_derived_file(self): + workflow(self.tmp, self.BODY) + b = next(i for i in instances(self.tmp).values() if i.job_id == "b") + cmd = ci_act.build_command(b, self.tmp, "act", event="push", + workflow_file=r"C:\tmp\derived.yml") + self.assertIn("C:/tmp/derived.yml", cmd) + self.assertIn("-j", cmd) + self.assertNotIn(".github/workflows/ci.yml", " ".join(cmd)) + + def test_run_job_refuses_before_starting_act_on_an_unresolved_need(self): + workflow(self.tmp, self.BODY) + b = next(i for i in instances(self.tmp).values() if i.job_id == "b") + outcome = ci_act.run_job(b, self.tmp, act_path="act-that-does-not-exist", + needs_ctx={"a": {"result": "success", "outputs": {}}}) + self.assertEqual(outcome["unresolved"], ["${{ needs.a.outputs.val }}"]) + self.assertEqual(outcome["command"], "") + + +def parse_workflow_text(text: str) -> dict: + import yaml + return yaml.safe_load(text) + + +class TestPullPolicy(TempProject): + BODY = """ +name: CI +on: [push] +jobs: + a: + runs-on: ubuntu-latest + steps: + - run: echo hi +""" + + def test_an_image_already_on_disk_is_not_pulled(self): + from unittest import mock + workflow(self.tmp, self.BODY) + a = instances(self.tmp)["a"] + with mock.patch.object(ci_act, "image_present", return_value=True): + outcome = ci_act.run_job(a, self.tmp, act_path="act-that-does-not-exist") + self.assertIn("--pull=false", outcome["command"]) + + def test_a_missing_image_is_pulled(self): + from unittest import mock + workflow(self.tmp, self.BODY) + a = instances(self.tmp)["a"] + with mock.patch.object(ci_act, "image_present", return_value=False): + outcome = ci_act.run_job(a, self.tmp, act_path="act-that-does-not-exist") + self.assertIn("--pull=true", outcome["command"]) + + def test_each_run_gets_its_own_artifact_server_port(self): + # act binds 34567 by default; two runs on one box collide there. + from unittest import mock + workflow(self.tmp, self.BODY) + a = instances(self.tmp)["a"] + with mock.patch.object(ci_act, "image_present", return_value=True), \ + mock.patch.object(ci_act, "free_port", return_value=45678): + outcome = ci_act.run_job(a, self.tmp, act_path="act-that-does-not-exist", + artifact_dir=str(self.tmp / "art")) + self.assertIn("--artifact-server-port 45678", outcome["command"]) + self.assertNotIn("34567", outcome["command"]) + port = ci_act.free_port() + self.assertTrue(1024 <= port <= 65535) + + +class TestActLogOutputs(unittest.TestCase): + def test_reads_set_output_lines_last_write_wins(self): + log = ( + "[probe/a] ⭐ Run Main echo\n" + "[probe/a] ⚙ ::set-output:: val=41\n" + "[probe/a] ⚙ ::set-output:: val=42\n" + "[probe/a] ⚙ ::set-output:: name=with spaces \n" + "[probe/a] 🏁 Job succeeded\n" + ) + self.assertEqual(ci_act.job_outputs_from_log(log), + {"val": "42", "name": "with spaces"}) + + +# ── The native runner: outputs flow into `needs` ──────────────────────────── + +class TestNativeOutputs(TempProject): + def test_outputs_reach_dependents_in_if_and_in_run_text(self): + workflow(self.tmp, f""" +name: CI +on: [push] +jobs: + a: + runs-on: {LOCAL_RUNNER} + outputs: + flag: ${{{{ steps.s.outputs.flag }}}} + steps: + - id: s + run: echo "flag=yes" >> "$GITHUB_OUTPUT" + b: + needs: a + if: needs.a.outputs.flag == 'yes' + runs-on: {LOCAL_RUNNER} + steps: + - run: echo "B-SAW result=${{{{ needs.a.result }}}} flag=${{{{ needs.a.outputs.flag }}}}" + c: + needs: a + if: needs.a.outputs.flag == 'no' + runs-on: {LOCAL_RUNNER} + steps: + - run: echo never +""") + res = cr.run_ci(self.tmp, engine="native") + jobs = {j.job_id: j for j in res.jobs} + self.assertEqual(jobs["a"].status, cr.PASSED, jobs["a"].reason) + self.assertEqual(jobs["a"].outputs, {"flag": "yes"}) + self.assertEqual(jobs["b"].status, cr.PASSED, jobs["b"].reason) + log = Path(jobs["b"].log_path).read_text(encoding="utf-8") + self.assertIn("B-SAW result=success flag=yes", log) + self.assertEqual(jobs["c"].status, cr.SKIPPED_IF) + + def test_a_declared_output_nobody_wrote_is_refused_not_blanked(self): + workflow(self.tmp, f""" +name: CI +on: [push] +jobs: + a: + runs-on: {LOCAL_RUNNER} + outputs: + flag: ${{{{ steps.s.outputs.flag }}}} + steps: + - id: s + run: echo nothing +""") + res = cr.run_ci(self.tmp, engine="native") + job = res.jobs[0] + self.assertEqual(job.status, cr.UNSUPPORTED) + self.assertIn("flag", job.reason) + + def test_event_field_is_blank_only_under_a_declared_event(self): + workflow(self.tmp, f""" +name: CI +on: [push] +jobs: + a: + runs-on: {LOCAL_RUNNER} + steps: + - run: echo "before=[${{{{ github.event.before }}}}]" +""") + undeclared = cr.run_ci(self.tmp, engine="native") + self.assertEqual(undeclared.jobs[0].status, cr.UNSUPPORTED) + declared = cr.run_ci(self.tmp, engine="native", event="push") + self.assertEqual(declared.jobs[0].status, cr.PASSED, declared.jobs[0].reason) + log = Path(declared.jobs[0].log_path).read_text(encoding="utf-8") + self.assertIn("before=[]", log) + + def test_selecting_a_dependent_job_runs_what_it_needs(self): + # `--job smokes` used to deselect `changes`, and then the `if:` that + # reads changes' output was unjudgeable. + workflow(self.tmp, f""" +name: CI +on: [push] +jobs: + changes: + runs-on: {LOCAL_RUNNER} + outputs: + code: ${{{{ steps.d.outputs.code }}}} + steps: + - id: d + run: echo "code=true" >> "$GITHUB_OUTPUT" + smokes: + needs: changes + if: needs.changes.outputs.code == 'true' + runs-on: {LOCAL_RUNNER} + steps: + - run: echo smoked + unrelated: + runs-on: {LOCAL_RUNNER} + steps: + - run: echo not me +""") + res = cr.run_ci(self.tmp, engine="native", selector="smokes") + jobs = {j.job_id: j for j in res.jobs} + self.assertEqual(jobs["changes"].status, cr.PASSED, jobs["changes"].reason) + self.assertEqual(jobs["smokes"].status, cr.PASSED, jobs["smokes"].reason) + self.assertEqual(jobs["unrelated"].status, cr.DESELECTED) + + def test_jobs_that_publish_outputs_are_never_served_from_cache(self): + workflow(self.tmp, f""" +name: CI +on: [push] +jobs: + a: + runs-on: {LOCAL_RUNNER} + outputs: + flag: ${{{{ steps.s.outputs.flag }}}} + steps: + - id: s + run: echo "flag=yes" >> "$GITHUB_OUTPUT" +""") + cr.run_ci(self.tmp, engine="native") + second = cr.run_ci(self.tmp, engine="native") + self.assertEqual(second.jobs[0].status, cr.PASSED) + self.assertNotEqual(second.jobs[0].status, cr.CACHED) + self.assertEqual(second.jobs[0].outputs, {"flag": "yes"}) + + +if __name__ == "__main__": + unittest.main()