refactor: Run the graph on the caller's thread

docs/concepts/graph-sdk.md

@@ -99,13 +99,6 @@ ceiling does not claim bounded memory usage or a hard transport-response size.
 Context retrieval, observations and other explicitly bounded contracts retain
 their separate limits.
 
-The qualified development detection/Redact profile derives count capacity from the selected
-source and declared work when no execution limits are supplied. Five retained
-artifacts per source is an upper bound: one root, three producer outputs and one
-assessment output. Final span count is bounded by source character count;
-validation membership is bounded by the admitted chunk capacity. Actual counts
-can be lower. Caller-supplied execution limits remain authoritative. The separate qualified development profile retains this five-artifact inventory.
-
 This counts-first policy records the 2026-10-13 operator clarification that
 ordinary runs should not impose incidental byte or per-string ceilings. It
 qualifies the RFC's earlier mandatory-retained-byte-bound wording; it does not
@@ -120,9 +113,7 @@ whole frame through `NddAdapter.run_workflow()`, with the existing DataDesigner
 retry, chunk scheduling and provider configuration. Authenticated source positions,
 typed spans and canonical text artifacts support the existing `PublishedFrame`;
 there is no inference from text, indexes or row order. This path makes no
-Protected claim and does not use the incomplete qualified request bridge.
-The qualified development profile remains available for its explicit supported
-controls; public execution does not fall back to it after an error.
+Protected claim.
 
 The ordinary profile reserves at most three semantic artifacts per selected
 source: initial text, final spans and transformed text. Span membership is bounded
@@ -225,11 +216,12 @@ thread that called them, at every depth, including built-in code reached through
 bodies, and every DataDesigner `create()`/`preview()` call and wait, run there too.
 Your code therefore sees that thread's thread-locals and can install signal handlers
 exactly when that thread is the main thread, and Ctrl-C reaches the run as in 0.4.0.
-Some Anonymizer internals still run on other threads: the per-call
-`anonymizer-graph-host` event-loop thread does bookkeeping such as stage completion,
-entity grouping and schema parsing, and DataDesigner's engine threads run
-Anonymizer's column generators and send model requests, as in 0.4.0. None of
-them runs your code. Each call runs in a copy of the caller's
+The graph's event loop runs on that thread as well, as a private loop that leaves a
+notebook's or `asyncio.run` caller's running loop untouched, and so does the graph
+bookkeeping those bodies perform. DataDesigner's engine threads run Anonymizer's
+column generators and send model requests, as in 0.4.0; the graph bookkeeping that
+DataDesigner's per-row hooks trigger runs there too, under the invocation's lock.
+None of those threads runs your code. Each call runs in a copy of the caller's
 context variables: values set before the call are visible inside, and values set
 inside are not visible to the caller afterwards.
 
@@ -362,29 +354,12 @@ and repeated observer cancellation. Physical transports, classification, Start,
 terminal facts and publication still execute on the invocation loop. A worker
 submission is not a physical request. Cancellation before worker entry submits
 nothing; cancellation after physical entry preserves the first terminal and
