diff --git a/packages/python/src/synapt/extract/batch.py b/packages/python/src/synapt/extract/batch.py index 5197565..46ad515 100644 --- a/packages/python/src/synapt/extract/batch.py +++ b/packages/python/src/synapt/extract/batch.py @@ -1,10 +1,10 @@ """Batch Stage-1 extraction primitive for SynaptExtraction. -SKELETON (recall#868 → extract_batch). API conformed to the pinned contract -(config/design/extract-batch-limits-characterization-2026-07-13.md §"Contract -decisions") AND to Sentinel's spec (extract#28, tests/python/test_extract_batch.py). -Every body raises NotImplementedError — the implementation lands in the follow-up -impl PR (TDD: this skeleton makes the spec COLLECT and run RED, not ImportError). +Implements the pinned contract (config/design/extract-batch-limits-characterization- +2026-07-13.md §"Contract decisions") and Sentinel's spec (extract#28, +tests/python/test_extract_batch.py). Reliability logic is per-unit: shaping + +per-item validation + fail-closed fallback, with every failure contained to its +own unit slot (count-invariant). Why this primitive exists ------------------------- @@ -34,18 +34,34 @@ Class-A PRE-parse text hygiene — strip ``` fences + `//` comments, STRING- LITERAL-AWARE (a `//` inside a JSON string, e.g. a URL, must survive). Class-B POST-parse coercion — capability set is the arbiter: in-scope fields - coerced (scalar→array, decided_at null→omit, category→valid/default), - out-of-scope dropped; temporal_refs → schema-valid raw/resolved only. + coerced (scalar→array; null/non-string OPTIONAL fields like category or + decided_at are omitted; an invalid REQUIRED field is kept so strict + validation rejects it), out-of-scope dropped; temporal_refs → schema-valid + raw/resolved only; non-dict leaves preserved into strict validation. Harvest map: scratchpad/extract_batch_craft_harvest.md. Boundary: OSS. """ from __future__ import annotations +import json from dataclasses import dataclass from typing import Any, Callable, Literal, TypedDict -from synapt.extract.finalize import finalize_extraction +from synapt.extract.builder import build_extraction_schema +from synapt.extract.finalize import FinalizeContext, finalize_extraction +from synapt.extract.prompt import ( + build_extraction_prompt, + profile_capabilities, + resolve_capabilities, +) + +# Container capabilities the finalized schema always requires, even when a caller +# did not request them (mirrors recall's backfill so validation does not fail on +# containers we deliberately did not request). +_ALWAYS_BACKFILL = ("entities", "goals", "themes") +# One deterministic retry per failed unit → 2 attempts total (Q-B, Sentinel). +_MAX_ATTEMPTS = 2 # Terminal per-unit failure reasons (Q5). A Literal (not an Enum) so the spec's # get_args(BatchFailureReason) reads the members. "merged" is reserved for a future @@ -104,27 +120,236 @@ async def extract_batch( per-unit calls with one deterministic retry per failed unit) driven through the injected ``infer`` seam, with zero dependency on any specific model client (Q4). ``capabilities`` defaults to the standard profile when omitted (Q3). - - SKELETON — body is NotImplementedError; the impl lands in the follow-up PR. """ - raise NotImplementedError( - "extract_batch skeleton conforms to the pinned contract + spec; the " - "implementation lands in the impl PR (recall#868)." + if not units: + return [] + ids = [unit.id for unit in units] + if len(set(ids)) != len(ids): + raise ValueError("duplicate unit id; BatchUnit ids must be unique") + + default_capabilities = ( + capabilities if capabilities is not None else profile_capabilities("standard") ) + results: list[BatchUnitResult] = [] + for unit in units: + unit_capabilities = ( + unit.capabilities if unit.capabilities is not None else default_capabilities + ) + results.append(_extract_unit(unit, infer, produced_by, unit_capabilities)) + return results + + +def _extract_unit( + unit: BatchUnit, + infer: Inferer, + produced_by: str, + capabilities: list[str], +) -> BatchUnitResult: + """Run one unit through the reliability ladder: build an out-of-band request → + infer → Class-A hygiene + parse → Class-B coerce → finalize/validate. One + deterministic retry on failure (2 attempts total); a persisting failure yields a + terminal marker carrying the last failure's reason (Q-B).""" + reason: BatchFailureReason = "dropped" + for _attempt in range(_MAX_ATTEMPTS): + # Out-of-band: the model sees the unit TEXT only — never its id or a boundary + # tag (Q-D). The id lives in bookkeeping and rides into the packet post-hoc. + prompt = build_extraction_prompt(unit.text, capabilities=list(capabilities), stage="stage1") + request: BatchInferRequest = { + "prompt": prompt, + "messages": [{"role": "user", "content": prompt}], + "capabilities": list(capabilities), + } + # Contain the injected seam per-unit: an infer failure (e.g. RuntimeError) + # must NOT escape and void the whole batch — it is this unit's failure, + # retried once then terminal, while neighbours still produce their slots. + # No output was produced, so the closest Q5 class is "dropped". + try: + completion = infer(request) + except Exception: + reason = "dropped" + continue + parsed = _parse_completion(completion) + if parsed is None: + reason = "unparseable" + continue + + coerced = _coerce_shape(parsed, capabilities) + for key in _ALWAYS_BACKFILL: + coerced.setdefault(key, []) + context = FinalizeContext( + produced_by=produced_by, + source_id=unit.id, + capabilities_hint=list(capabilities), + ) + try: + finalized = finalize_extraction(coerced, context) + except Exception: + reason = "schema_invalid" + continue + if not finalized.validation.valid: + reason = "schema_invalid" + continue + if _is_empty_extraction(finalized.extraction, capabilities): + reason = "dropped" + continue + return BatchUnitResult( + source_unit_id=unit.id, status="ok", extraction=finalized.extraction + ) + return BatchUnitResult(source_unit_id=unit.id, status="failed", reason=reason) + + +def _parse_completion(completion: str) -> dict | None: + """Class-A hygiene + JSON parse; None if the result is not a JSON object.""" + try: + parsed = json.loads(_strip_output_hygiene(completion)) + except (ValueError, TypeError): + return None + return parsed if isinstance(parsed, dict) else None -# --- Intended internal decomposition (stubs; bodies in the impl PR) ------------ def _strip_output_hygiene(raw: str) -> str: """Class-A PRE-parse (NET-NEW): strip ``` fences + ``//`` comments so grounded- but-wrapped JSON parses. STRING-LITERAL-AWARE — a ``//`` inside a JSON string value (e.g. ``https://…``) is preserved; only real line-comments are removed.""" - raise NotImplementedError + text = raw.strip() + # strip_markdown_fence: drop a leading ```/```json fence line, then the closing + # ``` and anything trailing it (e.g. a "Reasoning:" epilogue the model appends). + if text.startswith("```"): + newline = text.find("\n") + text = text[newline + 1:] if newline != -1 else "" + close = text.rfind("```") + if close != -1: + text = text[:close] + # strip_line_comments_outside_strings: remove `//` to end-of-line, but never when + # inside a JSON string literal (so a URL's `//` survives). Tracks string + escape. + out: list[str] = [] + in_string = False + escaped = False + i, n = 0, len(text) + while i < n: + ch = text[i] + if in_string: + out.append(ch) + if escaped: + escaped = False + elif ch == "\\": + escaped = True + elif ch == '"': + in_string = False + i += 1 + elif ch == '"': + in_string = True + out.append(ch) + i += 1 + elif ch == "/" and i + 1 < n and text[i + 1] == "/": + while i < n and text[i] != "\n": + i += 1 # drop the comment body; the newline (if any) is kept next loop + else: + out.append(ch) + i += 1 + return "".join(out).strip() def _coerce_shape(parsed: dict, capabilities: list[str]) -> dict: """Class-B POST-parse (harvest ``_sanitize_stage1_output`` whitelist backbone): - the capability set is the arbiter (Q2) — in-scope fields coerced (scalar→array, - ``decided_at`` null→omit, ``category``→valid/default), out-of-scope dropped; - ``temporal_refs`` coerced to schema-valid ``raw``/``resolved`` only.""" - raise NotImplementedError + the capability set is the arbiter (Q2). Per the Stage-1 schema for the requested + capabilities, whitelist each item type to its fields, coerce (scalar→array, + null/non-string optional → omit), and drop out-of-scope item types. ``entity_refs`` + is retained only when the ``entities`` capability is in scope; ``temporal_refs`` + keeps ``raw``/``resolved`` and drops schema-illegal extras (type/context/…).""" + resolved = set(resolve_capabilities(capabilities=list(capabilities))) + schema = build_extraction_schema(capabilities=list(capabilities)) + props = schema.get("properties", {}) + entities_in_scope = "entities" in resolved + + result: dict[str, Any] = {} + if "extracted_at" in parsed: + result["extracted_at"] = parsed["extracted_at"] + + for type_name, type_schema in props.items(): + if type_name == "extracted_at": + continue + if type_schema.get("type") != "array": + if type_name in parsed: + result[type_name] = parsed[type_name] + continue + items_schema = type_schema.get("items") + parsed_items = parsed.get(type_name) + if not isinstance(items_schema, dict) or "properties" not in items_schema: + result[type_name] = parsed_items if isinstance(parsed_items, list) else [] + continue + item_props = items_schema["properties"] + required = set(items_schema.get("required", [])) + coerced_items: list[Any] = [] + if isinstance(parsed_items, list): + for item in parsed_items: + if isinstance(item, dict): + coerced_items.append( + _coerce_item(item, item_props, required, entities_in_scope) + ) + else: + # Preserve non-dict leaves (null, 42, "str") verbatim so strict + # validation REJECTS them (→ schema_invalid) instead of silently + # dropping — a null sibling must fail its whole unit, not vanish. + coerced_items.append(item) + result[type_name] = coerced_items + return result + + +def _coerce_item( + item: dict, + item_props: dict, + required: set[str], + entities_in_scope: bool, +) -> dict: + """Whitelist + type-coerce one item to its schema fields. Null/non-string optional + fields are omitted (they were grounded but wrongly shaped); a scalar for an + array-typed field is wrapped; an invalid REQUIRED field is kept so finalize + rejects it (→ schema_invalid) rather than silently passing.""" + new_item: dict[str, Any] = {} + for field, field_schema in item_props.items(): + if field == "entity_refs" and not entities_in_scope: + continue # out-of-scope reference field → drop (Q2) + if field not in item: + continue + value = item[field] + field_type = field_schema.get("type") + if field_type == "array" and not isinstance(value, list): + if value is None: + continue # omit null optional array + value = [value] # scalar → array (in-scope coerce) + elif value is None: + if field in required: + new_item[field] = value # keep null required → finalize rejects + continue + elif field_type == "string" and not isinstance(value, str): + if field in required: + new_item[field] = value # keep invalid required → finalize rejects + continue + new_item[field] = value + return new_item + + +def _is_empty_extraction(extraction: Any, capabilities: list[str]) -> bool: + """True when the model produced NO payload for the unit across the REQUESTED + capabilities — every requested payload is empty. Type-aware over the Stage-1 + schema: an array payload (facts/decisions/entities/goals/…) counts when + non-empty; a scalar payload (summary/sentiment) counts when present and + non-empty. So an entities-only or summary-only extraction is NOT a false-drop. + Empty-but-valid is the 10/45 "dropped" mode: caught here and retried, never + silently absorbed.""" + schema = build_extraction_schema(capabilities=list(capabilities)) + for name, prop_schema in schema.get("properties", {}).items(): + if name == "extracted_at": + continue + value = ( + extraction.get(name) if isinstance(extraction, dict) + else getattr(extraction, name, None) + ) + if prop_schema.get("type") == "array": + if isinstance(value, list) and value: + return False + elif value not in (None, "", [], {}): + return False + return True diff --git a/tests/python/test_extract_batch.py b/tests/python/test_extract_batch.py index f8f0705..a31d09a 100644 --- a/tests/python/test_extract_batch.py +++ b/tests/python/test_extract_batch.py @@ -385,3 +385,74 @@ def test_empty_input_is_a_noop(): ) assert outputs == [] + + +# --- Fidelity-gate regression locks (Sentinel, extract#30 re-gate) ------------- + +def test_infer_exception_in_one_unit_never_voids_the_batch(): + """HIGH-1: an infer() exception must be contained to its own unit — neighbours + still produce their slots and the count-invariant holds (no output produced for + the failing unit → terminal "dropped" after retry).""" + units = [ + BatchUnit(id="good-before", text="The first durable fact is grounded."), + BatchUnit(id="raises", text="This unit's inference raises."), + BatchUnit(id="good-after", text="The final durable fact is grounded."), + ] + responses = { + units[0].text: json.dumps(_stage1(facts=[{"text": units[0].text}])), + units[2].text: json.dumps(_stage1(facts=[{"text": units[2].text}])), + } + + def infer(request): + if units[1].text in _request_prompt(request): + raise RuntimeError("inference exploded") + return _response_for_prompt(request, responses) + + outputs = _run_batch(units, infer) + + assert len(outputs) == len(units) + assert [output.source_unit_id for output in outputs] == [unit.id for unit in units] + _assert_success(outputs[0], "good-before") + _assert_failure(outputs[1], "raises", "dropped") + _assert_success(outputs[2], "good-after") + + +@pytest.mark.parametrize( + ("label", "malformed"), + [ + ("facts_null", _stage1(facts=[None])), + ("facts_null_beside_valid", _stage1(facts=[None, {"text": "A grounded durable fact."}])), + ("decisions_non_dict", _stage1(decisions=[42])), + ("temporal_non_dict", _stage1(temporal_refs=["tomorrow"])), + ], +) +def test_non_dict_leaves_reach_strict_validation(label, malformed): + """HIGH-2: a non-object leaf must not be silently deleted — it reaches strict + validation and fails its unit (schema_invalid), even beside a valid sibling.""" + unit = BatchUnit(id=f"nondict-{label}", text="This source unit stays attributable.") + + outputs = _run_batch([unit], lambda _request: json.dumps(malformed)) + + assert len(outputs) == 1 + _assert_failure(outputs[0], unit.id, "schema_invalid") + + +def test_valid_metadata_only_extraction_is_not_dropped(): + """HIGH-3: extract_batch is a GENERAL primitive (Q3) — an entities-only (array) + or summary-only (scalar) extraction is real content for its requested capability + set, not the empty "dropped" mode.""" + entities_unit = BatchUnit(id="entities-only", text="Synapt is an organization.") + entities_out = _run_batch( + [entities_unit], + lambda _request: json.dumps(_stage1(entities=[{"name": "Synapt", "type": "org"}])), + capabilities=["entities"], + ) + _assert_success(entities_out[0], "entities-only") + + summary_unit = BatchUnit(id="summary-only", text="A paragraph worth summarizing.") + summary_out = _run_batch( + [summary_unit], + lambda _request: json.dumps(_stage1(summary="A concise summary of the source.")), + capabilities=["summary"], + ) + _assert_success(summary_out[0], "summary-only")