Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions workers/api-deployment/tasks.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,11 @@
from shared.infrastructure.logging.workflow_logger import WorkerWorkflowLogger
from shared.patterns.retry.utils import retry
from shared.processing.files import FileProcessingUtils
from shared.utils.manual_review_factory import (
AUTO_ENABLE_HIGHLIGHT,
get_manual_review_workflow_util,
has_confidence_rules,
)
from shared.workflow.execution import WorkerExecutionContext, WorkflowOrchestrationUtils
from shared.workflow.execution.tool_validation import validate_workflow_tool_instances
from worker import app
Expand Down Expand Up @@ -665,6 +670,24 @@ def _run_workflow_api(
total_files=total_files,
)

# The destination evaluates API rules only for a hitl_queue_name request;
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
# a hitl_packet_id skips rule evaluation, so highlight would go unused.
# ponytail: the API + use_file_history path evaluates DB rules, which
# this check does not cover, so a DB confidence rule there gets no
# highlight. use_file_history is hidden from the API schema and meant for
# demos, but a caller can still send it; add a "DB" check if it matters.
# Highlight adds confidence_data to the result metadata and fills
# line_numbers and whisper_hash, which include_metadata callers receive
# (highlight_data too, when the org's API highlight flag is on). The keys
# are kept on purpose, so callers can see why a document was routed.
auto_enable_highlight = (
bool(kwargs.get("hitl_queue_name"))
and not kwargs.get("hitl_packet_id")
and has_confidence_rules(
get_manual_review_workflow_util(api_client, schema_name), workflow_id, "API"
)
)

# Create batch tasks following the exact Django pattern
batch_tasks = []
execution_mode_str = (
Expand All @@ -687,6 +710,8 @@ def _run_workflow_api(
total_files=total_files,
**kwargs,
)
if auto_enable_highlight:
file_data.manual_review_config[AUTO_ENABLE_HIGHLIGHT] = True

# Calculate manual review decisions for this specific batch
if file_data.manual_review_config.get("review_required", False):
Expand Down
18 changes: 18 additions & 0 deletions workers/file_processing/structure_tool_task.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,10 @@
)
from shared.enums.task_enums import TaskName
from shared.infrastructure.context import StateStore
from shared.utils.manual_review_factory import (
AUTO_ENABLE_HIGHLIGHT,
HIGHLIGHT_AUTO_ENABLED_MESSAGE,
)

from unstract.sdk1.constants import ToolEnv, UsageKwargs
from unstract.sdk1.execution.context import ExecutionContext
Expand Down Expand Up @@ -365,6 +369,20 @@ def _execute_structure_tool_impl(params: dict) -> dict:
execution_data_dir=execution_data_dir,
)

# HITL confidence rules need highlight data. Override a copy so the
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
# user's saved setting is unchanged. Agentic projects are skipped above:
# workflow runs give them no highlight data, so confidence can't work.
# Known limit: with an extractor other than LLMWhisperer V2, no highlight
# plugin, or summarize-as-source, highlight still gives no confidence and
# rules still score 100; this line is logged anyway.
# Highlight also changes extraction: the prompt gets the line-number
# postamble and LLMWhisperer V2 adds line numbers, so answers may differ
# from a run with highlight off. This is accepted: confidence needs the
# line numbers the postamble asks for.
if params.get(AUTO_ENABLE_HIGHLIGHT):
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
tool_instance_metadata = {**tool_instance_metadata, _SK.ENABLE_HIGHLIGHT: True}
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
shim.stream_log(HIGHLIGHT_AUTO_ENABLED_MESSAGE)

# ---- Step 3: Profile overrides ----
_handle_profile_overrides(exec_metadata, platform_helper, tool_metadata)

