Pre-publish check

# About scenario.json

`scenario.json` is an illustrative example request body in the shape of the public `evaluation.json`. It holds only fields the Evaluate request accepts: `profile_id`, `objective`, `context`, and `candidates` (each with `candidate_id`, `semantic_class`, `description`, and `operator_reported_support`).

The objective and the candidate descriptions say only what the decision is and what each step is. The illustrative marking lives in two short context facts, not in repeated prose:

- `context.evidence_kind` is `illustrative_example`.
- `context.support_basis` is `operator_judgment`.

The external candidate ends with "This external option must remain shadow only.", as in the docs example. Context values are short scalars (booleans, short strings, numbers).

Replace the objective, context, candidates, and every support value with your own. Evaluate does not verify context or support values. They are reported by you.

About scenario.json

scenario.json is an illustrative example request body in the shape of the public evaluation.json. It holds only fields the Evaluate request accepts: profile_id, objective, context, and candidates (each with candidate_id, semantic_class, description, and operator_reported_support).

The objective and the candidate descriptions say only what the decision is and what each step is. The illustrative marking lives in two short context facts, not in repeated prose:

• context.evidence_kind is illustrative_example.
• context.support_basis is operator_judgment.

The external candidate ends with "This external option must remain shadow only.", as in the docs example. Context values are short scalars (booleans, short strings, numbers).

Replace the objective, context, candidates, and every support value with your own. Evaluate does not verify context or support values. They are reported by you.

Illustrative example. Replace with your own values.

{
  "profile_id": "deltax-hosted-bounded-review-evaluation-v1",
  "objective": "Choose a bounded next step for a finished blog post draft that has no owner approval to publish.",
  "context": {
    "evidence_kind": "illustrative_example",
    "support_basis": "operator_judgment",
    "draft_status": "finished",
    "owner_approval_to_publish": false,
    "sources_listed": true,
    "sources_opened_by_agent": false
  },
  "candidates": [
    {
      "candidate_id": "validate-claims-against-sources",
      "semantic_class": "validate",
      "description": "Check each factual claim in the draft against the sources it lists and mark any claim without a source.",
      "operator_reported_support": 0.8
    },
    {
      "candidate_id": "draft-revision-notes",
      "semantic_class": "draft",
      "description": "Draft revision notes for the owner covering unsupported claims and unclear passages.",
      "operator_reported_support": 0.7
    },
    {
      "candidate_id": "compare-draft-to-style-rules",
      "semantic_class": "compare",
      "description": "Compare the draft to the written style rules and list the differences.",
      "operator_reported_support": 0.6
    },
    {
      "candidate_id": "analyze-open-questions",
      "semantic_class": "analyze",
      "description": "List the questions the draft raises but does not answer, with the paragraph for each.",
      "operator_reported_support": 0.5
    },
    {
      "candidate_id": "publish-post-now",
      "semantic_class": "external",
      "description": "Publish the post now without the owner's yes. This external option must remain shadow only.",
      "operator_reported_support": 0.9
    }
  ]
}

Reads credentials from environment variables named in the file. Never put credentials in the file.

#!/usr/bin/env python3
"""Call DeltaX Evaluate for one or more scenario files. Standard library only.

Illustrative integration snippet for this skill.

Credentials come only from the environment and are never printed or written:
  DELTAX_CLIENT_ID       account credential id
  DELTAX_CLIENT_SECRET   account credential secret
Optional:
  DELTAX_ACCESS_TOKEN        an access token already obtained (15 minute lifetime); skips the token call
  DELTAX_EVALUATE_BASE_URL   default https://access-api.deltaxevaluate.com

Usage:
  python3 integration.py scenario.json --dry-run   # checks the request locally, makes no call
  python3 integration.py scenario.json             # one evaluation (counts against your allowance)
  python3 integration.py a.json b.json c.json      # several, in one process: one token, calls 3 seconds apart

Requests follow the public account API reference: form-encoded token request with scope deltax:evaluate, the default
Python User-Agent, one token reused until shortly before its 15 minute expiry, Accept: application/json. A retry
(HTTP 202, 429, 5xx or a network error) always reuses the same Idempotency-Key and the same body, at most twice.
A scenario that already has a completed receipt for its key in --out is not sent again.

Contract: 2026-09-16.beta2, profile deltax-hosted-bounded-review-evaluation-v1.
A selection is never permission to act. Your own code performs any step.
"""
import argparse, datetime as dt, glob, hashlib, json, os, re, sys, time, urllib.error, urllib.parse, urllib.request

