"""Evidence packages (v3) in the answer service: per-unit, source-grounded, page-referenced. Text sources: each question unit (entity / label / boolean / choice) gets an evidence package (solomon.evidence_packages.build): its selected spans plus automatically attached governing context, all exact source ranges with code-point offsets, roles and reasons, and -- when a renderer page map is registered for the source -- rendered page IDs. `sufficiency` re-asks each unit with ONLY its package as a fresh document; `removal` re-asks the whole question with every selected evidence span removed. Budgets are checked before any model call. Evidence never promotes answer qualification and never establishes causal faithfulness. Image sources are delegated unchanged to solomon.service_evidence.EvidenceService (page selector path). """ import copy from solomon import evidence_packages as v3 from solomon.evidence import digest, validate_spans from solomon.retrieval import lexical_select, remove from solomon.service_evidence import EvidenceService, question_units, prediction CONTRACT = 'solomon-evidence-packages-v3' SIZE_CAP = v3.SIZE_CAP def unit_requests(task, request): """(question text, subject, restricted decision request) per unit, in answer-vector order.""" base = {k: request[k] for k in ('question', 'options', 'labels', 'template', 'entities') if k in request} qs = question_units(task, request) if task == 'entity': return [(q, e, {**base, 'entities': [e]}) for q, e in zip(qs, request['entities'])] if task == 'multilabel': return [(q, None, {**base, 'labels': [label]}) for q, label in zip(qs, request['labels'])] return [(qs[0], None, base)] class PackageEvidenceService(EvidenceService): """page_maps: optional {source_sha256: rendered page map} from source-bound renderer manifests.""" def __init__(self, answers, decide, selector=None, page_selector=None, *, page_maps=None, size_cap=SIZE_CAP, package_format='per_unit'): super().__init__(answers, decide, selector, page_selector) if package_format not in ('per_unit', 'union'): raise ValueError('package_format must be per_unit or union') self.package_format = package_format self.page_maps = dict(page_maps or {}) if not 0 < size_cap <= 1: raise ValueError('invalid package size cap') self.size_cap = size_cap def create(self, document): result = self.answers.create(document);result['contract'] = CONTRACT;return result def _unit_evidence(self, text, units, selection): """Per-unit selected spans. Selectors reporting `units` keep per-unit attribution; otherwise every unit receives the selector's (validated) evidence.""" union = validate_spans(text, selection['evidence']) if 'units' not in selection: return [union for _ in units] by = {u['unit']: u['evidence'] for u in selection['units']} if set(by) - set(range(len(units))): raise ValueError('selection reports unknown question units') out = [] for i in range(len(units)): spans = validate_spans(text, by.get(i, [])) if any(not any(u['start'] <= s['start'] and s['end'] <= u['end'] for u in union) for s in spans): raise ValueError('unit evidence outside the reported selection') out.append(spans) return out def ask(self, key, task, *, evidence='none', evidence_max_calls=2, **request): if evidence not in ('none', 'support', 'sufficiency', 'removal'): raise ValueError('invalid evidence verification level') if type(evidence_max_calls) is not int or not 0 <= evidence_max_calls <= 512: raise ValueError('invalid evidence call budget') with self.answers.lock: state = self.answers._warm(key) if any('image' in p for p in state['parts']): result = super().ask(key, task, evidence=evidence, evidence_max_calls=evidence_max_calls, **request) result['contract'] = CONTRACT if evidence != 'none' and isinstance(result.get('evidence'), dict): result['evidence']['package_format'] = 'pages_only' return result result = self.answers.ask(key, task, **request);result['contract'] = CONTRACT if evidence == 'none': return result parts = copy.deepcopy(state['parts']);text = ''.join(p['text'] for p in parts if 'text' in p) units = unit_requests(task, request);qs = [u[0] for u in units] only = len(units) if self.package_format == 'per_unit' else 1 needed = {'support': 0, 'sufficiency': only, 'removal': only + 1}[evidence] selector = self.selector or lexical_select;plan = None;planned = 0 if hasattr(selector, 'plan'): plan = (selector.plan(text, qs, task=task) if getattr(selector, 'task_aware', False) else selector.plan(text, qs)) if type(plan.get('calls')) is not int or plan['calls'] < 0: raise ValueError('invalid selection call estimate') planned = plan['calls'] if needed + planned > evidence_max_calls: result['evidence'] = {'status': 'budget_exhausted', 'calls': 0, 'required_calls': needed + planned, 'verification': 'none'} return result selection = selector.execute(plan) if plan is not None else selector(text, qs) cost = selection.get('cost', {'calls': 0, 'input_tokens': 0, 'branches': 0}) if any(type(cost.get(k)) is not int or cost[k] < 0 for k in ('calls', 'input_tokens', 'branches')): raise ValueError('invalid selection static cost') if cost['calls'] != planned: raise ValueError('selection exceeded declared inference plan') per_unit = self._unit_evidence(text, units, selection) st = v3.Structure(text);page_map = self.page_maps.get(digest(text)) packages = [] for i, ((q, subject, _), spans) in enumerate(zip(units, per_unit)): p = v3.build(text, q, spans, subject=subject, structure=st, cap=self.size_cap) if page_map is not None: p = v3.with_pages(p, text, page_map) packages.append({'unit': i, 'question': q, 'subject': subject, 'no_support_found': p['no_support_found'], 'spans': p['spans'], 'evidence_ranges': p['evidence_ranges'], 'context_reasons': p['context_reasons'], 'rendered_characters': p['rendered_characters'], 'covered_characters': p['covered_characters'], 'source_fraction': p['source_fraction'], 'within_size_cap': p['within_size_cap'], 'trimmed_for_cap': p['trimmed_for_cap'], **({'pages': p['pages']} if page_map is not None else {}), '_text': p['text']}) union_refs = [] for s in sorted({(s['start'], s['end']) for spans in per_unit for s in spans}): if union_refs and s[0] < union_refs[-1][1]: union_refs[-1] = (union_refs[-1][0], max(s[1], union_refs[-1][1])) else: union_refs.append(s) refs = [{'start': a, 'end': b, 'text': text[a:b], 'source_sha256': digest(text)} for a, b in union_refs] body = {'status': 'no_support_found' if not refs else 'found', 'references': refs, 'packages': [{k: v for k, v in p.items() if k != '_text'} for p in packages], 'package_format': self.package_format, 'package_schema': v3.SCHEMA, 'size_cap_source_fraction': self.size_cap, 'method': selection.get('method', 'external_selector'), 'verification': selection.get('verification', 'retrieval_only'), 'faithfulness_established': False, 'calls': 0, 'input_tokens': 0, 'branches': 0, 'selection_calls': cost['calls'], 'source_text_sha256': digest(text), 'answer_qualification_unchanged': True, **({'page_map_registered': True} if page_map is not None else {'page_map_registered': False})} body['calls'] += cost['calls'];body['input_tokens'] += cost['input_tokens'];body['branches'] += cost['branches'] for field in ('relevance_weights_sha256', 'dependency_rule'): if field in selection: body[field] = selection[field] baseline = prediction(result.get('answer') or result.get('candidate')) def call(document, req): output = self.decide(copy.deepcopy(document), task, copy.deepcopy(req)) if not isinstance(output, dict) or not {'answer', 'cost'} <= set(output): raise ValueError('invalid evidence decision callback') c = output['cost'] if any(type(c.get(k)) is not int or c[k] < 0 for k in ('input_tokens', 'branches')): raise ValueError('invalid evidence static cost') body['calls'] += 1;body['input_tokens'] += c['input_tokens'];body['branches'] += c['branches'] return prediction(output['answer']) if evidence in ('sufficiency', 'removal') and self.package_format == 'union': un = v3.union(text, [{'spans': p['spans']} for p in packages]) decision = {k: request[k] for k in ('question', 'options', 'labels', 'template', 'entities') if k in request} assembled = call([{'text': un['text']}], decision) body['union_package'] = {'spans': [{k: v for k, v in s.items() if k != 'pages'} for s in un['spans']], 'source_fraction': un['source_fraction'], 'rendered_characters': un['rendered_characters']} body['evidence_only'] = {'prediction': assembled, 'agrees_with_full': assembled == baseline, 'input': 'union of the question\'s unit packages, one fresh prefill'} elif evidence in ('sufficiency', 'removal'): predictions = [call([{'text': p['_text']}], req) for p, (_, _, req) in zip(packages, units)] if task in ('entity', 'multilabel'): assembled = [x for pr in predictions for x in (pr if isinstance(pr, list) else [pr])] else: assembled = predictions[0] body['evidence_only'] = {'prediction': assembled, 'agrees_with_full': assembled == baseline, 'per_unit': [{'unit': i, 'prediction': pr} for i, pr in enumerate(predictions)], 'input': 'per-unit evidence package (selected spans + governing context), fresh prefill per unit'} if evidence == 'removal': decision = {k: request[k] for k in ('question', 'options', 'labels', 'template', 'entities') if k in request} removed = call([{'text': remove(text, refs) if refs else text}], decision) body['evidence_removed'] = {'prediction': removed, 'agrees_with_full': removed == baseline, 'input': 'source with every selected evidence span removed (context kept)'} if body['calls'] > body['selection_calls']: body['verification'] = 'fresh_source_reencoding' result['evidence'] = body return result def handler(service): from solomon.service_evidence import handler as base_handler base = base_handler(service) class Handler(base): def do_GET(self): if self.path != '/health': return self.reply(404, {'error': 'not found'}) return self.reply(200, {'contract': CONTRACT, 'answer_types': ['boolean', 'single', 'ordered', 'multilabel', 'entity'], 'evidence_levels': ['none', 'support', 'sufficiency', 'removal'], 'evidence_package': v3.SCHEMA, 'package_format': self.package_format, 'strict_reliability_default': True, 'runtime_identity': self.server.service.runtime_identity if hasattr(self.server, 'service') else service.runtime_identity}) return Handler def service(store, engine, design, *, artifact=None, views_artifact=None, views_tasks=(), selector=None, page_selector=None, page_maps=None, size_cap=SIZE_CAP, package_format='per_unit'): """Serving constructor mirroring solomon.serving.service, with v3 evidence packages.""" from pathlib import Path from solomon.serving import ConfidenceHeadsService, FreshDecider answers = ConfidenceHeadsService(store, engine, design, artifact=artifact, views_artifact=views_artifact, views_tasks=views_tasks) return PackageEvidenceService(answers, FreshDecider(engine, design, Path(store)/'.evidence-readout'), selector, page_selector, page_maps=page_maps, size_cap=size_cap, package_format=package_format)