Bouw een agent voor een fictieve offerte

import copy
import hashlib
import json
import math
import re

def validate_quote_job(job):
    if not isinstance(job, dict) or set(job) != {'id', 'sku', 'quantity', 'budget_cents', 'ticket', 'version'}:
        raise ValueError('Exact offertecontract vereist')
    for key in ['id', 'sku', 'ticket']:
        if not isinstance(job[key], str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', job[key]):
            raise ValueError('Begrensde taak-, product- en ticket-ID vereist')
    for key, low, high in [('quantity', 1, 20), ('budget_cents', 0, 1000000), ('version', 0, 1000000)]:
        if type(job[key]) is not int or not low <= job[key] <= high:
            raise ValueError('Aantal, budget of versie buiten het domein')
    return dict(job)

job = {'id': 'quote-a', 'sku': 'lamp', 'quantity': 2,
       'budget_cents': 5000, 'ticket': 'repair-a', 'version': 0}

print(job['quantity'], job['budget_cents'])

# Gegeven helpers uit de eerdere agentlessen.
def validate_ticket_proposal(proposal):
    fields = {'call_id', 'actor', 'name', 'version', 'arguments'}
    if not isinstance(proposal, dict) or set(proposal) != fields:
        raise ValueError('Exact voorstelcontract vereist')
    for key in ['call_id', 'actor', 'name']:
        if not isinstance(proposal[key], str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,39}', proposal[key]):
            raise ValueError('Begrensde ASCII-identificatie vereist')
    if type(proposal['version']) is not int or not 0 <= proposal['version'] <= 1000000:
        raise ValueError('Gehele versie 0..1000000 vereist')
    args = proposal['arguments']
    if not isinstance(args, dict) or set(args) != {'ticket', 'status'}:
        raise ValueError('Exact ticket/status vereist')
    if not isinstance(args['ticket'], str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', args['ticket']) or args['status'] not in ['open', 'done']:
        raise ValueError('Ongeldige ticket-ID of status')
    return {**proposal, 'arguments': dict(args)}

def proposal_fingerprint(proposal):
    clean = validate_ticket_proposal(proposal)
    encoded = json.dumps(clean, sort_keys=True, ensure_ascii=True, separators=(',', ':'))
    return hashlib.sha256(encoded.encode('ascii')).hexdigest()

def dispatch_ticket(proposal, actor, approved, state, completed):
    proposal = validate_ticket_proposal(proposal)
    if not isinstance(actor, str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,39}', actor):
        raise ValueError('Hostactor vereist')
    if not isinstance(approved, set) or len(approved) > 64 or any(not isinstance(x, str) or not re.fullmatch(r'[0-9a-f]{64}', x) for x in approved):
        raise ValueError('Begrensde hostgoedkeuringen vereist')
    if not isinstance(state, dict) or set(state) != {'version', 'tickets'} or type(state['version']) is not int or not 0 <= state['version'] <= 1000001:
        raise ValueError('Ongeldige hosttoestand')
    if not isinstance(state['tickets'], dict) or not 1 <= len(state['tickets']) <= 16 or any(not isinstance(k, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', k) or v not in ['open', 'done'] for k, v in state['tickets'].items()):
        raise ValueError('Ongeldige eigen tickets')
    if not isinstance(completed, dict) or len(completed) > 64:
        raise ValueError('Begrensd intern uitvoerlog vereist')
    if proposal['actor'] != actor:
        return {'status': 'forbidden'}
    if proposal['name'] != 'set_ticket':
        return {'status': 'unknown-tool'}
    digest = proposal_fingerprint(proposal)
    prior = completed.get(proposal['call_id'])
    if prior is not None:
        if prior['fingerprint'] != digest:
            return {'status': 'call-id-conflict'}
        return {**copy.deepcopy(prior['result']), 'replayed': True}
    if proposal['version'] != state['version']:
        return {'status': 'stale-version'}
    if digest not in approved:
        return {'status': 'approval-required'}
    if proposal['arguments']['ticket'] not in state['tickets']:
        return {'status': 'missing-ticket'}
    if len(completed) == 64:
        return {'status': 'log-full'}
    state['tickets'][proposal['arguments']['ticket']] = proposal['arguments']['status']
    state['version'] += 1
    result = {'status': 'ok', 'version': state['version'], 'ticket': proposal['arguments']['ticket'], 'value': proposal['arguments']['status'], 'replayed': False}
    completed[proposal['call_id']] = {'fingerprint': digest, 'result': copy.deepcopy(result)}
    return result

def make_quote_tools(job, approved, state, completed):
    job = validate_quote_job(job)
    if not isinstance(approved, set) or len(approved) > 64 or any(not isinstance(x, str) or not re.fullmatch(r'[0-9a-f]{64}', x) for x in approved):
        raise ValueError('Begrensd hostregister met echte voorstelhashes vereist')
    if not isinstance(state, dict) or set(state) != {'version', 'tickets'} or type(state['version']) is not int or not 0 <= state['version'] <= 1000001:
        raise ValueError('Ongeldige hosttoestand')
    if not isinstance(state['tickets'], dict) or not 1 <= len(state['tickets']) <= 16 or job['ticket'] not in state['tickets'] or any(not isinstance(k, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', k) or v not in ['open', 'done'] for k, v in state['tickets'].items()):
        raise ValueError('Eigen ticket moet in de hosttoestand bestaan')
    if not isinstance(completed, dict) or len(completed) > 64:
        raise ValueError('Begrensd intern uitvoerlog vereist')
    # Frozen own classroom catalogue, in cents. No external store or purchase.
    catalogue = {'lamp': 2499, 'sensor': 1499, 'hub': 3999}
    approvals = set(approved)
    def lookup_price(sku):
        if sku != job['sku'] or not isinstance(sku, str):
            raise ValueError('Product buiten taakscope')
        return {'found': sku in catalogue, 'sku': sku, 'unit_cents': catalogue.get(sku)}
    def quote_total(unit_cents, quantity):
        if type(unit_cents) is not int or unit_cents != catalogue.get(job['sku']) or type(quantity) is not int or quantity != job['quantity']:
            raise ValueError('Prijs en aantal moeten bij de vastgelegde taak horen')
        return {'total_cents': unit_cents * quantity}
    def close_ticket(operation_id, version):
        if operation_id != job['id'] or type(version) is not int or version != job['version']:
            raise ValueError('Bewerking buiten taakscope')
        if job['sku'] not in catalogue:
            return {'status': 'unknown-product'}
        if catalogue[job['sku']] * job['quantity'] > job['budget_cents']:
            return {'status': 'over-budget'}
        proposal = {'call_id': operation_id, 'actor': 'learner', 'name': 'set_ticket', 'version': version, 'arguments': {'ticket': job['ticket'], 'status': 'done'}}
        return dispatch_ticket(proposal, 'learner', approvals, state, completed)
    return {'lookup_price': lookup_price, 'quote_total': quote_total, 'close_ticket': close_ticket}

# Gegeven hostfixture: uitsluitend dit concrete voorstel krijgt akkoord.
proposal = {'call_id': job['id'], 'actor': 'learner', 'name': 'set_ticket',
            'version': job['version'],
            'arguments': {'ticket': job['ticket'], 'status': 'done'}}
state = {'version': 0, 'tickets': {job['ticket']: 'open'}}
completed = {}
approved = {proposal_fingerprint(proposal)}

tools = make_quote_tools(job, approved, state, completed)
print(tools['lookup_price']('lamp')['unit_cents'])
print(tools['quote_total'](2499, 2)['total_cents'])

def make_quote_policy(job, replay=False):
    job = validate_quote_job(job)
    if type(replay) is not bool:
        raise ValueError('Replay is een expliciete proefkeuze')
    def policy(observations):
        def final(text):
            return {'kind': 'final', 'text': text}
        def call(name, arguments):
            return {'kind': 'tool', 'call_id': 'call'+str(len(observations)+1), 'name': name, 'arguments': arguments}
        if not observations:
            return call('lookup_price', {'sku': job['sku']})
        last = observations[-1]
        if not last['ok']:
            return final('stopped: tool-error')
        data = last['data']
        if last['name'] == 'lookup_price':
            if not data['found']:
                return final('stopped: unknown-product')
            return call('quote_total', {'unit_cents': data['unit_cents'], 'quantity': job['quantity']})
        if last['name'] == 'quote_total':
            if data['total_cents'] > job['budget_cents']:
                return final('stopped: over-budget')
            return call('close_ticket', {'operation_id': job['id'], 'version': job['version']})
        if last['name'] == 'close_ticket':
            if data['status'] != 'ok':
                return final('stopped: '+data['status'])
            if replay and not data['replayed']:
                return call('close_ticket', {'operation_id': job['id'], 'version': job['version']})
            return final('quote-checked: '+job['ticket'])
        raise ValueError('Onbekende observatiestap')
    return policy

policy = make_quote_policy(job)
print(policy([])['name'])

# Gegeven helpers uit de eerdere agentlessen.
def validate_agent_action(value):
    if not isinstance(value, dict):
        raise ValueError('Actie is een dictionary')
    if value.get('kind') == 'final':
        if set(value) != {'kind', 'text'} or not isinstance(value['text'], str) or not value['text'].strip() or len(value['text']) > 2000:
            raise ValueError('Final heeft uitsluitend niet-lege tekst tot 2000 tekens')
        return dict(value)
    if value.get('kind') != 'tool' or set(value) != {'kind', 'call_id', 'name', 'arguments'}:
        raise ValueError('Toolactie heeft kind/call_id/name/arguments')
    if any(not isinstance(value[k], str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,39}', value[k]) for k in ['call_id', 'name']):
        raise ValueError('Begrensde ASCII-call-ID en toolnaam vereist')
    args = value['arguments']
    if not isinstance(args, dict) or len(args) > 8:
        raise ValueError('Maximaal acht argumenten')
    for key, item in args.items():
        if not isinstance(key, str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_]{0,31}', key):
            raise ValueError('Ongeldige argumentnaam')
        if type(item) is str:
            if len(item) > 2000:
                raise ValueError('Argumenttekst te lang')
        elif type(item) in (int, float):
            if abs(item) > 1e6 or not math.isfinite(item):
                raise ValueError('Getal buiten bereik')
        elif item is not None and type(item) is not bool:
            raise ValueError('Alleen JSON-scalarargumenten in deze oefening')
    return {**value, 'arguments': dict(args)}

def run_bounded_agent(policy, tools, max_decisions=8, max_tool_calls=4):
    if not callable(policy) or not isinstance(tools, dict) or len(tools) > 8 or any(not isinstance(name, str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,39}', name) or not callable(tool) for name, tool in tools.items()):
        raise ValueError('Policy en maximaal acht benoemde tools vereist')
    if type(max_decisions) is not int or not 1 <= max_decisions <= 32 or type(max_tool_calls) is not int or not 0 <= max_tool_calls <= 16:
        raise ValueError('Beslisbudget 1–32, toolbudget 0–16')
    observations, trace, seen, attempts = [], [], set(), 0
    def finish(status, text=None):
        return {'status': status, 'answer': text, 'observations': observations, 'trace': trace, 'tool_attempts': attempts}
    for step in range(max_decisions):
        try:
            action = validate_agent_action(policy(copy.deepcopy(observations)))
        except Exception:
            trace.append({'step': step, 'kind': 'rejected', 'reason': 'invalid-policy-action'})
            return finish('policy-error')
        if action['kind'] == 'final':
            trace.append({'step': step, 'kind': 'final'})
            return finish('final', action['text'])
        if action['call_id'] in seen:
            trace.append({'step': step, 'kind': 'rejected', 'reason': 'duplicate-call-id'})
            return finish('duplicate-call')
        if attempts >= max_tool_calls:
            trace.append({'step': step, 'kind': 'rejected', 'reason': 'tool-budget'})
            return finish('tool-budget')
        seen.add(action['call_id']); attempts += 1
        tool = tools.get(action['name'])
        if tool is None:
            observation = {'call_id': action['call_id'], 'name': action['name'], 'ok': False, 'data': 'unknown-tool'}
        else:
            try:
                value = tool(**action['arguments'])
                encoded = json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(',', ':'))
                if len(encoded.encode('utf-8')) > 4096:
                    raise ValueError('Toolresultaat te groot')
                observation = {'call_id': action['call_id'], 'name': action['name'], 'ok': True, 'data': json.loads(encoded)}
            except Exception:
                observation = {'call_id': action['call_id'], 'name': action['name'], 'ok': False, 'data': 'tool-error'}
        observations.append(observation)
        trace.append({'step': step, 'kind': 'tool', 'call_id': action['call_id'], 'name': action['name'], 'executed': tool is not None, 'ok': observation['ok']})
    return finish('decision-budget')

def trace_work(trace, decision_units, tool_units):
    if not isinstance(trace, list) or len(trace) > 32 or type(decision_units) is not int or not 0 <= decision_units <= 1000 or not isinstance(tool_units, dict) or len(tool_units) > 8 or any(not isinstance(k, str) or type(v) is not int or not 0 <= v <= 1000 for k, v in tool_units.items()):
        raise ValueError('Begrensde trace en gehele oefenkosten vereist')
    decisions, calls, units = 0, 0, 0
    for i, row in enumerate(trace):
        if not isinstance(row, dict) or type(row.get('step')) is not int or row['step'] != i or row.get('kind') not in ['tool', 'final', 'rejected']:
            raise ValueError('Ongeldige opeenvolgende tracestappen')
        decisions += 1; units += decision_units
        if row['kind'] == 'tool':
            if type(row.get('executed')) is not bool or not isinstance(row.get('name'), str):
                raise ValueError('Tooltrace mist naam/uitvoering')
            if row['executed']:
                if row['name'] not in tool_units:
                    raise ValueError('Onbekende toolkosten')
                calls += 1; units += tool_units[row['name']]
    return {'decisions': decisions, 'executed_tools': calls, 'work_units': units}

def run_quote_case(job, grant=False, replay=False, max_decisions=8, max_tool_calls=4):
    job = validate_quote_job(job)
    if type(grant) is not bool:
        raise ValueError('Grant is een hostfixture, geen modelbesluit')
    state, completed = {'version': 0, 'tickets': {job['ticket']: 'open'}}, {}
    proposal = {'call_id': job['id'], 'actor': 'learner', 'name': 'set_ticket', 'version': job['version'], 'arguments': {'ticket': job['ticket'], 'status': 'done'}}
    approved = {proposal_fingerprint(proposal)} if grant else set()
    tools = make_quote_tools(job, approved, state, completed)
    policy = make_quote_policy(job, replay)
    result = run_bounded_agent(policy, tools, max_decisions, max_tool_calls)
    work = trace_work(result['trace'], 2, {'lookup_price': 1, 'quote_total': 2, 'close_ticket': 4})
    return {'run': result, 'work': work, 'state': copy.deepcopy(state), 'completed_operations': len(completed), 'ticket_done': state['tickets'][job['ticket']] == 'done'}

trial = run_quote_case(job, grant=True)
print(trial['run']['status'], trial['ticket_done'])
print(trial['work']['decisions'], trial['work']['executed_tools'], trial['completed_operations'])
print(trial['work']['work_units'])

# Gegeven onderwijsgevallen; expected stuurt de policy niet.
def make_quote_cases():
    job = {'id': 'quote-a', 'sku': 'lamp', 'quantity': 2, 'budget_cents': 5000, 'ticket': 'repair-a', 'version': 0}
    base = {'job': job, 'grant': True, 'replay': False, 'max_decisions': 8, 'max_tool_calls': 4, 'expected_status': 'final', 'expected_answer': 'quote-checked: repair-a', 'expected_done': True}
    cases = [
            {**base, 'id': 'ok'},
            {**base, 'id': 'approval', 'grant': False, 'expected_answer': 'stopped: approval-required', 'expected_done': False},
            {**base, 'id': 'budget', 'job': {**job, 'budget_cents': 4997}, 'expected_answer': 'stopped: over-budget', 'expected_done': False},
            {**base, 'id': 'missing', 'job': {**job, 'sku': 'missing'}, 'expected_answer': 'stopped: unknown-product', 'expected_done': False},
            {**base, 'id': 'stale', 'job': {**job, 'version': 1}, 'expected_answer': 'stopped: stale-version', 'expected_done': False},
            {**base, 'id': 'replay', 'replay': True},
            {**base, 'id': 'tool-cap', 'max_tool_calls': 2, 'expected_status': 'tool-budget', 'expected_answer': None, 'expected_done': False},
            {**base, 'id': 'step-cap', 'max_decisions': 3, 'expected_status': 'decision-budget', 'expected_answer': None},
        ]
    return cases

def evaluate_quote_cases(cases):
    if not isinstance(cases, list) or not 1 <= len(cases) <= 32:
        raise ValueError('1..32 eigen gevallen vereist')
    seen, rows = set(), []
    for case in cases:
        if not isinstance(case, dict) or set(case) != {'id', 'job', 'grant', 'replay', 'max_decisions', 'max_tool_calls', 'expected_status', 'expected_answer', 'expected_done'}:
            raise ValueError('Exact gevalcontract vereist')
        if not isinstance(case['id'], str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', case['id']) or case['id'] in seen:
            raise ValueError('Unieke geval-ID vereist')
        if case['expected_status'] not in ['final', 'policy-error', 'duplicate-call', 'tool-budget', 'decision-budget'] or type(case['expected_done']) is not bool:
            raise ValueError('Verwachte stopstatus en effect vereist')
        answer = case['expected_answer']
        if (case['expected_status'] == 'final' and (not isinstance(answer, str) or not answer.strip() or len(answer) > 2000)) or (case['expected_status'] != 'final' and answer is not None):
            raise ValueError('Final heeft antwoordtekst; andere stops hebben geen antwoord')
        seen.add(case['id'])
        result = run_quote_case(case['job'], case['grant'], case['replay'], case['max_decisions'], case['max_tool_calls'])
        expected_state = {'version': int(case['expected_done']),
                          'tickets': {case['job']['ticket']: 'done' if case['expected_done'] else 'open'}}
        effect_ok = result['state'] == expected_state and result['completed_operations'] == int(case['expected_done'])
        rows.append({'id': case['id'], 'status': result['run']['status'], 'answer': result['run']['answer'],
                     'ticket_done': result['ticket_done'], 'state': copy.deepcopy(result['state']),
                     'version': result['state']['version'], 'operations': result['completed_operations'],
                     **result['work'], 'passed': result['run']['status'] == case['expected_status']
                     and result['run']['answer'] == answer and result['ticket_done'] == case['expected_done'] and effect_ok})
    return {'rows': rows, 'passed': sum(row['passed'] for row in rows), 'n': len(rows)}

def plot_quote_report(report):
    """Draw actual counts; input is the internal bounded evaluation report."""
    import matplotlib.pyplot as plt
    rows = report['rows']
    labels = ['\n'.join(row['id'][i:i+12] for i in range(0, len(row['id']), 12)) for row in rows]
    line_count = sum(label.count('\n')+1 for label in labels)
    fields = [('decisions', 'Beslissingen', '#236b4f'),
              ('executed_tools', 'Functieaanroepen', '#af6a35'),
              ('operations', 'Mutaties', '#385b92')]
    maximum = max(row[field] for row in rows for field, _, _ in fields)
    figure, axes = plt.subplots(3, 1, figsize=(4.2, max(9.2, 1.3+.65*line_count)))
    figure.suptitle('Eigen toolagent', fontsize=12)
    positions = list(range(len(rows)))
    for axis, (field, title, color) in zip(axes, fields):
        values = [row[field] for row in rows]
        bars = axis.barh(positions, values, height=.64, color=color)
        axis.set_yticks(positions, labels)
        axis.invert_yaxis()
        axis.set_xlim(0, maximum+1)
        axis.set_xticks(sorted({0, maximum//2, maximum}))
        axis.set_title(title, fontsize=12)
        axis.tick_params(labelsize=12)
        for bar, value in zip(bars, values):
            axis.text(value+.08, bar.get_y()+bar.get_height()/2,
                      str(value), va='center', fontsize=12)
    axes[-1].set_xlabel('Aantal per proef', fontsize=12)
    figure.tight_layout(pad=1.4, rect=(0, 0, 1, .98))
    return figure

# Gegeven gemakshulp voor hetzelfde rapport.
def quote_lab_report():
    report = evaluate_quote_cases(make_quote_cases())
    return report, plot_quote_report(report)

cases = make_quote_cases()
report = evaluate_quote_cases(cases)
figure = plot_quote_report(report)
print(report['passed'], report['n'])
print([row['operations'] for row in report['rows']])