BASE = os.environ.get("DELTAX_EVALUATE_BASE_URL", "https://access-api.deltaxevaluate.com").rstrip("/")
TOKEN_URL = f"{BASE}/api/builder/oauth/token"
EVAL_URL = f"{BASE}/api/builder/evaluations"
API_VERSION = "2026-09-16.beta2"
PROFILE = "deltax-hosted-bounded-review-evaluation-v1"
SELECTABLE = {"analyze", "compare", "draft", "validate"}
CLASSES = SELECTABLE | {"external"}
AUTHORITY_FLAGS = ["selected_is_executed", "response_authorizes_downstream_action", "selected_is_correct_or_safe",
                   "response_is_professional_advice", "trace_is_certification_or_legal_non_repudiation"]
TRACE_FLAGS = ["external_effect", "submitted_candidate_executed", "learning_applied",
               "operator_reported_context_verified"]
RESERVED_CONTEXT_KEYS = {"authorization", "bearer", "callback", "command", "cookie", "credential", "credentials",
                         "file", "filepath", "password", "path", "plugin", "recovery_code", "secret", "session",
                         "shell", "subprocess", "token", "tool", "url", "webhook"}
MIN_SPACING_S = 3.0          # between requests from this process
MAX_RETRIES = 2              # same key and same body only
_state = {"token": None, "expires": 0.0, "last": 0.0}


def redact(text):
    for name in ("DELTAX_CLIENT_ID", "DELTAX_CLIENT_SECRET", "DELTAX_ACCESS_TOKEN"):
        value = os.environ.get(name)
        if value:
            text = text.replace(value, "[REDACTED]")
    text = re.sub(r'("?(?:access_token|client_secret|client_id|token)"?\s*[:=]\s*"?)[^",&\s]+', r"\1[REDACTED]", text, flags=re.I)
    return re.sub(r"Bearer\s+[A-Za-z0-9._~+/=-]+", "Bearer [REDACTED]", text)


def check_request(req):
    """Local checks against the published account limits. Returns a list of problems."""
    p = []
    if set(req) != {"profile_id", "objective", "context", "candidates"}:
        p.append("top-level fields must be exactly profile_id, objective, context, candidates")
    if req.get("profile_id") != PROFILE:
        p.append("profile_id must be " + PROFILE)
    obj = req.get("objective", "")
    if not obj.strip() or len(obj) > 1024:
        p.append("objective must be nonblank and at most 1024 characters")
    ctx = req.get("context", {})
    if len(ctx) > 16:
        p.append("context allows at most 16 properties")
    for k, v in ctx.items():
        if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]{0,63}", k):
            p.append(f"context key {k!r} has a bad format")
        if k.lower().replace("-", "_") in RESERVED_CONTEXT_KEYS:
            p.append(f"context key {k!r} is reserved")
        if isinstance(v, (dict, list)):
            p.append(f"context value for {k!r} must be a scalar")
        if isinstance(v, str) and len(v) > 256:
            p.append(f"context value for {k!r} is over 256 characters")
        if isinstance(v, (int, float)) and not isinstance(v, bool) and abs(v) > 1_000_000_000:
            p.append(f"context value for {k!r} is too large")
    cands = req.get("candidates", [])
    if not 1 <= len(cands) <= 8:
        p.append("candidates must contain 1 to 8 entries")
    ids = [c.get("candidate_id") for c in cands]
    if len(set(ids)) != len(ids):
        p.append("candidate_id values must be unique")
    for c in cands:
        if set(c) != {"candidate_id", "semantic_class", "description", "operator_reported_support"}:
            p.append(f"candidate {c.get('candidate_id')!r} must have exactly the four published fields")
        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._~-]{0,63}", str(c.get("candidate_id", ""))):
            p.append(f"candidate_id {c.get('candidate_id')!r} has a bad format")
        if c.get("semantic_class") not in CLASSES:
            p.append(f"candidate {c.get('candidate_id')!r} has an unknown semantic_class")
        d = c.get("description", "")
        if not d.strip() or len(d) > 512:
            p.append(f"candidate {c.get('candidate_id')!r} description must be nonblank and at most 512 characters")
        s = c.get("operator_reported_support")
        if isinstance(s, bool) or not isinstance(s, (int, float)) or not 0 <= s <= 1:
            p.append(f"candidate {c.get('candidate_id')!r} support must be a number from 0 to 1")
    return p


