Run the value-learning API workflow

Use this Python standard-library client to fit an initial Adapt-1 value model, stage a candidate, inspect predictions, and request promotion with your application’s measured outcomes. You supply the task’s configuration, observations, fitting options, and evaluation data as JSON files. The example uses target or outcome observations. It warm-starts the later candidate from the installed model. Native continuation rows have a separate installation restriction; this example does not demonstrate that installation path.

Before integration, check /version at the API host root and confirm the response shapes for your deployment. The client checks training.request_id, the installed and candidate model hashes, numeric predictions, and evaluator.model_sha256. It stops when these checks fail.

Prepare your environment and files

Create a fresh Domain using the Domain setup workflow, then set these environment variables. Use the same dedicated Domain for all three commands.

Environment variable	Value
`ADAPT1_API_BASE`	The deployment’s API base, including `/api/v1`, such as `https://rei-neuroadapt-api.reilabs.org/api/v1`
`ADAPT1_API_KEY`	The API key for that deployment
`ADAPT1_DOMAIN`	The existing Domain ID

Prepare these files from your application. The JSON files contain the object or array described below, without a session_id wrapper.

File	Contents
`value-config.json`	The `config` object from Configure, including your `relation`, `policy`, and representation. Set retention capacity for the complete selected dataset.
`initial-observations.json`	An array of complete observation rows, each with stable IDs, state inputs, and exactly one `target` or `outcome`.
`all-observations.json`	The original rows, unchanged, plus the new observations selected for the candidate fit. Identical replay uses the same `record_id`; a revised row needs a new ID.
`fit-options.json`	The `options` object from Train, with an explicit `model_type` and your fitting and validation settings. Omit `warm_start` and `supervision_scope`: the command sets those explicitly.
`probe-inputs.json`	Between 1 and 256 value inputs from your application, using the same encoding as the observations. Omit observation IDs, metadata, and labels.

Use informative observations and a suitable episode split. Four rows is the minimum for the explicit observation_ids array used here; it does not guarantee that a model passes validation. Choose a compatible representation and model and retain the same input meanings across the workflow. The client validates input-file roles, IDs, supervision, finite numeric values, and request sizes before writes. Deployment-specific schema and model checks still apply. Observation requests contain at most 50 rows and use a 900 KiB client margin. The 60-second request timeout and default 600-second fit wait are client settings, not service limits.

Save the Python client

Save the following code as value_client.py. Each command requires a new output directory. It saves request bodies, response receipts, and the last inspected state, without recording the authorization header. These local files contain your application data; redact them before sharing. It sends each mutation once and stops on a refusal or uncertain response.

value_client.py

import argparse
import json
import math
import os
from pathlib import Path
import re
import time
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request, urlopen
from uuid import uuid4

BODY_LIMIT = 900 * 1024  # Client margin, not a declared API quota.
INPUT_FIELDS = {"objects", "relations", "conditions", "context", "policy_features",
                "include_positions", "include_moments"}


def require(condition, message):
    if not condition:
        raise RuntimeError(message)


def finite(value):
    return type(value) in (int, float) and math.isfinite(value)


def unit(value):
    return finite(value) and 0 <= value <= 1


def encode(value):
    return json.dumps(value, ensure_ascii=False, allow_nan=False,
                      separators=(",", ":")).encode("utf-8")


def load(path):
    require(path is not None, "A required input-file argument is missing")
    value = json.loads(Path(path).read_text(encoding="utf-8"))
    encode(value)  # Reject NaN and infinity before making requests.
    return value


def model_hash(summary):
    require(isinstance(summary, dict), "Missing model summary")
    value = summary.get("model_sha256")
    require(isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value),
            "Missing or invalid model SHA-256")
    return value


