Add the giant/workflow b2luigi task graph (gitea #83)

One workflow TOML now parameterises a whole experiment and `giant workflow
run <spec.toml>` turns it into a b2luigi DAG whose targets are files on
/ceph: nothing already produced is recomputed, every step waits for its
inputs, and HTCondor submission/polling is b2luigi's job.

- spec.py: workflow TOML -> frozen dataclasses with name-uniqueness and
  cross-reference validation, unknown keys rejected the way giant.config
  rejects them, and a short spec_hash per task that folds in its transitive
  parents — so an edited spec re-runs exactly the affected subtree.
- htcondor.py: the CPU/GPU submit settings. The GPU requirement strings
  (ProvidesEtpCeph + optional device/memory pins) are ported from the
  condor-gpu-train-rollout branch rather than rewritten.
- tasks.py: DatasetTask, WarmCacheTask, GeometryOracleTask, TrainEpochTask
  (one short GPU job per epoch, chained via --resume, which the training
  loop already supports unchanged), TrainTask (publishes best.pt/last.pt and
  a concatenated metrics.csv so downstream never sees the epoch fan-out),
  RolloutTask, AnalysisPrepTask, AnalysisComputeTask (one job per plot x
  chunk, walltime sized from run_meta.json at submit time), AnalysisRenderTask
  (always local — the only step importing plotstyle/LaTeX), WorkflowTask.
  Task bodies call the existing entry points; none of them reimplement
  anything.
- run.py + `giant workflow run`: settings wiring and the script b2luigi
  re-executes on workers. add_filename_to_cmd is off because b2luigi passes
  only the script's basename, and --spec is forwarded via
  task_cmd_additional_args so a worker resolves the identical task graph.

configs/workflow_example.toml is the documented starting point.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/test_workflow_spec.py

@@ -0,0 +1,148 @@
+"""Workflow spec parsing, validation, and spec hashes (gitea #83)."""
+
+import pytest
+
+from giant.workflow.spec import (
+    WorkflowSpecError,
+    epoch_milestones,
+    load_spec,
+    parse_spec,
+    spec_hash,
+)
+
+MINIMAL = {
+    "workflow": {"name": "wf", "result_dir": "/tmp/wf"},
+    "condor": {"accounting_group": "cms", "repo_dir": "/work/lbogner/giant"},
+    "dataset": {"steps": "/data/train", "reference": "/data/holdout"},
+    "train": [{"name": "a", "epochs": 3}],
+    "rollout": [{"name": "a", "train": "a"}],
+    "analysis": [{"name": "cmp", "rollouts": ["a"], "chunks": 4}],
+}
+
+
+def _spec(**patch):
+    raw = {k: (v.copy() if isinstance(v, dict) else list(v)) for k, v in MINIMAL.items()}
+    raw.update(patch)
+    return parse_spec(raw)
+
+
+def test_parses_minimal_spec():
+    spec = _spec()
+    assert spec.name == "wf"
+    assert spec.log_dir == "/tmp/wf/logs"  # derived from result_dir
+    assert spec.train("a").epochs == 3
+    assert spec.rollout("a").train == "a"
+    assert spec.analysis("cmp").rollouts == ("a",)
+    # defaults come from the dataclasses, not the file
+    assert spec.geometry.method == "slab"
+    assert spec.condor.docker_image_gpu == "mschnepf/slc7-condocker"
+
+
+def test_example_config_is_valid():
+    spec = load_spec("configs/workflow_example.toml")
+    assert {t.name for t in spec.trains} == {"baseline", "router-balanced"}
+    assert spec.analysis("baseline-vs-router").rollouts == ("baseline", "router-balanced")
+
+
+@pytest.mark.parametrize(
+    "patch, message",
+    [
+        ({"train": [{"name": "a"}, {"name": "a"}]}, "unique"),
+        ({"rollout": [{"name": "r", "train": "nope"}]}, "names no"),
+        ({"analysis": [{"name": "c", "rollouts": ["nope"]}]}, "not defined"),
+        ({"analysis": [{"name": "c", "rollouts": []}]}, "at least one"),
+        ({"analysis": [{"name": "c", "rollouts": ["a"], "chunks": 0}]}, "chunks must be"),
+        ({"train": [{"name": "a", "epochs": 0}]}, "epochs must be"),
+        ({"train": [{"name": "a", "epchs": 3}]}, "unknown key"),
+        ({"geometry": {"methd": "slab"}}, "unknown key"),
+    ],
+)
+def test_validation_errors(patch, message):
+    with pytest.raises(WorkflowSpecError, match=message):
+        _spec(**patch)
+
+
+def test_unknown_top_level_table_rejected():
+    with pytest.raises(WorkflowSpecError, match="unknown top-level"):
+        _spec(nonsense={})
+
+
+def test_missing_required_table_rejected():
+    raw = {k: v for k, v in MINIMAL.items() if k != "dataset"}
+    with pytest.raises(WorkflowSpecError, match=r"missing required \[dataset\]"):
+        parse_spec(raw)
+
+
+def test_unknown_lookup_names_are_explicit():
+    spec = _spec()
+    with pytest.raises(WorkflowSpecError, match="no \\[\\[train\\]\\] named 'zzz'"):
+        spec.train("zzz")
+
+
+def test_hash_is_stable_and_order_independent():
+    a = _spec()
+    b = parse_spec(
+        {
+            "dataset": MINIMAL["dataset"],
+            "condor": MINIMAL["condor"],
+            "workflow": MINIMAL["workflow"],
+            "train": MINIMAL["train"],
+            "rollout": MINIMAL["rollout"],
+            "analysis": MINIMAL["analysis"],
+        }
+    )
+    assert a.train_hash("a") == b.train_hash("a")
+    assert a.analysis_hash("cmp") == b.analysis_hash("cmp")
+    assert len(a.train_hash("a")) == 8
+
+
+def test_hash_changes_with_own_settings():
+    base = _spec()
+    changed = _spec(train=[{"name": "a", "epochs": 4}])
+    assert base.train_hash("a") != changed.train_hash("a")
+
+
+def test_hash_propagates_from_parents():
+    """A dataset change must move every downstream task's directory."""
+    base = _spec()
+    changed = _spec(dataset={"steps": "/data/other", "reference": "/data/holdout"})
+    assert base.train_hash("a") != changed.train_hash("a")
+    assert base.rollout_hash("a") != changed.rollout_hash("a")
+    assert base.analysis_hash("cmp") != changed.analysis_hash("cmp")
+
+    # ... and so must a change to a training the analysis transitively uses.
+    retrained = _spec(train=[{"name": "a", "epochs": 9}])
+    assert retrained.analysis_hash("cmp") != base.analysis_hash("cmp")
+    # while an unrelated knob on the analysis leaves the training alone
+    rebinned = _spec(analysis=[{"name": "cmp", "rollouts": ["a"], "chunks": 4, "bins": 99}])
+    assert rebinned.train_hash("a") == base.train_hash("a")
+    assert rebinned.analysis_hash("cmp") != base.analysis_hash("cmp")
+
+
+def test_warm_cache_hash_ignores_epochs():
+    """Epoch count doesn't change the setup cache, so it must not re-warm it."""
+    base = _spec()
+    longer = _spec(train=[{"name": "a", "epochs": 50}])
+    assert base.warm_cache_hash("a") == longer.warm_cache_hash("a")
+    other_cfg = _spec(train=[{"name": "a", "epochs": 3, "config": "configs/router.toml"}])
+    assert base.warm_cache_hash("a") != other_cfg.warm_cache_hash("a")
+
+
+def test_spec_hash_expands_dataclasses():
+    spec = _spec()
+    assert spec_hash(spec.dataset) == spec_hash(spec.dataset)
+    assert spec_hash(spec.dataset) != spec_hash(spec.geometry)
+
+
+@pytest.mark.parametrize(
+    "epochs, per_job, expected",
+    [
+        (3, 1, [1, 2, 3]),
+        (10, 3, [3, 6, 9, 10]),
+        (9, 3, [3, 6, 9]),
+        (1, 5, [1]),
+    ],
+)
+def test_epoch_milestones(epochs, per_job, expected):
+    spec = _spec(train=[{"name": "a", "epochs": epochs, "epochs_per_job": per_job}])
+    assert epoch_milestones(spec.train("a")) == expected

tests/test_workflow_tasks.py

@@ -0,0 +1,172 @@
+"""Workflow task graph: dependencies, output paths, condor settings (gitea #83)."""
+
+import pytest
+
+from giant.analysis.catalog import catalog_ids, get_spec as get_plot_spec
+from giant.workflow import tasks
+from giant.workflow.spec import parse_spec
+
+CONDOR = {
+    "accounting_group": "cms",
+    "repo_dir": "/work/lbogner/giant",
+    "env_script": "/work/lbogner/giant/condor_env.sh",
+}
+
+RAW = {
+    "workflow": {"name": "wf", "result_dir": "/results/wf"},
+    "condor": CONDOR,
+    "dataset": {"steps": "/data/train", "reference": "/data/holdout"},
+    "train": [
+        {"name": "base", "epochs": 3, "gpu_memory_mb": 20000},
+        {"name": "router", "epochs": 2},
+    ],
+    "rollout": [
+        {"name": "base", "train": "base"},
+        {"name": "router", "train": "router"},
+    ],
+    "analysis": [{"name": "cmp", "rollouts": ["base", "router"], "chunks": 4}],
+}
+
+
+@pytest.fixture
+def spec():
+    s = parse_spec(RAW)
+    tasks.set_spec(s)
+    return s
+
+
+def _requires(task):
+    return list(task.requires() or [])
+
+
+def test_epoch_chain_is_linear_and_rooted_at_warm_cache(spec):
+    h = spec.train_hash("base")
+    third = tasks.TrainEpochTask(name="base", spec_hash=h, milestone=3)
+    second = _requires(third)
+    assert [type(t) for t in second] == [tasks.TrainEpochTask]
+    assert second[0].milestone == 2
+    first = _requires(second[0])[0]
+    assert first.milestone == 1
+    root = _requires(first)
+    assert [type(t) for t in root] == [tasks.WarmCacheTask]
+    # the warm cache is keyed by its own hash, not the training's
+    assert root[0].spec_hash == spec.warm_cache_hash("base")
+
+
+def test_epoch_task_outputs_last_pt_per_milestone(spec):
+    h = spec.train_hash("base")
+    path = tasks.TrainEpochTask(name="base", spec_hash=h, milestone=2).output().path
+    assert path == f"/results/wf/train_epoch/name=base/spec_hash={h}/epochs=2/last.pt"
+
+
+def test_train_task_requires_final_epoch_and_publishes_canonical_outputs(spec):
+    h = spec.train_hash("base")
+    train = tasks.TrainTask(name="base", spec_hash=h)
+    (dep,) = _requires(train)
+    assert isinstance(dep, tasks.TrainEpochTask) and dep.milestone == 3
+    out = train.output()
+    assert set(out) == {"best.pt", "last.pt", "metrics.csv"}
+    assert out["best.pt"].path == f"/results/wf/train/name=base/spec_hash={h}/best.pt"
+    # local: it only copies files around, no reason to queue a job for it
+    assert train.batch_system == "local"
+
+
+def test_rollout_requires_training_geometry_and_reference(spec):
+    ro = tasks.RolloutTask(name="base", spec_hash=spec.rollout_hash("base"))
+    deps = _requires(ro)
+    assert [type(d) for d in deps] == [tasks.TrainTask, tasks.GeometryOracleTask, tasks.DatasetTask]
+    assert deps[0].name == "base"
+    assert deps[2].path == "/data/holdout"
+    out = ro.output()
+    assert out["rollout.yaml"].path.endswith("rollout.yaml")
+    # the sidecar sits next to the parquet — the deterministic path
+    # `giant rollout --out` now produces
+    assert out["rollout.yaml"].path[: -len(".yaml")] == out["rollout.parquet"].path[: -len(".parquet")]
+
+
+def test_analysis_prep_requires_every_named_rollout(spec):
+    prep = tasks.AnalysisPrepTask(name="cmp", spec_hash=spec.analysis_hash("cmp"))
+    deps = _requires(prep)
+    assert [d.name for d in deps] == ["base", "router"]
+    assert all(isinstance(d, tasks.RolloutTask) for d in deps)
+    assert prep.batch_system == "local"
+
+
+def test_compute_job_enumeration_collapses_non_chunkable_specs(spec):
+    jobs = tasks.analysis_jobs(spec, "cmp")
+    non_chunkable = [i for i in catalog_ids() if not get_plot_spec(i).chunkable]
+    expected = (len(catalog_ids()) - len(non_chunkable)) * 4 + len(non_chunkable)
+    assert len(jobs) == expected
+    assert non_chunkable, "expected some chunkable=False specs in the catalog"
+    for spec_id in non_chunkable:
+        assert [c for i, c in jobs if i == spec_id] == [0]
+
+
+def test_compute_output_matches_the_on_disk_contract(spec):
+    h = spec.analysis_hash("cmp")
+    task = tasks.AnalysisComputeTask(name="cmp", spec_hash=h, plot_id="event_mean_length", chunk=2)
+    assert task.output().path == (
+        f"/results/wf/analysis/name=cmp/spec_hash={h}/reduced_partial/event_mean_length__2.json"
+    )
+    (dep,) = _requires(task)
+    assert isinstance(dep, tasks.AnalysisPrepTask)
+
+
+def test_render_requires_every_compute_job_and_runs_locally(spec):
+    render = tasks.AnalysisRenderTask(name="cmp", spec_hash=spec.analysis_hash("cmp"))
+    deps = _requires(render)
+    assert len(deps) == len(tasks.analysis_jobs(spec, "cmp"))
+    assert render.batch_system == "local"  # the only step importing plotstyle/LaTeX
+    assert render.output().path.endswith("/plots/metadata.yaml")
+
+
+def test_workflow_task_wraps_every_analysis(spec):
+    deps = _requires(tasks.WorkflowTask(workflow_name="wf"))
+    assert [(type(d), d.name) for d in deps] == [(tasks.AnalysisRenderTask, "cmp")]
+
+
+def test_workflow_without_analysis_falls_back_to_rollouts():
+    raw = {k: v for k, v in RAW.items() if k != "analysis"}
+    tasks.set_spec(parse_spec(raw))
+    deps = _requires(tasks.WorkflowTask(workflow_name="wf"))
+    assert [type(d) for d in deps] == [tasks.RolloutTask, tasks.RolloutTask]
+
+
+def test_gpu_settings_carry_remote_ceph_and_pins(spec):
+    settings = tasks.TrainEpochTask(name="base", spec_hash=spec.train_hash("base"), milestone=1).htcondor_settings
+    assert settings["+RemoteJob"] == "True"
+    assert settings["RequestGPUs"] == 1
+    assert "TARGET.ProvidesEtpCeph =?= True" in settings["requirements"]
+    assert "TARGET.GPUs_GlobalMemoryMb >= 20000" in settings["requirements"]
+    assert settings["accounting_group"] == "cms"
+    assert settings["docker_image"] == "mschnepf/slc7-condocker"
+
+
+def test_cpu_settings_used_for_analysis_compute(spec):
+    task = tasks.AnalysisComputeTask(
+        name="cmp", spec_hash=spec.analysis_hash("cmp"), plot_id="event_mean_length", chunk=0
+    )
+    settings = task.htcondor_settings
+    assert settings["docker_image"] == "cverstege/alma9-gridjob"
+    assert "RequestGPUs" not in settings
+    # no run_meta.json yet (prep hasn't run), so no walltime is claimed
+    assert "+RequestWalltime" not in settings
+
+
+def test_cpu_settings_local_files_use_provides_etp_resources():
+    raw = {**RAW, "condor": {**CONDOR, "remote": False}}
+    spec = parse_spec(raw)
+    tasks.set_spec(spec)
+    settings = tasks.GeometryOracleTask(spec_hash=spec.geometry_hash()).htcondor_settings
+    assert settings["requirements"] == "TARGET.ProvidesETPResources"
+    assert "+RemoteJob" not in settings
+
+
+def test_missing_dataset_fails_immediately(spec):
+    with pytest.raises(FileNotFoundError, match="/ceph"):
+        tasks.DatasetTask(path="/data/train").complete()
+
+
+def test_dataset_that_exists_is_complete(tmp_path, spec):
+    (tmp_path / "steps.parquet").write_text("")
+    assert tasks.DatasetTask(path=str(tmp_path / "steps.parquet")).complete()