def post(url, data, headers):
    """One request, spaced at least MIN_SPACING_S after the previous one. Default User-Agent."""
    wait = MIN_SPACING_S - (time.monotonic() - _state["last"])
    if _state["last"] and wait > 0:
        time.sleep(wait)
    request = urllib.request.Request(url, data=data, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=20) as r:
            return r.status, r.read().decode("utf-8", "replace"), dict(r.headers)
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", "replace"), dict(e.headers)
    finally:
        _state["last"] = time.monotonic()


def get_token(refresh=False):
    """One access token per process, reused until about a minute before its 900 second expiry."""
    existing = os.environ.get("DELTAX_ACCESS_TOKEN")
    if existing and not refresh:
        return existing
    if _state["token"] and not refresh and time.monotonic() < _state["expires"]:
        return _state["token"]
    cid, secret = os.environ.get("DELTAX_CLIENT_ID"), os.environ.get("DELTAX_CLIENT_SECRET")
    if not cid or not secret:
        sys.exit("Set DELTAX_CLIENT_ID and DELTAX_CLIENT_SECRET (or DELTAX_ACCESS_TOKEN) in the environment.")
    form = urllib.parse.urlencode({"grant_type": "client_credentials", "client_id": cid,
                                   "client_secret": secret, "scope": "deltax:evaluate"}).encode("utf-8")
    status, text, _ = post(TOKEN_URL, form, {"Content-Type": "application/x-www-form-urlencoded"})
    if status != 200:
        sys.exit(f"Token request failed with HTTP {status}: {redact(text)[:500]}")
    data = json.loads(text)
    _state["token"] = data["access_token"]
    _state["expires"] = time.monotonic() + max(30, int(data.get("expires_in", 900)) - 60)
    return _state["token"]


def send_evaluation(body, key):
    """POST the evaluation. On 202, 429, 5xx or a network error, retry with the SAME key and body, with backoff,
    at most MAX_RETRIES times. Returns (status, text, headers, attempts)."""
    attempts, refreshed, need_fresh, n = [], False, False, 0
    while True:
        headers = {"Authorization": f"Bearer {get_token(refresh=need_fresh)}",
                   "Content-Type": "application/json", "X-DeltaX-API-Version": API_VERSION,
                   "Idempotency-Key": key, "Accept": "application/json"}
        need_fresh = False
        started = time.monotonic()
        try:
            status, text, resp_headers = post(EVAL_URL, body, headers)
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            status, text, resp_headers = 0, redact(f"{type(exc).__name__}: {exc}"), {}
        keep = {k.lower(): v for k, v in resp_headers.items()
                if k.lower() in ("retry-after", "content-type", "idempotent-replayed", "x-request-id", "date")}
        attempts.append({"attempt": len(attempts) + 1, "http_status": status, "elapsed_s": round(time.monotonic() - started, 2),
                         "headers_subset": keep})
        if status == 401 and not refreshed:  # expired or revoked token: one fresh token, same key and body
            refreshed = need_fresh = True
            continue
        if (status in (202, 429, 500, 502, 503, 504) or status == 0) and n < MAX_RETRIES:
            retry_after = keep.get("retry-after")
            delay = min(120, int(retry_after)) if retry_after and retry_after.isdigit() else (10, 30)[n]
            print(f"  HTTP {status}; sending the same body with the same key again in {delay}s", file=sys.stderr)
            time.sleep(delay)
            n += 1
            continue
        return status, text, resp_headers, attempts