def descriptor(value):
    require(isinstance(value, dict) and set(value) <= INPUT_FIELDS, "Invalid value input")
    for field, limit in (("conditions", 16), ("context", 16), ("policy_features", 4096)):
        fields = value.get(field, {})
        require(isinstance(fields, dict) and len(fields) <= limit and
                all(finite(v) for v in fields.values()), f"Invalid {field} numeric map")
    for field in ("include_positions", "include_moments"):
        require(field not in value or type(value[field]) is bool, f"Invalid {field}")


def sized(body):
    payload = {"session_id": "ignored", **body}
    require(len(encode(payload)) <= BODY_LIMIT, "Request exceeds the client body margin")
    return payload


def batches(rows):
    result, batch = [], []
    for row in rows:
        sized({"observations": [row]})
        proposed = batch + [row]
        if batch and (len(proposed) > 50 or
                      len(encode({"session_id": "ignored", "observations": proposed})) > BODY_LIMIT):
            result.append(batch)
            batch = []
        batch.append(row)
    if batch:
        result.append(batch)
    return result


def training_inputs(args):
    rows, options, inputs = load(args.observations), load(args.options), load(args.inputs)
    require(isinstance(rows, list) and 4 <= len(rows) <= 20000, "Select 4 to 20000 rows")
    ids = []
    for row in rows:
        require(isinstance(row, dict), "Observation must be an object")
        for field in ("record_id", "episode_id"):
            require(isinstance(row.get(field), str) and 1 <= len(row[field]) <= 256,
                    f"Invalid {field}")
        ids.append(row["record_id"])
        require(finite(row.get("step")) and row["step"] >= 0, "Invalid observation step")
        labels = set(row) & {"target", "outcome", "continuation"}
        require(labels in ({"target"}, {"outcome"}), "Use one target or outcome; no continuation rows")
        descriptor({k: v for k, v in row.items() if k in INPUT_FIELDS})
        require(set(row) <= INPUT_FIELDS | {"record_id", "episode_id", "step", "metadata"} | labels,
                "Unexpected observation field")
        if "target" in row:
            require(unit(row["target"]), "Target must be a finite number in [0,1]")
        else:
            outcome = row["outcome"]
            require(isinstance(outcome, dict) and unit(outcome.get("value")) and
                    finite(outcome.get("terminal_step")) and outcome["terminal_step"] >= 0 and
                    all(unit(outcome[k]) for k in ("discount", "neutral") if k in outcome),
                    "Invalid observed outcome")
    require(len(set(ids)) == len(ids), "Each selected record_id must be unique")
    require(isinstance(options, dict) and options.get("model_type") in
            ("linear", "mlp_v2", "object_value"), "Specify the intended model_type in options")
    require(not ({"warm_start", "supervision_scope"} & set(options)),
            "This client sets warm_start and supervision_scope; omit them from options")
    require(isinstance(inputs, list) and 1 <= len(inputs) <= 256, "Supply 1 to 256 probe inputs")
    for value in inputs:
        descriptor(value)
    options = {**options, "warm_start": args.phase == "candidate", "supervision_scope": "all"}
    # Fixed-length sizing placeholders are never submitted.
    sized({"inputs": inputs, "model_source": "candidate", "expected_model_sha256": "0" * 64})
    sized({"options": options, "observation_ids": ids, "request_id": "0" * 32,
           "install": False, "warm_start_source": "installed", "expected_initial_model_sha256": "0" * 64})
    return rows, options, inputs, ids, batches(rows)


