Skip to content

memory.runner

memory.runner

angelo-runner: poll the job queue and execute experiment reruns.

Runs on the machine that has compute and DVC/S3 access (typically the AWS box). Claims jobs from the S3 queue, executes sandbox reruns via memory.artifacts.rerun_experiment, streams logs back to the queue, and pushes the resulting run manifest to the shared git remote so every machine sees the new run after a pull.

Usage::

angelo-runner --repo /path/to/angelo [--interval 15] [--once]

process_job 

process_job(queue: JobQueue, job: dict[str, Any], repo_root: Path) -> dict[str, Any]

Execute one claimed job and report the result to the queue.

Source code in memory/runner.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def process_job(queue: JobQueue, job: dict[str, Any], repo_root: Path) -> dict[str, Any]:
    """Execute one claimed job and report the result to the queue."""

    job_id = str(job.get("job_id") or "")

    def log(message: str) -> None:
        text = message if message.endswith("\n") else message + "\n"
        queue.append_log(job_id, f"[{_now()}] {text}")
        queue.heartbeat(job_id)

    try:
        kind = str(job.get("kind") or "")
        if kind != "rerun":
            raise ValueError(f"Unsupported job kind: {kind!r}")
        experiment_id = str(job.get("experiment_id") or "")
        if not experiment_id:
            raise ValueError("Job is missing experiment_id.")

        log(f"Claimed job {job_id} for experiment {experiment_id}")

        # Sync so the manifest referenced by the job is present locally.
        sync = artifacts._run(("git", "pull", "--ff-only"), cwd=repo_root, check=False)
        if not sync.ok:
            log(f"git pull --ff-only failed (continuing with local state): {sync.stderr.strip()}")

        outcome = artifacts.rerun_experiment(
            experiment_id,
            repo_root,
            executor=str(job.get("executor") or "aws-runner"),
            dry_run=False,
            log_fn=log,
        )

        _publish_run(repo_root, outcome, log)

        result = {
            "status": outcome["status"],
            "replication": outcome["replication"],
            "manifest_path": outcome["manifest_path"],
            "run_id": str(outcome["manifest"].get("id") or ""),
            "artifact_state": outcome.get("artifact_state", ""),
        }
        if outcome["status"] == "success":
            queue.complete(job_id, result)
            log(f"Job {job_id} done (replication: {outcome['replication']})")
        else:
            queue.fail(job_id, "Rerun command exited non-zero.", result)
            log(f"Job {job_id} failed: command exited non-zero")
        return result
    except Exception as exc:
        detail = f"{exc}\n{traceback.format_exc()}"
        queue.append_log(job_id, f"[{_now()}] ERROR: {detail}\n")
        queue.fail(job_id, str(exc))
        return {"status": "failed", "error": str(exc)}