""" Simulated four-week run of a gated CRM-update agent. Pipeline per run (one proposed write to Salesforce): 1. Intake + field extraction DeepSeek V4.1-Flash 2. Draft the write DeepSeek V4.1-Flash 3. Gate Jev typed judgment, returns p(write is correct) p >= auto_threshold -> auto-write person_threshold <= p < auto_threshold -> retry on Sonnet 5, re-gate p < person_threshold -> to a person 4. Retry (if gated) Claude Sonnet 5 redraft, re-gate with Jev p >= retry_auto_threshold -> auto-write, else -> to a person 5. Person review 3.5 min at $48/hr loaded; 2% spot-check on auto-writes Events that skew it (kept on purpose): - Day 9: drafting provider outage for one afternoon, ~40% of that day's runs skip DeepSeek and draft straight on Sonnet. - Day 15: person_threshold raised from 0.70 to 0.75 after a wrong write got through in week 2. Stays at 0.75 for weeks 3 and 4. Pricing (USD per million tokens, October 2026 list prices): Jev 0.042 in, 0 out DeepSeek V4.1-Flash 0.30 in, 1.20 out (peak) Claude Sonnet 5 2.00 in, 10.00 out Seed is fixed. Change SEED and re-run; the numbers move a little, the shape does not. """ import numpy as np SEED = 7102026 rng = np.random.default_rng(SEED) DAYS = 28 PRICE = { # per token "jev": (0.042 / 1e6, 0.0), "deepseek": (0.30 / 1e6, 1.20 / 1e6), "sonnet": (2.00 / 1e6, 10.00 / 1e6), } REVIEW_MIN = 3.5 SPOT_MIN = 2.0 SPOT_RATE = 0.02 HOURLY = 48.0 AUTO_T = 0.88 RETRY_AUTO_T = 0.85 PERSON_T_BEFORE = 0.70 PERSON_T_AFTER = 0.75 THRESHOLD_CHANGE_DAY = 15 OUTAGE_DAY = 9 BASE_WRONG_P = 0.065 # latent prob a DeepSeek draft is wrong SONNET_FIX_P = 0.72 # Sonnet retry fixes a wrong draft this often SONNET_WRONG_P = 0.025 # Sonnet drafting from scratch is wrong this often def tokens(mean, sd, n): return np.clip(rng.normal(mean, sd, n), mean * 0.4, None).round() def gate_prob(is_wrong, n): # Jev-style calibrated probability of "this write is correct". # Two populations each side, so the gate has real false alarms and real misses. ambiguous = rng.random(n) < 0.10 # correct but the record looks messy looks_fine = rng.random(n) < 0.18 # wrong but the draft reads cleanly p_correct = np.where(ambiguous, rng.beta(4.0, 2.0, n), rng.beta(24.0, 1.45, n)) p_wrong = np.where(looks_fine, rng.beta(7.0, 1.8, n), rng.beta(2.0, 4.5, n)) return np.where(is_wrong, p_wrong, p_correct) rows = [] for day in range(1, DAYS + 1): weekday = (day - 1) % 7 < 5 lam = 95 if weekday else 20 n = rng.poisson(lam) if n == 0: continue person_t = PERSON_T_BEFORE if day < THRESHOLD_CHANGE_DAY else PERSON_T_AFTER # difficulty varies per run; harder records are wrong more often difficulty = rng.beta(2, 5, n) wrong_p = np.clip(BASE_WRONG_P * (0.5 + difficulty * 2.0), 0.01, 0.35) outage = np.zeros(n, dtype=bool) if day == OUTAGE_DAY: outage = rng.random(n) < 0.40 cost_model = np.zeros(n) cost_review = np.zeros(n) sonnet_used = np.zeros(n, dtype=bool) retried = np.zeros(n, dtype=bool) to_person = np.zeros(n, dtype=bool) draft_wrong = np.zeros(n, dtype=bool) final_wrong_if_written = np.zeros(n, dtype=bool) # Step 1: intake/extraction on DeepSeek (skipped if outage -> Sonnet) t_in = tokens(2200, 450, n) t_out = tokens(350, 70, n) for i in range(n): m = "sonnet" if outage[i] else "deepseek" cost_model[i] += t_in[i] * PRICE[m][0] + t_out[i] * PRICE[m][1] sonnet_used[i] |= outage[i] # Step 2: draft t_in = tokens(3000, 600, n) t_out = tokens(450, 90, n) for i in range(n): m = "sonnet" if outage[i] else "deepseek" cost_model[i] += t_in[i] * PRICE[m][0] + t_out[i] * PRICE[m][1] draft_wrong[i] = rng.random() < (SONNET_WRONG_P if outage[i] else wrong_p[i]) # Step 3: gate t_in = tokens(1800, 300, n) cost_model += t_in * PRICE["jev"][0] p = gate_prob(draft_wrong, n) auto = p >= AUTO_T retry = (p >= person_t) & (p < AUTO_T) person = p < person_t final_wrong_if_written[auto] = draft_wrong[auto] # Step 4: retry on Sonnet then re-gate idx = np.where(retry)[0] if idx.size: retried[idx] = True sonnet_used[idx] = True t_in = tokens(4500, 700, idx.size) t_out = tokens(500, 100, idx.size) cost_model[idx] += t_in * PRICE["sonnet"][0] + t_out * PRICE["sonnet"][1] fixed = rng.random(idx.size) < SONNET_FIX_P still_wrong = np.where(draft_wrong[idx], ~fixed, rng.random(idx.size) < SONNET_WRONG_P) t_in = tokens(1800, 300, idx.size) cost_model[idx] += t_in * PRICE["jev"][0] p2 = gate_prob(still_wrong, idx.size) auto2 = p2 >= RETRY_AUTO_T final_wrong_if_written[idx[auto2]] = still_wrong[auto2] person[idx[~auto2]] = True # for items that go to a person after retry, record whether the write would have been wrong draft_wrong[idx] = still_wrong # what the person would have seen to_person = person cost_review[to_person] = REVIEW_MIN / 60 * HOURLY # spot checks on auto-writes written = ~to_person spot = written & (rng.random(n) < SPOT_RATE) cost_review[spot] += SPOT_MIN / 60 * HOURLY for i in range(n): rows.append(dict( day=day, week=(day - 1) // 7 + 1, outage=outage[i], retried=retried[i], to_person=to_person[i], sonnet_used=sonnet_used[i], person_would_be_wrong=bool(to_person[i] and draft_wrong[i]), wrong_written=bool((not to_person[i]) and final_wrong_if_written[i]), cost_model=cost_model[i], cost_review=cost_review[i], )) import pandas as pd df = pd.DataFrame(rows) N = len(df) to_person = df.to_person.sum() retried = df.retried.sum() escalated_any = (df.to_person | df.retried).sum() wrong_caught = df.person_would_be_wrong.sum() wrong_written = df.wrong_written.sum() model_spend = df.cost_model.sum() review_spend = df.cost_review.sum() resolved = N # every run ends resolved: auto-written or handled by a person print(f"SEED {SEED}") print(f"Runs (gated writes), 28 days: {N}") print(f"To a person: {to_person} ({to_person/N:.1%})") print(f"Retried on Sonnet: {retried} ({retried/N:.1%})") print(f"Escalated (person or retry): {escalated_any} ({escalated_any/N:.1%})") print(f"Of those sent to a person, would have been wrong: {wrong_caught} ({wrong_caught/to_person:.1%} of person-sends)") print(f"Wrong writes that got through the gate: {wrong_written} ({wrong_written/N:.2%} of runs)") print(f"Model spend total: ${model_spend:,.2f} -> ${model_spend/resolved:.4f} per resolved write (model only)") print(f"Review time total: ${review_spend:,.2f} -> ${(model_spend+review_spend)/resolved:.4f} per resolved write (with review)") print(f"Share of model spend on Sonnet runs: {df[df.sonnet_used].cost_model.sum()/model_spend:.0%} across {df.sonnet_used.mean():.1%} of runs") print() print("By week:") wk = df.groupby("week").agg( runs=("day", "size"), to_person=("to_person", "sum"), retried=("retried", "sum"), wrong_caught=("person_would_be_wrong", "sum"), wrong_written=("wrong_written", "sum"), model=("cost_model", "sum"), review=("cost_review", "sum"), ) wk["person_%"] = (wk.to_person / wk.runs * 100).round(1) wk["escalated_%"] = ((wk.to_person + wk.retried) / wk.runs * 100).round(1) wk["model_per_write"] = (wk.model / wk.runs).round(4) wk["all_in_per_write"] = ((wk.model + wk.review) / wk.runs).round(4) print(wk.to_string()) print() print("Day 9 (outage):", df[df.day == 9].outage.sum(), "of", (df.day == 9).sum(), "runs drafted on Sonnet") # Counterfactuals for the post all_sonnet_cost = ( (tokens(2200, 450, N) * PRICE["sonnet"][0] + tokens(350, 70, N) * PRICE["sonnet"][1]).sum() + (tokens(3000, 600, N) * PRICE["sonnet"][0] + tokens(450, 90, N) * PRICE["sonnet"][1]).sum() ) print() print(f"Counterfactual, every step on Sonnet, no gate: ${all_sonnet_cost/N:.4f} per write (model only), ~{SONNET_WRONG_P:.1%} wrong writes with nothing catching them") no_gate_wrong = int(round(BASE_WRONG_P * 1.1 * N)) print(f"Counterfactual, DeepSeek only, no gate: roughly {no_gate_wrong} wrong writes reach Salesforce in the month") df.to_csv("runs.csv", index=False) wk.to_csv("weekly.csv")