class Client:
    def __init__(self, output, opener=urlopen):
        self.base = os.environ["ADAPT1_API_BASE"].rstrip("/")
        self.domain, self.key = os.environ["ADAPT1_DOMAIN"], os.environ["ADAPT1_API_KEY"]
        parsed = urlsplit(self.base)
        require(parsed.scheme == "https" and parsed.netloc and not parsed.username and
                not parsed.password and not parsed.query and not parsed.fragment,
                "Use an HTTPS API base without credentials, query, or fragment")
        require(self.domain and self.key, "Domain and API key must be nonempty")
        self.url = self.base + "/domains/" + quote(self.domain, safe="") + "/value/"
        self.output, self.opener, self.sequence = Path(output), opener, 0
        self.output.mkdir(parents=True, exist_ok=False)

    def save(self, name, value):
        (self.output / name).write_bytes(encode(value))

    def post(self, operation, body=None, expected=200, timeout=60):
        payload = sized(body or {})
        self.sequence += 1
        stem = f"{self.sequence:04d}-{operation}"
        self.save(stem + ".request.json", payload)  # Headers and API key are never journaled.
        request = Request(self.url + operation, data=encode(payload), method="POST", headers={
            "Authorization": "Bearer " + self.key, "Content-Type": "application/json",
            "User-Agent": "adapt1-value-example/1.0"})
        try:
            with self.opener(request, timeout=timeout) as response:
                status, raw = response.status, response.read().decode("utf-8")
        except HTTPError as error:
            status, raw = error.code, error.read().decode("utf-8", errors="replace")
        except (URLError, TimeoutError, OSError) as error:
            self.save("last-response.json", {"operation": operation, "error_type": type(error).__name__})
            raise RuntimeError("Uncertain request outcome; inspect the journal and value/state") from None
        receipt = {"operation": operation, "http_status": status, "body": raw}
        self.save(stem + ".response.json", receipt)
        self.save("last-response.json", receipt)
        require(status == expected, f"{operation}: HTTP {status}; inspect last-response.json")
        value = json.loads(raw)
        require(isinstance(value, dict), "Expected a JSON response object")
        return value

    def state(self, timeout=60):
        value = self.post("state", timeout=timeout)
        self.save("last-state.json", value)
        require(value.get("domain_id") == self.domain, "State belongs to a different Domain")
        return value

    def wait(self, request_id, seconds):
        deadline = time.monotonic() + seconds
        while True:
            remaining = deadline - time.monotonic()
            require(remaining > 0, "Client wait ended; inspect value/state before another write")
            state = self.state(timeout=min(60, remaining))
            job = state.get("training")
            require(isinstance(job, dict) and job.get("request_id") == request_id,
                    "Unexpected training job; inspect state instead of resubmitting")
            status = job.get("status")
            require(status in ("queued", "running", "trained", "candidate", "rejected"), "Unknown training status")
            require(status != "rejected", "Fit rejected; inspect last-state.json and its reason")
            if status not in ("queued", "running"):
                return state
            time.sleep(min(1, max(0, deadline - time.monotonic())))

    def predict(self, source, expected_hash, inputs):
        model_hash({"model_sha256": expected_hash})
        result = self.post("predict", {"model_source": source, "expected_model_sha256": expected_hash,
                                       "inputs": inputs})
        require(result.get("domain_id") == self.domain, "Prediction Domain mismatch")
        require(model_hash(result.get("evaluator")) == expected_hash, "Prediction model changed")
        values = result.get("predictions")
        require(isinstance(values, list) and len(values) == len(inputs) and all(finite(v) for v in values),
                "Invalid prediction values or result count")
        return values


