feat(ci): AgentCI owns the DAG — job outputs, run-time needs.*, one job per act (2.130.0)

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.<job>.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<<EOF` forms), a job's `outputs:` mapping is resolved from
+  `steps.<id>.outputs.<k>`, and dependents read `needs.<job>.outputs.<name>`
+  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 <job>` 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

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):

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"

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 <job>` 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