-usable late settlement while draining local work. The DataDesigner physical
-transport cancels its in-flight HTTP exchange when asked to stop; the provider
-gives no stop evidence, so such a request is recorded as lost.
-
-The Substitute map's DataDesigner bridge declares retry follow-ups. DataDesigner
-0.9.4 keeps its own retry policy: its request executor retries 502/503/504 and
-connection errors with backoff, and its scheduler re-runs a task that failed
-with a 5xx, 429, timeout or connection error in later salvage rounds. Each of
-those attempts is a separate leaf with its own dispatch and settlement, linked to
-the failed predecessor as a `retry` child. A refused connection settles as
-`rejected_before_acceptance`; any other transport error leaves the provider side
-unknown. Each record admits its parser attempts plus
-`(RetryConfig.max_retries + 1) * (salvage_max_rounds + 1) - 1` retries as
-accounted leaves, which covers DataDesigner's bounded 5xx, timeout and
-connection policy: exhausting it drops that record, as in 0.4.0. DataDesigner
-never drops a rate-limited (429) task and keeps retrying it; attempts beyond the
-admitted leaves run as DataDesigner-owned requests outside graph request
-accounting, and the bridge logs their count. This is honest only because the
-Substitute invocation already declares no request upper bound and admits no hard
-request budget; a run that selects one could not use this path. DataDesigner runs that the graph does not
-bridge (for example Rewrite's stages) are cancelled through DataDesigner's own
-scheduler cancellation when the owning invocation is cancelled, as a Ctrl-C does
-in 0.4.0. The caller owns
+usable late settlement while draining local work.
+
+The Substitute map, like every other facade stage, runs on DataDesigner's own
+transport, retry policy and scheduling, as in 0.4.0; the graph accounts it as one
+opaque invocation with unknown physical requests. A Ctrl-C cancels DataDesigner's
+runs through DataDesigner's own scheduler cancellation, as in 0.4.0. The caller owns
 the pool and borrowed loop; the executor neither closes nor replaces them. The
 pool must be open at service admission. These are live execution resources, not stateless
 handles: this invocation drains its retained worker and children without waiting

docs/development/graph-sdk-cutover.md

@@ -27,7 +27,7 @@ M2 and M3 can proceed independently after their M1 prerequisites. The campaign c
 
 ## Rewrite semantic controller checkpoint
 
-Built-in Rewrite now uses one admitted controller for iterative batches and combined DataDesigner columns. Default facade calls, direct calls and supplied instances that inherit the library entry bind the actual runner, adapter, arguments and source before a single lowering entry. Injected detection callbacks share that Rewrite owner while preserving their caller thread and original argument objects. Every synchronous body runs on the thread that called the public entry point, with a copy of its context variables; values set inside are not visible to the caller afterwards. This covers the Rewrite body, built-in detection, Redact, Annotate, Hash and Substitute, built-in code reached from a caller override through `super()`, `build_graph` overrides, the built-in detection lowered inside Rewrite (including its stage overrides and its detection adapter's `run_workflow`), and the DataDesigner calls they make. No caller code and no DataDesigner wait runs off the caller thread. The per-invocation `anonymizer-graph-host` loop thread still runs Anonymizer internals (stage completion, entity grouping, schema parsing), and DataDesigner's engine threads run Anonymizer column generators, as on main. Arbitrary runner overrides remain opaque. The previous iterative repair loop, duplicate combined `run()` controller and last-populated-row finalizer have been removed. Physical batching and conditional column unrolling remain lowerings.
+Built-in Rewrite now uses one admitted controller for iterative batches and combined DataDesigner columns. Default facade calls, direct calls and supplied instances that inherit the library entry bind the actual runner, adapter, arguments and source before a single lowering entry. Injected detection callbacks share that Rewrite owner while preserving their caller thread and original argument objects. Every synchronous body runs on the thread that called the public entry point, with a copy of its context variables; values set inside are not visible to the caller afterwards. This covers the Rewrite body, built-in detection, Redact, Annotate, Hash and Substitute, built-in code reached from a caller override through `super()`, `build_graph` overrides, the built-in detection lowered inside Rewrite (including its stage overrides and its detection adapter's `run_workflow`), and the DataDesigner calls they make. No caller code and no DataDesigner wait runs off the caller thread. The graph's private event loop and the bookkeeping the bodies perform run on the caller thread too. DataDesigner's engine threads run Anonymizer column generators, as on main, and the graph bookkeeping that DataDesigner's per-row hooks trigger runs on those threads under the invocation's lock. Arbitrary runner overrides remain opaque. The previous iterative repair loop, duplicate combined `run()` controller and last-populated-row finalizer have been removed. Physical batching and conditional column unrolling remain lowerings.
 
 Native generators publish their observed typed stages. Caller callbacks with opaque internals publish source-bound observations without invented QA assessments, repair history or physical request counts. Retained candidate values and presentation dtypes cannot change the accepted observation. Source snapshots detach structured semantic payloads, including supported model and NumPy cells, while retaining opaque caller metadata references. Rows of an admitted Rewrite are finalized by the lowering's binding instead of the combined graph's own finalization column. That column finalizes rows in two other cases: combined columns run outside an admitted Rewrite, and the built-in combined Rewrite under an adapter whose `run_workflow` is overridden. It projects the final state directly from the observed evaluation states without running the admitted transitions: the final state is the first evaluation state whose repair decision is false, or the last state when every decision calls for repair. A missing evaluation-state column, or a missing, null or non-boolean repair decision, in an active state fails explicitly rather than falling back to the first state. The projection reads no original text. Arbitrary injected runners retain their explicitly opaque compatibility contract.
 

src/anonymizer/engine/detection/_graph_invocation.py

@@ -3,15 +3,12 @@
 """Whole-frame detection/Redact consumer of an admitted graph invocation.
 
 Source frames always bind authenticated positions and typed span/text outputs.
-An optional physical bridge selects the separately qualified development path;
-ordinary public execution leaves DataDesigner's retry and scheduling intact.
+DataDesigner keeps its own transport, retries and scheduling.
 """
 
 from __future__ import annotations
 
-import asyncio
-from collections.abc import Callable, Mapping
-from contextvars import copy_context
+from collections.abc import Mapping
 from dataclasses import dataclass, replace
 from typing import Any, cast
 
@@ -21,28 +18,23 @@
 from anonymizer.config.models import DetectionModelSelection
 from anonymizer.engine.constants import COL_FINAL_ENTITIES, COL_REPLACED_TEXT
 from anonymizer.engine.detection.artifacts import ENTITY_SPANS_TYPE, EntitySpansValue
-from anonymizer.engine.detection.chunked_validation import validation_chunks
 from anonymizer.engine.detection.detection_workflow import (
-    _DEFAULT_VALIDATION_MAX_ENTITIES_PER_CALL,
     EntityDetectionWorkflow,
 )
 from anonymizer.engine.detection.postprocess import EntitySpan
 from anonymizer.engine.detection.run_arguments import detection_run_arguments
-from anonymizer.engine.graph_sdk._execution_values import DelegatedRequests, LocalCompleted, LocalFailure
+from anonymizer.engine.graph_sdk._execution_delegated import drive
+from anonymizer.engine.graph_sdk._execution_values import DelegatedRequests, LocalCompleted
 from anonymizer.engine.graph_sdk._graph_values import ArtifactRef, DatumId
-from anonymizer.engine.graph_sdk._workflow_values import ArtifactType
 from anonymizer.engine.graph_sdk.capabilities import FrozenConfig
 from anonymizer.engine.graph_sdk.requests import (
     AssociationInput,
     AssociationResult,
     PortArtifact,
     SemanticAssociation,
     TextArtifactValue,
-    TextCollectionItem,
-    TextCollectionValue,
 )
 from anonymizer.engine.harness.frames import FramePublication, InvocationFrame
-from anonymizer.engine.ndd._graph_requests import NddPhysicalTransport, NddRequestBridge
 from anonymizer.engine.replace._graph_redact import local_strategy
 from anonymizer.engine.replace.strategies import apply_local_replace_strategy
 from anonymizer.engine.schemas import EntitiesSchema
@@ -59,32 +51,14 @@ class DetectionRedactInvocation:
     model_configs: list[ModelConfig]
     selected_models: DetectionModelSelection
     redact_configuration: FrozenConfig
-    physical: NddPhysicalTransport | None
-    slots: Mapping[tuple[str, int], int | tuple[int, ...]]
     detection_options: Mapping[str, Any]
     output_port: str
     outcome: str
-    execution: str = "same_loop"
     semantic_slots: Mapping[str, int] | None = None
-    semantic_loop: asyncio.AbstractEventLoop | None = None
     semantic_transform_slot: int | None = None
     semantic_entry_slot: int | None = None
-    handoff: Callable[[Callable[[], LocalCompleted | LocalFailure]], LocalCompleted | LocalFailure] | None = None
 
-    async def submit(
-        self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests
-    ) -> LocalCompleted | LocalFailure:
-        return await asyncio.to_thread(self.run, inputs, requests)
-
-    def run(self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests) -> LocalCompleted | LocalFailure:
-        context = copy_context()
-        return (
-            self._run(inputs, requests)
-            if self.handoff is None
-            else self.handoff(lambda: context.run(self._run, inputs, requests))
-        )
-
-    def _run(self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests) -> LocalCompleted:
+    def run(self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests) -> LocalCompleted:
         if any(
             type(item.association) is not SemanticAssociation
             or len(item.inputs) != 1
@@ -93,12 +67,10 @@ def _run(self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests
         ):
             raise AnonymizerWorkflowError("DD graph detection requires one source-text input per association")
         frame = requests.bind_frame(self.source, targets=self.targets, source_limit=len(self.targets))
-        bridge = None if self.physical is None else NddRequestBridge(requests, self.slots, self.physical)
         arguments: dict[str, Any] = {
             "model_configs": self.model_configs,
             "selected_models": self.selected_models,
             "source_frame": frame,
-            "request_bridge": bridge,
             **self.detection_options,
         }
         from anonymizer.engine.detection._semantic_detection import (
@@ -109,20 +81,14 @@ def _run(self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests
 
         # The submitting profile classified the runner: only a builtin one has detection children.
         if self.semantic_slots is not None:
-            if self.semantic_loop is None:
-                raise AnonymizerWorkflowError("Detection binding has no owning loop")
-            if self.physical is None:
-                asyncio.run_coroutine_threadsafe(
-                    complete_detection_entry(requests, inputs), self.semantic_loop
-                ).result()
+            drive(complete_detection_entry(requests, inputs))
             detected = lower_detection(
                 self.detection,
                 self.source,
                 arguments,
                 requests,
                 frame,
                 dict(self.semantic_slots),
-                self.semantic_loop,
                 self.semantic_entry_slot,
             )
         else:
@@ -164,37 +130,6 @@ def _run(self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests
                 outcome=self.outcome,
                 consumed_context_ports=frozenset(),
                 outputs=(
-                    *(
-                        (
-                            PortArtifact(
-                                port="validation_chunks",
-                                artifact_type=ArtifactType(name="anonymizer.validation-chunks", revision=1),
-                                artifact=None,
-                                value=TextCollectionValue(
-                                    items=tuple(
-                                        TextCollectionItem(
-                                            key=index,
-                                            version=1,
-                                            value=TextArtifactValue(
-                                                text="\n".join(candidate.id for candidate, _ in chunk)
-                                            ),
-                                        )
-                                        for index, chunk in enumerate(
-                                            validation_chunks(
-                                                detected.dataframe.iloc[rows[item.association]].to_dict(),
-                                                self.detection_options.get(
-                                                    "validation_max_entities_per_call",
-                                                    _DEFAULT_VALIDATION_MAX_ENTITIES_PER_CALL,
-                                                ),
-                                            )[2]
-                                        )
-                                    )
-                                ),
-                            ),
-                        )
-                        if self.physical is not None
-                        else ()
-                    ),
                     PortArtifact(
                         port="spans", artifact_type=ENTITY_SPANS_TYPE, artifact=None, value=spans[item.association]
                     ),
@@ -210,8 +145,6 @@ def _run(self, inputs: tuple[AssociationInput, ...], requests: DelegatedRequests
             if item.association in rows
         )
         if self.semantic_transform_slot is not None:
-            if self.semantic_loop is None:
-                raise AnonymizerWorkflowError("Detection binding has no owning loop")
             transform_slot = self.semantic_transform_slot
             completed_sources = {}
 
@@ -248,7 +181,7 @@ async def finish() -> None:
                     await complete_empty_stage(requests, [item])
                 results = accepted
 
-            asyncio.run_coroutine_threadsafe(finish(), self.semantic_loop).result()
+            drive(finish())
             successful = tuple(completed_sources[association] for association in successful)
         results_by_source = {item.association: item for item in results}
         results = tuple(results_by_source[association] for association in successful)