def run(args, opener=urlopen):
    require(finite(args.wait_seconds) and args.wait_seconds > 0, "Choose a positive client wait")
    if args.phase == "promote":
        pair, measured = load(args.pair), load(args.cases)
        require(isinstance(pair, dict) and isinstance(measured, dict), "Invalid model pair or cases")
        for name in ("incumbent", "candidate"):
            model_hash({"model_sha256": pair.get(name)})
            require(measured.get(name) == pair[name], "Measured outcomes belong to another model pair")
        cases = measured.get("cases")
        require(isinstance(cases, list) and 16 <= len(cases) <= 10000, "This example requires 16 to 10000 cases")
        for case in cases:
            require(isinstance(case, dict) and set(case) == {"case_id", "incumbent_utility", "candidate_utility"}
                    and isinstance(case["case_id"], str) and 1 <= len(case["case_id"]) <= 256
                    and unit(case["incumbent_utility"]) and unit(case["candidate_utility"]), "Invalid measured case")
        require(len({c["case_id"] for c in cases}) == len(cases), "Duplicate case_id")
        body = {"expected_current_model_sha256": pair["incumbent"], "expected_model_sha256": pair["candidate"],
                "canary": {"minimum_cases": 16, "required_improvement": 0, "cases": cases}}
        sized(body)
    else:
        rows, options, inputs, ids, uploads = training_inputs(args)
        config = load(args.config) if args.phase == "initial" else None
        if args.phase == "initial":
            require(isinstance(config, dict) and all(isinstance(config.get(k), str) and config[k]
                    for k in ("relation", "policy")), "Provide a value config with relation and policy")
            sized({"config": config})
    client = Client(args.output, opener)
    state = client.state()
    if args.phase != "initial":
        require(state.get("frozen") is False, "Use an unfrozen Domain")
    require((state.get("training") or {}).get("status") not in ("queued", "running"), "A fit is already running")
    if args.phase == "promote":
        require(pair.get("base") == client.base and pair.get("domain") == client.domain, "Model pair target changed")
        require(model_hash(state.get("model")) == pair["incumbent"] and
                model_hash(state.get("candidate")) == pair["candidate"], "Current model pair changed")
        client.post("promote", body)
        require(model_hash(client.state().get("model")) == pair["candidate"], "Promotion did not install the candidate")
        return client.output
    initial = args.phase == "initial"
    if initial:
        require(not state.get("config") and not state.get("model") and not state.get("candidate") and
                state.get("sample_count") == 0, "Initial mode requires a fresh existing Domain")
    else:
        config = state.get("config")
        require(isinstance(config, dict), "Missing value configuration")
    capacity = config.get("max_samples", 20000)
    require(type(capacity) is int and capacity >= len(rows), "Configuration cannot retain the selected rows")
    graph = config.get("representation", "projected") == "object_graph"
    require((graph and options["model_type"] == "object_value") or
            (not graph and options["model_type"] in ("linear", "mlp_v2")), "Representation/model mismatch")
    require(not graph or options.get("selection_objective") != "adaptation_prediction", "Unsupported graph selection")
    incumbent = None if initial else model_hash(state.get("model"))
    if initial:
        require(client.post("configure", {"config": config}).get("status") == "configured", "Configuration not confirmed")
        require(client.state().get("frozen") is False, "Configured Domain is not confirmed unfrozen")
    for batch in uploads:
        admitted = client.post("observe", {"observations": batch})
        counts = [admitted.get("accepted"), admitted.get("duplicates", 0)]
        require(all(type(v) is int and v >= 0 for v in counts) and sum(counts) == len(batch), "Observation admission mismatch")
    request_id = uuid4().hex
    body = {"request_id": request_id, "options": options, "install": initial, "observation_ids": ids}
    if not initial:
        body.update(warm_start_source="installed", expected_initial_model_sha256=incumbent)
    client.post("train", body, expected=202)
    state = client.wait(request_id, args.wait_seconds)
    if initial:
        require(state["training"]["status"] == "trained" and state["training"].get("installed") is True,
                "Initial fit did not install a model")
        result = {"installed": client.predict("installed", model_hash(state.get("model")), inputs)}
    else:
        require(state["training"]["status"] == "candidate" and state["training"].get("installed") is not True and
                model_hash(state.get("model")) == incumbent, "Candidate did not preserve the incumbent")
        candidate = model_hash(state.get("candidate"))
        result = {"incumbent": client.predict("installed", incumbent, inputs),
                  "candidate": client.predict("candidate", candidate, inputs)}
        client.save("model-pair.json", {"base": client.base, "domain": client.domain,
                                        "incumbent": incumbent, "candidate": candidate})
    client.save("probe-results.json", {"inputs": inputs, "predictions": result})
    return client.output


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("phase", choices=("initial", "candidate", "promote"))
    for name in ("config", "observations", "options", "inputs", "pair", "cases"):
        parser.add_argument("--" + name)
    parser.add_argument("--output", required=True)
    parser.add_argument("--wait-seconds", type=float, default=600)
    try:
        print("Completed; inspect", run(parser.parse_args()))
    except Exception as error:
        raise SystemExit(str(error)) from None

