memory.dashboard.backend.routes.experiments

memory.dashboard.backend.routes.experiments

Experiment + rerun-job routes.

Split out of the former flat routes.py; behaviour is unchanged.

list_experiments

list_experiments(request: Request = None, response: Response = None)

Return experiment manifests joined with memory experiment entries.

Answers conditional GETs: an ETag from :func:_experiments_etag_key (the entry/manifest/peer fileset signature) lets an unchanged poll collapse to a bodyless 304, skipping the manifest read + entry join + serialization.

Source code in memory/dashboard/backend/routes/experiments.py

@router.get("/experiments")
def list_experiments(request: Request = None, response: Response = None):
    """Return experiment manifests joined with memory experiment entries.

    Answers conditional GETs: an ETag from :func:`_experiments_etag_key` (the
    entry/manifest/peer fileset signature) lets an unchanged poll collapse to a
    bodyless 304, skipping the manifest read + entry join + serialization.
    """
    # Refresh the loader from the local ``.memory/`` fileset FIRST, then compute
    # the ETag — so the validator derives from the SAME post-rebuild
    # ``loader.version`` the body does (mirrors ``get_tree``). Computing the ETag
    # first (over a pre-rebuild loader) meant a cold load (version 0.0) or a
    # version-change rebuild returned an ETag the next identical poll couldn't
    # match -> a missed 304. The rebuild also guards a delete/older-mtime swap the
    # mtime-gated reload misses from leaving a stale experiment entry in the body.
    _ensure_loader_local_fresh()

    try:
        graph = loader.graph
    except FileNotFoundError:
        graph = None

    # 304 short-circuit BEFORE the expensive body assembly (manifest read + entry
    # join + serialization). The key now reflects the post-refresh loader state.
    not_modified = _conditional_etag(request, response, _experiments_etag_key())
    if not_modified is not None:
        return not_modified

    entries: list[dict] = []
    if graph is not None:
        try:
            proj = _get_project()
            entries = list(graph.cypher(
                "MATCH (e:entry) WHERE e.project = $proj AND e.type = $type AND e.status = $st "
                "RETURN e.id AS id, e.title AS title, e.body AS body, e.tags AS tags, "
                "e.success AS success, e.parent_id AS parent_id, e.created_at AS created_at, "
                "e.updated_at AS updated_at",
                params={"proj": proj["id"], "type": "experiment", "st": "active"},
            ))
        except HTTPException:
            entries = []

    entries_by_id = {entry["id"]: entry for entry in entries}
    manifests = _read_experiment_manifests()
    experiments: list[dict] = []
    seen_entries: set[str] = set()
    local_meta = _local_repo_meta() if graph is not None else {
        "repo_id": "local",
        "repo_name": Path.cwd().name,
        "repo_path": str(Path.cwd()),
        "read_only": False,
    }

    for manifest in manifests:
        entry = _find_entry_for_manifest(manifest, entries, entries_by_id)
        memory_entry_id = manifest.get("memory_entry") or (entry["id"] if entry else "")
        if entry:
            seen_entries.add(entry["id"])
        manifest_data = manifest.get("manifest") or {}
        kind = str(manifest_data.get("kind") or "test")
        subtype = str(manifest_data.get("subtype") or "")
        initiated_by = str(manifest_data.get("initiated_by") or "agent")
        family = str(manifest_data.get("experiment") or manifest_data.get("rerun_of") or manifest["id"])
        experiments.append({
            "experiment_family": family,
            "rerun_of": str(manifest_data.get("rerun_of") or ""),
            "replication": str(manifest_data.get("replication") or ""),
            "executor": str(manifest_data.get("executor") or ""),
            "run_stamp": str(manifest_data.get("run_stamp") or ""),
            "id": manifest["id"],
            "title": manifest["title"] if manifest["title"] != manifest["id"] else (entry["title"] if entry else manifest["title"]),
            "status": manifest["status"],
            "created_at": manifest["created_at"] or (entry.get("created_at") if entry else ""),
            "updated_at": entry.get("updated_at") if entry else "",
            "command": manifest["command"],
            "git_commit": manifest["git_commit"],
            "dirty": manifest["dirty"],
            "artifact_count": manifest["artifact_count"],
            "artifacts": manifest["artifacts"],
            "manifest_path": manifest["manifest_path"],
            "memory_entry_id": memory_entry_id,
            "memory_entry_title": entry["title"] if entry else "",
            "remote_status": manifest["remote_status"],
            "summary": manifest["summary"] or (entry.get("body", "")[:240] if entry else ""),
            "tags": entry.get("tags", "") if entry else "",
            "source": "manifest",
            "manifest": manifest_data,
            "kind": kind,
            "subtype": subtype,
            "initiated_by": initiated_by,
            "notes": manifest["notes"],
            **local_meta,
        })

    for entry in entries:
        if entry["id"] in seen_entries:
            continue
        tags = entry.get("tags", "")
        subtype = _infer_subtype(tags)
        kind = "experiment" if subtype in RESEARCH_SUBTYPES else "test"
        experiments.append({
            "experiment_family": entry["id"],
            "rerun_of": "",
            "replication": "",
            "executor": "",
            "run_stamp": "",
            "id": entry["id"],
            "title": entry["title"],
            "status": _experiment_status_from_success(entry.get("success")),
            "created_at": entry.get("created_at", ""),
            "updated_at": entry.get("updated_at", ""),
            "command": "",
            "git_commit": "",
            "dirty": False,
            "artifact_count": 0,
            "artifacts": [],
            "manifest_path": "",
            "memory_entry_id": entry["id"],
            "memory_entry_title": entry["title"],
            "remote_status": "",
            "summary": entry.get("body", "")[:240],
            "tags": tags,
            "source": "memory",
            "manifest": {},
            "kind": kind,
            "subtype": subtype,
            "initiated_by": "user",
            "notes": _entry_notes(entry["id"]),
            **local_meta,
        })

    experiments.extend(_federated_experiment_records())
    cards = _group_experiment_runs(experiments)

    def sort_key(exp: dict) -> str:
        return exp.get("latest_run_at") or exp.get("created_at") or exp.get("updated_at") or ""

    cards.sort(key=sort_key, reverse=True)
    return {
        "experiments": [e for e in cards if e["kind"] == "experiment"],
        "tests": [e for e in cards if e["kind"] != "experiment"],
    }

