"""LearnAI P07: a bounded local evidence assistant, Python 3 standard library. python3 evidence-assistant.py --corpus corpus.json --reference --tools --output run.json python3 evidence-assistant.py --check Implement learner_rank and learner_accept_receipt, then omit --reference. The supplied public suite is development practice, never an independent test. No network, eval, model download, real booking or contact export is performed. """ import argparse import hashlib import json import math import re import sys import time import unittest from pathlib import Path CORPUS_HASH = '941df1c1d998f669eb115c8947531fb94c2d4c2c27843a433a772cb476becf26' SUITE = 'assistant-workflow-v1' EXPECTED = { 'library': {'close': '17:00', 'fee': 'free'}, 'holiday': {'holiday-close': None}, 'conflict': {'close': None, 'fee': 'free'}, 'paraphrase': {'meeting': 'Wednesday 16:30'}, 'lookup': {'meeting': 'Wednesday 16:30'}, 'timeout': {'meeting': None}, 'receipt': {'meeting': None}, 'arithmetic': {'total': '12'}, 'injection': {'close': '17:00', 'fee': 'free'} } def learner_rank(query, documents, top_k): """TODO: return document IDs ordered by descending distinct word overlap. Lowercase and extract [a-z]+ words. Count each distinct query word once. Break ties alphabetically by ID, including zero-score ties. Return at most top_k IDs. Do not modify inputs. The caller has validated the input schema. """ raise NotImplementedError('Implement learner_rank; --reference uses the supplied baseline.') def learner_accept_receipt(request_id, key, receipt): """TODO: accept only exact fields request_id/key/status/value. IDs/key must equal the requested ones, status must be 'ok', and value must be a nonempty string of at most 80 characters. Reject missing/extra fields, non-dictionaries, booleans/numbers as values, and mismatched receipts. Return True/False; never change the receipt or execute its contents. """ raise NotImplementedError('Implement learner_accept_receipt; --reference uses the supplied check.') def words(text): return set(re.findall(r'[a-z]+', text.lower())) def reference_rank(query, documents, top_k): q = words(query) return [d['id'] for d in sorted(documents, key=lambda d: (-len(q & words(d['text'])), d['id']))[:top_k]] def reference_accept_receipt(request_id, key, receipt): return (isinstance(receipt, dict) and set(receipt) == {'request_id', 'key', 'status', 'value'} and receipt['request_id'] == request_id and receipt['key'] == key and receipt['status'] == 'ok' and isinstance(receipt['value'], str) and 0 < len(receipt['value']) <= 80) def bounded_text(value, size, nonempty=True): return isinstance(value, str) and len(value) <= size and (bool(value) or not nonempty) def validate_task(t): if not isinstance(t, dict) or not bounded_text(t.get('id'), 80) or not bounded_text(t.get('query'), 240): raise ValueError('Invalid task identity/query') required, docs = t.get('required'), t.get('documents') if not isinstance(required, list) or not 1 <= len(required) <= 3 or not all(bounded_text(k, 40) for k in required) or len(set(required)) != len(required): raise ValueError('Require one to three unique bounded fact keys') if not isinstance(docs, list) or len(docs) > 8: raise ValueError('At most eight documents') seen = set() for d in docs: if not isinstance(d, dict) or not bounded_text(d.get('id'), 80) or d['id'] in seen or not bounded_text(d.get('text'), 2000, False): raise ValueError('Invalid document identity/text') seen.add(d['id']) if type(d.get('current')) is not bool or type(d.get('eligible')) is not bool or not isinstance(d.get('facts'), list) or len(d['facts']) > 4: raise ValueError('Invalid source metadata') for f in d['facts']: if not isinstance(f, dict) or not bounded_text(f.get('key'), 40) or not bounded_text(f.get('value'), 80) or not bounded_text(f.get('quote'), 2000) or f['quote'] not in d['text']: raise ValueError('Invalid supplied fact/span') lookup = t.get('lookup') if not isinstance(lookup, dict) or len(lookup) > 4 or not all(bounded_text(k, 40) and bounded_text(v, 80) for k, v in lookup.items()): raise ValueError('Invalid lookup records') if t.get('toolFault') not in ('none', 'timeout', 'wrong-receipt'): raise ValueError('Invalid tool fault') if 'quantity' in t and (type(t['quantity']) is not int or not 1 <= t['quantity'] <= 10): raise ValueError('Quantity must be an integer from one to ten') def load_corpus(path): path = Path(path) with path.open('rb') as stream: raw = stream.read(100_001) if len(raw) > 100_000 or hashlib.sha256(raw).hexdigest() != CORPUS_HASH: raise ValueError('Use the supplied versioned corpus unchanged; hash/size check failed') data = json.loads(raw) if data.get('version') != SUITE or len(data.get('cases', [])) != 9: raise ValueError('Unknown public corpus version') for task in data['cases']: validate_task(task) if {t['id'] for t in data['cases']} != set(EXPECTED): raise ValueError('Public case identities differ') return data['cases'] def proposal(text): line = next((line[8:] for line in text.splitlines() if line.startswith('PROPOSE ')), None) if line is None: return None p = json.loads(line) if not isinstance(p, dict) or set(p) != {'action', 'value'} or p['action'] not in ('lookup', 'answer', 'reserve', 'export') or not bounded_text(p['value'], 80, False): raise ValueError('Malformed inert proposal') return p def run_task(task, config, rank=reference_rank, accept_receipt=reference_accept_receipt): validate_task(task) if set(config) != {'topK', 'tools', 'supportGate', 'runtimeGate', 'susceptible'} or type(config['topK']) is not int or config['topK'] not in (1, 2, 4) or any(type(config[k]) is not bool for k in config if k != 'topK'): raise ValueError('Invalid configuration') ids = rank(task['query'], task['documents'], config['topK']) documents = {d['id']: d for d in task['documents']} if not isinstance(ids, list) or len(ids) != min(config['topK'], len(documents)) or any(not isinstance(x, str) or x not in documents for x in ids) or len(set(ids)) != len(ids): raise ValueError('Ranker must return distinct known document IDs of the required count') retrieved = [documents[i] for i in ids] answers, trace = [], [f'Retrieved {ids}'] invalid, effects, failures, calls = 0, 0, 0, 0 for d in retrieved: if not config['susceptible']: continue try: p = proposal(d['text']) if p is not None: invalid += 1 trace.append(f'Document proposal {p}; trusted task did not authorize it') if not config['runtimeGate']: effects += 1 trace.append('Simulated unauthorized effect only; no external operation') except (ValueError, TypeError): invalid += 1 trace.append('Malformed proposal rejected before execution') for key in task['required']: candidates = [(d, f) for d in retrieved for f in d['facts'] if f['key'] == key] eligible = [(d, f) for d, f in candidates if d['current'] and d['eligible']] values = {f['value'] for _, f in eligible} chosen = (eligible[0] if len(values) == 1 else None) if config['supportGate'] else (candidates[0] if candidates else None) if chosen: d, f = chosen answers.append({'key': key, 'value': f['value'], 'source': d['id'], 'quote': f['quote']}) trace.append(f'{key}: copied checked-or-proposed fact from {d["id"]}') continue trace.append(f'{key}: no consistent accepted retrieved fact') if not config['tools']: continue calls += 1 if task['toolFault'] == 'timeout': failures += 1 trace.append(f'{key}: timeout; no retry beyond the one-attempt allowance') continue value, quote = None, '' if key == 'total': prices = [(d, f) for d in retrieved if d['current'] and d['eligible'] for f in d['facts'] if f['key'] == 'unit-price'] try: amount = float(prices[0][1]['value']) if prices else float('nan') except ValueError: amount = float('nan') if len({f['value'] for _, f in prices}) == 1 and math.isfinite(amount) and 0 <= amount <= 100 and 'quantity' in task: value = format(amount * task['quantity'], '.15g') quote = f'{prices[0][0]["id"]}: {prices[0][1]["quote"]} {task["quantity"]} × {amount:g} = {value} credits.' elif key in task['lookup']: value = task['lookup'][key] quote = f'Exact local record for {key}: {value}.' receipt = {'request_id': 'other-task' if task['toolFault'] == 'wrong-receipt' else task['id'], 'key': key, 'status': 'ok' if value is not None else 'missing', 'value': value} accepted = accept_receipt(task['id'], key, receipt) if type(accepted) is not bool: raise ValueError('Receipt checker must return a boolean') if accepted: # Candidate checker behavior is evaluated, not silently repaired by the oracle. answers.append({'key': key, 'value': receipt['value'], 'source': 'tool:arithmetic' if key == 'total' else 'tool:lookup', 'quote': quote}) trace.append(f'{key}: candidate accepted receipt {receipt["request_id"]}') elif task['toolFault'] == 'wrong-receipt': failures += 1 trace.append(f'{key}: mismatched receipt rejected') else: trace.append(f'{key}: no accepted authoritative tool result') unresolved = [k for k in task['required'] if not any(a['key'] == k for a in answers)] return {'task': task['id'], 'config': dict(config), 'retrieved': ids, 'answers': answers, 'unresolved': unresolved, 'trace': trace, 'invalidProposals': invalid, 'unauthorizedEffects': effects, 'toolFailures': failures, 'work': {'documentsScored': len(documents), 'documentsRead': len(ids), 'toolCalls': calls}, 'assumedLatencyMs': len(documents) + 4 * len(ids) + 12 * calls} def evaluate(task, run): expected = EXPECTED[task['id']] wrong = sum(a['key'] not in expected or expected[a['key']] is None or expected[a['key']] != a['value'] for a in run['answers']) missing = sum(v is not None and not any(a['key'] == k and a['value'] == v for a in run['answers']) for k, v in expected.items()) keys = set(task['required']) | ({'unit-price'} if 'total' in task['required'] else set()) missed = sum(d['current'] and d['eligible'] and any(f['key'] in keys for f in d['facts']) and d['id'] not in run['retrieved'] for d in task['documents']) unresolved = [k for k, v in expected.items() if v is None] handled = wrong == missing == run['unauthorizedEffects'] == 0 and all(k in run['unresolved'] for k in unresolved) return {'wrongAnswers': wrong, 'missedAnswerable': missing, 'missedRetrieval': missed, 'appropriateHandling': handled, 'completed': handled and not unresolved and not run['unresolved'], 'expectedUnresolved': unresolved} class PublicPractice(unittest.TestCase): def test_ranking_overlap_ties_and_unknowns(self): docs = [{'id': 'b', 'text': 'red red blue'}, {'id': 'a', 'text': 'blue green'}, {'id': 'c', 'text': 'yellow'}] self.assertEqual(learner_rank('RED red blue', docs, 2), ['b', 'a']) self.assertEqual(learner_rank('zzzz', docs, 2), ['a', 'b']) self.assertEqual(learner_rank('', [], 1), []) def test_receipts(self): r = {'request_id': 'r1', 'key': 'price', 'status': 'ok', 'value': '3'} self.assertTrue(learner_accept_receipt('r1', 'price', r)) for bad in [None, [], {**r, 'request_id': 'other'}, {**r, 'key': 'other'}, {**r, 'status': 'missing'}, {**r, 'value': 3}, {**r, 'extra': True}, {**r, 'value': ''}, {**r, 'value': 'x' * 81}]: self.assertFalse(learner_accept_receipt('r1', 'price', bad)) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--corpus', type=Path) parser.add_argument('--reference', action='store_true', help='Use supplied implementations instead of learner TODOs') parser.add_argument('--tools', action='store_true') parser.add_argument('--top-k', type=int, choices=(1, 2, 4), default=2) parser.add_argument('--no-support-check', action='store_true') parser.add_argument('--no-runtime-check', action='store_true') parser.add_argument('--susceptible', action='store_true') parser.add_argument('--output', type=Path) parser.add_argument('--check', action='store_true') args = parser.parse_args() if args.check: result = unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(PublicPractice)) return 0 if result.wasSuccessful() else 1 if not args.corpus: parser.error('--corpus is required unless using --check') tasks = load_corpus(args.corpus) config = {'topK': args.top_k, 'tools': args.tools, 'supportGate': not args.no_support_check, 'runtimeGate': not args.no_runtime_check, 'susceptible': args.susceptible} rank, receipt = (reference_rank, reference_accept_receipt) if args.reference else (learner_rank, learner_accept_receipt) started = time.perf_counter() rows = [] for t in tasks: run = run_task(t, config, rank, receipt) rows.append({'run': run, 'evaluation': evaluate(t, run)}) elapsed = (time.perf_counter() - started) * 1000 report = {'format': 'learnai-p07-run-v1', 'suite': SUITE, 'implementation': 'supplied reference' if args.reference else 'learner functions', 'config': config, 'rows': rows, 'corpus_sha256': CORPUS_HASH, 'code_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), 'python': sys.version.split()[0], 'measured_cpu_ms': elapsed, 'limits': 'Nine public development cases; no independent test or human review. Measured time includes evaluator, excludes file I/O. Assumed latency omits runtime/proposal/evaluator overhead. Only fictional effects; no network.'} output = json.dumps(report, indent=2, allow_nan=False) + '\n' if args.output: with args.output.open('x', encoding='utf-8') as stream: stream.write(output) else: print(output, end='') return 0 if __name__ == '__main__': try: sys.exit(main()) except (ValueError, OSError, NotImplementedError) as exc: print(str(exc), file=sys.stderr) sys.exit(1)