def completed_receipt(out_dir, key):
    """Return a saved completed (HTTP 200) result for this key in out_dir, so an identical body is never sent twice."""
    for path in sorted(glob.glob(os.path.join(out_dir, "*.json"))):
        try:
            with open(path, encoding="utf-8") as f:
                saved = json.load(f)
        except (OSError, json.JSONDecodeError):
            continue
        resp = saved.get("response") or {}
        if saved.get("idempotency_key") == key and saved.get("http_status", 200) == 200 and resp.get("receipt_id") \
                and isinstance(resp.get("result"), dict):
            return path, resp
    return None, None


def reconcile(req, resp):
    """Fail-closed reconciliation of a completed response. Returns a list of problems."""
    p = []
    res = resp.get("result", resp)
    if res.get("api_version") != API_VERSION:
        p.append(f"unexpected api_version {res.get('api_version')}")
    if res.get("profile_id") != PROFILE:
        p.append("unexpected profile_id")
    outcome = res.get("outcome")
    if outcome not in ("selected", "governed_noop_refusal"):
        p.append(f"unknown outcome {outcome}")
    selected = res.get("selected_candidate_id")
    eligible = {c["candidate_id"] for c in req["candidates"] if c["semantic_class"] in SELECTABLE}
    if outcome == "selected" and selected not in eligible:
        p.append("selection is outside the submitted review candidates")
    if outcome == "governed_noop_refusal" and selected is not None:
        p.append("refusal carries a selection")
    for block, flags in (("authority", AUTHORITY_FLAGS), ("trace", TRACE_FLAGS)):
        found = res.get(block) or {}
        for flag in flags:
            if flag not in found:
                p.append(f"{block}.{flag} missing")
            elif found[flag] is not False:
                p.append(f"{block}.{flag} is not false")
    return p


