"""Contract tests for every API route, on a generated tiny artifact set.""" import json import os import time from types import SimpleNamespace import pytest from fastapi.testclient import TestClient from plastic.api.app import create_app from plastic.config import ModelConfig from plastic.data.text import encode_documents_to_bin from plastic.harness.canary import CanarySuite from plastic.store import ArtifactStore from plastic.tokenizer.bpe import Tokenizer from plastic.train.loop import TrainConfig, train DOCS = ["alpha beta gamma delta epsilon " * 80, "one two three four five six " * 80] PROMPT = "alpha beta gamma delta epsilon alpha beta gamma delta epsilon one two three four" TRANSACTION_KEYS = {"index", "t_unix", "pos_start", "pos_end", "decision", "requested", "signals", "accepted", "read_only", "read_only_reason", "seconds"} ACCEPTED_KEYS = {"delta_norm", "budget_charge", "budget_used", "budget_remaining"} SUMMARY_KEYS = {"pos", "pending", "budget_used", "read_only", "n_transactions", "cusum", "state_norms", "drift_from_anchor"} @pytest.fixture(scope="module") def api(tmp_path_factory): root = str(tmp_path_factory.mktemp("api")) data_dir = os.path.join(root, "data", "tiny") os.makedirs(data_dir) tok = Tokenizer.train(DOCS, vocab_size=300) tok.save(os.path.join(data_dir, "tokenizer.json")) n_train = encode_documents_to_bin(tok, DOCS, os.path.join(data_dir, "train.bin")) n_val = encode_documents_to_bin(tok, DOCS, os.path.join(data_dir, "validation.bin")) with open(os.path.join(data_dir, "meta.json"), "w", encoding="utf-8") as f: json.dump( { "corpus": "tiny", "vocab_size": tok.vocab_size, "splits": {"train": n_train, "validation": n_val, "test": 0}, "created_at_unix": int(time.time()), "tokenizer_lines": len(DOCS), }, f, ) text_id = train( TrainConfig( domain="text", model=ModelConfig(d_model=32, n_heads=2, n_layers=2, chunk=8, vocab_size=tok.vocab_size), artifacts_root=root, model_id="lm_tiny", data_dir=data_dir, steps=3, batch_size=2, seq_len=32, warmup_steps=1, eval_every=0, save_every=0, eval_batches=1, log_every=1, device="cpu", mqar_frac=0.0, ), log=lambda s: None, ) physics_id = train( TrainConfig( domain="physics", model=ModelConfig(domain="physics", d_model=32, n_heads=2, n_layers=1, chunk=8), artifacts_root=root, model_id="phys_tiny", steps=3, batch_size=2, seq_len=32, episodes_per_seq=2, warmup_steps=1, eval_every=0, save_every=0, eval_batches=1, log_every=1, device="cpu", use_muon=False, ), log=lambda s: None, ) store = ArtifactStore(root) CanarySuite.default_text( os.path.join(data_dir, "validation.bin"), vocab_size=tok.vocab_size, n_probe=2, probe_len=16 ).save(store.canary_path(text_id)) CanarySuite.default_physics(n_probe=2, steps=8).save(store.canary_path(physics_id)) app = create_app(root, device="cpu") with TestClient(app) as client: yield SimpleNamespace(client=client, store=store, root=root, data=data_dir, text=text_id, physics=physics_id) # ---------------------------------------------------------------------- health and data def test_health(api): body = api.client.get("/api/health").json() assert body["capabilities"] == {"create_session": True, "fork": True, "reset": True, "delete": True, "resume": True, "calibrate": True, "sleep": True} assert body["public"] is False assert body["ok"] is True assert body["artifacts_root"] == api.store.root and body["device"] == "cpu" assert body["n_models"] >= 2 and body["n_sessions"] >= 0 # ---------------------------------------------------------------------- models @pytest.mark.parametrize('backend', ['plastic', 'qwen']) def test_model_identity_survives_list_and_detail_without_inventing_config(tmp_path, backend): store = ArtifactStore(str(tmp_path)) record = {'domain': 'text', 'status': 'completed', 'params': 12} if backend == 'qwen': record['backend'] = backend store.register_model('model', record) with TestClient(create_app(str(tmp_path), device='cpu')) as client: assert client.get('/api/models').json()[0]['backend'] == backend detail = client.get('/api/models/model').json() assert detail['record']['backend'] == backend assert detail['config'] == {} def test_models_listing(api): body = api.client.get("/api/models").json() by_id = {m["model_id"]: m for m in body} assert {api.text, api.physics} <= set(by_id) text = by_id[api.text] assert text["domain"] == "text" and text["status"] == "completed" assert text["params"] > 0 and text["steps"] == 3 and text["has_canary"] is True assert text["calibrated"] is False assert text["eval"]["heldout_loss"] > 0 and "memory_value" in text["eval"] assert by_id[api.physics]["domain"] == "physics" def test_models_listing_survives_non_finite_eval_and_sleep(api): # A model whose consolidation ran against an empty canary set carries NaN in its sleep # manifest (score_suite returns NaN for an empty probe set); an unevaluated model's eval # can be NaN/Inf too. GET /api/models is the first call every dashboard tab makes, so a # single non-finite float there must render as null, not 500 the whole response. api.store.register_model( "nan_probe", { "domain": "text", "status": "completed", "params": 1, "eval": {"heldout_loss": float("nan"), "memory_value": float("inf")}, "sleep": { "accepted": True, "canary_before": {"coherence": float("nan"), "poison": 0.1}, "canary_after": {"coherence": float("nan"), "poison": float("-inf")}, }, }, ) resp = api.client.get("/api/models") assert resp.status_code == 200 rec = {m["model_id"]: m for m in resp.json()}["nan_probe"] assert rec["eval"]["heldout_loss"] is None and rec["eval"]["memory_value"] is None assert rec["sleep"]["canary_before"]["coherence"] is None and rec["sleep"]["canary_before"]["poison"] == 0.1 assert rec["sleep"]["canary_after"]["coherence"] is None and rec["sleep"]["canary_after"]["poison"] is None def test_model_detail(api): body = api.client.get(f"/api/models/{api.text}").json() assert body["record"]["model_id"] == api.text assert body["config"]["d_model"] == 32 and body["config"]["chunk"] == 8 assert body["eval"]["heldout_loss"] > 0 and "beta_hist" in body["eval"] # the detail eval keeps the histogram assert body["calibration"] is None assert body["canary"] == {"n_coherence": 2, "n_poison": 3} assert body["log"] and any("loss" in rec for rec in body["log"]) def test_calibrate_dispatches_pretrained_records_to_the_real_chat_calibration(api, tmp_path, monkeypatch): from plastic.harness import calibrate as calibrate_mod from plastic.harness.calibration_prompts import DEFAULT_CALIBRATION_PROMPTS ckpt = tmp_path / "ckpt" ckpt.mkdir() api.store.register_model("chat_pre", {"backend": "ttt", "domain": "text", "checkpoint_dir": str(ckpt), "params": 1}) calls = [] def fake_qwen(store, model_id, prompts, **kw): calls.append((model_id, list(prompts), kw)) return calibrate_mod.Calibration(thresholds={"chunk_loss": 2.0}, achievable_fpr={"chunk_loss": 0.1}, reference={"chunk_loss": [1.0, 2.0]}, n_chunks=2, target_fpr=kw["target_fpr"]) monkeypatch.setattr(calibrate_mod, "calibrate_qwen", fake_qwen) monkeypatch.setattr(calibrate_mod, "calibrate_model", lambda *a, **k: pytest.fail("toy path used for a pretrained record")) r = api.client.post("/api/models/chat_pre/calibrate", json={}) assert r.status_code == 200, r.text assert r.json()["thresholds"] == {"chunk_loss": 2.0} and r.json()["n_chunks"] == 2 assert calls[-1][1] == list(DEFAULT_CALIBRATION_PROMPTS) and calls[-1][2]["max_new_tokens"] == 32 r = api.client.post("/api/models/chat_pre/calibrate", json={"prompts": ["hi there"], "cusum_prompts": ["again"], "max_new_tokens": 4, "fpr": 0.2}) assert r.status_code == 200 and calls[-1][1] == ["hi there"] assert calls[-1][2]["cusum_prompts"] == ["again"] and calls[-1][2]["max_new_tokens"] == 4 and calls[-1][2]["target_fpr"] == 0.2 # request-shape family: empty prompt list and an out-of-range generation cap are rejected before any work assert api.client.post("/api/models/chat_pre/calibrate", json={"prompts": []}).status_code == 422 assert api.client.post("/api/models/chat_pre/calibrate", json={"max_new_tokens": 0}).status_code == 422 # a pretrained record whose checkpoint directory is gone is "no checkpoint yet" api.store.register_model("chat_gone", {"backend": "ttt", "domain": "text", "checkpoint_dir": str(tmp_path / "missing")}) assert api.client.post("/api/models/chat_gone/calibrate", json={}).status_code == 400 def test_sleep_routes_start_a_process_and_report_its_outcome(api, tmp_path, monkeypatch): """The route validates the record and body, starts one process per run, and reads the outcome back from the report the process writes. The process here is a stand-in that writes an accepted/rejected report.""" import sys from plastic.harness.config import HarnessConfig from plastic.api import sleep_jobs ckpt = tmp_path / "ckpt" ckpt.mkdir() api.store.register_model("chat_sleep", {"backend": "ttt", "domain": "text", "checkpoint_dir": str(ckpt), "params": 1}) api.store.create_session("teach", model_id="chat_sleep", domain="text", harness_cfg=HarnessConfig()) seen = {} def fake_argv(model_id, run_dir, artifacts_root, device, options, sessions, probes_path): seen.update(model_id=model_id, options=options, sessions=sessions, probes_path=probes_path, device=device) outcome = "rejected" if options.get("method") == "anchor" else "accepted" script = ( "import json,sys,os; d=sys.argv[1]; open(os.path.join(d,'log.txt'),'a').write('[sleep] stand-in\\n');" f"json.dump({{'run_id': os.path.basename(d), 'parent_model_id': {model_id!r}, 'status': {outcome!r}, 'model_id': 'sleep_child' if {outcome!r}=='accepted' else None," " 'gate': {'passed': " + ("False" if outcome == "rejected" else "True") + ", 'checks': []}, 'losses': [1.0]}, open(os.path.join(d,'sleep_report.json'),'w'))" ) return [sys.executable, "-c", script, run_dir] monkeypatch.setattr(sleep_jobs, "build_sleep_argv", fake_argv) # a toy record has no fast weights to consolidate assert api.client.post(f"/api/models/{api.text}/sleep", json={}).status_code == 400 assert api.client.post("/api/models/nope/sleep", json={}).status_code == 404 # request-shape family for bad in ({"method": "nap"}, {"steps": 0}, {"replay_ratio": 2}, {"sessions": []}, {"probes": [{"question": "", "answer": "x"}]}): assert api.client.post("/api/models/chat_sleep/sleep", json=bad).status_code == 422, bad assert api.client.post("/api/models/chat_sleep/sleep", json={"sessions": ["missing"]}).status_code == 400 r = api.client.post("/api/models/chat_sleep/sleep", json={"method": "distill", "steps": 3, "sessions": ["teach"], "probes": [{"question": "q", "answer": "a", "paraphrase": "q2"}]}) assert r.status_code == 200, r.text job = r.json() assert job["model_id"] == "chat_sleep" and job["status"] in ("running", "accepted") and job["n_probes"] == 1 assert seen["options"]["method"] == "distill" and seen["options"]["steps"] == 3 and seen["sessions"] == ["teach"] assert seen["probes_path"] and seen["probes_path"].endswith("probes.json") and seen["device"] == "cpu" run_id = job["run_id"] for _ in range(100): s = api.client.get(f"/api/sleep/{run_id}").json() if s["status"] != "running": break time.sleep(0.05) assert s["status"] == "accepted" and s["report"]["model_id"] == "sleep_child" and s["log_tail"] == ["[sleep] stand-in"] r2 = api.client.post("/api/models/chat_sleep/sleep", json={"method": "anchor"}).json() for _ in range(100): s2 = api.client.get(f"/api/sleep/{r2['run_id']}").json() if s2["status"] != "running": break time.sleep(0.05) assert s2["status"] == "rejected" and s2["report"]["gate"]["passed"] is False listing = api.client.get("/api/sleep").json() assert {j["run_id"] for j in listing} >= {run_id, r2["run_id"]} and all("losses" not in (j["report"] or {}) for j in listing) assert api.client.get("/api/sleep/nope").status_code == 404 def test_model_unknown_is_404(api): assert api.client.get("/api/models/nope").status_code == 404 assert api.client.post("/api/models/nope/calibrate", json={}).status_code == 404 def test_calibrate(api): body = api.client.post( f"/api/models/{api.text}/calibrate", json={"data_dir": api.data, "chunks": 16, "fisher_chunks": 2, "fpr": 0.1}, ).json() assert body["n_chunks"] >= 16 and body["target_fpr"] == 0.1 assert body["thresholds"] and all(v is None or isinstance(v, float) for v in body["thresholds"].values()) assert body["reference_sizes"] and "reference" not in body assert body["created_at_unix"] > 0 # the rate the sample can support, which is not the requested target_fpr assert set(body["achievable_fpr"]) <= set(body["thresholds"]) assert body["achievable_fpr"] and all(v > 0 for v in body["achievable_fpr"].values()) detail = api.client.get(f"/api/models/{api.text}").json() assert detail["record"]["calibrated"] is True assert detail["calibration"]["thresholds"] == body["thresholds"] assert detail["calibration"]["achievable_fpr"] == body["achievable_fpr"] def test_an_unbounded_threshold_is_null(api): """A threshold can be infinite (an unbounded bound); JSON carries it as null.""" path = os.path.join(api.store.model_dir(api.text), "calibration.json") original = open(path, "rb").read() payload = json.loads(original) signal = sorted(payload["thresholds"])[0] payload["thresholds"][signal] = float("inf") try: with open(path, "w", encoding="utf-8") as f: json.dump(payload, f) body = api.client.get(f"/api/models/{api.text}") assert body.status_code == 200 assert body.json()["calibration"]["thresholds"][signal] is None finally: with open(path, "wb") as f: f.write(original) # ---------------------------------------------------------------------- sessions def test_create_sessions(api): text = api.client.post("/api/sessions", json={"model_id": api.text, "session_id": "t1", "harness": {"enable_projection": False}}) assert text.status_code == 200 body = text.json() assert body["session_id"] == "t1" and body["domain"] == "text" and body["model_id"] == api.text assert body["pos"] == 0 and body["n_transactions"] == 0 and body["read_only"] is False physics = api.client.post("/api/sessions", json={"model_id": api.physics, "session_id": "p1"}).json() assert physics["session_id"] == "p1" and physics["domain"] == "physics" auto = api.client.post("/api/sessions", json={"model_id": api.text}).json() assert auto["session_id"] and auto["session_id"] not in ("t1", "p1") api.client.delete(f"/api/sessions/{auto['session_id']}") def test_create_session_errors(api): assert api.client.post("/api/sessions", json={"model_id": "nope"}).status_code == 404 assert api.client.post("/api/sessions", json={"model_id": api.text, "session_id": "t1"}).status_code == 409 bad = api.client.post("/api/sessions", json={"model_id": api.text, "session_id": "bad", "harness": {"nope": 1}}) assert bad.status_code == 400 and "nope" in bad.json()["detail"] def test_sessions_listing(api): body = api.client.get("/api/sessions").json() by_id = {s["session_id"]: s for s in body} assert {"t1", "p1"} <= set(by_id) assert by_id["t1"]["model_id"] == api.text and by_id["p1"]["domain"] == "physics" def test_chat(api): body = api.client.post("/api/sessions/t1/chat", json={"prompt": PROMPT, "max_new_tokens": 4, "seed": 0}).json() assert body["prompt"] == PROMPT and isinstance(body["completion"], str) assert body["n_tokens_in"] > 0 and body["n_tokens_out"] <= 4 assert body["transactions"] and TRANSACTION_KEYS <= set(body["transactions"][0]) assert body["transactions"][0]["decision"]["kind"] in ("commit", "rollback", "scale", "project", "readonly") signals = body["transactions"][0]["signals"] assert signals["n_tokens"] > 0 and "log_delta_norm" in signals and "z" in signals # signals are the proposed update; accepted is what the decision actually committed accepted = body["transactions"][0]["accepted"] assert ACCEPTED_KEYS <= set(accepted) and accepted["delta_norm"] >= 0.0 assert "canary_delta_coherence" in accepted assert SUMMARY_KEYS <= set(body["summary"]) def test_session_detail(api): body = api.client.get("/api/sessions/t1").json() assert body["meta"]["session_id"] == "t1" assert body["meta"]["harness"]["enable_projection"] is False and body["meta"]["model_signature"] assert body["lineage"] == ["t1"] assert body["summary"]["n_transactions"] == body["meta"]["n_transactions"] assert body["summary"]["state_norms"]["s_norm"] and body["summary"]["cusum"] assert body["transactions"] and TRANSACTION_KEYS <= set(body["transactions"][0]) assert body["trace"] and body["trace"][0]["kind"] == "chat" def test_transactions_paging(api): full = api.client.get("/api/sessions/t1/transactions", params={"limit": 100}).json() assert full["total"] == len(full["items"]) >= 1 page = api.client.get("/api/sessions/t1/transactions", params={"limit": 1, "offset": 0}).json() assert page["total"] == full["total"] and len(page["items"]) == 1 assert page["items"][0]["index"] == full["items"][0]["index"] empty = api.client.get("/api/sessions/t1/transactions", params={"limit": 5, "offset": full["total"] + 10}).json() assert empty["items"] == [] and empty["total"] == full["total"] def test_state(api): body = api.client.get("/api/sessions/t1/state").json() assert body["kind"] == "plastic" # the toy per-layer S/h shape, discriminated for the client assert body["pos"] > 0 and len(body["layers"]) == 2 layer = body["layers"][0] assert len(layer["s_norm_per_head"]) == 2 and layer["h_norm"] >= 0.0 assert len(layer["singular_values"]) == 2 and len(layer["singular_values"][0]) <= 8 assert layer["drift_from_anchor"] >= 0.0 def test_state_payload_native_backend_has_no_plastic_layers(): # ASTRA-081 #5: a pretrained (Qwen) session's committed state has no per-layer S/h shape, so the # /state route must return an honest recurrent payload instead of a 500 (committed.layers) or # fabricated tensors. Driven with a fake session so it needs no model/isolated deps. import torch from plastic.api.service import state_payload class _QwenLikeState: # no `.layers`, exactly like QwenState pass class _FakeBackend: def state_norms(self, _s): return {"recurrent_norm": [1.5, 2.0], "recurrent_norm_total": 2.5} def state_delta(self, _a, _b): return [torch.tensor([3.0, 4.0]), torch.tensor([0.0])] # per-unit norms 5.0, 0.0 class _FakeRunner: committed = _QwenLikeState() anchor = _QwenLikeState() pos = 12 backend = _FakeBackend() class _FakeSession: runner = _FakeRunner() backend_kind = "qwen" body = state_payload(_FakeSession()) assert body["kind"] == "recurrent" and body["backend"] == "qwen" and body["pos"] == 12 assert body["recurrent_norm_total"] == 2.5 assert [u["recurrent_norm"] for u in body["units"]] == [1.5, 2.0] assert body["units"][0]["drift_from_anchor"] == 5.0 and body["units"][1]["drift_from_anchor"] == 0.0 assert "layers" not in body # never the plastic shape, never fabricated S/h tensors def test_fork(api): child = api.client.post("/api/sessions/t1/fork", json={"child_session_id": "t1_fork"}).json() assert child["session_id"] == "t1_fork" and child["parent_session_id"] == "t1" assert child["root_session_id"] == "t1" and child["pos"] == api.store.load_session_meta("t1")["pos"] grand = api.client.post("/api/sessions/t1_fork/fork", json={}).json() detail = api.client.get(f"/api/sessions/{grand['session_id']}").json() assert detail["lineage"] == ["t1", "t1_fork", grand["session_id"]] assert api.client.post("/api/sessions/t1/fork", json={"child_session_id": "t1_fork"}).status_code == 409 assert api.client.post("/api/sessions/nope/fork", json={}).status_code == 404 def test_reset_and_resume(api): before = api.client.get("/api/sessions/t1_fork").json()["meta"] assert before["pos"] > 0 reset = api.client.post("/api/sessions/t1_fork/reset").json() assert reset["pos"] == 0 and reset["read_only"] is False resumed = api.client.post("/api/sessions/t1_fork/resume").json() assert resumed["read_only"] is False and resumed["session_id"] == "t1_fork" assert api.client.post("/api/sessions/nope/reset").status_code == 404 assert api.client.post("/api/sessions/nope/resume").status_code == 404 def test_delete(api): created = api.client.post("/api/sessions", json={"model_id": api.text, "session_id": "doomed"}).json() assert created["session_id"] == "doomed" assert api.client.delete("/api/sessions/doomed").json() == {"deleted": True} assert api.client.get("/api/sessions/doomed").status_code == 404 assert api.client.delete("/api/sessions/doomed").status_code == 404 assert "doomed" not in {s["session_id"] for s in api.client.get("/api/sessions").json()} def test_session_unknown_is_404(api): assert api.client.get("/api/sessions/nope").status_code == 404 assert api.client.get("/api/sessions/nope/state").status_code == 404 assert api.client.get("/api/sessions/nope/transactions").status_code == 404 assert api.client.post("/api/sessions/nope/chat", json={"prompt": "x"}).status_code == 404 # ---------------------------------------------------------------------- training jobs # ---------------------------------------------------------------------- red team # ---------------------------------------------------------------------- sleep def test_model_summary_exposes_sleep_field(): from plastic.api.service import model_summary from plastic.store import ArtifactStore store = ArtifactStore("/nonexistent-root-for-summary-only") rec = {"model_id": "m", "domain": "text", "type": "sleep_consolidation", "parent_model_id": "base", "sleep": {"accepted": True, "delta_coherence": 0.01, "delta_poison": -0.2}} out = model_summary(store, rec) assert out["sleep"] == rec["sleep"] assert out["type"] == "sleep_consolidation" and out["parent_model_id"] == "base" # a model without a sleep record reports None, not a fabricated value assert model_summary(store, {"model_id": "m2", "domain": "text"})["sleep"] is None def test_state_payload_native_missing_or_failed_drift_is_null(): # ASTRA-094: a failed / missing / nonfinite measurement is null (unavailable), never a measured # zero; a genuine zero is preserved. Driven with fakes, no model. import torch from plastic.api.service import state_payload class _QwenLikeState: pass def _sess(norms, delta): class _Backend: def state_norms(self, _s): return norms def state_delta(self, _a, _b): if isinstance(delta, Exception): raise delta return delta class _Runner: committed = _QwenLikeState() anchor = _QwenLikeState() pos = 5 backend = _Backend() class _Session: runner = _Runner() backend_kind = "qwen" return _Session() # state_delta raises -> every drift null, but the recurrent norms are still reported body = state_payload(_sess({"recurrent_norm": [1.0, 2.0], "recurrent_norm_total": 3.0}, RuntimeError("boom"))) assert [u["drift_from_anchor"] for u in body["units"]] == [None, None] assert [u["recurrent_norm"] for u in body["units"]] == [1.0, 2.0] # a short delta list -> the unmatched unit is null; a genuine zero drift is preserved as 0.0 body = state_payload(_sess({"recurrent_norm": [1.0, 2.0, 3.0], "recurrent_norm_total": 4.0}, [torch.tensor([0.0]), torch.tensor([4.0])])) assert [u["drift_from_anchor"] for u in body["units"]] == [0.0, 4.0, None] # an absent total is null, not zero body = state_payload(_sess({"recurrent_norm": [1.0]}, [torch.tensor([2.0])])) assert body["recurrent_norm_total"] is None # nonfinite norm / total / drift are all null body = state_payload(_sess({"recurrent_norm": [float("inf")], "recurrent_norm_total": float("nan")}, [torch.tensor([float("inf")])])) assert body["units"][0]["recurrent_norm"] is None assert body["units"][0]["drift_from_anchor"] is None assert body["recurrent_norm_total"] is None def test_session_detail_exposes_loaded_calibration_not_replaced_artifact(api): # ASTRA-096 #2: a session's active calibration is the one it verified at open, not the model's # CURRENT saved artifact. A separate calibration process replacing the artifact same-model must # not change what the open session (and its running policy) uses or what the UI labels active. from plastic.harness.calibrate import Calibration mid = api.text api.client.post(f"/api/models/{mid}/calibrate", json={"data_dir": api.data, "chunks": 16, "fisher_chunks": 2, "fpr": 0.1}) model_dir = api.store.model_dir(mid) sid = api.client.post("/api/sessions", json={"model_id": mid, "session_id": "cal_identity"}).json()["session_id"] detail = api.client.get(f"/api/sessions/{sid}").json() assert detail["summary"]["calibration"] == "installed" assert detail["calibration"] is not None loaded_thresholds = detail["calibration"]["thresholds"] assert loaded_thresholds # the session exposes its loaded calibration's thresholds # a SEPARATE calibration process replaces the saved artifact with a valid same-signature one whose # thresholds differ (bumped by 100); the model signature is unchanged so it is still "installable" replacement = Calibration.load(model_dir) replacement.thresholds = {k: (None if v is None else float(v) + 100.0) for k, v in replacement.thresholds.items()} replacement.save(model_dir) model_after = api.client.get(f"/api/models/{mid}").json() assert model_after["calibration"]["thresholds"] != loaded_thresholds # the model detail shows the replacement # the OPEN session still exposes the calibration it actually loaded, not the replacement detail_after = api.client.get(f"/api/sessions/{sid}").json() assert detail_after["summary"]["calibration"] == "installed" assert detail_after["calibration"]["thresholds"] == loaded_thresholds assert detail_after["calibration"]["thresholds"] != model_after["calibration"]["thresholds"] def test_session_summary_reports_effective_generation_write_policy(api): # ASTRA-101: writes_generation is the EFFECTIVE policy (backend source capability OR # learn_from_generation). A plastic session freezes generation by default; the flag makes it # write-eligible. (Qwen's generation writes regardless -- that path is model-gated.) sid = api.client.post("/api/sessions", json={"model_id": api.text}).json()["session_id"] body = api.client.get(f"/api/sessions/{sid}").json() assert body["summary"]["backend"] == "plastic" assert body["summary"]["writes_generation"] is False # default: generation frozen read-only sid2 = api.client.post( "/api/sessions", json={"model_id": api.text, "harness": {"learn_from_generation": True}} ).json()["session_id"] body2 = api.client.get(f"/api/sessions/{sid2}").json() assert body2["summary"]["writes_generation"] is True # the override makes generation write-eligible def test_sleep_argv_forwards_the_replay_revision_and_rejects_a_malformed_one(api): """ASTRA-181: the API's replay_revision reaches the sleep process; a non-hex value is a request-shape error.""" from plastic.api.schemas import SleepRequest from plastic.api.sleep_jobs import build_sleep_argv argv = build_sleep_argv("m", "/run", "/root", "cpu", {"method": "replay", "replay_revision": "5feaf2fd3ffca7"}, None, None) assert argv[argv.index("--replay-revision") + 1] == "5feaf2fd3ffca7" assert "--replay-revision" not in build_sleep_argv("m", "/run", "/root", "cpu", {"method": "replay"}, None, None) assert SleepRequest().model_dump(exclude_none=True).get("replay_revision") is None assert api.client.post("/api/models/nope/sleep", json={"replay_revision": "not-a-sha!"}).status_code == 422