Operations workbench delivery proposal

Workbench stakeholder brief

Operations lead: Start each morning with orders that are already overdue and still undelivered, then inspect the hub and route. Historical late deliveries support the review but should not hide the active queue. Filter by hub, service and cohort, open an order's event timeline, and show the review cutoff and last successful data update.

Hub lead: Do not rank staff or hubs using overall rates with different service mixes. Southbay express has too little data for a stable comparison. A route note is an investigation lead, not proof of an employee's responsibility.

Security lead: Operators may read only their assigned hubs. Exports require supervisor or service permissions, checked on the server for every request together with session validity. Use the existing corporate identity service and do not add a password store. Choosing a hub in the UI does not grant access.

Delivery lead: Phase one is a read-only workbench, order evidence timeline and controlled export design. Refunds, dispatch writes and customer notifications are excluded. Integration depends on client identity, gateway and order interfaces. Estimate effort, define acceptance, failure handling and rollback. No budget, contractual response SLA, concurrency target or launch date has been agreed; list them as decisions to confirm.

{
  "session_ttl_minutes": 30,
  "export_roles": [
    "supervisor",
    "service"
  ],
  "operator_hub": {
    "op-1": "Harbour",
    "op-2": "Eastgate",
    "op-3": "Westhill",
    "op-4": "Southbay",
    "op-7": "Harbour"
  },
  "network_note": "Access is through a corporate VPN gateway. Documentation IP addresses do not establish location or personal identity.",
  "service_accounts": [
    "svc-archive"
  ]
}

{
  "version": "1.0",
  "as_of": "2026-09-15T00:00:00Z",
  "dataset_version": "coldchain-v1",
  "routes": [
    {
      "method": "GET",
      "path": "/v1/orders",
      "query": [
        "hub",
        "service",
        "cohort",
        "as_of",
        "cursor",
        "limit"
      ],
      "limit": {
        "min": 1,
        "max": 100
      },
      "response": [
        "items",
        "total",
        "next_cursor",
        "as_of",
        "dataset_version",
        "updated_at"
      ],
      "errors": [
        400,
        401,
        403,
        409,
        429,
        503
      ]
    },
    {
      "method": "GET",
      "path": "/v1/orders/{order_id}/events",
      "response": [
        "events",
        "visible_at_cutoff"
      ],
      "errors": [
        401,
        403,
        404,
        503
      ]
    },
    {
      "method": "GET",
      "path": "/v1/metrics",
      "response": [
        "eligible",
        "late",
        "by_hub",
        "by_service",
        "sample_status"
      ],
      "errors": [
        400,
        401,
        403,
        409,
        503
      ]
    },
    {
      "method": "POST",
      "path": "/v1/exports",
      "roles": [
        "supervisor",
        "service"
      ],
      "body": [
        "scope",
        "as_of",
        "dataset_version",
        "request_key"
      ],
      "accepted_status": 202,
      "errors": [
        400,
        401,
        403,
        409,
        429,
        503
      ]
    }
  ]
}

"""Reference implementation for the Ape Intern cold-chain operations lab."""
import csv,json
from collections import defaultdict,Counter
from datetime import datetime,timezone
from pathlib import Path
from urllib.parse import urlparse,parse_qs
UTC=timezone.utc
KINDS={'OPEN':'created','COLLECT':'picked_up','DROP':'delivered','VOID':'cancelled'}
HUBS={'Harbour','Eastgate','Westhill','Southbay'}
def parse_time(value):
    t=datetime.fromisoformat(value.replace('Z','+00:00'))
    if t.tzinfo is None:raise ValueError('timezone_required')
    return t.astimezone(UTC)
def iso(t):return t.isoformat(timespec='seconds').replace('+00:00','Z')
def readcsv(p):
    with Path(p).open(encoding='utf-8-sig',newline='') as f:return list(csv.DictReader(f))
def writecsv(p,rows,fields=None):
    rows=list(rows)
    with Path(p).open('w',encoding='utf-8-sig',newline='') as f:
        w=csv.DictWriter(f,fieldnames=fields or list(rows[0]));w.writeheader();w.writerows(rows)