Run the workflow

Fit and inspect the first model

Configure the fresh Domain, upload the selected observations, fit with warm_start: false and install: true, and predict on your probe inputs.

python value_client.py initial \
  --config value-config.json \
  --observations initial-observations.json \
  --options fit-options.json \
  --inputs probe-inputs.json \
  --output run-initial

The command checks that the reported job matches its generated request_id, completed as trained, and installed a model. Inspect run-initial/last-state.json and run-initial/probe-results.json before preparing another fit.

Stage a candidate and compare predictions

Supply the complete earlier-plus-new observation selection. The client replays the rows unchanged, checks admission counts, and sends their IDs explicitly to training. It uses warm_start: true, warm_start_source: installed, the incumbent hash, and install: false.

python value_client.py candidate \
  --observations all-observations.json \
  --options fit-options.json \
  --inputs probe-inputs.json \
  --output run-candidate

The command requires a completed candidate result with the incumbent unchanged. It predicts with both models on the same inputs, checks the returned hashes and prediction counts, and saves model-pair.json and probe-results.json.Compare the probe predictions to identify functional changes. A changed SHA alone does not establish changed predictions or better decisions.

Measure application outcomes and request promotion

Evaluate the saved incumbent and candidate on paired application cases under the same conditions. Use the saved hashes to bind each evaluation. Keep the actual outcomes and the mapping from outcomes to utility in your application record.Write paired-cases.json as an object with the following fields:

Field	Contents
`incumbent`	The exact `incumbent` hash from `run-candidate/model-pair.json`
`candidate`	The exact `candidate` hash from that same file
`cases`	An array of objects containing a unique `case_id`, measured `incumbent_utility`, and measured `candidate_utility`

This is a client file format. The script maps those hashes to the API’s expected_current_model_sha256 and expected_model_sha256 fields. Supply finite utility values in [0,1]. The example chooses minimum_cases: 16 and required_improvement: 0; the server still decides whether the canary demonstrates sufficient improvement. Prediction values from the previous step are not measured application outcomes.

python value_client.py promote \
  --pair run-candidate/model-pair.json \
  --cases paired-cases.json \
  --output run-promotion

The command verifies that the current model pair still matches the evaluated pair, submits the canary once, and reads state back to confirm that the candidate became the installed model. Follow promotion checks when selecting your application measure and interpreting a refusal.

Handle a stopped command

Inspect the output directory before another mutation. last-response.json preserves the most recent response or transport failure, and last-state.json preserves the latest successful state inspection. The numbered training request file records the generated request_id.

Result	Next action
A different job ID, model hash, Domain, or prediction count	Stop and reconcile the request with the returned state. Do not evaluate or promote a different model under the saved identity.
`training.status: rejected`	Read the reason and fit report, then correct the data or configuration issue before an intentional new fit.
Explicit owner-fit `409`	Wait for the active fit to finish and follow the owner scheduling rules.
Promotion `409`	Preserve the canary result and inspect the installed identity. Do not treat a completed fit as adopted.
A transport timeout or the client fit-wait deadline	Read `value/state` and reconcile the journal before resubmitting. Stopping the client does not cancel an accepted server job.

Keep the selected observations, actual model pair, and evaluation outcomes together when continuing to another candidate round. The explicit promotion command does not implement the separate registered application_cases and application_outcomes automatic-adoption flow.

Get started

Concepts

Guides

Value learning

Planning

Discovery

Operations

API reference

Prepare your environment and files

Save the Python client

Run the workflow

Handle a stopped command

​Prepare your environment and files

​Save the Python client

​Run the workflow

​Handle a stopped command

Prepare your environment and files

Save the Python client

Run the workflow

Handle a stopped command