debug_subscription_billing_pipeline — Multi-Report Consistency Debugging
ranked by score ↓Source
Paste as source: in your trap.yaml
git+https://github.com/trapstreet/trapstreet-tasks@4d7400a7e5c9ace5b4db3f9d6c89b73777419dbc#subdirectory=tasks/debug_subscription_billing_pipelineShare
debug-subscription-billing-pipeline
An open-source evaluation task for cross-file, cross-report consistency debugging — when a ticket asks for a change to a SaaS billing pipeline, does the agent identify ALL the places that need updating so that FOUR reports (each resolving the same underlying facts through a d
6 cases
Each case feeds files from inputs/<id>/ to the solution, expects files in expected/<id>/, and is scored by judge.py then aggregated by grader.py.
cases (6)
▸case_01Customer region migration effective a specific date; must fix live current-state AND one wrongly-baked historical invoice, leave the prior invoice untouched
input
README.md
# Task
A colleague filed a ticket asking for a change to the subscription billing pipeline.
**Your goal:** ensure all FOUR reports (`billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports these four scripts produce match the reports they would produce after a correct fix**. The judge applies your edits, runs all four scripts, and compares stdout to the gold reports.
## Business context
This is a SaaS company's subscription billing system. Customers subscribe to plans (which belong to tiers), can add optional add-ons, and may have a discount code applied. Each billing period an invoice is generated and its financial fields are BAKED into `invoices.csv` at that point in time -- they are a historical record, not something that gets silently recomputed later.
The four reports read the same underlying facts through DIFFERENT paths:
- `billing_summary.py` -- LIVE: "if we billed today," fully resolved from current customers/plans/tiers/addons/discount_codes/regions. No invoice history involved.
- `invoice_detail.py` -- pure BAKED: reads `invoices.csv`'s own columns directly, no lookups at all. This is the historical record customers see.
- `finance_ledger.py` -- MIXED: pretax subtotal always from baked invoice columns; tax is baked-tax-if-present, else a LIVE fallback lookup (invoice -> subscription -> customer -> region).
- `customer_statement.py` -- MIXED differently: base price stays BAKED (grandfather pricing), but add-on names/prices, discount, and tax are all resolved LIVE against current tables.
Because each report resolves differently, the SAME underlying change can require touching different combinations of files. A change to "current state" tables (plans/addons/discount_codes/customers) flows automatically into the two LIVE reports. A change that must also correct HISTORICAL invoices requires directly editing the affected row(s) in `invoices.csv` -- and only the affected row(s); untouched historical invoices must stay untouched.
## How to approach this
- The data tables are meant to be read in full, but do not touch rows the ticket doesn't call for -- edits to invoices outside the ticket's stated scope will be scored as incorrect, even if well-intentioned.
- Read all four report scripts to understand exactly which table/column each one reads and whether that path is baked, live, or a fallback of one to the other.
- If a ticket changes a rate/price/code definition, decide: does this only affect the live view going forward, or does a specific historical invoice also need its baked figures corrected?
## Files in this directory
- `ticket.md` -- the change request
- `customers.csv`, `subscriptions.csv`, `plans.csv`, `tiers.csv`, `addons.csv`, `discount_codes.csv`, `regions.csv` -- current-state master tables
- `invoices.csv` -- historical baked invoice records
- `support_tickets.csv`, `marketing_campaigns.csv` -- unrelated auxiliary tables (not used by any report)
- `billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py` -- the four report scripts
## Output format
Emit a JSON array of edits to stdout. Nothing else -- no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Only your first 30 edits will be applied by the judge -- padding the list with extra guesses past that point does not help.
Output ONLY the JSON array.
addons.csv
addon_id,addon_name,addon_price_usd
ADDON-SSO,Single Sign-On,49.0
ADDON-API,API Access,19.0
ADDON-PRIORITY,Priority Support,39.0
ADDON-AUDIT,Audit Log Export,25.0
billing_summary.py
"""Billing summary -- LIVE current-state view of what each active
subscription would be billed TODAY.
Reads: customers.csv, subscriptions.csv, plans.csv, tiers.csv, addons.csv,
discount_codes.csv, regions.csv
Everything here is resolved via LOOKUP against the CURRENT tables (no
invoice history involved). If a plan's price changes, or a customer's
region changes, or an add-on is renamed/repriced, this report reflects
the new value immediately for every active subscription -- it is the
"if we billed today" view, not a historical record.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
plans = index_by(load_csv(here / "plans.csv"), "plan_id")
tiers = index_by(load_csv(here / "tiers.csv"), "tier_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
subscriptions = load_csv(here / "subscriptions.csv")
rows = []
for sub in subscriptions:
if sub.get("status") != "active":
continue
customer = customers.get(sub["customer_id"])
plan = plans.get(sub["plan_id"])
if not customer or not plan:
continue
tier = tiers.get(plan.get("tier_id"), {})
region = regions.get(customer.get("region_id"), {})
base_price = to_num(plan.get("base_price_usd")) or 0.0
addon_total = 0.0
for aid in split_ids(sub.get("addon_ids")):
addon = addons.get(aid)
if addon:
addon_total += to_num(addon.get("addon_price_usd")) or 0.0
subtotal = base_price + addon_total
discount = 0.0
code = (sub.get("discount_code") or "").strip()
if code and code in discount_codes:
dc = discount_codes[code]
pct = to_num(dc.get("discount_pct"))
fixed = to_num(dc.get("discount_fixed_usd"))
if pct:
discount = subtotal * pct
elif fixed:
discount = fixed
after_discount = subtotal - discount
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"subscription_id": sub["subscription_id"],
"customer_name": customer.get("customer_name", ""),
"plan_name": plan.get("plan_name", ""),
"tier_name": tier.get("tier_name", ""),
"region_name": region.get("region_name", ""),
"base_price_usd": f"{base_price:.2f}",
"addon_total_usd": f"{addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: r["subscription_id"])
cols = ["subscription_id", "customer_name", "plan_name", "tier_name",
"region_name", "base_price_usd", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customer_statement.py
"""Customer statement -- all-time invoice history re-rendered with
CURRENT add-on names/prices and CURRENT customer region, but the
historical BASE PRICE stays locked at what was actually baked (grandfather
pricing: a plan price hike does not retroactively re-bill old invoices).
Reads: invoices.csv, subscriptions.csv, addons.csv, discount_codes.csv,
customers.csv, regions.csv
Resolution per invoice:
- base price: BAKED (invoice's own baked_base_price_usd) -- historical lock-in
- add-on names/total: LIVE lookup via subscription.addon_ids -> addons.csv
(a rename or reprice shows up on EVERY invoice immediately, including old ones)
- discount: LIVE lookup via subscription.discount_code -> discount_codes.csv,
using whichever of pct-discount or fixed-discount is LARGER for the
customer (best-of rule, distinct from billing_summary's pct-or-fixed rule)
- tax: LIVE lookup via subscription -> customer -> CURRENT region (a
customer's region change re-taxes their ENTIRE history when this report
is regenerated, not just future invoices)
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
sub = subscriptions.get(inv.get("subscription_id"), {})
addon_ids = split_ids(sub.get("addon_ids"))
live_addon_names = "; ".join(
addons[aid]["addon_name"] for aid in sorted(addon_ids) if aid in addons
)
live_addon_total = sum(
to_num(addons[aid].get("addon_price_usd")) or 0.0
for aid in addon_ids if aid in addons
)
base = to_num(inv.get("baked_base_price_usd")) or 0.0
subtotal = base + live_addon_total
code = (sub.get("discount_code") or "").strip()
discount = 0.0
if code and code in discount_codes:
dc = discount_codes[code]
pct_discount = (subtotal * to_num(dc.get("discount_pct"))) if to_num(dc.get("discount_pct")) else 0.0
fixed_discount = to_num(dc.get("discount_fixed_usd")) or 0.0
discount = max(pct_discount, fixed_discount)
after_discount = subtotal - discount
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"invoice_id": inv["invoice_id"],
"period": inv.get("period", ""),
"addon_names": live_addon_names,
"addon_total_usd": f"{live_addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "period", "addon_names", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customers.csv
customer_id,customer_name,region_id,signup_date
CUST-001,Acme Corp,REG-CA,2022-01-10
CUST-002,Globex Ltd,REG-UK,2022-03-22
discount_codes.csv
code,discount_pct,discount_fixed_usd,expires_on
WELCOME10,0.1,,2099-12-31
LAUNCH50,,50.0,2099-12-31
PARTNER20,0.2,,2099-12-31
finance_ledger.py
"""Finance ledger -- reconciliation report Finance runs against invoice_detail.
Reads: invoices.csv, subscriptions.csv, customers.csv, regions.csv
The pretax subtotal ALWAYS comes from the invoice's own BAKED columns
(base + addon - discount, all baked). Tax is COALESCE(baked_tax_usd, LIVE
recompute): if the invoice has a baked tax figure, trust it; only when
baked_tax_usd is blank does this report fall back to a LIVE lookup
(invoice -> subscription -> customer -> region -> tax_rate_pct) to fill
the gap. This means a baked tax correction on one invoice is picked up
here automatically (same source column as invoice_detail), while a
missing/never-baked tax figure gets computed fresh from current customer
region.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
base = to_num(inv.get("baked_base_price_usd")) or 0.0
addon_total = to_num(inv.get("baked_addon_total_usd")) or 0.0
discount = to_num(inv.get("baked_discount_applied_usd")) or 0.0
pretax = base + addon_total - discount
tax = to_num(inv.get("baked_tax_usd"))
if tax is None:
sub = subscriptions.get(inv.get("subscription_id"), {})
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = pretax * tax_rate / 100.0
total = pretax + tax
rows.append({
"invoice_id": inv["invoice_id"],
"subscription_id": inv.get("subscription_id", ""),
"period": inv.get("period", ""),
"pretax_subtotal_usd": f"{pretax:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "subscription_id", "period",
"pretax_subtotal_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
print(",".join(str(r[c]) for c in cols))
if __name__ == "__main__":
main()
invoice_detail.py
"""Invoice detail -- the historical record customers see on their invoices.
Reads: invoices.csv ONLY.
Every field is read directly from the invoice's own BAKED columns -- no
lookups against customers/plans/addons/discount_codes/regions. This is
what actually went out to the customer at billing time, so it must NOT
change just because current-state tables (plan prices, addon names,
customer region, discount code definitions) change later. Only editing
a specific invoice row changes what this report shows for that row.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
invoices.sort(key=lambda r: (r.get("period", ""), r.get("invoice_id", "")))
cols = ["invoice_id", "period", "baked_customer_name", "baked_plan_name",
"baked_addon_names", "baked_base_price_usd", "baked_addon_total_usd",
"baked_discount_applied_usd", "baked_tax_usd", "baked_total_usd", "status"]
print(",".join(cols))
for r in invoices:
vals = [r.get(c, "") for c in cols]
vals = [f'"{v}"' if "," in v else v for v in vals]
print(",".join(vals))
if __name__ == "__main__":
main()
invoices.csv
invoice_id,subscription_id,period,baked_customer_name,baked_plan_name,baked_addon_names,baked_base_price_usd,baked_addon_total_usd,baked_discount_applied_usd,baked_tax_usd,baked_total_usd,status
INV-1001,SUB-001,2024-01,Acme Corp,Growth Monthly,API Access; Single Sign-On,99.0,68.0,0.0,14.2,181.19,paid
INV-1002,SUB-001,2024-02,Acme Corp,Growth Monthly,API Access; Single Sign-On,99.0,68.0,0.0,14.2,181.19,paid
INV-1003,SUB-002,2024-02,Globex Ltd,Scale Monthly,Priority Support,299.0,39.0,0.0,67.6,405.6,paid
marketing_campaigns.csv
campaign_id,region_id,campaign_name,start_date
CMP-001,REG-CA,Spring Launch,2024-02-01
CMP-002,REG-UK,EU Expansion,2024-03-01
plans.csv
plan_id,plan_name,tier_id,base_price_usd
PLAN-STARTER,Starter Monthly,TIER-STARTER,29.0
PLAN-GROWTH,Growth Monthly,TIER-GROWTH,99.0
PLAN-SCALE,Scale Monthly,TIER-SCALE,299.0
PLAN-ENTERPRISE,Enterprise Monthly,TIER-ENTERPRISE,899.0
regions.csv
region_id,region_name,tax_rate_pct
REG-CA,California,8.5
REG-TX,Texas,0.0
REG-NY,New York,8.0
REG-UK,United Kingdom,20.0
REG-DE,Germany,19.0
subscriptions.csv
subscription_id,customer_id,plan_id,addon_ids,discount_code,status,start_date
SUB-001,CUST-001,PLAN-GROWTH,ADDON-SSO;ADDON-API,,active,2023-01-01
SUB-002,CUST-002,PLAN-SCALE,ADDON-PRIORITY,,active,2023-04-01
support_tickets.csv
ticket_id,customer_id,subject,opened_on
TCK-001,CUST-001,Login issue,2024-01-15
TCK-002,CUST-003,Billing question,2024-02-10
ticket.md
# Ticket
Acme Corp (CUST-001) relocated its HQ from California to Texas, effective 2024-02-01. Please update our records to reflect the new region. Their February invoice (INV-1002) already went out still taxed at the old California rate -- please correct it to the Texas rate (0%). The January invoice was correctly taxed under California and should NOT be touched.
tiers.csv
tier_id,tier_name,seat_limit
TIER-STARTER,Starter,5
TIER-GROWTH,Growth,25
TIER-SCALE,Scale,100
TIER-ENTERPRISE,Enterprise,1000
expected output
answer.json
{
"id": "case_01",
"category": "region_migration_dated",
"gold_edits": [
{
"file": "customers.csv",
"op": "update",
"match": {
"customer_id": "CUST-001"
},
"set": {
"region_id": "REG-TX"
}
},
{
"file": "invoices.csv",
"op": "update",
"match": {
"invoice_id": "INV-1002"
},
"set": {
"baked_tax_usd": 0,
"baked_total_usd": 167
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_02Plan price increase effective from a date; percentage discount must recalc off the new subtotal while a fixed discount does not, invoices before the date are untouched
input
README.md
# Task
A colleague filed a ticket asking for a change to the subscription billing pipeline.
**Your goal:** ensure all FOUR reports (`billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports these four scripts produce match the reports they would produce after a correct fix**. The judge applies your edits, runs all four scripts, and compares stdout to the gold reports.
## Business context
This is a SaaS company's subscription billing system. Customers subscribe to plans (which belong to tiers), can add optional add-ons, and may have a discount code applied. Each billing period an invoice is generated and its financial fields are BAKED into `invoices.csv` at that point in time -- they are a historical record, not something that gets silently recomputed later.
The four reports read the same underlying facts through DIFFERENT paths:
- `billing_summary.py` -- LIVE: "if we billed today," fully resolved from current customers/plans/tiers/addons/discount_codes/regions. No invoice history involved.
- `invoice_detail.py` -- pure BAKED: reads `invoices.csv`'s own columns directly, no lookups at all. This is the historical record customers see.
- `finance_ledger.py` -- MIXED: pretax subtotal always from baked invoice columns; tax is baked-tax-if-present, else a LIVE fallback lookup (invoice -> subscription -> customer -> region).
- `customer_statement.py` -- MIXED differently: base price stays BAKED (grandfather pricing), but add-on names/prices, discount, and tax are all resolved LIVE against current tables.
Because each report resolves differently, the SAME underlying change can require touching different combinations of files. A change to "current state" tables (plans/addons/discount_codes/customers) flows automatically into the two LIVE reports. A change that must also correct HISTORICAL invoices requires directly editing the affected row(s) in `invoices.csv` -- and only the affected row(s); untouched historical invoices must stay untouched.
## How to approach this
- The data tables are meant to be read in full, but do not touch rows the ticket doesn't call for -- edits to invoices outside the ticket's stated scope will be scored as incorrect, even if well-intentioned.
- Read all four report scripts to understand exactly which table/column each one reads and whether that path is baked, live, or a fallback of one to the other.
- If a ticket changes a rate/price/code definition, decide: does this only affect the live view going forward, or does a specific historical invoice also need its baked figures corrected?
## Files in this directory
- `ticket.md` -- the change request
- `customers.csv`, `subscriptions.csv`, `plans.csv`, `tiers.csv`, `addons.csv`, `discount_codes.csv`, `regions.csv` -- current-state master tables
- `invoices.csv` -- historical baked invoice records
- `support_tickets.csv`, `marketing_campaigns.csv` -- unrelated auxiliary tables (not used by any report)
- `billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py` -- the four report scripts
## Output format
Emit a JSON array of edits to stdout. Nothing else -- no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Only your first 30 edits will be applied by the judge -- padding the list with extra guesses past that point does not help.
Output ONLY the JSON array.
addons.csv
addon_id,addon_name,addon_price_usd
ADDON-SSO,Single Sign-On,49.0
ADDON-API,API Access,19.0
ADDON-PRIORITY,Priority Support,39.0
ADDON-AUDIT,Audit Log Export,25.0
billing_summary.py
"""Billing summary -- LIVE current-state view of what each active
subscription would be billed TODAY.
Reads: customers.csv, subscriptions.csv, plans.csv, tiers.csv, addons.csv,
discount_codes.csv, regions.csv
Everything here is resolved via LOOKUP against the CURRENT tables (no
invoice history involved). If a plan's price changes, or a customer's
region changes, or an add-on is renamed/repriced, this report reflects
the new value immediately for every active subscription -- it is the
"if we billed today" view, not a historical record.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
plans = index_by(load_csv(here / "plans.csv"), "plan_id")
tiers = index_by(load_csv(here / "tiers.csv"), "tier_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
subscriptions = load_csv(here / "subscriptions.csv")
rows = []
for sub in subscriptions:
if sub.get("status") != "active":
continue
customer = customers.get(sub["customer_id"])
plan = plans.get(sub["plan_id"])
if not customer or not plan:
continue
tier = tiers.get(plan.get("tier_id"), {})
region = regions.get(customer.get("region_id"), {})
base_price = to_num(plan.get("base_price_usd")) or 0.0
addon_total = 0.0
for aid in split_ids(sub.get("addon_ids")):
addon = addons.get(aid)
if addon:
addon_total += to_num(addon.get("addon_price_usd")) or 0.0
subtotal = base_price + addon_total
discount = 0.0
code = (sub.get("discount_code") or "").strip()
if code and code in discount_codes:
dc = discount_codes[code]
pct = to_num(dc.get("discount_pct"))
fixed = to_num(dc.get("discount_fixed_usd"))
if pct:
discount = subtotal * pct
elif fixed:
discount = fixed
after_discount = subtotal - discount
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"subscription_id": sub["subscription_id"],
"customer_name": customer.get("customer_name", ""),
"plan_name": plan.get("plan_name", ""),
"tier_name": tier.get("tier_name", ""),
"region_name": region.get("region_name", ""),
"base_price_usd": f"{base_price:.2f}",
"addon_total_usd": f"{addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: r["subscription_id"])
cols = ["subscription_id", "customer_name", "plan_name", "tier_name",
"region_name", "base_price_usd", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customer_statement.py
"""Customer statement -- all-time invoice history re-rendered with
CURRENT add-on names/prices and CURRENT customer region, but the
historical BASE PRICE stays locked at what was actually baked (grandfather
pricing: a plan price hike does not retroactively re-bill old invoices).
Reads: invoices.csv, subscriptions.csv, addons.csv, discount_codes.csv,
customers.csv, regions.csv
Resolution per invoice:
- base price: BAKED (invoice's own baked_base_price_usd) -- historical lock-in
- add-on names/total: LIVE lookup via subscription.addon_ids -> addons.csv
(a rename or reprice shows up on EVERY invoice immediately, including old ones)
- discount: LIVE lookup via subscription.discount_code -> discount_codes.csv,
using whichever of pct-discount or fixed-discount is LARGER for the
customer (best-of rule, distinct from billing_summary's pct-or-fixed rule)
- tax: LIVE lookup via subscription -> customer -> CURRENT region (a
customer's region change re-taxes their ENTIRE history when this report
is regenerated, not just future invoices)
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
sub = subscriptions.get(inv.get("subscription_id"), {})
addon_ids = split_ids(sub.get("addon_ids"))
live_addon_names = "; ".join(
addons[aid]["addon_name"] for aid in sorted(addon_ids) if aid in addons
)
live_addon_total = sum(
to_num(addons[aid].get("addon_price_usd")) or 0.0
for aid in addon_ids if aid in addons
)
base = to_num(inv.get("baked_base_price_usd")) or 0.0
subtotal = base + live_addon_total
code = (sub.get("discount_code") or "").strip()
discount = 0.0
if code and code in discount_codes:
dc = discount_codes[code]
pct_discount = (subtotal * to_num(dc.get("discount_pct"))) if to_num(dc.get("discount_pct")) else 0.0
fixed_discount = to_num(dc.get("discount_fixed_usd")) or 0.0
discount = max(pct_discount, fixed_discount)
after_discount = subtotal - discount
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"invoice_id": inv["invoice_id"],
"period": inv.get("period", ""),
"addon_names": live_addon_names,
"addon_total_usd": f"{live_addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "period", "addon_names", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customers.csv
customer_id,customer_name,region_id,signup_date
CUST-001,Acme Corp,REG-CA,2022-01-10
CUST-005,Soylent Co,REG-NY,2022-09-18
discount_codes.csv
code,discount_pct,discount_fixed_usd,expires_on
WELCOME10,0.1,,2099-12-31
LAUNCH50,,50.0,2099-12-31
PARTNER20,0.2,,2099-12-31
finance_ledger.py
"""Finance ledger -- reconciliation report Finance runs against invoice_detail.
Reads: invoices.csv, subscriptions.csv, customers.csv, regions.csv
The pretax subtotal ALWAYS comes from the invoice's own BAKED columns
(base + addon - discount, all baked). Tax is COALESCE(baked_tax_usd, LIVE
recompute): if the invoice has a baked tax figure, trust it; only when
baked_tax_usd is blank does this report fall back to a LIVE lookup
(invoice -> subscription -> customer -> region -> tax_rate_pct) to fill
the gap. This means a baked tax correction on one invoice is picked up
here automatically (same source column as invoice_detail), while a
missing/never-baked tax figure gets computed fresh from current customer
region.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
base = to_num(inv.get("baked_base_price_usd")) or 0.0
addon_total = to_num(inv.get("baked_addon_total_usd")) or 0.0
discount = to_num(inv.get("baked_discount_applied_usd")) or 0.0
pretax = base + addon_total - discount
tax = to_num(inv.get("baked_tax_usd"))
if tax is None:
sub = subscriptions.get(inv.get("subscription_id"), {})
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = pretax * tax_rate / 100.0
total = pretax + tax
rows.append({
"invoice_id": inv["invoice_id"],
"subscription_id": inv.get("subscription_id", ""),
"period": inv.get("period", ""),
"pretax_subtotal_usd": f"{pretax:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "subscription_id", "period",
"pretax_subtotal_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
print(",".join(str(r[c]) for c in cols))
if __name__ == "__main__":
main()
invoice_detail.py
"""Invoice detail -- the historical record customers see on their invoices.
Reads: invoices.csv ONLY.
Every field is read directly from the invoice's own BAKED columns -- no
lookups against customers/plans/addons/discount_codes/regions. This is
what actually went out to the customer at billing time, so it must NOT
change just because current-state tables (plan prices, addon names,
customer region, discount code definitions) change later. Only editing
a specific invoice row changes what this report shows for that row.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
invoices.sort(key=lambda r: (r.get("period", ""), r.get("invoice_id", "")))
cols = ["invoice_id", "period", "baked_customer_name", "baked_plan_name",
"baked_addon_names", "baked_base_price_usd", "baked_addon_total_usd",
"baked_discount_applied_usd", "baked_tax_usd", "baked_total_usd", "status"]
print(",".join(cols))
for r in invoices:
vals = [r.get(c, "") for c in cols]
vals = [f'"{v}"' if "," in v else v for v in vals]
print(",".join(vals))
if __name__ == "__main__":
main()
invoices.csv
invoice_id,subscription_id,period,baked_customer_name,baked_plan_name,baked_addon_names,baked_base_price_usd,baked_addon_total_usd,baked_discount_applied_usd,baked_tax_usd,baked_total_usd,status
INV-2001,SUB-001,2024-01,Acme Corp,Growth Monthly,,99.0,0,9.9,7.57,96.67,paid
INV-2002,SUB-001,2024-03,Acme Corp,Growth Monthly,,99.0,0,9.9,7.57,96.67,paid
INV-2003,SUB-005,2024-01,Soylent Co,Growth Monthly,API Access,99.0,19.0,50.0,5.44,73.44,paid
INV-2004,SUB-005,2024-03,Soylent Co,Growth Monthly,API Access,99.0,19.0,50.0,5.44,73.44,paid
marketing_campaigns.csv
campaign_id,region_id,campaign_name,start_date
CMP-001,REG-CA,Spring Launch,2024-02-01
CMP-002,REG-UK,EU Expansion,2024-03-01
plans.csv
plan_id,plan_name,tier_id,base_price_usd
PLAN-STARTER,Starter Monthly,TIER-STARTER,29.0
PLAN-GROWTH,Growth Monthly,TIER-GROWTH,99.0
PLAN-SCALE,Scale Monthly,TIER-SCALE,299.0
PLAN-ENTERPRISE,Enterprise Monthly,TIER-ENTERPRISE,899.0
regions.csv
region_id,region_name,tax_rate_pct
REG-CA,California,8.5
REG-TX,Texas,0.0
REG-NY,New York,8.0
REG-UK,United Kingdom,20.0
REG-DE,Germany,19.0
subscriptions.csv
subscription_id,customer_id,plan_id,addon_ids,discount_code,status,start_date
SUB-001,CUST-001,PLAN-GROWTH,,WELCOME10,active,2023-01-01
SUB-005,CUST-005,PLAN-GROWTH,ADDON-API,LAUNCH50,active,2023-10-01
support_tickets.csv
ticket_id,customer_id,subject,opened_on
TCK-001,CUST-001,Login issue,2024-01-15
TCK-002,CUST-003,Billing question,2024-02-10
ticket.md
# Ticket
We're increasing the Growth Monthly plan price from $99 to $129/month, effective for invoices from 2024-03-01 onwards. Invoices before that date must NOT be touched -- customers were grandfathered at the old price for anything already billed. Remember that a percentage discount recalculates off the new subtotal; a fixed-dollar discount does not.
tiers.csv
tier_id,tier_name,seat_limit
TIER-STARTER,Starter,5
TIER-GROWTH,Growth,25
TIER-SCALE,Scale,100
TIER-ENTERPRISE,Enterprise,1000
expected output
answer.json
{
"id": "case_02",
"category": "plan_price_change_dated",
"gold_edits": [
{
"file": "plans.csv",
"op": "update",
"match": {
"plan_id": "PLAN-GROWTH"
},
"set": {
"base_price_usd": 129
}
},
{
"file": "invoices.csv",
"op": "update",
"match": {
"invoice_id": "INV-2002"
},
"set": {
"baked_base_price_usd": 129,
"baked_discount_applied_usd": 12.9,
"baked_tax_usd": 9.87,
"baked_total_usd": 125.97
}
},
{
"file": "invoices.csv",
"op": "update",
"match": {
"invoice_id": "INV-2004"
},
"set": {
"baked_base_price_usd": 129,
"baked_discount_applied_usd": 50,
"baked_tax_usd": 7.84,
"baked_total_usd": 105.84
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_03Add-on rename and reprice with NO backfill required; tests restraint against over-editing historical invoices that should not change
input
README.md
# Task
A colleague filed a ticket asking for a change to the subscription billing pipeline.
**Your goal:** ensure all FOUR reports (`billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports these four scripts produce match the reports they would produce after a correct fix**. The judge applies your edits, runs all four scripts, and compares stdout to the gold reports.
## Business context
This is a SaaS company's subscription billing system. Customers subscribe to plans (which belong to tiers), can add optional add-ons, and may have a discount code applied. Each billing period an invoice is generated and its financial fields are BAKED into `invoices.csv` at that point in time -- they are a historical record, not something that gets silently recomputed later.
The four reports read the same underlying facts through DIFFERENT paths:
- `billing_summary.py` -- LIVE: "if we billed today," fully resolved from current customers/plans/tiers/addons/discount_codes/regions. No invoice history involved.
- `invoice_detail.py` -- pure BAKED: reads `invoices.csv`'s own columns directly, no lookups at all. This is the historical record customers see.
- `finance_ledger.py` -- MIXED: pretax subtotal always from baked invoice columns; tax is baked-tax-if-present, else a LIVE fallback lookup (invoice -> subscription -> customer -> region).
- `customer_statement.py` -- MIXED differently: base price stays BAKED (grandfather pricing), but add-on names/prices, discount, and tax are all resolved LIVE against current tables.
Because each report resolves differently, the SAME underlying change can require touching different combinations of files. A change to "current state" tables (plans/addons/discount_codes/customers) flows automatically into the two LIVE reports. A change that must also correct HISTORICAL invoices requires directly editing the affected row(s) in `invoices.csv` -- and only the affected row(s); untouched historical invoices must stay untouched.
## How to approach this
- The data tables are meant to be read in full, but do not touch rows the ticket doesn't call for -- edits to invoices outside the ticket's stated scope will be scored as incorrect, even if well-intentioned.
- Read all four report scripts to understand exactly which table/column each one reads and whether that path is baked, live, or a fallback of one to the other.
- If a ticket changes a rate/price/code definition, decide: does this only affect the live view going forward, or does a specific historical invoice also need its baked figures corrected?
## Files in this directory
- `ticket.md` -- the change request
- `customers.csv`, `subscriptions.csv`, `plans.csv`, `tiers.csv`, `addons.csv`, `discount_codes.csv`, `regions.csv` -- current-state master tables
- `invoices.csv` -- historical baked invoice records
- `support_tickets.csv`, `marketing_campaigns.csv` -- unrelated auxiliary tables (not used by any report)
- `billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py` -- the four report scripts
## Output format
Emit a JSON array of edits to stdout. Nothing else -- no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Only your first 30 edits will be applied by the judge -- padding the list with extra guesses past that point does not help.
Output ONLY the JSON array.
addons.csv
addon_id,addon_name,addon_price_usd
ADDON-SSO,Single Sign-On,49.0
ADDON-API,API Access,19.0
ADDON-PRIORITY,Priority Support,39.0
ADDON-AUDIT,Audit Log Export,25.0
billing_summary.py
"""Billing summary -- LIVE current-state view of what each active
subscription would be billed TODAY.
Reads: customers.csv, subscriptions.csv, plans.csv, tiers.csv, addons.csv,
discount_codes.csv, regions.csv
Everything here is resolved via LOOKUP against the CURRENT tables (no
invoice history involved). If a plan's price changes, or a customer's
region changes, or an add-on is renamed/repriced, this report reflects
the new value immediately for every active subscription -- it is the
"if we billed today" view, not a historical record.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
plans = index_by(load_csv(here / "plans.csv"), "plan_id")
tiers = index_by(load_csv(here / "tiers.csv"), "tier_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
subscriptions = load_csv(here / "subscriptions.csv")
rows = []
for sub in subscriptions:
if sub.get("status") != "active":
continue
customer = customers.get(sub["customer_id"])
plan = plans.get(sub["plan_id"])
if not customer or not plan:
continue
tier = tiers.get(plan.get("tier_id"), {})
region = regions.get(customer.get("region_id"), {})
base_price = to_num(plan.get("base_price_usd")) or 0.0
addon_total = 0.0
for aid in split_ids(sub.get("addon_ids")):
addon = addons.get(aid)
if addon:
addon_total += to_num(addon.get("addon_price_usd")) or 0.0
subtotal = base_price + addon_total
discount = 0.0
code = (sub.get("discount_code") or "").strip()
if code and code in discount_codes:
dc = discount_codes[code]
pct = to_num(dc.get("discount_pct"))
fixed = to_num(dc.get("discount_fixed_usd"))
if pct:
discount = subtotal * pct
elif fixed:
discount = fixed
after_discount = subtotal - discount
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"subscription_id": sub["subscription_id"],
"customer_name": customer.get("customer_name", ""),
"plan_name": plan.get("plan_name", ""),
"tier_name": tier.get("tier_name", ""),
"region_name": region.get("region_name", ""),
"base_price_usd": f"{base_price:.2f}",
"addon_total_usd": f"{addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: r["subscription_id"])
cols = ["subscription_id", "customer_name", "plan_name", "tier_name",
"region_name", "base_price_usd", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customer_statement.py
"""Customer statement -- all-time invoice history re-rendered with
CURRENT add-on names/prices and CURRENT customer region, but the
historical BASE PRICE stays locked at what was actually baked (grandfather
pricing: a plan price hike does not retroactively re-bill old invoices).
Reads: invoices.csv, subscriptions.csv, addons.csv, discount_codes.csv,
customers.csv, regions.csv
Resolution per invoice:
- base price: BAKED (invoice's own baked_base_price_usd) -- historical lock-in
- add-on names/total: LIVE lookup via subscription.addon_ids -> addons.csv
(a rename or reprice shows up on EVERY invoice immediately, including old ones)
- discount: LIVE lookup via subscription.discount_code -> discount_codes.csv,
using whichever of pct-discount or fixed-discount is LARGER for the
customer (best-of rule, distinct from billing_summary's pct-or-fixed rule)
- tax: LIVE lookup via subscription -> customer -> CURRENT region (a
customer's region change re-taxes their ENTIRE history when this report
is regenerated, not just future invoices)
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
sub = subscriptions.get(inv.get("subscription_id"), {})
addon_ids = split_ids(sub.get("addon_ids"))
live_addon_names = "; ".join(
addons[aid]["addon_name"] for aid in sorted(addon_ids) if aid in addons
)
live_addon_total = sum(
to_num(addons[aid].get("addon_price_usd")) or 0.0
for aid in addon_ids if aid in addons
)
base = to_num(inv.get("baked_base_price_usd")) or 0.0
subtotal = base + live_addon_total
code = (sub.get("discount_code") or "").strip()
discount = 0.0
if code and code in discount_codes:
dc = discount_codes[code]
pct_discount = (subtotal * to_num(dc.get("discount_pct"))) if to_num(dc.get("discount_pct")) else 0.0
fixed_discount = to_num(dc.get("discount_fixed_usd")) or 0.0
discount = max(pct_discount, fixed_discount)
after_discount = subtotal - discount
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"invoice_id": inv["invoice_id"],
"period": inv.get("period", ""),
"addon_names": live_addon_names,
"addon_total_usd": f"{live_addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "period", "addon_names", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customers.csv
customer_id,customer_name,region_id,signup_date
CUST-002,Globex Ltd,REG-UK,2022-03-22
CUST-004,Umbrella GmbH,REG-DE,2022-07-01
discount_codes.csv
code,discount_pct,discount_fixed_usd,expires_on
WELCOME10,0.1,,2099-12-31
LAUNCH50,,50.0,2099-12-31
PARTNER20,0.2,,2099-12-31
finance_ledger.py
"""Finance ledger -- reconciliation report Finance runs against invoice_detail.
Reads: invoices.csv, subscriptions.csv, customers.csv, regions.csv
The pretax subtotal ALWAYS comes from the invoice's own BAKED columns
(base + addon - discount, all baked). Tax is COALESCE(baked_tax_usd, LIVE
recompute): if the invoice has a baked tax figure, trust it; only when
baked_tax_usd is blank does this report fall back to a LIVE lookup
(invoice -> subscription -> customer -> region -> tax_rate_pct) to fill
the gap. This means a baked tax correction on one invoice is picked up
here automatically (same source column as invoice_detail), while a
missing/never-baked tax figure gets computed fresh from current customer
region.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
base = to_num(inv.get("baked_base_price_usd")) or 0.0
addon_total = to_num(inv.get("baked_addon_total_usd")) or 0.0
discount = to_num(inv.get("baked_discount_applied_usd")) or 0.0
pretax = base + addon_total - discount
tax = to_num(inv.get("baked_tax_usd"))
if tax is None:
sub = subscriptions.get(inv.get("subscription_id"), {})
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = pretax * tax_rate / 100.0
total = pretax + tax
rows.append({
"invoice_id": inv["invoice_id"],
"subscription_id": inv.get("subscription_id", ""),
"period": inv.get("period", ""),
"pretax_subtotal_usd": f"{pretax:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "subscription_id", "period",
"pretax_subtotal_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
print(",".join(str(r[c]) for c in cols))
if __name__ == "__main__":
main()
invoice_detail.py
"""Invoice detail -- the historical record customers see on their invoices.
Reads: invoices.csv ONLY.
Every field is read directly from the invoice's own BAKED columns -- no
lookups against customers/plans/addons/discount_codes/regions. This is
what actually went out to the customer at billing time, so it must NOT
change just because current-state tables (plan prices, addon names,
customer region, discount code definitions) change later. Only editing
a specific invoice row changes what this report shows for that row.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
invoices.sort(key=lambda r: (r.get("period", ""), r.get("invoice_id", "")))
cols = ["invoice_id", "period", "baked_customer_name", "baked_plan_name",
"baked_addon_names", "baked_base_price_usd", "baked_addon_total_usd",
"baked_discount_applied_usd", "baked_tax_usd", "baked_total_usd", "status"]
print(",".join(cols))
for r in invoices:
vals = [r.get(c, "") for c in cols]
vals = [f'"{v}"' if "," in v else v for v in vals]
print(",".join(vals))
if __name__ == "__main__":
main()
invoices.csv
invoice_id,subscription_id,period,baked_customer_name,baked_plan_name,baked_addon_names,baked_base_price_usd,baked_addon_total_usd,baked_discount_applied_usd,baked_tax_usd,baked_total_usd,status
INV-3001,SUB-002,2024-01,Globex Ltd,Scale Monthly,Priority Support,299.0,39.0,0.0,67.6,405.6,paid
INV-3002,SUB-004,2024-01,Umbrella GmbH,Enterprise Monthly,Audit Log Export; Priority Support; Single Sign-On,899.0,113.0,202.4,153.82,963.42,paid
marketing_campaigns.csv
campaign_id,region_id,campaign_name,start_date
CMP-001,REG-CA,Spring Launch,2024-02-01
CMP-002,REG-UK,EU Expansion,2024-03-01
plans.csv
plan_id,plan_name,tier_id,base_price_usd
PLAN-STARTER,Starter Monthly,TIER-STARTER,29.0
PLAN-GROWTH,Growth Monthly,TIER-GROWTH,99.0
PLAN-SCALE,Scale Monthly,TIER-SCALE,299.0
PLAN-ENTERPRISE,Enterprise Monthly,TIER-ENTERPRISE,899.0
regions.csv
region_id,region_name,tax_rate_pct
REG-CA,California,8.5
REG-TX,Texas,0.0
REG-NY,New York,8.0
REG-UK,United Kingdom,20.0
REG-DE,Germany,19.0
subscriptions.csv
subscription_id,customer_id,plan_id,addon_ids,discount_code,status,start_date
SUB-002,CUST-002,PLAN-SCALE,ADDON-PRIORITY,,active,2023-04-01
SUB-004,CUST-004,PLAN-ENTERPRISE,ADDON-SSO;ADDON-AUDIT;ADDON-PRIORITY,PARTNER20,active,2023-08-01
support_tickets.csv
ticket_id,customer_id,subject,opened_on
TCK-001,CUST-001,Login issue,2024-01-15
TCK-002,CUST-003,Billing question,2024-02-10
ticket.md
# Ticket
Rebrand the 'Priority Support' add-on to 'Premium Support' and raise its price from $39 to $59/month, effective immediately for current and future billing. Do NOT alter any past invoices -- customers already billed under the old name/price keep their historical record as-is.
tiers.csv
tier_id,tier_name,seat_limit
TIER-STARTER,Starter,5
TIER-GROWTH,Growth,25
TIER-SCALE,Scale,100
TIER-ENTERPRISE,Enterprise,1000
expected output
answer.json
{
"id": "case_03",
"category": "addon_rename_reprice_no_backfill",
"gold_edits": [
{
"file": "addons.csv",
"op": "update",
"match": {
"addon_id": "ADDON-PRIORITY"
},
"set": {
"addon_name": "Premium Support",
"addon_price_usd": 59
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_04Discount code percentage correction plus one specific historical invoice fix; a distractor invoice on a different code must not be touched
input
README.md
# Task
A colleague filed a ticket asking for a change to the subscription billing pipeline.
**Your goal:** ensure all FOUR reports (`billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports these four scripts produce match the reports they would produce after a correct fix**. The judge applies your edits, runs all four scripts, and compares stdout to the gold reports.
## Business context
This is a SaaS company's subscription billing system. Customers subscribe to plans (which belong to tiers), can add optional add-ons, and may have a discount code applied. Each billing period an invoice is generated and its financial fields are BAKED into `invoices.csv` at that point in time -- they are a historical record, not something that gets silently recomputed later.
The four reports read the same underlying facts through DIFFERENT paths:
- `billing_summary.py` -- LIVE: "if we billed today," fully resolved from current customers/plans/tiers/addons/discount_codes/regions. No invoice history involved.
- `invoice_detail.py` -- pure BAKED: reads `invoices.csv`'s own columns directly, no lookups at all. This is the historical record customers see.
- `finance_ledger.py` -- MIXED: pretax subtotal always from baked invoice columns; tax is baked-tax-if-present, else a LIVE fallback lookup (invoice -> subscription -> customer -> region).
- `customer_statement.py` -- MIXED differently: base price stays BAKED (grandfather pricing), but add-on names/prices, discount, and tax are all resolved LIVE against current tables.
Because each report resolves differently, the SAME underlying change can require touching different combinations of files. A change to "current state" tables (plans/addons/discount_codes/customers) flows automatically into the two LIVE reports. A change that must also correct HISTORICAL invoices requires directly editing the affected row(s) in `invoices.csv` -- and only the affected row(s); untouched historical invoices must stay untouched.
## How to approach this
- The data tables are meant to be read in full, but do not touch rows the ticket doesn't call for -- edits to invoices outside the ticket's stated scope will be scored as incorrect, even if well-intentioned.
- Read all four report scripts to understand exactly which table/column each one reads and whether that path is baked, live, or a fallback of one to the other.
- If a ticket changes a rate/price/code definition, decide: does this only affect the live view going forward, or does a specific historical invoice also need its baked figures corrected?
## Files in this directory
- `ticket.md` -- the change request
- `customers.csv`, `subscriptions.csv`, `plans.csv`, `tiers.csv`, `addons.csv`, `discount_codes.csv`, `regions.csv` -- current-state master tables
- `invoices.csv` -- historical baked invoice records
- `support_tickets.csv`, `marketing_campaigns.csv` -- unrelated auxiliary tables (not used by any report)
- `billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py` -- the four report scripts
## Output format
Emit a JSON array of edits to stdout. Nothing else -- no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Only your first 30 edits will be applied by the judge -- padding the list with extra guesses past that point does not help.
Output ONLY the JSON array.
addons.csv
addon_id,addon_name,addon_price_usd
ADDON-SSO,Single Sign-On,49.0
ADDON-API,API Access,19.0
ADDON-PRIORITY,Priority Support,39.0
ADDON-AUDIT,Audit Log Export,25.0
billing_summary.py
"""Billing summary -- LIVE current-state view of what each active
subscription would be billed TODAY.
Reads: customers.csv, subscriptions.csv, plans.csv, tiers.csv, addons.csv,
discount_codes.csv, regions.csv
Everything here is resolved via LOOKUP against the CURRENT tables (no
invoice history involved). If a plan's price changes, or a customer's
region changes, or an add-on is renamed/repriced, this report reflects
the new value immediately for every active subscription -- it is the
"if we billed today" view, not a historical record.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
plans = index_by(load_csv(here / "plans.csv"), "plan_id")
tiers = index_by(load_csv(here / "tiers.csv"), "tier_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
subscriptions = load_csv(here / "subscriptions.csv")
rows = []
for sub in subscriptions:
if sub.get("status") != "active":
continue
customer = customers.get(sub["customer_id"])
plan = plans.get(sub["plan_id"])
if not customer or not plan:
continue
tier = tiers.get(plan.get("tier_id"), {})
region = regions.get(customer.get("region_id"), {})
base_price = to_num(plan.get("base_price_usd")) or 0.0
addon_total = 0.0
for aid in split_ids(sub.get("addon_ids")):
addon = addons.get(aid)
if addon:
addon_total += to_num(addon.get("addon_price_usd")) or 0.0
subtotal = base_price + addon_total
discount = 0.0
code = (sub.get("discount_code") or "").strip()
if code and code in discount_codes:
dc = discount_codes[code]
pct = to_num(dc.get("discount_pct"))
fixed = to_num(dc.get("discount_fixed_usd"))
if pct:
discount = subtotal * pct
elif fixed:
discount = fixed
after_discount = subtotal - discount
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"subscription_id": sub["subscription_id"],
"customer_name": customer.get("customer_name", ""),
"plan_name": plan.get("plan_name", ""),
"tier_name": tier.get("tier_name", ""),
"region_name": region.get("region_name", ""),
"base_price_usd": f"{base_price:.2f}",
"addon_total_usd": f"{addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: r["subscription_id"])
cols = ["subscription_id", "customer_name", "plan_name", "tier_name",
"region_name", "base_price_usd", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customer_statement.py
"""Customer statement -- all-time invoice history re-rendered with
CURRENT add-on names/prices and CURRENT customer region, but the
historical BASE PRICE stays locked at what was actually baked (grandfather
pricing: a plan price hike does not retroactively re-bill old invoices).
Reads: invoices.csv, subscriptions.csv, addons.csv, discount_codes.csv,
customers.csv, regions.csv
Resolution per invoice:
- base price: BAKED (invoice's own baked_base_price_usd) -- historical lock-in
- add-on names/total: LIVE lookup via subscription.addon_ids -> addons.csv
(a rename or reprice shows up on EVERY invoice immediately, including old ones)
- discount: LIVE lookup via subscription.discount_code -> discount_codes.csv,
using whichever of pct-discount or fixed-discount is LARGER for the
customer (best-of rule, distinct from billing_summary's pct-or-fixed rule)
- tax: LIVE lookup via subscription -> customer -> CURRENT region (a
customer's region change re-taxes their ENTIRE history when this report
is regenerated, not just future invoices)
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
sub = subscriptions.get(inv.get("subscription_id"), {})
addon_ids = split_ids(sub.get("addon_ids"))
live_addon_names = "; ".join(
addons[aid]["addon_name"] for aid in sorted(addon_ids) if aid in addons
)
live_addon_total = sum(
to_num(addons[aid].get("addon_price_usd")) or 0.0
for aid in addon_ids if aid in addons
)
base = to_num(inv.get("baked_base_price_usd")) or 0.0
subtotal = base + live_addon_total
code = (sub.get("discount_code") or "").strip()
discount = 0.0
if code and code in discount_codes:
dc = discount_codes[code]
pct_discount = (subtotal * to_num(dc.get("discount_pct"))) if to_num(dc.get("discount_pct")) else 0.0
fixed_discount = to_num(dc.get("discount_fixed_usd")) or 0.0
discount = max(pct_discount, fixed_discount)
after_discount = subtotal - discount
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"invoice_id": inv["invoice_id"],
"period": inv.get("period", ""),
"addon_names": live_addon_names,
"addon_total_usd": f"{live_addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "period", "addon_names", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customers.csv
customer_id,customer_name,region_id,signup_date
CUST-004,Umbrella GmbH,REG-DE,2022-07-01
discount_codes.csv
code,discount_pct,discount_fixed_usd,expires_on
WELCOME10,0.1,,2099-12-31
LAUNCH50,,50.0,2099-12-31
PARTNER20,0.2,,2099-12-31
finance_ledger.py
"""Finance ledger -- reconciliation report Finance runs against invoice_detail.
Reads: invoices.csv, subscriptions.csv, customers.csv, regions.csv
The pretax subtotal ALWAYS comes from the invoice's own BAKED columns
(base + addon - discount, all baked). Tax is COALESCE(baked_tax_usd, LIVE
recompute): if the invoice has a baked tax figure, trust it; only when
baked_tax_usd is blank does this report fall back to a LIVE lookup
(invoice -> subscription -> customer -> region -> tax_rate_pct) to fill
the gap. This means a baked tax correction on one invoice is picked up
here automatically (same source column as invoice_detail), while a
missing/never-baked tax figure gets computed fresh from current customer
region.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
base = to_num(inv.get("baked_base_price_usd")) or 0.0
addon_total = to_num(inv.get("baked_addon_total_usd")) or 0.0
discount = to_num(inv.get("baked_discount_applied_usd")) or 0.0
pretax = base + addon_total - discount
tax = to_num(inv.get("baked_tax_usd"))
if tax is None:
sub = subscriptions.get(inv.get("subscription_id"), {})
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = pretax * tax_rate / 100.0
total = pretax + tax
rows.append({
"invoice_id": inv["invoice_id"],
"subscription_id": inv.get("subscription_id", ""),
"period": inv.get("period", ""),
"pretax_subtotal_usd": f"{pretax:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "subscription_id", "period",
"pretax_subtotal_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
print(",".join(str(r[c]) for c in cols))
if __name__ == "__main__":
main()
invoice_detail.py
"""Invoice detail -- the historical record customers see on their invoices.
Reads: invoices.csv ONLY.
Every field is read directly from the invoice's own BAKED columns -- no
lookups against customers/plans/addons/discount_codes/regions. This is
what actually went out to the customer at billing time, so it must NOT
change just because current-state tables (plan prices, addon names,
customer region, discount code definitions) change later. Only editing
a specific invoice row changes what this report shows for that row.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
invoices.sort(key=lambda r: (r.get("period", ""), r.get("invoice_id", "")))
cols = ["invoice_id", "period", "baked_customer_name", "baked_plan_name",
"baked_addon_names", "baked_base_price_usd", "baked_addon_total_usd",
"baked_discount_applied_usd", "baked_tax_usd", "baked_total_usd", "status"]
print(",".join(cols))
for r in invoices:
vals = [r.get(c, "") for c in cols]
vals = [f'"{v}"' if "," in v else v for v in vals]
print(",".join(vals))
if __name__ == "__main__":
main()
invoices.csv
invoice_id,subscription_id,period,baked_customer_name,baked_plan_name,baked_addon_names,baked_base_price_usd,baked_addon_total_usd,baked_discount_applied_usd,baked_tax_usd,baked_total_usd,status
INV-4001,SUB-004,2024-06,Umbrella GmbH,Enterprise Monthly,Audit Log Export; Priority Support; Single Sign-On,899.0,113.0,101.2,173.05,1083.85,paid
INV-4002,SUB-004,2024-08,Umbrella GmbH,Enterprise Monthly,Audit Log Export; Priority Support; Single Sign-On,899.0,113.0,202.4,153.82,963.42,paid
marketing_campaigns.csv
campaign_id,region_id,campaign_name,start_date
CMP-001,REG-CA,Spring Launch,2024-02-01
CMP-002,REG-UK,EU Expansion,2024-03-01
plans.csv
plan_id,plan_name,tier_id,base_price_usd
PLAN-STARTER,Starter Monthly,TIER-STARTER,29.0
PLAN-GROWTH,Growth Monthly,TIER-GROWTH,99.0
PLAN-SCALE,Scale Monthly,TIER-SCALE,299.0
PLAN-ENTERPRISE,Enterprise Monthly,TIER-ENTERPRISE,899.0
regions.csv
region_id,region_name,tax_rate_pct
REG-CA,California,8.5
REG-TX,Texas,0.0
REG-NY,New York,8.0
REG-UK,United Kingdom,20.0
REG-DE,Germany,19.0
subscriptions.csv
subscription_id,customer_id,plan_id,addon_ids,discount_code,status,start_date
SUB-004,CUST-004,PLAN-ENTERPRISE,ADDON-SSO;ADDON-AUDIT;ADDON-PRIORITY,PARTNER20,active,2023-08-01
support_tickets.csv
ticket_id,customer_id,subject,opened_on
TCK-001,CUST-001,Login issue,2024-01-15
TCK-002,CUST-003,Billing question,2024-02-10
ticket.md
# Ticket
PARTNER20 was supposed to be a 25% discount, not 20% -- please fix the code definition. Umbrella GmbH's (CUST-004, SUB-004) August invoice (INV-4002) already went out billed at the wrong 20%; please correct that invoice too. Their June invoice (INV-4001) used a different code and is unaffected -- leave it alone.
tiers.csv
tier_id,tier_name,seat_limit
TIER-STARTER,Starter,5
TIER-GROWTH,Growth,25
TIER-SCALE,Scale,100
TIER-ENTERPRISE,Enterprise,1000
expected output
answer.json
{
"id": "case_04",
"category": "discount_code_pct_correction",
"gold_edits": [
{
"file": "discount_codes.csv",
"op": "update",
"match": {
"code": "PARTNER20"
},
"set": {
"discount_pct": 0.25
}
},
{
"file": "invoices.csv",
"op": "update",
"match": {
"invoice_id": "INV-4002"
},
"set": {
"baked_discount_applied_usd": 253,
"baked_tax_usd": 144.21,
"baked_total_usd": 903.21
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_05Subscription transfers to a NEW customer entity in a different region (insert + update + baked tax recompute), deepest hop chain
input
README.md
# Task
A colleague filed a ticket asking for a change to the subscription billing pipeline.
**Your goal:** ensure all FOUR reports (`billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports these four scripts produce match the reports they would produce after a correct fix**. The judge applies your edits, runs all four scripts, and compares stdout to the gold reports.
## Business context
This is a SaaS company's subscription billing system. Customers subscribe to plans (which belong to tiers), can add optional add-ons, and may have a discount code applied. Each billing period an invoice is generated and its financial fields are BAKED into `invoices.csv` at that point in time -- they are a historical record, not something that gets silently recomputed later.
The four reports read the same underlying facts through DIFFERENT paths:
- `billing_summary.py` -- LIVE: "if we billed today," fully resolved from current customers/plans/tiers/addons/discount_codes/regions. No invoice history involved.
- `invoice_detail.py` -- pure BAKED: reads `invoices.csv`'s own columns directly, no lookups at all. This is the historical record customers see.
- `finance_ledger.py` -- MIXED: pretax subtotal always from baked invoice columns; tax is baked-tax-if-present, else a LIVE fallback lookup (invoice -> subscription -> customer -> region).
- `customer_statement.py` -- MIXED differently: base price stays BAKED (grandfather pricing), but add-on names/prices, discount, and tax are all resolved LIVE against current tables.
Because each report resolves differently, the SAME underlying change can require touching different combinations of files. A change to "current state" tables (plans/addons/discount_codes/customers) flows automatically into the two LIVE reports. A change that must also correct HISTORICAL invoices requires directly editing the affected row(s) in `invoices.csv` -- and only the affected row(s); untouched historical invoices must stay untouched.
## How to approach this
- The data tables are meant to be read in full, but do not touch rows the ticket doesn't call for -- edits to invoices outside the ticket's stated scope will be scored as incorrect, even if well-intentioned.
- Read all four report scripts to understand exactly which table/column each one reads and whether that path is baked, live, or a fallback of one to the other.
- If a ticket changes a rate/price/code definition, decide: does this only affect the live view going forward, or does a specific historical invoice also need its baked figures corrected?
## Files in this directory
- `ticket.md` -- the change request
- `customers.csv`, `subscriptions.csv`, `plans.csv`, `tiers.csv`, `addons.csv`, `discount_codes.csv`, `regions.csv` -- current-state master tables
- `invoices.csv` -- historical baked invoice records
- `support_tickets.csv`, `marketing_campaigns.csv` -- unrelated auxiliary tables (not used by any report)
- `billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py` -- the four report scripts
## Output format
Emit a JSON array of edits to stdout. Nothing else -- no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Only your first 30 edits will be applied by the judge -- padding the list with extra guesses past that point does not help.
Output ONLY the JSON array.
addons.csv
addon_id,addon_name,addon_price_usd
ADDON-SSO,Single Sign-On,49.0
ADDON-API,API Access,19.0
ADDON-PRIORITY,Priority Support,39.0
ADDON-AUDIT,Audit Log Export,25.0
billing_summary.py
"""Billing summary -- LIVE current-state view of what each active
subscription would be billed TODAY.
Reads: customers.csv, subscriptions.csv, plans.csv, tiers.csv, addons.csv,
discount_codes.csv, regions.csv
Everything here is resolved via LOOKUP against the CURRENT tables (no
invoice history involved). If a plan's price changes, or a customer's
region changes, or an add-on is renamed/repriced, this report reflects
the new value immediately for every active subscription -- it is the
"if we billed today" view, not a historical record.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
plans = index_by(load_csv(here / "plans.csv"), "plan_id")
tiers = index_by(load_csv(here / "tiers.csv"), "tier_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
subscriptions = load_csv(here / "subscriptions.csv")
rows = []
for sub in subscriptions:
if sub.get("status") != "active":
continue
customer = customers.get(sub["customer_id"])
plan = plans.get(sub["plan_id"])
if not customer or not plan:
continue
tier = tiers.get(plan.get("tier_id"), {})
region = regions.get(customer.get("region_id"), {})
base_price = to_num(plan.get("base_price_usd")) or 0.0
addon_total = 0.0
for aid in split_ids(sub.get("addon_ids")):
addon = addons.get(aid)
if addon:
addon_total += to_num(addon.get("addon_price_usd")) or 0.0
subtotal = base_price + addon_total
discount = 0.0
code = (sub.get("discount_code") or "").strip()
if code and code in discount_codes:
dc = discount_codes[code]
pct = to_num(dc.get("discount_pct"))
fixed = to_num(dc.get("discount_fixed_usd"))
if pct:
discount = subtotal * pct
elif fixed:
discount = fixed
after_discount = subtotal - discount
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"subscription_id": sub["subscription_id"],
"customer_name": customer.get("customer_name", ""),
"plan_name": plan.get("plan_name", ""),
"tier_name": tier.get("tier_name", ""),
"region_name": region.get("region_name", ""),
"base_price_usd": f"{base_price:.2f}",
"addon_total_usd": f"{addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: r["subscription_id"])
cols = ["subscription_id", "customer_name", "plan_name", "tier_name",
"region_name", "base_price_usd", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customer_statement.py
"""Customer statement -- all-time invoice history re-rendered with
CURRENT add-on names/prices and CURRENT customer region, but the
historical BASE PRICE stays locked at what was actually baked (grandfather
pricing: a plan price hike does not retroactively re-bill old invoices).
Reads: invoices.csv, subscriptions.csv, addons.csv, discount_codes.csv,
customers.csv, regions.csv
Resolution per invoice:
- base price: BAKED (invoice's own baked_base_price_usd) -- historical lock-in
- add-on names/total: LIVE lookup via subscription.addon_ids -> addons.csv
(a rename or reprice shows up on EVERY invoice immediately, including old ones)
- discount: LIVE lookup via subscription.discount_code -> discount_codes.csv,
using whichever of pct-discount or fixed-discount is LARGER for the
customer (best-of rule, distinct from billing_summary's pct-or-fixed rule)
- tax: LIVE lookup via subscription -> customer -> CURRENT region (a
customer's region change re-taxes their ENTIRE history when this report
is regenerated, not just future invoices)
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
sub = subscriptions.get(inv.get("subscription_id"), {})
addon_ids = split_ids(sub.get("addon_ids"))
live_addon_names = "; ".join(
addons[aid]["addon_name"] for aid in sorted(addon_ids) if aid in addons
)
live_addon_total = sum(
to_num(addons[aid].get("addon_price_usd")) or 0.0
for aid in addon_ids if aid in addons
)
base = to_num(inv.get("baked_base_price_usd")) or 0.0
subtotal = base + live_addon_total
code = (sub.get("discount_code") or "").strip()
discount = 0.0
if code and code in discount_codes:
dc = discount_codes[code]
pct_discount = (subtotal * to_num(dc.get("discount_pct"))) if to_num(dc.get("discount_pct")) else 0.0
fixed_discount = to_num(dc.get("discount_fixed_usd")) or 0.0
discount = max(pct_discount, fixed_discount)
after_discount = subtotal - discount
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"invoice_id": inv["invoice_id"],
"period": inv.get("period", ""),
"addon_names": live_addon_names,
"addon_total_usd": f"{live_addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "period", "addon_names", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customers.csv
customer_id,customer_name,region_id,signup_date
CUST-002,Globex Ltd,REG-UK,2022-03-22
discount_codes.csv
code,discount_pct,discount_fixed_usd,expires_on
WELCOME10,0.1,,2099-12-31
LAUNCH50,,50.0,2099-12-31
PARTNER20,0.2,,2099-12-31
finance_ledger.py
"""Finance ledger -- reconciliation report Finance runs against invoice_detail.
Reads: invoices.csv, subscriptions.csv, customers.csv, regions.csv
The pretax subtotal ALWAYS comes from the invoice's own BAKED columns
(base + addon - discount, all baked). Tax is COALESCE(baked_tax_usd, LIVE
recompute): if the invoice has a baked tax figure, trust it; only when
baked_tax_usd is blank does this report fall back to a LIVE lookup
(invoice -> subscription -> customer -> region -> tax_rate_pct) to fill
the gap. This means a baked tax correction on one invoice is picked up
here automatically (same source column as invoice_detail), while a
missing/never-baked tax figure gets computed fresh from current customer
region.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
base = to_num(inv.get("baked_base_price_usd")) or 0.0
addon_total = to_num(inv.get("baked_addon_total_usd")) or 0.0
discount = to_num(inv.get("baked_discount_applied_usd")) or 0.0
pretax = base + addon_total - discount
tax = to_num(inv.get("baked_tax_usd"))
if tax is None:
sub = subscriptions.get(inv.get("subscription_id"), {})
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = pretax * tax_rate / 100.0
total = pretax + tax
rows.append({
"invoice_id": inv["invoice_id"],
"subscription_id": inv.get("subscription_id", ""),
"period": inv.get("period", ""),
"pretax_subtotal_usd": f"{pretax:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "subscription_id", "period",
"pretax_subtotal_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
print(",".join(str(r[c]) for c in cols))
if __name__ == "__main__":
main()
invoice_detail.py
"""Invoice detail -- the historical record customers see on their invoices.
Reads: invoices.csv ONLY.
Every field is read directly from the invoice's own BAKED columns -- no
lookups against customers/plans/addons/discount_codes/regions. This is
what actually went out to the customer at billing time, so it must NOT
change just because current-state tables (plan prices, addon names,
customer region, discount code definitions) change later. Only editing
a specific invoice row changes what this report shows for that row.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
invoices.sort(key=lambda r: (r.get("period", ""), r.get("invoice_id", "")))
cols = ["invoice_id", "period", "baked_customer_name", "baked_plan_name",
"baked_addon_names", "baked_base_price_usd", "baked_addon_total_usd",
"baked_discount_applied_usd", "baked_tax_usd", "baked_total_usd", "status"]
print(",".join(cols))
for r in invoices:
vals = [r.get(c, "") for c in cols]
vals = [f'"{v}"' if "," in v else v for v in vals]
print(",".join(vals))
if __name__ == "__main__":
main()
invoices.csv
invoice_id,subscription_id,period,baked_customer_name,baked_plan_name,baked_addon_names,baked_base_price_usd,baked_addon_total_usd,baked_discount_applied_usd,baked_tax_usd,baked_total_usd,status
INV-5001,SUB-002,2024-02,Globex Ltd,Scale Monthly,Priority Support,299.0,39.0,0.0,67.6,405.6,paid
INV-5002,SUB-002,2024-05,Globex Ltd,Scale Monthly,Priority Support,299.0,39.0,0.0,67.6,405.6,paid
marketing_campaigns.csv
campaign_id,region_id,campaign_name,start_date
CMP-001,REG-CA,Spring Launch,2024-02-01
CMP-002,REG-UK,EU Expansion,2024-03-01
plans.csv
plan_id,plan_name,tier_id,base_price_usd
PLAN-STARTER,Starter Monthly,TIER-STARTER,29.0
PLAN-GROWTH,Growth Monthly,TIER-GROWTH,99.0
PLAN-SCALE,Scale Monthly,TIER-SCALE,299.0
PLAN-ENTERPRISE,Enterprise Monthly,TIER-ENTERPRISE,899.0
regions.csv
region_id,region_name,tax_rate_pct
REG-CA,California,8.5
REG-TX,Texas,0.0
REG-NY,New York,8.0
REG-UK,United Kingdom,20.0
REG-DE,Germany,19.0
subscriptions.csv
subscription_id,customer_id,plan_id,addon_ids,discount_code,status,start_date
SUB-002,CUST-002,PLAN-SCALE,ADDON-PRIORITY,,active,2023-04-01
support_tickets.csv
ticket_id,customer_id,subject,opened_on
TCK-001,CUST-001,Login issue,2024-01-15
TCK-002,CUST-003,Billing question,2024-02-10
ticket.md
# Ticket
Globex Ltd (CUST-002) is being acquired; its subscription (SUB-002) transfers to the parent company, Initrode Global -- a NEW customer record (billing entity registered in Germany, not the UK), effective 2024-04-01. Historical invoices before the transfer stay attributed to Globex Ltd under UK tax; the May invoice (INV-5002, after the transfer date) must show Initrode Global and be re-taxed at the German rate. The February invoice must NOT be touched.
tiers.csv
tier_id,tier_name,seat_limit
TIER-STARTER,Starter,5
TIER-GROWTH,Growth,25
TIER-SCALE,Scale,100
TIER-ENTERPRISE,Enterprise,1000
expected output
answer.json
{
"id": "case_05",
"category": "subscription_transfer_new_customer_new_region",
"gold_edits": [
{
"file": "customers.csv",
"op": "insert",
"row": {
"customer_id": "CUST-007",
"customer_name": "Initrode Global",
"region_id": "REG-DE",
"signup_date": "2024-04-01"
}
},
{
"file": "subscriptions.csv",
"op": "update",
"match": {
"subscription_id": "SUB-002"
},
"set": {
"customer_id": "CUST-007"
}
},
{
"file": "invoices.csv",
"op": "update",
"match": {
"invoice_id": "INV-5002"
},
"set": {
"baked_customer_name": "Initrode Global",
"baked_tax_usd": 64.22,
"baked_total_usd": 402.22
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_06Compound ticket: tier upgrade and discount code applied simultaneously on one invoice without cross-contaminating the prior period
input
README.md
# Task
A colleague filed a ticket asking for a change to the subscription billing pipeline.
**Your goal:** ensure all FOUR reports (`billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py`) come out CORRECT after the fix is applied.
You will not be graded on your intermediate reasoning or on the structure of your edits. You will be graded on **whether the reports these four scripts produce match the reports they would produce after a correct fix**. The judge applies your edits, runs all four scripts, and compares stdout to the gold reports.
## Business context
This is a SaaS company's subscription billing system. Customers subscribe to plans (which belong to tiers), can add optional add-ons, and may have a discount code applied. Each billing period an invoice is generated and its financial fields are BAKED into `invoices.csv` at that point in time -- they are a historical record, not something that gets silently recomputed later.
The four reports read the same underlying facts through DIFFERENT paths:
- `billing_summary.py` -- LIVE: "if we billed today," fully resolved from current customers/plans/tiers/addons/discount_codes/regions. No invoice history involved.
- `invoice_detail.py` -- pure BAKED: reads `invoices.csv`'s own columns directly, no lookups at all. This is the historical record customers see.
- `finance_ledger.py` -- MIXED: pretax subtotal always from baked invoice columns; tax is baked-tax-if-present, else a LIVE fallback lookup (invoice -> subscription -> customer -> region).
- `customer_statement.py` -- MIXED differently: base price stays BAKED (grandfather pricing), but add-on names/prices, discount, and tax are all resolved LIVE against current tables.
Because each report resolves differently, the SAME underlying change can require touching different combinations of files. A change to "current state" tables (plans/addons/discount_codes/customers) flows automatically into the two LIVE reports. A change that must also correct HISTORICAL invoices requires directly editing the affected row(s) in `invoices.csv` -- and only the affected row(s); untouched historical invoices must stay untouched.
## How to approach this
- The data tables are meant to be read in full, but do not touch rows the ticket doesn't call for -- edits to invoices outside the ticket's stated scope will be scored as incorrect, even if well-intentioned.
- Read all four report scripts to understand exactly which table/column each one reads and whether that path is baked, live, or a fallback of one to the other.
- If a ticket changes a rate/price/code definition, decide: does this only affect the live view going forward, or does a specific historical invoice also need its baked figures corrected?
## Files in this directory
- `ticket.md` -- the change request
- `customers.csv`, `subscriptions.csv`, `plans.csv`, `tiers.csv`, `addons.csv`, `discount_codes.csv`, `regions.csv` -- current-state master tables
- `invoices.csv` -- historical baked invoice records
- `support_tickets.csv`, `marketing_campaigns.csv` -- unrelated auxiliary tables (not used by any report)
- `billing_summary.py`, `invoice_detail.py`, `finance_ledger.py`, `customer_statement.py` -- the four report scripts
## Output format
Emit a JSON array of edits to stdout. Nothing else -- no explanation, no markdown fences.
Each edit is one of:
```
{"file": "<name>.csv", "op": "update", "match": {"<col>": "<value>"}, "set": {"<col>": <value>, ...}}
{"file": "<name>.csv", "op": "insert", "row": {"<col>": <value>, ...}}
```
For updates: `match` specifies which rows to change; `set` specifies which columns to update to which values.
For inserts: `row` is the new row to add (include all columns for that table).
Use `null` (JSON literal) for empty values.
Only your first 30 edits will be applied by the judge -- padding the list with extra guesses past that point does not help.
Output ONLY the JSON array.
addons.csv
addon_id,addon_name,addon_price_usd
ADDON-SSO,Single Sign-On,49.0
ADDON-API,API Access,19.0
ADDON-PRIORITY,Priority Support,39.0
ADDON-AUDIT,Audit Log Export,25.0
billing_summary.py
"""Billing summary -- LIVE current-state view of what each active
subscription would be billed TODAY.
Reads: customers.csv, subscriptions.csv, plans.csv, tiers.csv, addons.csv,
discount_codes.csv, regions.csv
Everything here is resolved via LOOKUP against the CURRENT tables (no
invoice history involved). If a plan's price changes, or a customer's
region changes, or an add-on is renamed/repriced, this report reflects
the new value immediately for every active subscription -- it is the
"if we billed today" view, not a historical record.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
plans = index_by(load_csv(here / "plans.csv"), "plan_id")
tiers = index_by(load_csv(here / "tiers.csv"), "tier_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
subscriptions = load_csv(here / "subscriptions.csv")
rows = []
for sub in subscriptions:
if sub.get("status") != "active":
continue
customer = customers.get(sub["customer_id"])
plan = plans.get(sub["plan_id"])
if not customer or not plan:
continue
tier = tiers.get(plan.get("tier_id"), {})
region = regions.get(customer.get("region_id"), {})
base_price = to_num(plan.get("base_price_usd")) or 0.0
addon_total = 0.0
for aid in split_ids(sub.get("addon_ids")):
addon = addons.get(aid)
if addon:
addon_total += to_num(addon.get("addon_price_usd")) or 0.0
subtotal = base_price + addon_total
discount = 0.0
code = (sub.get("discount_code") or "").strip()
if code and code in discount_codes:
dc = discount_codes[code]
pct = to_num(dc.get("discount_pct"))
fixed = to_num(dc.get("discount_fixed_usd"))
if pct:
discount = subtotal * pct
elif fixed:
discount = fixed
after_discount = subtotal - discount
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"subscription_id": sub["subscription_id"],
"customer_name": customer.get("customer_name", ""),
"plan_name": plan.get("plan_name", ""),
"tier_name": tier.get("tier_name", ""),
"region_name": region.get("region_name", ""),
"base_price_usd": f"{base_price:.2f}",
"addon_total_usd": f"{addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: r["subscription_id"])
cols = ["subscription_id", "customer_name", "plan_name", "tier_name",
"region_name", "base_price_usd", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customer_statement.py
"""Customer statement -- all-time invoice history re-rendered with
CURRENT add-on names/prices and CURRENT customer region, but the
historical BASE PRICE stays locked at what was actually baked (grandfather
pricing: a plan price hike does not retroactively re-bill old invoices).
Reads: invoices.csv, subscriptions.csv, addons.csv, discount_codes.csv,
customers.csv, regions.csv
Resolution per invoice:
- base price: BAKED (invoice's own baked_base_price_usd) -- historical lock-in
- add-on names/total: LIVE lookup via subscription.addon_ids -> addons.csv
(a rename or reprice shows up on EVERY invoice immediately, including old ones)
- discount: LIVE lookup via subscription.discount_code -> discount_codes.csv,
using whichever of pct-discount or fixed-discount is LARGER for the
customer (best-of rule, distinct from billing_summary's pct-or-fixed rule)
- tax: LIVE lookup via subscription -> customer -> CURRENT region (a
customer's region change re-taxes their ENTIRE history when this report
is regenerated, not just future invoices)
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def split_ids(s):
return [x.strip() for x in (s or "").split(";") if x.strip()]
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
addons = index_by(load_csv(here / "addons.csv"), "addon_id")
discount_codes = index_by(load_csv(here / "discount_codes.csv"), "code")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
sub = subscriptions.get(inv.get("subscription_id"), {})
addon_ids = split_ids(sub.get("addon_ids"))
live_addon_names = "; ".join(
addons[aid]["addon_name"] for aid in sorted(addon_ids) if aid in addons
)
live_addon_total = sum(
to_num(addons[aid].get("addon_price_usd")) or 0.0
for aid in addon_ids if aid in addons
)
base = to_num(inv.get("baked_base_price_usd")) or 0.0
subtotal = base + live_addon_total
code = (sub.get("discount_code") or "").strip()
discount = 0.0
if code and code in discount_codes:
dc = discount_codes[code]
pct_discount = (subtotal * to_num(dc.get("discount_pct"))) if to_num(dc.get("discount_pct")) else 0.0
fixed_discount = to_num(dc.get("discount_fixed_usd")) or 0.0
discount = max(pct_discount, fixed_discount)
after_discount = subtotal - discount
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = after_discount * tax_rate / 100.0
total = after_discount + tax
rows.append({
"invoice_id": inv["invoice_id"],
"period": inv.get("period", ""),
"addon_names": live_addon_names,
"addon_total_usd": f"{live_addon_total:.2f}",
"discount_usd": f"{discount:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "period", "addon_names", "addon_total_usd",
"discount_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
vals = [f'"{r[c]}"' if "," in r[c] else r[c] for c in cols]
print(",".join(vals))
if __name__ == "__main__":
main()
customers.csv
customer_id,customer_name,region_id,signup_date
CUST-003,Initech,REG-TX,2022-05-14
discount_codes.csv
code,discount_pct,discount_fixed_usd,expires_on
WELCOME10,0.1,,2099-12-31
LAUNCH50,,50.0,2099-12-31
PARTNER20,0.2,,2099-12-31
finance_ledger.py
"""Finance ledger -- reconciliation report Finance runs against invoice_detail.
Reads: invoices.csv, subscriptions.csv, customers.csv, regions.csv
The pretax subtotal ALWAYS comes from the invoice's own BAKED columns
(base + addon - discount, all baked). Tax is COALESCE(baked_tax_usd, LIVE
recompute): if the invoice has a baked tax figure, trust it; only when
baked_tax_usd is blank does this report fall back to a LIVE lookup
(invoice -> subscription -> customer -> region -> tax_rate_pct) to fill
the gap. This means a baked tax correction on one invoice is picked up
here automatically (same source column as invoice_detail), while a
missing/never-baked tax figure gets computed fresh from current customer
region.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def to_num(v):
if v is None or v == "":
return None
try:
return float(v)
except (ValueError, TypeError):
return None
def index_by(rows, key):
return {r[key]: r for r in rows}
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
subscriptions = index_by(load_csv(here / "subscriptions.csv"), "subscription_id")
customers = index_by(load_csv(here / "customers.csv"), "customer_id")
regions = index_by(load_csv(here / "regions.csv"), "region_id")
rows = []
for inv in invoices:
base = to_num(inv.get("baked_base_price_usd")) or 0.0
addon_total = to_num(inv.get("baked_addon_total_usd")) or 0.0
discount = to_num(inv.get("baked_discount_applied_usd")) or 0.0
pretax = base + addon_total - discount
tax = to_num(inv.get("baked_tax_usd"))
if tax is None:
sub = subscriptions.get(inv.get("subscription_id"), {})
customer = customers.get(sub.get("customer_id"), {})
region = regions.get(customer.get("region_id"), {})
tax_rate = to_num(region.get("tax_rate_pct")) or 0.0
tax = pretax * tax_rate / 100.0
total = pretax + tax
rows.append({
"invoice_id": inv["invoice_id"],
"subscription_id": inv.get("subscription_id", ""),
"period": inv.get("period", ""),
"pretax_subtotal_usd": f"{pretax:.2f}",
"tax_usd": f"{tax:.2f}",
"total_usd": f"{total:.2f}",
})
rows.sort(key=lambda r: (r["period"], r["invoice_id"]))
cols = ["invoice_id", "subscription_id", "period",
"pretax_subtotal_usd", "tax_usd", "total_usd"]
print(",".join(cols))
for r in rows:
print(",".join(str(r[c]) for c in cols))
if __name__ == "__main__":
main()
invoice_detail.py
"""Invoice detail -- the historical record customers see on their invoices.
Reads: invoices.csv ONLY.
Every field is read directly from the invoice's own BAKED columns -- no
lookups against customers/plans/addons/discount_codes/regions. This is
what actually went out to the customer at billing time, so it must NOT
change just because current-state tables (plan prices, addon names,
customer region, discount code definitions) change later. Only editing
a specific invoice row changes what this report shows for that row.
"""
import csv
from pathlib import Path
def load_csv(path):
if not path.exists():
return []
with open(path, newline="") as f:
return list(csv.DictReader(f))
def main():
here = Path(__file__).parent
invoices = load_csv(here / "invoices.csv")
invoices.sort(key=lambda r: (r.get("period", ""), r.get("invoice_id", "")))
cols = ["invoice_id", "period", "baked_customer_name", "baked_plan_name",
"baked_addon_names", "baked_base_price_usd", "baked_addon_total_usd",
"baked_discount_applied_usd", "baked_tax_usd", "baked_total_usd", "status"]
print(",".join(cols))
for r in invoices:
vals = [r.get(c, "") for c in cols]
vals = [f'"{v}"' if "," in v else v for v in vals]
print(",".join(vals))
if __name__ == "__main__":
main()
invoices.csv
invoice_id,subscription_id,period,baked_customer_name,baked_plan_name,baked_addon_names,baked_base_price_usd,baked_addon_total_usd,baked_discount_applied_usd,baked_tax_usd,baked_total_usd,status
INV-6001,SUB-003,2024-04,Initech,Starter Monthly,,29.0,0,0.0,0.0,29.0,paid
INV-6002,SUB-003,2024-05,Initech,Starter Monthly,,29.0,0,0.0,0.0,29.0,paid
marketing_campaigns.csv
campaign_id,region_id,campaign_name,start_date
CMP-001,REG-CA,Spring Launch,2024-02-01
CMP-002,REG-UK,EU Expansion,2024-03-01
plans.csv
plan_id,plan_name,tier_id,base_price_usd
PLAN-STARTER,Starter Monthly,TIER-STARTER,29.0
PLAN-GROWTH,Growth Monthly,TIER-GROWTH,99.0
PLAN-SCALE,Scale Monthly,TIER-SCALE,299.0
PLAN-ENTERPRISE,Enterprise Monthly,TIER-ENTERPRISE,899.0
regions.csv
region_id,region_name,tax_rate_pct
REG-CA,California,8.5
REG-TX,Texas,0.0
REG-NY,New York,8.0
REG-UK,United Kingdom,20.0
REG-DE,Germany,19.0
subscriptions.csv
subscription_id,customer_id,plan_id,addon_ids,discount_code,status,start_date
SUB-003,CUST-003,PLAN-STARTER,,,active,2023-06-01
support_tickets.csv
ticket_id,customer_id,subject,opened_on
TCK-001,CUST-001,Login issue,2024-01-15
TCK-002,CUST-003,Billing question,2024-02-10
ticket.md
# Ticket
Two changes for Initech (CUST-003, SUB-003), both effective 2024-05-01: (1) they're upgrading from Starter to Growth Monthly, and (2) apply the WELCOME10 code starting with the May invoice (INV-6002) -- they should have gotten it at signup-of-upgrade but didn't. The April invoice (INV-6001) is on the old Starter plan with no discount and must NOT change.
tiers.csv
tier_id,tier_name,seat_limit
TIER-STARTER,Starter,5
TIER-GROWTH,Growth,25
TIER-SCALE,Scale,100
TIER-ENTERPRISE,Enterprise,1000
expected output
answer.json
{
"id": "case_06",
"category": "compound_tier_upgrade_plus_discount",
"gold_edits": [
{
"file": "subscriptions.csv",
"op": "update",
"match": {
"subscription_id": "SUB-003"
},
"set": {
"plan_id": "PLAN-GROWTH",
"discount_code": "WELCOME10"
}
},
{
"file": "invoices.csv",
"op": "update",
"match": {
"invoice_id": "INV-6002"
},
"set": {
"baked_plan_name": "Growth Monthly",
"baked_base_price_usd": 99,
"baked_discount_applied_usd": 9.9,
"baked_tax_usd": 0,
"baked_total_usd": 89.1
}
}
]
}Scored by judge.py — see Scoring logic below for the full rule.
scoring logic
judge.py runs once per case and prints a score per case. grader.py runs once at the end and folds case scores into a run-level summary. Without grader.py, the run's score is simply the average of case scores.
▸judge.py247 lines · view on GitHub
"""Per-case judge for debug_subscription_billing_pipeline.
Report-output-based scoring: apply the agent's edits and the gold edits to
SEPARATE copies of the input tables, RUN all four report scripts against
each, and compare stdout. Score 1.0 only if all four reports produce
identical output -- multiple equivalent fix paths all pass, since scoring
never inspects edit structure, only the resulting report output.
I/O contract: reads TRAPTASK_MANIFEST (trap-cli). See
references/traptask-contract.md for the exact manifest shape.
"""
from __future__ import annotations
import json
import os
import re
from pathlib import Path
from typing import Any
REPORT_SCRIPTS = ["billing_summary.py", "invoice_detail.py",
"finance_ledger.py", "customer_statement.py"]
MAX_EDITS_SCORED = 30 # anti-shotgun: edits beyond this are dropped before applying
def strip_fences(s: str) -> str:
s = s.strip()
s = re.sub(r"<think>.*?</think>\s*", "", s, flags=re.DOTALL)
s = re.sub(r"^```(?:json)?\s*", "", s)
s = re.sub(r"\s*```$", "", s)
return s.strip()
def parse_agent_edits(stdout: str) -> list | None:
"""Return list of edit dicts, or None if unparseable."""
s = strip_fences(stdout)
try:
parsed = json.loads(s)
if isinstance(parsed, list):
return parsed
except Exception:
pass
for start in range(len(s)):
if s[start] != '[':
continue
depth = 0
for end in range(start, len(s)):
if s[end] == '[':
depth += 1
elif s[end] == ']':
depth -= 1
if depth == 0:
try:
parsed = json.loads(s[start:end + 1])
if isinstance(parsed, list):
return parsed
except Exception:
break
m = re.search(r"```(?:json)?\s*(\[.*?\])\s*```", s, re.DOTALL)
if m:
try:
parsed = json.loads(m.group(1))
if isinstance(parsed, list):
return parsed
except Exception:
pass
return None
def load_input_csvs(inputs_dir: Path) -> dict:
import csv
tables = {}
for path in inputs_dir.glob("*.csv"):
with path.open() as f:
tables[path.name] = list(csv.DictReader(f))
return tables
def apply_edits(tables: dict, edits: list) -> dict:
"""Apply a list of edits to a COPY of tables. Returns new state.
Malformed entries (non-dict, unknown file, missing keys) are skipped,
not raised -- an agent emitting garbage should score 0 via output
mismatch, not crash the judge."""
import copy
new_tables = {name: copy.deepcopy(rows) for name, rows in tables.items()}
for edit in edits[:MAX_EDITS_SCORED]:
if not isinstance(edit, dict):
continue
file = edit.get("file")
op = edit.get("op")
if file not in new_tables:
continue
if op == "update":
match = edit.get("match", {})
set_ = edit.get("set", {})
if not isinstance(match, dict) or not isinstance(set_, dict):
continue
for row in new_tables[file]:
if all(
str(row.get(k, "")) == str(v) if v is not None else (row.get(k) in (None, "", "None"))
for k, v in match.items()
):
for sk, sv in set_.items():
row[sk] = "" if sv is None else str(sv)
elif op == "insert":
row_data = edit.get("row", {})
if isinstance(row_data, dict):
new_row = {k: ("" if v is None else str(v)) for k, v in row_data.items()}
new_tables[file].append(new_row)
return new_tables
def write_tables_to_dir(tables: dict, out_dir: Path) -> None:
import csv
for fname, rows in tables.items():
if not rows:
continue
cols = list(rows[0].keys())
with (out_dir / fname).open("w", newline="") as f:
w = csv.DictWriter(f, fieldnames=cols)
w.writeheader()
for row in rows:
w.writerow({c: row.get(c, "") for c in cols})
def run_reports(work_dir: Path) -> dict:
import subprocess
import sys
out = {}
for script in REPORT_SCRIPTS:
try:
r = subprocess.run(
[sys.executable, script],
cwd=work_dir, capture_output=True, text=True, timeout=30,
)
out[script] = r.stdout
except Exception as e:
out[script] = f"__JUDGE_EXEC_ERROR__: {e}"
return out
def score_case(stdout: str, expected: dict, inputs_dir: Path = None) -> dict[str, Any]:
import shutil
import tempfile
gold_edits = expected.get("gold_edits", [])
agent_edits = parse_agent_edits(stdout)
if agent_edits is None:
return {
"score": 0.0,
"reason": "failed to parse agent stdout as JSON list of edits",
"category": expected.get("category"),
"gold_edit_count": len(gold_edits),
}
if not isinstance(agent_edits, list):
return {
"score": 0.0,
"reason": "agent output not a JSON list",
"category": expected.get("category"),
}
if inputs_dir is None or not inputs_dir.exists():
return {
"score": 0.0,
"reason": "inputs_dir not available for data-aware scoring",
"category": expected.get("category"),
}
initial = load_input_csvs(inputs_dir)
try:
with tempfile.TemporaryDirectory() as td:
td_path = Path(td)
gold_dir = td_path / "gold"
agent_dir = td_path / "agent"
shutil.copytree(inputs_dir, gold_dir)
gold_state = apply_edits(initial, gold_edits)
write_tables_to_dir(gold_state, gold_dir)
gold_out = run_reports(gold_dir)
shutil.copytree(inputs_dir, agent_dir)
agent_state = apply_edits(initial, agent_edits)
write_tables_to_dir(agent_state, agent_dir)
agent_out = run_reports(agent_dir)
mismatches = []
for script in REPORT_SCRIPTS:
if gold_out[script].strip() != agent_out[script].strip():
mismatches.append(script)
if not mismatches:
return {
"score": 1.0,
"reason": "all four reports match gold output",
"category": expected.get("category"),
"gold_edit_count": len(gold_edits),
"agent_edit_count": len(agent_edits),
}
diffs = [
f"{script} differs; agent output first 300 chars: {agent_out[script][:300]!r}"
for script in mismatches
]
return {
"score": 0.0,
"reason": " | ".join(diffs),
"category": expected.get("category"),
"mismatched_reports": mismatches,
"gold_edit_count": len(gold_edits),
"agent_edit_count": len(agent_edits),
}
except Exception as e:
return {
"score": 0.0,
"reason": f"judge crashed while running scripts: {e}",
"category": expected.get("category"),
}
def main() -> None:
m = json.loads(os.environ["TRAPTASK_MANIFEST"])
stdout = Path(m["run"]["stdout"]).read_text()
exit_code = json.loads(Path(m["run"]["meta"]).read_text())["exit_code"]
expected = json.loads((Path(m["expected_dir"]) / "answer.json").read_text())
inputs_dir = Path(m["inputs_dir"])
base = {"id": expected.get("id")}
if exit_code != 0:
print(json.dumps({**base, "score": 0.0, "reason": f"solution exited {exit_code}",
"agent_output": stdout.strip()[:500]}))
return
if not stdout.strip():
print(json.dumps({**base, "score": 0.0, "reason": "agent produced no output",
"agent_output": ""}))
return
metrics = score_case(stdout, expected, inputs_dir)
metrics.update(base)
metrics["agent_output"] = stdout.strip()[:500]
print(json.dumps(metrics))
if __name__ == "__main__":
main()
▸grader.py79 lines · view on GitHub
"""Overall grader for debug_subscription_billing_pipeline.
Aggregates per-case judge results (the trap-cli TRAPTASK_MANIFEST list)
into a run-level verdict. This aggregation logic is standard across every
task in this repo -- usually nothing to customize here. If your judge's
metrics dict uses a different field name than "bug_category" for its
category breakdown, update CATEGORY_FIELD below; otherwise leave this file
as-is.
"""
from __future__ import annotations
import json
import os
from collections import Counter
PASS_THRESHOLD = 0.5
CATEGORY_FIELD = "category" # change to match your judge.py's metrics dict, or None to disable
def main() -> None:
cases = json.loads(os.environ["TRAPTASK_MANIFEST"])
scored = [c for c in cases if c.get("metrics") and c["metrics"].get("score") is not None]
skipped = [c for c in cases if not c.get("metrics") or c["metrics"].get("score") is None]
accuracy = sum(c["metrics"]["score"] for c in scored) / len(scored) if scored else 0.0
by_category_pct = {}
if CATEGORY_FIELD:
by_category_score: Counter[str] = Counter()
by_category_total: Counter[str] = Counter()
for c in scored:
cat = c["metrics"].get(CATEGORY_FIELD)
if cat:
by_category_total[cat] += 1
by_category_score[cat] += c["metrics"]["score"]
by_category_pct = {
k: round(by_category_score[k] / by_category_total[k], 3) for k in by_category_total
}
durations = [c.get("duration", 0.0) for c in cases if c.get("duration") is not None]
if durations:
ds = sorted(durations)
latency_ms_median = round(ds[len(ds) // 2] * 1000, 1)
latency_ms_p95 = round(ds[int(0.95 * len(ds))] * 1000, 1) if len(ds) > 1 else latency_ms_median
latency_ms_total = round(sum(ds) * 1000, 1)
else:
latency_ms_median = latency_ms_p95 = latency_ms_total = 0.0
case_costs = [
c["cost"]["cost_usd"]
for c in cases
if isinstance(c.get("cost"), dict) and c["cost"].get("cost_usd") is not None
]
cost_usd_total = round(sum(case_costs), 4) if case_costs else None
n_passed = sum(1 for c in scored if c["metrics"]["score"] == 1.0)
passed = bool(scored) and accuracy >= PASS_THRESHOLD
print(json.dumps({
"passed": passed,
"score": round(accuracy, 3),
"n_passed": n_passed,
"n_total": len(cases),
"n_scored": len(scored),
"n_skipped_no_gold": len(skipped),
"threshold": PASS_THRESHOLD,
"by_category": by_category_pct,
"latency_ms_median": latency_ms_median,
"latency_ms_p95": latency_ms_p95,
"latency_ms_total": latency_ms_total,
"cost_usd_total": cost_usd_total,
}))
if __name__ == "__main__":
main()