Scientific instrument provenance

A spectrometer takes a reading. A mass spectrometer ingests a sample. A telescope captures a frame. All of these are scientific instruments where the provenance of each reading — when, what instrument, what calibration, what operator — matters as much as the reading itself.

Pattern: one state machine per instrument. Each reading writes:

• A reading_raw event — hash of the raw sensor bytes.
• A reading_processed event — hash of the calibrated, processed result.
• A compute row — the operator sequence that did the processing, with a CKO (ComputedKnowledgeObject) proof.

A reviewer reading a paper that cites a result can validate the entangled state forward from the published row to the live instrument and confirm: this reading came from this instrument at this Zeqond, processed with these operators, against this calibration.

---

1. Spin up the instrument

curl -sS https://zeqapi.com/api/chain/state-machines \
  -H "Authorization: Bearer ${ZSM_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"slug":"spectrometer-bench-3","is_public":true,
       "display_name":"Bench 3 NMR spectrometer",
       "purpose":"Audited NMR spectra, calibration logged, operator id on every reading"}'

is_public: true — peer reviewers should be able to read without an account.

2. Record a reading (Python on the lab PC)

# instrument-driver.py — runs on the lab PC, beside the instrument
import hashlib, json, requests, time, os

ZEQ_HOST = "https://zeqapi.com"
SLUG     = "spectrometer-bench-3"
ZSM_KEY  = os.environ["ZSM_KEY"]

def sha256(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()

def chain_event(type_: str, hash_: str, envelope: dict | None = None):
    body = {"type": type_, "hash": hash_}
    if envelope: body["envelope"] = envelope  # for sealed events
    r = requests.post(
        f"{ZEQ_HOST}/api/chain/{SLUG}/event",
        headers={"Authorization": f"Bearer {ZSM_KEY}", "Content-Type": "application/json"},
        json=body,
    )
    return r.json()

def chain_file(file_hash: str, name: str, bytes_: int, mime: str):
    r = requests.post(
        f"{ZEQ_HOST}/api/chain/{SLUG}/file",
        headers={"Authorization": f"Bearer {ZSM_KEY}", "Content-Type": "application/json"},
        json={"file_hash": file_hash, "name": name, "bytes": bytes_, "mime": mime},
    )
    return r.json()

def compute(operators: list[str], inputs: dict) -> dict:
    """Run a wizard composition through /api/zeq/prove. Returns the CKO."""
    r = requests.post(
        f"{ZEQ_HOST}/api/zeq/prove",
        headers={"Authorization": f"Bearer {ZSM_KEY}", "Content-Type": "application/json"},
        json={"operators": operators, "domain": "spectroscopy", "inputs": inputs},
    )
    return r.json()

# ─── Run a reading ───────────────────────────────────────────────────────
sample_id     = "P-2026-04-28-#142"
calibration_id = "cal-2026-04-28-am-001"

# Capture from instrument (placeholder; replace with your driver call)
raw_bytes = bytes(open("/tmp/last_capture.fid","rb").read())
raw_hash  = sha256(raw_bytes)

# 1. raw reading on chain
chain_file(raw_hash, "reading.fid", len(raw_bytes), "application/octet-stream")

# 2. process — phase correction, baseline, peak picking
processed = {
    "sample_id": sample_id,
    "calibration_id": calibration_id,
    "peaks": [{"ppm": 7.26, "area": 1.0}, {"ppm": 4.18, "area": 0.97}],
    "raw_hash": raw_hash,
}
processed_hash = sha256(json.dumps(processed, sort_keys=True).encode())
chain_event("reading_processed", processed_hash)

# 3. CKO via /api/zeq/prove — uses QM operators relevant to NMR
cko = compute(
    operators=["KO42", "QM10", "QM5"],   # KO42 + Planck-Einstein + Hamiltonian
    inputs={"sample_id": sample_id, "peak_ppm": 7.26},
)

# 4. compute event on chain — links operator sequence + CKO digest
chain_event("compute", cko["zeqProof"], envelope={
    "operators": cko["operators"],
    "result_digest": cko["resultDigest"],
    "precision": cko["precision"],
    "linked_processed_hash": processed_hash,
})

print(f"reading committed: raw={raw_hash[:16]}... processed={processed_hash[:16]}... cko={cko['zeqProof'][:16]}...")
print(f"zeqond: {cko['zeqond']}, precision: {cko['precision']}")

Three rows on the entangled state per reading: raw bytes hash, processed JSON hash, compute CKO.

3. Reviewer-side validation

A peer reviewer reading a paper that cites the reading at Zeqond Z:

# Fetch the block at the claimed Zeqond
curl -sS "https://zeqapi.com/api/chain/spectrometer-bench-3/block?zeqond=${Z}" | jq .

# Validate the entire chain integrity
curl -sS "https://zeqapi.com/api/chain/spectrometer-bench-3/pohc/validate?from=${Z}&to=${Z}+10"

# Verify the CKO independently
curl -sS https://zeqapi.com/api/zeq/verify \
  -H "Content-Type: application/json" \
  -d "{\"zeqProof\":\"${PROOF}\",\"operators\":[\"KO42\",\"QM10\",\"QM5\"],\"R_t\":${R_T},\"zeqond\":${Z}}"

The reviewer gets:

• A signed CKO proof, verifiable without the instrument.
• A hash-linked log showing the row hasn't been tampered.
• The exact operator sequence that processed the raw bytes.
• A precision number ≤ 0.001 (the framework's gate; published readings should never have a higher residual).

4. Calibration as a contract

For instruments where calibration drift matters, deploy the iot-device template (lifecycle: provisioned → active → maintenance → retired) on the same machine. Every calibration is a maintenance → active cycle, recorded as a contract transition with the calibration certificate hash as input.

A reviewer can query:

curl -sS "https://zeqapi.com/api/chain/spectrometer-bench-3/contracts" \
  -H "Authorization: Bearer ${ZSM_KEY}" | jq '.contracts[] | select(.name == "iot_device_lifecycle")'

…then walk the lifecycle to find the active calibration at the time of any reading.

---

What gets proven

Claim	Verification
"This reading came from this instrument"	state_machines row's slug + owner_zid
"…at this Zeqond"	audit_log.zeqondNumber of the reading_processed row
"…with this calibration"	The iot-device contract's current_state at that Zeqond was active for the named calibration
"…processed with these operators"	The compute row's envelope.operators
"…within ≤0.1% precision"	The CKO's precision field, verifiable via POST /api/zeq/verify
"The entangled state hasn't been tampered"	pohc/validate returns valid: true

The framework deletes the bytes (free tier) but keeps the proofs. The lab keeps the bytes; the entangled state proves what they had. That's the deal.

---

Scaling to a lab

For 10+ instruments, a typical layout:

• One state machine per instrument, all under the lab's owner ZID.
• One oracle contract template per data feed (calibration sources, ambient temperature, magnetic field) — they publish to chain on a schedule.
• One compliance-audit contract per regulated study, transitioning on every reading and tying together the instrument slug + calibration contract + reading hash.

The lab's audit dashboard reads the public registry (/api/chain/aggregate/list) and subscribes to /api/chain/aggregate/sse for one stream of all instruments. The framework already ships that aggregator.