def dump(p,value):Path(p).write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n')
def _normalize(record,source):
    if source=='A':
        e=dict(event_id=record['event_id'],order_id=record['order_ref'],hub=record['hub'],service=record['service'],kind=record['event'],happened_at=iso(parse_time(record['event_time'])))
    else:
        ms=record['at_ms']
        if type(ms) is not int:raise ValueError('epoch_must_be_integer')
        e=dict(event_id=record['id'],order_id=record['shipment']['number'],hub=record['shipment']['station'],service=record['shipment']['product'],kind=KINDS[record['action']],happened_at=iso(datetime.fromtimestamp(ms/1000,UTC)))
    if e['kind'] not in KINDS.values():raise ValueError('unsupported_event')
    if e['hub'] not in HUBS or e['service'] not in {'standard','express'}:raise ValueError('unknown_dimension')
    if not e['event_id'] or not e['order_id']:raise ValueError('empty_key')
    return e

def normalize(record,source):
    try:return _normalize(record,source)
    except (KeyError,TypeError,OverflowError) as error:raise ValueError(str(error).strip("'")) from error

def ingest(data):
    grouped=defaultdict(list);rejected=[];duplicates=[]
    for source,name in [('A','dispatch-a.jsonl'),('B','courier-b.jsonl')]:
        for line,text in enumerate((data/name).read_text().splitlines(),1):
            loc={'source':name,'line':line}
            try:
                event=normalize(json.loads(text),source);grouped[event['event_id']].append((event,loc))
            except (ValueError,KeyError,TypeError,OverflowError) as ex:
                rejected.append(dict(loc,reason=str(ex).strip("'")))
    accepted=[]
    for eid,items in grouped.items():
        distinct={json.dumps(e,sort_keys=True,ensure_ascii=False) for e,loc in items}
        if len(distinct)>1:
            rejected.extend(dict(loc,reason='immutable_id_conflict') for e,loc in items);continue
        accepted.append(items[0][0])
        duplicates.extend(dict(loc,event_id=eid,reason='exact_replay') for e,loc in items[1:])
    accepted.sort(key=lambda e:(e['happened_at'],e['event_id']))
    return accepted,rejected,duplicates

def classify_order(order,events,cutoff):
    # Only facts available at cutoff can influence a retrospective result.
    visible=[e for e in events if parse_time(e['happened_at'])<=cutoff]
    visible.sort(key=lambda e:parse_time(e['happened_at']))
    delivered=[parse_time(e['happened_at']) for e in visible if e['kind']=='delivered']
    cancelled=any(e['kind']=='cancelled' for e in visible)
    due=parse_time(order['due_at'])
    if cancelled:state='cancelled'
    elif due>cutoff:state='not_mature'
    elif not delivered:state='overdue_open'
    elif min(delivered)>due:state='delivered_late'
    else:state='on_time'
    delay=round((min(delivered)-due).total_seconds()/60) if state=='delivered_late' else 0
    overdue=round((cutoff-due).total_seconds()/60) if state=='overdue_open' else 0
    return dict(order,cohort=state,late_minutes=delay,overdue_minutes=overdue)

def aggregate(rows,keys):
    groups=defaultdict(list)
    for row in rows:groups[tuple(row[k] for k in keys)].append(row)
    result=[]
    for dims,records in sorted(groups.items()):
        counts=Counter(r['cohort'] for r in records)
        eligible=sum(counts[x] for x in ['on_time','delivered_late','overdue_open'])
        late=counts['delivered_late']+counts['overdue_open']
        result.append(dict(zip(keys,dims),total=len(records),eligible=eligible,late=late,late_rate=round(late/eligible,6) if eligible else None,on_time=counts['on_time'],delivered_late=counts['delivered_late'],overdue_open=counts['overdue_open'],cancelled=counts['cancelled'],not_mature=counts['not_mature'],delivered_late_minutes=sum(r['late_minutes'] for r in records),open_overdue_minutes=sum(r['overdue_minutes'] for r in records)))
    return result