Expand Down
11 changes: 10 additions & 1 deletion workers/general/tasks.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,11 @@
from shared.patterns.retry.utils import circuit_breaker
from shared.processing.files import FileProcessingUtils
from shared.processing.types import FileDataValidator, TypeConverter
from shared.utils.manual_review_factory import get_manual_review_service
from shared.utils.manual_review_factory import (
AUTO_ENABLE_HIGHLIGHT,
get_manual_review_service,
has_confidence_rules,
)
from shared.workflow.connectors.source import WorkerSourceConnector
from shared.workflow.execution import (
WorkerExecutionContext,
Expand Down Expand Up @@ -978,6 +982,11 @@ def _orchestrate_file_processing_general(
total_files=len(source_files),
)
logger.debug("Global file data configured for manual review")
# Run-wide: rules are evaluated after extraction, so every file needs
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
# confidence data, including files not sent to review. Where highlight
# can't produce it, see the override in structure_tool_task.py.
if has_confidence_rules(workflow_util, workflow_id, "DB"):
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
global_file_data.manual_review_config[AUTO_ENABLE_HIGHLIGHT] = True
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
# Pre-calculate file decisions for ALL files based on total count - not per batch!
q_file_no_list = global_file_data.manual_review_config.get("q_file_no_list", [])
logger.info(
Expand Down
17 changes: 17 additions & 0 deletions workers/shared/utils/manual_review_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,13 @@

MANUAL_REVIEW_PLUGIN_NAME = "manual_review"

# manual_review_config key: the workflow has HITL confidence rules, so the
# structure tool runs with highlight on to produce confidence scores.
AUTO_ENABLE_HIGHLIGHT = "auto_enable_highlight"
HIGHLIGHT_AUTO_ENABLED_MESSAGE = (
"Highlight enabled automatically because HITL confidence rules are configured"
)


class ManualReviewServiceProtocol(Protocol):
"""Protocol defining the interface for manual review services."""
Expand Down Expand Up @@ -383,6 +390,16 @@ def get_manual_review_workflow_util(api_client: Any, organization_id: str) -> An
return _create_null_workflow_util()


def has_confidence_rules(workflow_util: Any, workflow_id: str, rule_type: str) -> bool:
"""Whether the workflow's HITL rules of ``rule_type`` ("DB" or "API")
contain a confidence condition.

False in OSS, and with a plugin that lacks the method.
"""
check = getattr(workflow_util, "has_confidence_rules", None)
return bool(check and check(workflow_id, rule_type))


def has_manual_review_plugin():
return has_client_plugin(MANUAL_REVIEW_PLUGIN_NAME)

Expand Down
23 changes: 19 additions & 4 deletions workers/shared/workflow/execution/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@
from ...api.internal_client import InternalAPIClient
from ...infrastructure.logging import WorkerLogger
from ...utils.error_utils import get_user_friendly_error_message
from ...utils.manual_review_factory import AUTO_ENABLE_HIGHLIGHT
from ..destination_connector import (
DestinationConfig,
WorkerDestinationConnector,
Expand Down Expand Up @@ -847,7 +848,12 @@ def _execute_workflow_with_service(
# Continue with normal execution if check fails

# Step 3: Build and execute workflow
self._build_and_execute_workflow(execution_service, file_name)
review_config = file_processing_context.file_data.manual_review_config
self._build_and_execute_workflow(
execution_service,
file_name,
auto_enable_highlight=review_config.get(AUTO_ENABLE_HIGHLIGHT, False),
Comment thread
chandrasekharan-zipstack marked this conversation as resolved.
Resolved
)

return True

Expand Down Expand Up @@ -969,15 +975,20 @@ def _prepare_workflow_input_file(
raise file_prep_error

def _build_and_execute_workflow(
self, execution_service: WorkflowExecutionService, file_name: str
self,
execution_service: WorkflowExecutionService,
file_name: str,
auto_enable_highlight: bool = False,
) -> None:
"""Build and execute the workflow.

Detects structure tool workflows and routes them to the Celery-based
execute_structure_tool task instead of the Docker container flow.
"""
if self._is_structure_tool_workflow(execution_service):
self._execute_structure_tool_workflow(execution_service, file_name)
self._execute_structure_tool_workflow(
execution_service, file_name, auto_enable_highlight
)
else:
# Original Docker-based flow (unchanged)
execution_service.build_workflow()
Expand Down Expand Up @@ -1015,7 +1026,10 @@ def _is_structure_tool_workflow(
return False

def _execute_structure_tool_workflow(
self, execution_service: WorkflowExecutionService, file_name: str
self,
execution_service: WorkflowExecutionService,
file_name: str,
auto_enable_highlight: bool = False,
) -> None:
"""Execute structure tool as a task instead of a Docker container.

Expand Down Expand Up @@ -1061,6 +1075,7 @@ def _execute_structure_tool_workflow(
"messaging_channel": getattr(execution_service, "messaging_channel", ""),
"file_hash": metadata.get("source_hash", ""),
"exec_metadata": metadata,
AUTO_ENABLE_HIGHLIGHT: auto_enable_highlight,
}

# Call synchronously (same process, in-band)
Expand Down
130 changes: 130 additions & 0 deletions workers/tests/test_barrier.py
Original file line number Diff line number Diff line change
Expand Up @@ -773,5 +773,135 @@ def test_manual_review_required_invokes_decision_helper_per_batch(
)


class TestAutoEnableHighlightOrchestration:
"""HITL confidence rules set ``auto_enable_highlight``.

The orchestrators check their own flow's rules once (API deployments the
API rules, only for a ``hitl_queue_name`` request without a
``hitl_packet_id``; the rest the DB rules) and put the flag in
``manual_review_config``, which the file workers read.
"""

@staticmethod
def _util(confidence_rule_type):
"""A workflow util with a confidence rule of one type, or none."""
util = MagicMock(name="workflow_util")
util.has_confidence_rules.side_effect = (
lambda workflow_id, rule_type: rule_type == confidence_rule_type
)
return util

@pytest.mark.parametrize(
"confidence_rule_type, hitl_kwargs, has_rules",
[
("API", {"hitl_queue_name": "q"}, True),
("DB", {"hitl_queue_name": "q"}, False),
(None, {"hitl_queue_name": "q"}, False),
# The destination evaluates no API rules for these requests.
("API", {}, False),
("API", {"hitl_queue_name": "q", "hitl_packet_id": "p"}, False),
],
)
def test_api_deployment_sets_flag_on_every_batch(
self, monkeypatch, confidence_rule_type, hitl_kwargs, has_rules
):
from shared.utils.manual_review_factory import AUTO_ENABLE_HIGHLIGHT

mocks = _setup_workflow_api_mocks(monkeypatch, num_batches=2)
util = self._util(confidence_rule_type)
monkeypatch.setattr(
mocks.api_tasks, "get_manual_review_workflow_util", lambda *a: util
)
configs = []
monkeypatch.setattr(
mocks.api_tasks,
"_create_file_data",
lambda **kwargs: MagicMock(manual_review_config={}),
)
monkeypatch.setattr(
mocks.api_tasks,
"_create_batch_data",
lambda **kwargs: configs.append(
dict(kwargs["file_data"].manual_review_config)
),
)

mocks.api_tasks._run_workflow_api(
api_client=mocks.api_client,
schema_name="org_test",
workflow_id="wf-1",
execution_id="exec-1",
hash_values_of_files={"f1": MagicMock(name="file_1")},
scheduled=False,
execution_mode=None,
pipeline_id="pipe-1",
use_file_history=False,
task_id="task-1",
**hitl_kwargs,
)

if hitl_kwargs.get("hitl_queue_name") and not hitl_kwargs.get("hitl_packet_id"):
util.has_confidence_rules.assert_called_once_with("wf-1", "API")
else:
util.has_confidence_rules.assert_not_called()
assert len(configs) == 2
assert all(c.get(AUTO_ENABLE_HIGHLIGHT, False) is has_rules for c in configs)

@pytest.mark.parametrize(
"confidence_rule_type, has_rules", [("DB", True), ("API", False), (None, False)]
)
def test_general_sets_flag_on_global_file_data(
self, monkeypatch, confidence_rule_type, has_rules
):
from general import tasks as general_tasks
from shared.utils.manual_review_factory import AUTO_ENABLE_HIGHLIGHT

api_client = MagicMock(name="api_client")
util = self._util(confidence_rule_type)
util.create_workflow_file_data_with_manual_review.return_value = MagicMock(
manual_review_config={}
)
service = MagicMock(name="manual_review_service")
service.get_workflow_util.return_value = util
monkeypatch.setattr(
general_tasks, "get_manual_review_service", lambda **kwargs: service
)
monkeypatch.setattr(
general_tasks,
"_get_file_batches_general",
lambda **kwargs: [MagicMock(name="batch_1")],
)
configs = []

def create_batch_data(**kwargs):
configs.append(dict(kwargs["file_data"].manual_review_config))
return MagicMock(files=[])

monkeypatch.setattr(
general_tasks, "_create_batch_data_general", create_batch_data
)
monkeypatch.setattr(general_tasks, "_log_batch_statistics_to_ui", MagicMock())
monkeypatch.setattr(
general_tasks.WorkflowOrchestrationUtils,
"create_chord_execution",
MagicMock(return_value=MagicMock(id="chord-id")),
)

general_tasks._orchestrate_file_processing_general(
api_client=api_client,
workflow_id="wf-1",
execution_id="exec-1",
source_files={"f1": MagicMock(name="file_1")},
pipeline_id="pipe-1",
scheduled=False,
execution_mode=None,
use_file_history=False,
organization_id="org_test",
)

util.has_confidence_rules.assert_called_once_with("wf-1", "DB")
assert configs[0].get(AUTO_ENABLE_HIGHLIGHT, False) is has_rules


if __name__ == "__main__":
pytest.main([__file__, "-v"])
Loading
Loading