def run_one(scenario_path, args):
    with open(scenario_path, encoding="utf-8") as f:
        sc = json.load(f)
    req = {"profile_id": sc.get("profile_id", PROFILE), "objective": sc["objective"],
           "context": sc.get("context", {}), "candidates": sc["candidates"]}
    body = json.dumps(req, separators=(",", ":"), sort_keys=True, ensure_ascii=False).encode("utf-8")
    problems = check_request(req)
    if len(body) > 32768:
        problems.append("request body is over 32768 bytes")
    if problems:
        sys.exit(f"Request problems in {scenario_path}:\n- " + "\n- ".join(problems))
    key = args.key or "skill-" + hashlib.sha256(body).hexdigest()[:40]
    if not re.fullmatch(r"[A-Za-z0-9._~-]{16,128}", key):
        sys.exit("Bad idempotency key.")
    if args.dry_run:
        print(json.dumps({"endpoint": EVAL_URL, "idempotency_key": key, "request": req}, indent=2))
        return 0

    saved_path, saved = completed_receipt(args.out, key)
    if saved:
        res = saved.get("result", {})
        print(f"already completed for this key; stored result reused, no call made: {saved_path}")
        print(f"outcome={res.get('outcome')} selected={res.get('selected_candidate_id')} receipt_id={saved.get('receipt_id')}")
        return 0

    status, text, resp_headers, attempts = send_evaluation(body, key)
    text = redact(text)
    os.makedirs(args.out, exist_ok=True)
    stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    if status != 200:
        # 202: keep the account record id so it can be given to support, as the account API reference describes.
        # Any other status: keep the (redacted) error body and attempt timings; do not start a new request.
        try:
            payload = json.loads(text)
        except json.JSONDecodeError:
            payload = {"raw": text[:2000]}
        rid = payload.get("receipt_id") if isinstance(payload, dict) else None
        path = os.path.join(args.out, f"{stamp}-{rid or 'http-' + str(status)}.json")
        with open(path, "w", encoding="utf-8") as f:
            json.dump({"idempotency_key": key, "api_version": API_VERSION, "http_status": status, "attempts": attempts,
                       "request": req, "response": payload,
                       "reconciliation_problems": ["outcome unconfirmed; do not act"]}, f, indent=2)
        if status == 202:
            extra = f" Account record id: {rid}." if rid else ""
            print(f"Unconfirmed (HTTP 202). Do not start a new request. Re-send the same body with key {key} to read its status.{extra} Saved {path}.", file=sys.stderr)
        else:
            print(f"Evaluation failed with HTTP {status} after {len(attempts)} attempt(s): {text[:500]} Saved {path}.", file=sys.stderr)
        return 1
    resp = json.loads(text)
    bad = reconcile(req, resp)

    path = os.path.join(args.out, f"{stamp}-{resp.get('receipt_id', 'receipt')}.json")
    with open(path, "w", encoding="utf-8") as f:
        json.dump({"idempotency_key": key, "api_version": API_VERSION, "http_status": 200, "attempts": attempts,
                   "request": req, "response": resp, "reconciliation_problems": bad}, f, indent=2)

    res = resp.get("result", {})
    print(f"outcome={res.get('outcome')} selected={res.get('selected_candidate_id')} counted={resp.get('counted')}")
    print(f"receipt_id={resp.get('receipt_id')} trace_id={(res.get('trace') or {}).get('trace_id')}")
    print(f"saved {path}")
    if bad:
        print("RECONCILIATION PROBLEMS, do not act on this result:", "; ".join(bad))
        return 1
    # Next: if outcome is "selected", your own code performs that review step and keeps the
    # result for the owner. If it is "governed_noop_refusal", stop this path and use the
    # fallback you declared before calling. Neither outcome permits an external action.
    return 0


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("scenario", nargs="+", help="one or more scenario files; they share one token and are spaced")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--key", help="Idempotency-Key, 16 to 128 characters from A-Z a-z 0-9 . _ ~ - (single scenario only)")
    ap.add_argument("--out", default="receipts")
    args = ap.parse_args()
    if args.key and len(args.scenario) > 1:
        sys.exit("--key works with a single scenario")
    for path in args.scenario:
        code = run_one(path, args)
        if code:
            if len(args.scenario) > 1:
                print("stopped at the first scenario that did not complete; nothing further sent", file=sys.stderr)
            return code
    return 0


if __name__ == "__main__":
    sys.exit(main())

---
name: pre-publish-check
description: "Use when a draft is ready to publish and an agent must pick the next review step before anyone publishes, with publishing left to the owner."
---

# Pre-publish check

Any flow where an agent prepared something that would go out under the owner's name and you want one review step chosen first, with a receipt kept beside the draft.

DeltaX Evaluate is the checkpoint in the middle of this recipe. It selects one step from the eligible candidates you submit, or refuses, and returns an auditable trace. It does not give a reason for its choice. The scenario that ships with this skill is illustrative: replace its values with your own.

## Setup

- Create a Free account at https://deltaxevaluate.com/account and create an API credential. Allowance and plans are on https://deltaxevaluate.com/pricing.
- Set `DELTAX_CLIENT_ID` and `DELTAX_CLIENT_SECRET` in your server or agent environment. `DELTAX_ACCESS_TOKEN` is also read if you already have a token. Never paste any of them into chat, files, repositories, or logs.
- `integration.py` needs Python 3.9 or newer and no packages. Contract: `2026-09-16.beta2`, profile `deltax-hosted-bounded-review-evaluation-v1`. If a request is blocked, a descriptive User-Agent (`DELTAX_USER_AGENT`) may help.