def service_band(rate,n,rules):
    if n<rules['min_sample']:return 'More data needed'
    gap=(rate-rules['late_target'])*100
    if gap>rules['priority_gap_pp']:return 'Priority review'
    if gap>rules['review_gap_pp']:return 'Review needed'
    return 'Within threshold'
def fairness(hub_tiers,rules):
    rows=[]
    for x in hub_tiers:
        rate=x['late_rate'];rows.append(dict(hub=x['hub'],service=x['service'],eligible=x['eligible'],late=x['late'],late_rate=rate,target=rules['late_target'],gap_pp=round((rate-rules['late_target'])*100,3) if rate is not None else None,review=service_band(rate,x['eligible'],rules)))
    standardized=[]
    for hub in sorted(HUBS):
        parts=[r for r in rows if r['hub']==hub]
        enough=len(parts)==2 and all(r['eligible']>=rules['min_sample'] for r in parts)
        value=sum(r['late_rate']*rules[r['service']+'_weight'] for r in parts) if enough else None
        standardized.append(dict(hub=hub,standardized_late_rate=round(value,6) if value is not None else None,status='Comparable' if enough else 'More data needed'))
    return rows,standardized

def investigate(logs,policy):
    logins={};findings=[]
    for row in sorted(logs,key=lambda r:(r['at'],r['request_id'])):
        t=parse_time(row['at']);status=int(row['status']);route=urlparse(row['path']);query=parse_qs(route.query)
        if route.path=='/login' and status==200:logins[(row['user'],row['session'])]=t;continue
        if status!=200:continue
        types=[]
        if route.path=='/export' and row['role'] not in policy['export_roles']:types.append('unauthorised_export')
        hub=query.get('hub',[''])[0]
        if row['role']=='operator' and hub and hub!=policy['operator_hub'].get(row['user']):types.append('hub_scope_violation')
        login=logins.get((row['user'],row['session']))
        if login is None or (t-login).total_seconds()>policy['session_ttl_minutes']*60:types.append('expired_or_missing_session')
        if types:findings.append(dict(request_id=row['request_id'],at=row['at'],user=row['user'],session=row['session'],finding=';'.join(types),status=status,rows=int(row['rows']),bytes=int(row['bytes'])))
    return findings

def solve(data,out):
    data=Path(data);out=Path(out);out.mkdir(parents=True,exist_ok=True)
    events,rejects,dups=ingest(data)
    (out/'canonical-events.jsonl').write_text(''.join(json.dumps(e,ensure_ascii=False)+'\n' for e in events))
    writecsv(out/'rejected.csv',rejects);writecsv(out/'duplicates.csv',dups)
    rules=json.loads((data/'rules.json').read_text());cutoff=parse_time(rules['cutoff'])
    by_order=defaultdict(list)
    for e in events:by_order[e['order_id']].append(e)
    orders=readcsv(data/'orders.csv')
    records=[classify_order(o,by_order[o['order_id']],cutoff) for o in orders]
    writecsv(out/'order-review.csv',records)
    hubs=aggregate(records,['hub']);tiers=aggregate(records,['hub','service']);routes=aggregate(records,['route'])
    writecsv(out/'hub-metrics.csv',hubs);writecsv(out/'route-metrics.csv',routes)
    bands,std=fairness(tiers,rules)
    writecsv(out/'service-review.csv',bands);writecsv(out/'standardized-comparison.csv',std)
    findings=investigate(readcsv(data/'access-log.csv'),json.loads((data/'access-policy.json').read_text()))
    writecsv(out/'security-findings.csv',findings)
    summary=dict(orders=len(orders),accepted_events=len(events),rejected=len(rejects),duplicates=len(dups),cohorts=dict(Counter(r['cohort'] for r in records)),hubs=hubs,service_review=bands,standardized=std,security=dict(flagged_requests=len(findings),potential_export_rows=sum(f['rows'] for f in findings if 'unauthorised_export' in f['finding']),potential_export_bytes=sum(f['bytes'] for f in findings if 'unauthorised_export' in f['finding']),findings=findings))
    dump(out/'results.json',summary)
    return summary
if __name__=='__main__':
    import sys
    print(json.dumps(solve(sys.argv[1],sys.argv[2]),ensure_ascii=False,indent=2))