update_experiment

update_experiment(record_id: str, updates: ExperimentUpdate)

Update researcher-editable fields (title, notes) on an experiment or test.

Manifest-backed records write to the manifest YAML; memory-backed records write to the entry file and patch the graph cache.

Source code in memory/dashboard/backend/routes/experiments.py

@router.patch("/experiments/{record_id}")
def update_experiment(record_id: str, updates: ExperimentUpdate):
    """Update researcher-editable fields (title, notes) on an experiment or test.

    Manifest-backed records write to the manifest YAML; memory-backed records
    write to the entry file and patch the graph cache.
    """
    from memory import storage

    if federation.split_namespace(record_id) is not None:
        raise HTTPException(status_code=403, detail="Federated experiments are read-only")

    if updates.title is None and updates.notes is None:
        raise HTTPException(status_code=400, detail="No updates provided")

    # Manifest-backed record: update the YAML file
    manifests_dir = Path.cwd() / artifacts.MANIFESTS_DIR
    if manifests_dir.exists():
        for path in manifests_dir.glob("*.yaml"):
            try:
                data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
            except (OSError, yaml.YAMLError):
                continue
            if not isinstance(data, dict) or str(data.get("id") or path.stem) != record_id:
                continue
            if updates.title is not None:
                data["title"] = updates.title
            if updates.notes is not None:
                data["notes"] = updates.notes
            path.write_text(
                yaml.safe_dump(data, sort_keys=False, allow_unicode=True),
                encoding="utf-8",
            )
            return {"id": record_id, "source": "manifest", "status": "updated"}

    # Memory-backed record: update the entry file + graph cache
    graph = loader.graph
    rows = list(graph.cypher(
        "MATCH (e:entry) WHERE e.id = $eid RETURN e.id AS id",
        params={"eid": record_id},
    ))
    if not rows:
        raise HTTPException(status_code=404, detail=f"Experiment not found: {record_id}")

    now = datetime.now(timezone.utc).isoformat()
    field_updates: dict = {"updated_at": now}
    if updates.title is not None:
        field_updates["title"] = updates.title
        field_updates["node_label"] = updates.title[:50]
    if updates.notes is not None:
        field_updates["notes"] = updates.notes

    if storage.update_entry_field(record_id, field_updates) is None:
        raise HTTPException(status_code=404, detail=f"Entry file not found: {record_id}")

    # Title changes must be reflected in the cached graph so the tree updates
    if updates.title is not None:
        full = list(graph.cypher(
            "MATCH (e:entry) WHERE e.id = $eid RETURN e.id AS id, e.type AS type, "
            "e.project AS project, e.parent_id AS parent_id, e.title AS title, "
            "e.node_label AS node_label, e.tags AS tags, e.status AS status, "
            "e.success AS success, e.files AS files, e.open_threads AS open_threads, "
            "e.external_plan AS external_plan, e.session_id AS session_id, "
            "e.body AS body, e.created_at AS created_at",
            params={"eid": record_id},
        ))
        if full:
            updated = {
                **full[0],
                "title": updates.title,
                "node_label": updates.title[:50],
                "updated_at": now,
            }
            df = pd.DataFrame([updated])
            graph.add_nodes(df, node_type="entry", unique_id_field="id", node_title_field="title")
            atomic_graph_save(graph, MEMORY_FILE)
            loader._mtime = MEMORY_FILE.stat().st_mtime

    return {"id": record_id, "source": "memory", "status": "updated"}

