diff --git a/workers/executor/executor_tool_shim.py b/workers/executor/executor_tool_shim.py index 828e641398..e6346152dd 100644 --- a/workers/executor/executor_tool_shim.py +++ b/workers/executor/executor_tool_shim.py @@ -93,6 +93,9 @@ def __init__( # silently swallowing every subsequent log line at DEBUG. self._progress_publish_failed = False self._log_publish_failed = False + # Adapters whose name+model has already been surfaced to the UI; + # later mentions skip repeating the model line. + self._adapters_logged: set[str] = set() # Initialize StreamMixin. EXECUTION_BY_TOOL is not set in # the worker environment, so _exec_by_tool will be False. super().__init__(log_level=LogLevel.INFO) @@ -161,41 +164,36 @@ def stream_log( if _levels.index(level) < _levels.index(self.log_level): return - # Publish progress to frontend via the log consumer queue. - if not self.log_events_id: - return - wf_level = _SDK_TO_WF_LEVEL.get(level, "INFO") - # PROGRESS payload — IDE prompt-card live updates only. Dropped at - # the DB persist layer because LogPublisher.publish only stores - # payloads whose `type == "LOG"`. - try: - progress_payload = LogPublisher.log_progress( - component=self.component, - level=wf_level, - state=stage, - message=log, - ) - LogPublisher.publish( - channel_id=self.log_events_id, - payload=progress_payload, - ) - except Exception: - first_failure = not self._progress_publish_failed - self._progress_publish_failed = True - logger.log( - logging.WARNING if first_failure else logging.DEBUG, - "Failed to publish progress log (non-fatal)", - exc_info=first_failure, - ) - - # LOG payload — feeds workflow logs UI and persists to execution_log. - # LogDataDTO validation requires `execution_id` and `organization_id`; - # `file_execution_id` is optional. + # PROGRESS payload routes via the websocket room; skip when absent. + if self.log_events_id: + try: + progress_payload = LogPublisher.log_progress( + component=self.component, + level=wf_level, + state=stage, + message=log, + ) + LogPublisher.publish( + channel_id=self.log_events_id, + payload=progress_payload, + ) + except Exception: + first_failure = not self._progress_publish_failed + self._progress_publish_failed = True + logger.log( + logging.WARNING if first_failure else logging.DEBUG, + "Failed to publish progress log (non-fatal)", + exc_info=first_failure, + ) + + # LOG payload persists to execution_log; falls back to execution_id + # as the routing channel so it survives without a websocket subscriber. if not (self.execution_id and self.organization_id): return + log_channel = self.log_events_id or self.execution_id try: log_payload = LogPublisher.log_workflow( stage=stage, @@ -206,7 +204,7 @@ def stream_log( organization_id=self.organization_id, ) LogPublisher.publish( - channel_id=self.log_events_id, + channel_id=log_channel, payload=log_payload, ) except Exception: @@ -234,3 +232,26 @@ def stream_error_and_exit(self, message: str, err: Exception | None = None) -> N """ logger.error(message) raise SdkError(message, actual_err=err) + + def log_adapter_once( + self, + kind: str, + adapter_id: str, + adapter: Any, + ) -> None: + """Surface adapter identity to the UI on first use only. + + ``kind`` is the human label ("LLM", "Embedding", "Vector DB"). + ``adapter`` is an SDK instance — read only for non-sensitive + identity (model name or adapter display name); ``adapter_id`` + is the dedup key. Subsequent calls for the same id are no-ops. + """ + if not adapter_id or adapter_id in self._adapters_logged: + return + self._adapters_logged.add(adapter_id) + get_model = getattr(adapter, "get_model_name", None) + if callable(get_model): + label = get_model() or adapter_id + else: + label = getattr(adapter, "_adapter_name", "") or adapter_id + self.stream_log(f"Using {kind}: `{label}`") diff --git a/workers/executor/executors/index.py b/workers/executor/executors/index.py index ebb4f6d599..4bbbaf8839 100644 --- a/workers/executor/executors/index.py +++ b/workers/executor/executors/index.py @@ -155,7 +155,6 @@ def perform_indexing( ): return doc_id - self.tool.stream_log("Indexing file...") full_text = [ { "section": "full", @@ -171,7 +170,6 @@ def perform_indexing( def _trigger_indexing(self, vector_db: Any, documents: list) -> None: import openai - self.tool.stream_log("Adding nodes to vector db...") try: vector_db.index_document( documents, diff --git a/workers/executor/executors/legacy_executor.py b/workers/executor/executors/legacy_executor.py index 0ddb29e261..f319e1d24d 100644 --- a/workers/executor/executors/legacy_executor.py +++ b/workers/executor/executors/legacy_executor.py @@ -13,6 +13,7 @@ from executor.executor_tool_shim import ExecutorToolShim from executor.executors.constants import ExecutionSource from executor.executors.constants import IndexingConstants as IKeys +from executor.executors.constants import PromptServiceConstants as PSKeys from executor.executors.dto import ( ChunkingConfig, FileInfo, @@ -242,15 +243,15 @@ def _handle_extract(self, context: ExecutionContext) -> ExecutionResult: Path(file_path).name, context.run_id, ) - shim.stream_log("Initializing text extractor...") - shim.stream_log(f"Using text extractor: {type(x2text.x2text_instance).__name__}") - + extractor_name = type(x2text.x2text_instance).__name__ try: - shim.stream_log("Extracting text from document...") + shim.stream_log( + f"Extracting text using `{extractor_name}`" + + (" (with highlight)" if enable_highlight else "") + ) if enable_highlight and isinstance( x2text.x2text_instance, (LLMWhisperer, LLMWhispererV2) ): - shim.stream_log("Extracting text with highlight support enabled...") process_response: TextExtractionResult = x2text.process( input_file_path=file_path, output_file_path=output_file_path, @@ -301,7 +302,6 @@ def _handle_extract(self, context: ExecutionContext) -> ExecutionResult: process_response.extraction_metadata and process_response.extraction_metadata.line_metadata ): - shim.stream_log("Saving extraction metadata...") result_data["highlight_metadata"] = ( process_response.extraction_metadata.line_metadata ) @@ -605,12 +605,23 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: shim = self._build_shim( platform_api_key=extract_params.get("platform_api_key", ""), ) + + # One-shot run-config line — non-sensitive flags only; adapter + # identities are emitted inline on first use with full model info. + tool_settings = answer_params.get(PSKeys.TOOL_SETTINGS, {}) + outputs = answer_params.get(PSKeys.OUTPUTS, []) + shim.stream_log( + f"Run config: prompts={len(outputs)} | " + f"single_pass={'on' if is_single_pass else 'off'} | " + f"summarize={'on' if is_summarization else 'off'} | " + f"challenge=" + f"{'on' if tool_settings.get(PSKeys.ENABLE_CHALLENGE) else 'off'}" + ) step = 1 try: # ---- Step 1: Extract ---- if not skip_extraction: - shim.stream_log(f"Pipeline step {step}: Extracting text from document...") step += 1 extract_ctx = ExecutionContext( executor_name=context.executor_name, @@ -632,7 +643,6 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: # ---- Step 2: Summarize (if enabled) ---- if is_summarization: - shim.stream_log(f"Pipeline step {step}: Summarizing extracted text...") step += 1 summarize_result = self._run_pipeline_summarize( context=context, @@ -648,9 +658,6 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: answer_params["file_path"] = input_file_path elif not is_single_pass: # ---- Step 3: Index per output with dedup ---- - shim.stream_log( - f"Pipeline step {step}: Indexing document into vector store..." - ) step += 1 index_metrics = self._run_pipeline_index( context=context, @@ -693,7 +700,9 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: index_metrics=index_metrics, ) - shim.stream_log("Pipeline completed successfully") + output_map = structured_output.get(PSKeys.OUTPUT, {}) or {} + answered = sum(1 for v in output_map.values() if v not in (None, "", [], {})) + shim.stream_log(f"Pipeline completed: {answered}/{len(outputs)} prompts answered") out_metadata = { k: v for k, v in (answer_result.metadata or {}).items() @@ -728,12 +737,9 @@ def _run_pipeline_answer_step( output["chunk-size"] = 0 output["chunk-overlap"] = 0 operation = Operation.SINGLE_PASS_EXTRACTION.value - mode_label = "single pass" else: operation = Operation.ANSWER_PROMPT.value - mode_label = "prompt" - shim.stream_log(f"Pipeline step {step}: Running {mode_label} execution...") answer_ctx = ExecutionContext( executor_name=context.executor_name, operation=operation, @@ -1075,8 +1081,6 @@ def _handle_index(self, context: ExecutionContext) -> ExecutionResult: Path(file_path).name, context.run_id, ) - shim.stream_log("Initializing indexing pipeline...") - # Skip indexing when chunk_size is 0 — no vector operations needed. # ChunkingConfig raises ValueError for 0, so handle before DTO. if chunk_size == 0: @@ -1117,7 +1121,6 @@ def _handle_index(self, context: ExecutionContext) -> ExecutionResult: ) doc_id = index.generate_index_key(file_info=file_info, fs=fs_instance) logger.debug("Generated index key: doc_id=%s", doc_id) - shim.stream_log("Checking document index status...") embedding = embedding_compat( adapter_instance_id=embedding_instance_id, @@ -1129,7 +1132,8 @@ def _handle_index(self, context: ExecutionContext) -> ExecutionResult: adapter_instance_id=vector_db_instance_id, embedding=embedding, ) - shim.stream_log("Initialized embedding and vector DB adapters") + shim.log_adapter_once("Embedding", embedding_instance_id, embedding) + shim.log_adapter_once("Vector DB", vector_db_instance_id, vector_db) doc_id_found = index.is_document_indexed( doc_id=doc_id, embedding=embedding, vector_db=vector_db @@ -1150,11 +1154,9 @@ def _handle_index(self, context: ExecutionContext) -> ExecutionResult: ) return ExecutionResult(success=True, data={IKeys.DOC_ID: doc_id}) - if doc_id_found and reindex: - shim.stream_log("Document already indexed, re-indexing...") - else: - shim.stream_log("Indexing document for the first time...") - shim.stream_log("Indexing document into vector store...") + shim.stream_log( + "Re-indexing document" if doc_id_found else "Indexing document" + ) index.perform_indexing( vector_db=vector_db, doc_id=doc_id, @@ -1689,7 +1691,8 @@ def _execute_single_prompt( retrieval_strategy = output.get(PSKeys.RETRIEVAL_STRATEGY) valid_strategies = {s.value for s in RetrievalStrategy} if retrieval_strategy in valid_strategies: - shim.stream_log(f"Retrieving context for: `{prompt_name}`") + if chunk_size > 0: + shim.stream_log(f"Retrieving context for: `{prompt_name}`") logger.info( "Performing retrieval: prompt=%s strategy=%s chunk_size=%d", prompt_name, @@ -1713,9 +1716,11 @@ def _execute_single_prompt( context_retrieval_metrics=context_retrieval_metrics, ) metadata[PSKeys.CONTEXT][prompt_name] = context_list - shim.stream_log( - f"Retrieved {len(context_list)} context chunks for: `{prompt_name}`" - ) + if chunk_size > 0: + shim.stream_log( + f"Retrieved {len(context_list)} chunks via RAG " + f"for `{prompt_name}`" + ) logger.debug( "Retrieved %d context chunks for prompt: %s", len(context_list), @@ -1861,6 +1866,11 @@ def _init_llm_and_retrieval( adapter_instance_id=output[PSKeys.VECTOR_DB], embedding=embedding, ) + shim.log_adapter_once("LLM", output[PSKeys.LLM], llm) + if embedding is not None: + shim.log_adapter_once("Embedding", output[PSKeys.EMBEDDING], embedding) + if vector_db is not None: + shim.log_adapter_once("Vector DB", output[PSKeys.VECTOR_DB], vector_db) shim.stream_log( f"Initialized LLM and retrieval adapters for: `{prompt_name}`" ) @@ -2274,7 +2284,6 @@ def _handle_summarize(self, context: ExecutionContext) -> ExecutionResult: _, _, _, _, llm_cls, _, _ = self._get_prompt_deps() - shim.stream_log("Initializing LLM for summarization...") llm: Any = None try: llm = llm_cls( @@ -2286,7 +2295,9 @@ def _handle_summarize(self, context: ExecutionContext) -> ExecutionResult: AnswerPromptService as answer_prompt_svc, ) - shim.stream_log("Running document summarization...") + shim.stream_log( + f"Summarizing extracted text using LLM: `{llm.get_model_name()}`" + ) summary = answer_prompt_svc.run_completion(llm=llm, prompt=prompt) records = list(llm.flush_pending_usage()) logger.info("Summarization completed: run_id=%s", context.run_id) diff --git a/workers/file_processing/structure_tool_task.py b/workers/file_processing/structure_tool_task.py index 502e48ff32..3fcf1999c4 100644 --- a/workers/file_processing/structure_tool_task.py +++ b/workers/file_processing/structure_tool_task.py @@ -459,6 +459,7 @@ def _execute_structure_tool_impl(params: dict) -> dict: execution_source="tool", organization_id=organization_id, request_id=file_execution_id, + log_events_id=log_events_id, execution_id=execution_id, file_execution_id=file_execution_id, executor_params=agentic_params, @@ -490,6 +491,7 @@ def _execute_structure_tool_impl(params: dict) -> dict: execution_source="tool", organization_id=organization_id, request_id=file_execution_id, + log_events_id=log_events_id, execution_id=execution_id, file_execution_id=file_execution_id, executor_params={