feat(extract): implement extract_batch — 49/49 spec green (recall#868)

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

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")