enqueue_rerun

enqueue_rerun(record_id: str)

Enqueue a cloud rerun job for a manifest-backed experiment.

Source code in memory/dashboard/backend/routes/experiments.py

@router.post("/experiments/{record_id}/rerun")
def enqueue_rerun(record_id: str):
    """Enqueue a cloud rerun job for a manifest-backed experiment."""
    from memory import jobs

    if federation.split_namespace(record_id) is not None:
        raise HTTPException(status_code=403, detail="Federated experiments are read-only")

    try:
        # Cards carry the latest run's id, which may be a failed run with no
        # usable provenance — resolve back to the nearest successful run.
        manifest = artifacts.resolve_rerun_target(record_id, Path.cwd())
    except FileNotFoundError:
        raise HTTPException(
            status_code=404,
            detail=f"No manifest for {record_id}; only manifest-backed experiments can be rerun.",
        )
    if not str(manifest.get("git_commit") or "").strip():
        raise HTTPException(status_code=409, detail="Manifest has no git_commit; cannot sandbox-rerun.")
    if not str(manifest.get("command") or "").strip():
        raise HTTPException(status_code=409, detail="Manifest has no command; cannot rerun.")

    queue = _job_queue()
    resolved_id = str(manifest.get("id") or record_id)
    family = str(manifest.get("experiment") or manifest.get("rerun_of") or resolved_id)

    # Debounce: if an active job already exists for this family, return it
    # instead of stacking a duplicate (rapid double-clicks, laggy polling).
    try:
        active = [
            job for job in queue.find_jobs(states=("pending", "running"))
            if family in (str(job.get("experiment") or ""), str(job.get("experiment_id") or ""))
            and not _annotate_job_staleness(dict(job)).get("stale")
        ]
    except Exception:
        active = []
    if active:
        existing = active[0]
        return {
            "job_id": existing["job_id"],
            "experiment_id": resolved_id,
            "state": str(existing.get("state") or "pending"),
            "queue": queue.base,
            "deduplicated": True,
        }

    job = jobs.build_rerun_job(manifest, requested_by="dashboard")
    try:
        queue.enqueue(job)
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"Failed to enqueue rerun job: {exc}")
    _jobs_response_cache.clear()
    return {
        "job_id": job["job_id"],
        "experiment_id": resolved_id,
        "state": "pending",
        "queue": queue.base,
    }

list_experiment_jobs

list_experiment_jobs(record_id: str)

Return rerun jobs for an experiment family, so job state survives navigation.

Source code in memory/dashboard/backend/routes/experiments.py

@router.get("/experiments/{record_id}/jobs")
def list_experiment_jobs(record_id: str):
    """Return rerun jobs for an experiment family, so job state survives navigation."""

    def build() -> dict:
        queue = _job_queue()
        family = record_id
        try:
            manifest = artifacts.read_manifest(record_id, Path.cwd())
            family = str(manifest.get("experiment") or record_id)
        except (FileNotFoundError, ValueError, OSError):
            pass
        matches = [_annotate_job_staleness(job) for job in queue.find_jobs(experiment=family)]
        active = next(
            (job for job in matches if job["state"] in ("pending", "running") and not job.get("stale")),
            None,
        )
        return {
            "experiment_id": record_id,
            "family": family,
            "active": active,
            "latest": matches[0] if matches else None,
            "jobs": matches[:10],
        }

    return _jobs_cached(f"experiment:{record_id}", build)

list_active_jobs

list_active_jobs()

Return all pending/running rerun jobs (for in-flight badges in the list view).

Source code in memory/dashboard/backend/routes/experiments.py

@router.get("/jobs")
def list_active_jobs():
    """Return all pending/running rerun jobs (for in-flight badges in the list view)."""

    def build() -> dict:
        queue = _job_queue()
        matches = [
            _annotate_job_staleness(job)
            for job in queue.find_jobs(states=("pending", "running"))
        ]
        return {"jobs": [job for job in matches if not job.get("stale")]}

    return _jobs_cached("active", build)

get_job_status

get_job_status(job_id: str, tail: int = 8000)

Return job state plus a log tail for live status in the UI.

Source code in memory/dashboard/backend/routes/experiments.py

@router.get("/jobs/{job_id}")
def get_job_status(job_id: str, tail: int = 8000):
    """Return job state plus a log tail for live status in the UI."""
    queue = _job_queue()
    try:
        found = queue.get_job(job_id)
        log = queue.read_log(job_id, tail_chars=tail)
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"Failed to read job queue: {exc}")
    if found is None:
        raise HTTPException(status_code=404, detail=f"Job not found: {job_id}")
    state, job = found
    return {"job_id": job_id, "state": state, "job": job, "log": log}