## Steps

1. **Freeze the draft.** Take the exact text that would be published and note its version or hash in your own records. Do not put the draft text, secrets, or personal data into the request.
2. **State the situation.** Write the objective (at most 1024 characters): what is being decided and what is not allowed. Add context as scalar values only (string up to 256 characters, number, boolean, or null; at most 16 entries). No secrets, no personal data, and no reserved keys such as `url`, `path`, `token`, or `tool`. Include `evidence_kind` and a date. A changed situation needs a changed request.
3. **Build the candidates.** List 1 to 8 candidates with unique ids of at most 64 characters. Use `analyze`, `compare`, `draft`, or `validate` for steps your agent can perform and the owner can review. Describe exactly what happens and state that nothing is sent, merged, or published. Add every consequential outside action you are considering as one `external` candidate. External candidates are shadow only and cannot be selected.
4. **Set support.** Use your revision log if you keep one, otherwise report labeled judgment and say so in each description (the sample labels its values illustrative).
5. **Declare your fallback before calling.** Decide now what your own system does if Evaluate refuses (for example, hand the case to a person). The fallback is your action, with its own risks. Evaluate does not perform it.
6. **Run it.** Run `python3 integration.py scenario.json --dry-run` to check the request locally, then run it without `--dry-run`. It sends one evaluation with an `Idempotency-Key` and credentials from the environment. Reuse the same key and body to retry. Never turn a timeout into a new request.
7. **Reconcile before you act.** Reject the result if `api_version` is not `2026-09-16.beta2`, if `profile_id` is not the profile you sent, if the selected id is not one of your submitted review candidates, or if any `authority` or `trace` flag is not `false`. On HTTP 202 stop and reuse the original key. On 429 stop.
8. **Act within your own rules.** If the outcome is `selected`, perform that one review step yourself and hand the result to the owner. If it is `governed_noop_refusal`, stop this path and use your declared fallback. A selection is never permission or authority. Anything with an outside effect needs its own separate yes from the owner, outside Evaluate.
9. **Record it.** Store the request, the full response, the top-level `receipt_id`, `result.commercial_receipt.receipt_id`, and `result.trace.trace_id` with the work. Keep the receipt with the draft version it was run on. A changed draft is a new situation and needs a new request.

## Sample candidates

The files `scenario.json` and `integration.py` ship with this skill. Every support value in the sample is an illustrative placeholder, labeled as such (see `scenario.README.md`).

| candidate_id | semantic_class | operator_reported_support |
| --- | --- | --- |
| `validate-claims-against-sources` | validate | 0.8 |
| `draft-revision-notes` | draft | 0.7 |
| `compare-draft-to-style-rules` | compare | 0.6 |
| `analyze-open-questions` | analyze | 0.5 |
| `publish-post-now` | external | 0.9 |

## Setting support

- **Labeled judgment unless you keep a revision log.** Report your own judgment for each review step and say so in its description. The sample's own values are labeled illustrative; replace them.
- **Measured data when you have it.** If you log, per published piece, which review step found a change that the owner kept, use that dated and sourced rate as the support value and state the window. Do not use the agent's own confidence.
- Context such as `sources_opened_by_agent` is reported by you. Evaluate does not verify it.

## Reading the response

The result sits inside `result`. `outcome` is `selected` or `governed_noop_refusal`, and `selected_candidate_id` is a submitted id or `null`. `reason_codes` holds bounded outcome codes, such as `candidate_selected_within_bounded_evaluation`; store it with the receipt, and do not present it, or any entry in `trace.lambda_decisions`, as a reason for or against any single candidate. The account record id (`receipt_id`) and the evaluator receipt id (`result.commercial_receipt.receipt_id`) are different records. Keep both.