diff --git a/application/single_app/config.py b/application/single_app/config.py index 10d13658b..9d6d17910 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -101,7 +101,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.261.304" +VERSION = "0.261.305" IS_DEVELOPMENT = is_development_env_enabled() # Opt-out for deployments where App Service Easy Auth is active but the platform diff --git a/application/single_app/functions_mixed_source_orchestration.py b/application/single_app/functions_mixed_source_orchestration.py index 9a45a71a1..a15e37671 100644 --- a/application/single_app/functions_mixed_source_orchestration.py +++ b/application/single_app/functions_mixed_source_orchestration.py @@ -140,6 +140,7 @@ def log_event(*args, **kwargs): "canceled_source_count", "citation_count", "completed_source_count", + "completion_tokens", "duplicate_evidence_count", "engine_call_count", "evidence_omitted_count", @@ -165,6 +166,7 @@ def log_event(*args, **kwargs): "unexpected_evidence_count", "unsupported_source_count", "unresolved_source_count", + "xml_schema_source_count", }) MIXED_SOURCE_TELEMETRY_DIMENSIONS = frozenset({ "cancellation_phase", diff --git a/application/single_app/functions_saved_analysis.py b/application/single_app/functions_saved_analysis.py index b5e28c932..a4a87a282 100644 --- a/application/single_app/functions_saved_analysis.py +++ b/application/single_app/functions_saved_analysis.py @@ -19,7 +19,7 @@ from functions_workflow_context import WorkflowContextBudgetError, calculate_workflow_context_budget from functions_workflow_identity import normalize_workflow_iteration_path from functions_workflow_result_masking import message_uses_workflow_result, withhold_workflow_result_message -from functions_workflow_result_store import WorkflowResultStorageUnavailableError, _quota_bytes +from functions_workflow_result_store import AnalysisWorkUnitConflictError, WorkflowResultStorageUnavailableError, _quota_bytes from functions_workflow_runtime_store import WorkflowRuntimeConflict from functions_workflow_results import ( ANALYSIS_SOURCE_ACCESS_VERSION, @@ -1927,7 +1927,10 @@ def sanitize_workflow_analysis_history(workflow, run_record, user_id, *, items=N bound_workflow or workflow, producer["run_id"], producer.get("task_id"), reference, reader_user_id=user_id, **selectors, ) - except (PermissionError, LookupError, ValueError, AzureError, WorkflowResultStorageUnavailableError, WorkflowRuntimeConflict) as exc: + except ( + PermissionError, LookupError, ValueError, AzureError, AnalysisWorkUnitConflictError, + WorkflowResultStorageUnavailableError, WorkflowRuntimeConflict, + ) as exc: log_event( "[DOCUMENT_ANALYSIS] Workflow analysis preview withheld.", extra={"run_id": run_record.get("id"), "task_id": task_id, "error_type": type(exc).__name__}, diff --git a/application/single_app/functions_workflow_result_store.py b/application/single_app/functions_workflow_result_store.py index 820f1db66..02b67b929 100644 --- a/application/single_app/functions_workflow_result_store.py +++ b/application/single_app/functions_workflow_result_store.py @@ -249,6 +249,21 @@ def _orchestration_blob_prefix(scope): return prefix +def _is_hierarchical_directory(properties): + """Return True for an ADLS Gen2 (hierarchical namespace) directory entry. + + Flat Blob listings on HNS-enabled accounts also yield zero-byte directory + objects for every path segment; they carry no result payload or result-specific metadata. + """ + metadata = getattr(properties, "metadata", None) or {} + metadata_is_folder = isinstance(metadata, Mapping) and any( + isinstance(key, str) and key.lower() == "hdi_isfolder" and str(value).strip().lower() == "true" + for key, value in metadata.items() + ) + resource_type = str(getattr(properties, "resource_type", "") or "").lower() + return metadata_is_folder or resource_type == "directory" + + def _blob_name(identity, reference): if identity["scope_type"] == "chat": return f"{_chat_blob_prefix(identity)}{reference['sha256']}.json" @@ -1505,10 +1520,26 @@ def _delete_scoped_blobs(self, scope): prefix = _orchestration_blob_prefix(scope) else: prefix = _run_blob_prefix(scope) + if is_orchestration and "run_id" not in scope: + directory_pattern = r"[0-9a-f]{64}(?:/[0-9a-f]{64})?" + elif is_chat and "message_id" in scope: + directory_pattern = None + else: + directory_pattern = r"[0-9a-f]{64}" + directories = [] container = self.blob_client.get_container_client(self.blob_container_name) for properties in container.list_blobs(name_starts_with=prefix, include=["metadata"]): name = properties.name suffix = name[len(prefix):] if isinstance(name, str) and name.startswith(prefix) else "" + if _is_hierarchical_directory(properties): + directory_suffix = suffix[:-1] if suffix.endswith("/") else suffix + if ( + directory_pattern is None or getattr(properties, "size", None) != 0 + or not re.fullmatch(directory_pattern, directory_suffix) + ): + raise WorkflowResultIntegrityError("Unexpected object in the private result prefix.") + directories.append(name) + continue if is_orchestration and "run_id" not in scope: if not re.fullmatch(r"[0-9a-f]{64}/[0-9a-f]{64}/[0-9a-f]{64}\.json", suffix): raise WorkflowResultIntegrityError("Unexpected object in the private result prefix.") @@ -1566,6 +1597,17 @@ def _delete_scoped_blobs(self, scope): ) except ResourceNotFoundError: continue + # Directories are removed only after their files, deepest first. Each one is + # re-read so the conditional delete uses its current ETag and type. + for name in sorted(set(directories), key=lambda value: (-value.rstrip("/").count("/"), value)): + blob = self.blob_client.get_blob_client(container=self.blob_container_name, blob=name) + try: + current = blob.get_blob_properties() + if not _is_hierarchical_directory(current) or getattr(current, "size", None) != 0: + raise WorkflowResultIntegrityError("Unexpected object in the private result prefix.") + blob.delete_blob(etag=current.etag, match_condition=MatchConditions.IfNotModified) + except ResourceNotFoundError: + continue def delete_run_results(self, workflow, run_id): """Stream all private records and run-prefix blobs, including interrupted writes. diff --git a/application/single_app/route_backend_workflows.py b/application/single_app/route_backend_workflows.py index bd90577e0..616be5186 100644 --- a/application/single_app/route_backend_workflows.py +++ b/application/single_app/route_backend_workflows.py @@ -91,7 +91,12 @@ validate_url_access_request, ) from functions_workflow_runner import _workflow_task_run_item_id, create_workflow_run_id, run_group_workflow, run_personal_workflow -from functions_workflow_result_store import WorkflowResultStorageUnavailableError, read_workflow_task_result_page +from functions_workflow_result_store import ( + AnalysisWorkUnitConflictError, + WorkflowResultIntegrityError, + WorkflowResultStorageUnavailableError, + read_workflow_task_result_page, +) from functions_workflow_definitions import ( WorkflowDefinitionConflict, WorkflowDefinitionError, @@ -134,9 +139,51 @@ WORKFLOW_INSTRUCTION_FIELD_LIMIT = 6000 +WORKFLOW_DELETE_PENDING_MESSAGE = ( + 'This workflow is being deleted, so its runs can no longer be cancelled. ' + 'Delete the workflow again to finish cleanup.' +) +WORKFLOW_RUNTIME_MISSING_MESSAGE = ( + 'This workflow run can no longer be cancelled because its runtime state was removed. ' + 'If you were deleting the workflow, delete it again to finish cleanup.' +) +# Cleanup failures that leave the workflow retryable; anything else still surfaces as a 500. +WORKFLOW_DELETE_FAILURE_ERRORS = ( + AnalysisWorkUnitConflictError, + AzureError, + RuntimeUnavailable, + WorkflowResultIntegrityError, + WorkflowResultStorageUnavailableError, + WorkflowRuntimeConflict, +) +WORKFLOW_DELETE_FAILED_MESSAGE = ( + 'The workflow could not be fully deleted. Try deleting it again; ' + 'if the problem continues, contact an administrator.' +) + + class WorkflowCancellationConflictError(RuntimeError): """Raised when a workflow run cannot transition into cancellation.""" + def __init__(self, public_message='Workflow run cannot be cancelled in its current state.'): + self.public_message = public_message + super().__init__(public_message) + + +def _workflow_delete_failure_response(exc, workflow_id, *, workspace_type, group_id=None): + log_event( + '[WORKFLOW_ROUTES] Workflow delete failed', + extra={ + 'workflow_id': workflow_id, + 'workspace_type': workspace_type, + 'group_id': group_id or '', + 'error_type': type(exc).__name__, + }, + level=logging.ERROR, + exceptionTraceback=True, + ) + return jsonify({'error': WORKFLOW_DELETE_FAILED_MESSAGE}), 500 + def _normalize_identifier(value): return str(value or '').strip() @@ -241,6 +288,8 @@ def _request_workflow_run_cancellation( workflow_id = _normalize_identifier(workflow.get('id')) active_run_id = _normalize_identifier(workflow.get('active_run_id')) target_run_id = _normalize_identifier(run_id) or active_run_id + if workflow.get('deleting') is True or _normalize_identifier(workflow.get('status')).lower() == 'deleting': + raise WorkflowCancellationConflictError(WORKFLOW_DELETE_PENDING_MESSAGE) if not target_run_id: raise WorkflowCancellationConflictError('No active workflow run is available to cancel.') if active_run_id and active_run_id != target_run_id: @@ -252,7 +301,13 @@ def _request_workflow_run_cancellation( if not run_record and target_run_id != active_run_id: raise LookupError('Workflow run not found.') if run_record and run_record.get('durable_execution') is True: - cancelled = cancel_durable_workflow_run(workflow, target_run_id, actor_user_id=requested_by) + try: + cancelled = cancel_durable_workflow_run(workflow, target_run_id, actor_user_id=requested_by) + except WorkflowRuntimeConflict as exc: + # A deleted runtime control usually means an earlier delete already fenced this run. + if exc.code == 'not_found': + raise WorkflowCancellationConflictError(WORKFLOW_RUNTIME_MISSING_MESSAGE) from exc + raise WorkflowCancellationConflictError(exc.public_message) from exc safe_run = {key: cancelled['run'].get(key) for key in ('id', 'workflow_id', 'status', 'durable_execution', 'runtime')} return _workflow_definition_response(cancelled['workflow']), safe_run @@ -1840,7 +1895,10 @@ def delete_user_workflow(workflow_id): if not workflow: return jsonify({'error': 'Workflow not found.'}), 404 - deleted = delete_personal_workflow(user_id, workflow_id) + try: + deleted = delete_personal_workflow(user_id, workflow_id) + except WORKFLOW_DELETE_FAILURE_ERRORS as exc: + return _workflow_delete_failure_response(exc, workflow_id, workspace_type='personal') if not deleted: return jsonify({'error': 'Workflow not found.'}), 404 @@ -1901,7 +1959,7 @@ def cancel_active_user_workflow_run(workflow_id): return jsonify({'error': 'Workflow run not found.'}), 404 except WorkflowCancellationConflictError as exc: logging.exception("Workflow cancellation conflict while cancelling active user workflow run.") - return jsonify({'error': 'Workflow run cannot be cancelled in its current state.'}), 409 + return jsonify({'error': exc.public_message}), 409 return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202 @@ -1935,14 +1993,14 @@ def cancel_user_workflow_run(workflow_id, run_id): user_id, ) return jsonify({'error': 'Workflow run not found.'}), 404 - except WorkflowCancellationConflictError: + except WorkflowCancellationConflictError as exc: logging.exception( "WorkflowCancellationConflictError while cancelling personal workflow run. workflow_id=%s run_id=%s user_id=%s", sanitize_log_message(workflow_id), sanitize_log_message(run_id), user_id, ) - return jsonify({'error': 'Workflow run cancellation conflict.'}), 409 + return jsonify({'error': exc.public_message}), 409 return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202 @@ -2284,7 +2342,10 @@ def delete_group_workflow_route(workflow_id): if not workflow: return jsonify({'error': 'Workflow not found.'}), 404 - deleted = delete_group_workflow(group_id, workflow_id) + try: + deleted = delete_group_workflow(group_id, workflow_id) + except WORKFLOW_DELETE_FAILURE_ERRORS as exc: + return _workflow_delete_failure_response(exc, workflow_id, workspace_type='group', group_id=group_id) if not deleted: return jsonify({'error': 'Workflow not found.'}), 404 @@ -2379,7 +2440,7 @@ def cancel_active_group_workflow_run(workflow_id): logging.exception('Group workflow run cancellation failed: run not found.', exc_info=exc) return jsonify({'error': 'Workflow run not found.'}), 404 except WorkflowCancellationConflictError as exc: - return jsonify({'error': str(exc)}), 409 + return jsonify({'error': exc.public_message}), 409 return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202 @@ -2422,7 +2483,7 @@ def cancel_group_workflow_run(workflow_id, run_id): return jsonify({'error': 'Run not found.'}), 404 except WorkflowCancellationConflictError as exc: logging.exception('Group workflow run cancellation conflict.') - return jsonify({'error': 'Unable to cancel run in its current state.'}), 409 + return jsonify({'error': exc.public_message}), 409 return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202 diff --git a/docs/explanation/features/WORKFLOW_DURABLE_EXECUTION.md b/docs/explanation/features/WORKFLOW_DURABLE_EXECUTION.md index acebf15ce..f34945b07 100644 --- a/docs/explanation/features/WORKFLOW_DURABLE_EXECUTION.md +++ b/docs/explanation/features/WORKFLOW_DURABLE_EXECUTION.md @@ -2,7 +2,7 @@ Implemented in version: **0.261.111** -Updated in version: **0.261.120**. +Updated in version: **0.261.305**. Application version tracking: `application/single_app/config.py`. @@ -182,6 +182,12 @@ records are excluded from ordinary task-item lists. Private lifecycle tombstones survive run cleanup. Deletion removes result payloads without allowing a late worker to recreate the deleted run. +Since **0.261.305**, private cleanup also handles explicitly marked, empty HNS +directories, deleting them after their result files. If cleanup fails after a +workflow is marked for deletion, retry **Delete** to finish it; **Cancel** cannot +restore a tombstoned runtime. Deletion-fenced saved outputs are withheld from +history. See [Workflow deletion recovery](../fixes/WORKFLOW_DELETE_HNS_DIRECTORY_FIX.md). + ## Runtime API Use the corresponding personal or group prefix: diff --git a/docs/explanation/fixes/WORKFLOW_DELETE_HNS_DIRECTORY_FIX.md b/docs/explanation/fixes/WORKFLOW_DELETE_HNS_DIRECTORY_FIX.md new file mode 100644 index 000000000..75288d367 --- /dev/null +++ b/docs/explanation/fixes/WORKFLOW_DELETE_HNS_DIRECTORY_FIX.md @@ -0,0 +1,138 @@ +# Workflow deletion and partial-cleanup recovery + +Fixed in version: **0.261.305** + +Application version tracking: `application/single_app/config.py`. + +## Issue and evidence + +Personal workflow DELETE requests returned HTTP 500 with +`WorkflowResultIntegrityError: Unexpected object in the private result prefix.` +The failing operation was the private result Blob cleanup, not workflow ownership +validation. An earlier deletion had already marked the workflow as `deleting` +and tombstoned its runtime control. Later Cancel requests therefore encountered +`WorkflowRuntimeConflict("not_found")`. Reading run history could also raise +`AnalysisWorkUnitConflictError("analysis_work_deleted")`. + +The affected storage account has hierarchical namespace (HNS) enabled. Flat +Blob listings on these accounts can include directory entries as well as files. +The cleanup previously accepted only result-file paths, rejecting legitimate +directories before cleanup could complete. + +A bounded, read-only inspection of both affected workflow prefixes confirmed +the directory entries. The email prefix contained four marked directories and +three result files; the Analyze prefix contained four marked directories and +24 result files. Entries inside the run cleanup scope had `hdi_isfolder=true` +and size zero in both the listing and a fresh properties read. Those entries +fail the previous result-file-only path validation. No result payloads were read +and no storage objects were changed during the inspection. + +## Root cause and related run failure + +Result cleanup checks hashed paths and stored identity metadata to avoid deleting +unrelated data. HNS directory objects do not have result-file suffixes or +result-specific metadata. Treating them as files caused cleanup to fail while +retaining the workflow and its history for retry. + +Separately, the failed Analyze workflow emitted aggregate `completion_tokens` +while development telemetry was enabled. That metric was missing from the +telemetry allowlist, causing `Mixed-source telemetry contains non-allowlisted +fields`. Coverage telemetry also emitted the missing aggregate +`xml_schema_source_count`. Both fields are now allowlisted; arbitrary telemetry +fields remain rejected. + +The email workflow's telemetry showed a Microsoft 365 wait and notification, +not continuous worker execution. A workflow can remain active while waiting +for sign-in or approval. The precise persisted wait state was not inspected, +and this fix does not force waiting workflows into an idle state. + +## Code changes and safety boundaries + +`application/single_app/functions_workflow_result_store.py` recognizes directory +entries only when an explicit `hdi_isfolder` metadata flag or `resource_type` +identifies a directory. A trailing slash alone is not sufficient. + +Accepted directories must be zero bytes with a known size and use the exact +hash-path depth permitted by the workflow-run, chat, or orchestration cleanup +scope. Result files retain their existing path, identity, and metadata checks. +Files are deleted first; directories are then re-read and removed deepest-first +using their current ETag and a non-recursive conditional delete. Changed types, +changed ETags, unexpected objects, and storage failures still fail cleanup. +Already missing objects are safe to retry. Scope-root and ancestor directories +remain outside the prefix sweep. + +`application/single_app/route_backend_workflows.py` now returns a stable JSON +500 for recognized cleanup/storage failures and logs the failure server-side. +It does not report deletion success when cleanup fails. All four personal/group +Cancel endpoints return JSON 409 for cancellation/runtime conflicts. Deleting +workflows receive guidance to retry Delete; a missing runtime receives a neutral +message because missing state does not always prove deletion. + +`application/single_app/functions_saved_analysis.py` withholds unverifiable +previews when their saved analysis is deletion-fenced, preserving the existing +history redaction behavior rather than returning an unhandled exception. + +`application/single_app/functions_mixed_source_orchestration.py` includes the two +aggregate telemetry metrics. `application/single_app/config.py` records +**0.261.305**. No new settings, routes, or deployment configuration are introduced. + +## Recovery after deployment + +Deploy **0.261.305** or later before retrying deletion on an affected site. +Retry **Delete** for a partially deleted workflow rather than **Cancel**. +Existing runtime tombstones continue to prevent late workers from publishing, +while cleanup can finish and remove the workflow's run history and definition. + +If deletion still fails, an administrator should inspect the workflow-delete +event under `[WORKFLOW_ROUTES]` and the corresponding storage exception. Do not +bypass integrity checks, recursively delete broad prefixes, or delete original +source documents or independently published reports. + +This change has not been deployed as part of the investigation. No live workflow, +conversation, or Blob was manually deleted. + +## Testing and validation + +Regression coverage: + +- `functional_tests/test_workflow_result_store.py`: HNS files-before-directories + ordering, personal/group/chat/orchestration path depths, case-insensitive + metadata, trailing-slash entries, explicit directory identity, missing size, + type replacement, ETag races, concurrent children, missing directories, + idempotent retry, and existing result isolation checks. +- `functional_tests/test_workflow_delete_recovery_routes.py`: safe JSON errors, + retry after partial deletion, 404 behavior, all four cancellation endpoints, + successful scoped cancellation, and group management access before storage. +- `functional_tests/test_saved_analysis_service.py`: deletion-fenced result + previews are withheld without leaking saved output. +- `functional_tests/test_workflow_cancellation.py` and + `functional_tests/test_mixed_source_hardening.py`: runtime conflict translation + and the exact aggregate token/coverage telemetry metric sets. +- `functional_tests/test_v2_workflow_run_action_clients.mjs`: existing V2 + cancellation and resume contracts remain compatible. + +The focused result-store, recovery-route, saved-analysis, group-fixture, +run-history, and repeat-policy selection passed **211 tests and 112 subtests**. +The cancellation +log-injection/cold-import checks, V2 action tests, and all three required route +policy scripts also pass. + +The complete mixed-source-hardening/cancellation run still has ten failures +previously reproduced on the baseline: stale AST harness dependencies, missing +Azure configuration, and older evidence/runner expectations. The focused +telemetry and route-helper selection passes. These baseline failures are not +reported as passing and are outside this fix. + +Local route checks used isolated, repository-pinned Flask/Werkzeug packages +and a compatible OpenSSL dependency; no shared Python installation was changed. + +## Before and after + +Before, a legitimate HNS directory could interrupt deletion and leave Cancel +and Runs returning unhandled errors. After, validated empty directories are +cleaned after their files, partial cleanup remains retryable, cancellation +conflicts return JSON, and fenced saved previews are withheld. + +See [Durable workflow execution](../features/WORKFLOW_DURABLE_EXECUTION.md) +and [Trigger a workflow](../../guides/trigger-a-workflow.md) for run states and +recovery guidance. diff --git a/docs/guides/trigger-a-workflow.md b/docs/guides/trigger-a-workflow.md index 89bad1ccd..ddb02a261 100644 --- a/docs/guides/trigger-a-workflow.md +++ b/docs/guides/trigger-a-workflow.md @@ -4,7 +4,7 @@ title: "Trigger a workflow" description: "Run a workflow now, schedule future runs, and inspect run activity." section: "Guides" audience: user -version: "0.261.122" +version: "0.261.305" --- ## What this does @@ -358,6 +358,7 @@ See [Workflow publication completion](../explanation/features/WORKFLOW_PUBLICATI | A scheduled workflow does not run | It is disabled or still configured for manual trigger | Edit the trigger and confirm the workflow is enabled. | | A calendar-scheduled workflow stopped running | Its saved time zone is no longer in the server's time zone database, so no next run can be worked out | Open it in the V2 editor, choose a time zone from the list, and save. | | A run fails immediately | A runner, action, document, or File Sync source is unavailable | Open run details, fix the dependency, and run again. | +| Delete failed and Cancel no longer works | Deletion already fenced the run, but result cleanup did not finish | On version **0.261.305** or later, retry **Delete** to finish cleanup. If it still fails, ask an administrator to inspect the workflow-delete logs. | | Chat never plans to run your workflow | **Run Workflows From Chat** is off, the conversation is shared, or the request didn't ask to run the workflow now | Ask from your own conversation, name the workflow and ask to run it now, or ask your administrator about the setting. | | The answer says only workflows with durable execution can be started from chat | The workflow runs synchronously | Turn on **Durable execution** in the V2 editor and save, or select **Run** in Workflows. | | The answer says the workflow is waiting for a Microsoft 365 approval or sign-in | An earlier run is waiting on Microsoft 365 | Finish that approval or sign-in, then ask again. | diff --git a/functional_tests/test_mixed_source_hardening.py b/functional_tests/test_mixed_source_hardening.py index d0e7d3bf8..be29c9f49 100644 --- a/functional_tests/test_mixed_source_hardening.py +++ b/functional_tests/test_mixed_source_hardening.py @@ -2,8 +2,8 @@ # test_mixed_source_hardening.py """ Functional tests for mixed-source hardening, extraction, and rollout. -Version: 0.250.167 -Implemented in: 0.250.070; direct-run telemetry isolation updated in 0.250.160; aggregate tabular parity harness coverage updated in 0.250.166 +Version: 0.261.305 +Implemented in: 0.250.070; direct-run telemetry isolation updated in 0.250.160; aggregate tabular parity harness coverage updated in 0.250.166; telemetry allowlist coverage updated in 0.261.305 This test ensures Phase 6 of #1061 preserves the bounded Phase 1-5 evidence contracts from #1056, #1057, #1058, #1059, and #1060 under parent #1055. @@ -1320,6 +1320,73 @@ def test_development_telemetry_is_default_off_allowlisted_and_privacy_safe(): raise AssertionError("Source-shaped telemetry fields must be rejected") +def _literal_dict_keys(call, keyword_name): + for keyword in call.keywords: + if keyword.arg == keyword_name and isinstance(keyword.value, ast.Dict): + return [key.value for key in keyword.value.keys if isinstance(key, ast.Constant)] + return [] + + +def test_every_emitted_telemetry_field_is_allowlisted(): + """Every literal metric and dimension sent by the app must pass the allowlist.""" + unapproved = [] + call_count = 0 + for path in sorted(APP_ROOT.glob("*.py")): + source = path.read_text(encoding="utf-8") + if "emit_mixed_source_telemetry(" not in source: + continue + for node in ast.walk(ast.parse(source, filename=str(path))): + if not isinstance(node, ast.Call): + continue + function = node.func + name = function.attr if isinstance(function, ast.Attribute) else getattr(function, "id", "") + if name != "emit_mixed_source_telemetry": + continue + call_count += 1 + for keyword_name, allowlist in ( + ("metrics", orchestration.MIXED_SOURCE_TELEMETRY_METRICS), + ("dimensions", orchestration.MIXED_SOURCE_TELEMETRY_DIMENSIONS), + ): + for key in _literal_dict_keys(node, keyword_name): + if key not in allowlist: + unapproved.append(f"{path.name}:{node.lineno} {keyword_name}.{key}") + assert call_count > 0, "Expected to find mixed-source telemetry calls." + assert unapproved == [], f"Telemetry fields missing from the allowlist: {unapproved}" + + +def test_native_execution_and_coverage_telemetry_emit_when_enabled(): + """Token usage and coverage telemetry must not fail a run when development telemetry is on.""" + captured_events = [] + original_log_event = orchestration.log_event + orchestration.log_event = lambda message, **kwargs: captured_events.append(kwargs) + settings = {"enable_mixed_source_development_telemetry": True} + try: + emitted_tokens = orchestration.emit_mixed_source_telemetry( + settings, + "native_execution", + "analyze", + metrics={ + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + "token_request_count": 1, + "request_count": 1, + }, + ) + emitted_coverage = orchestration.emit_mixed_source_coverage_telemetry( + settings, + "analyze", + {"requested_source_count": 1, "successful_source_count": 1, "completed_source_count": 1}, + ) + finally: + orchestration.log_event = original_log_event + + assert emitted_tokens is True + assert emitted_coverage is True + assert captured_events[0]["extra"]["completion_tokens"] == 5 + assert captured_events[1]["extra"]["xml_schema_source_count"] == 0 + + if __name__ == "__main__": original_main_log_event = orchestration.log_event orchestration.log_event = lambda *args, **kwargs: None @@ -1343,6 +1410,8 @@ def test_development_telemetry_is_default_off_allowlisted_and_privacy_safe(): test_bounded_analyze_all_catalog_rejects_over_limit_without_truncation() test_analyze_all_action_is_analyze_only_and_manifest_is_fresh() test_development_telemetry_is_default_off_allowlisted_and_privacy_safe() + test_every_emitted_telemetry_field_is_allowlisted() + test_native_execution_and_coverage_telemetry_emit_when_enabled() finally: orchestration.log_event = original_main_log_event print("Mixed-source hardening functional tests passed.") \ No newline at end of file diff --git a/functional_tests/test_saved_analysis_service.py b/functional_tests/test_saved_analysis_service.py index d75f53493..0a556bd0f 100644 --- a/functional_tests/test_saved_analysis_service.py +++ b/functional_tests/test_saved_analysis_service.py @@ -1,9 +1,10 @@ # test_saved_analysis_service.py """ Functional tests for saved Analyze data in chat and follow-up reads. -Version: 0.261.232 +Version: 0.261.305 Implemented in: 0.261.109 Container-only saved-result access covered in: 0.261.232 +Deletion-fenced workflow previews covered in: 0.261.305 The production result builder/readers use serialized sections and authorized message seams without re-uploading, indexing, or analyzing source files. A saved @@ -233,7 +234,11 @@ def test_invalid_page_bounds_do_not_load_results(saved_chat, offset, limit): assert fixture["store"].reads == [] -def test_workflow_history_cannot_reveal_a_blocked_result_preview(): +@pytest.mark.parametrize("failure", [ + PermissionError("source revoked"), + saved.AnalysisWorkUnitConflictError("analysis_work_deleted"), +]) +def test_workflow_history_cannot_reveal_a_blocked_result_preview(failure): workflow = {"id": "workflow-1", "user_id": "owner"} summary = { "analysis_result": True, @@ -248,7 +253,7 @@ def test_workflow_history_cannot_reveal_a_blocked_result_preview(): item = {**deepcopy(task), "output_summary": "PRIVATE FINDING"} def deny(*args, **kwargs): - raise PermissionError("source revoked") + raise failure filtered_run, items, allowed = saved.sanitize_workflow_analysis_history( workflow, run, "viewer", items=[item], result_reader=deny, diff --git a/functional_tests/test_v2_workflow_run_action_clients.mjs b/functional_tests/test_v2_workflow_run_action_clients.mjs index 9d2aa3823..75bfb2da5 100644 --- a/functional_tests/test_v2_workflow_run_action_clients.mjs +++ b/functional_tests/test_v2_workflow_run_action_clients.mjs @@ -391,7 +391,7 @@ test('The run-level cancel route answers 202, 404 or a code-less 409', () => { assert.ok(cancel.includes('_request_workflow_run_cancellation(')); assert.ok(cancel.includes("return jsonify({'error': 'Workflow not found.'}), 404")); assert.ok(cancel.includes("return jsonify({'error': 'Workflow run not found.'}), 404")); - assert.ok(cancel.includes("return jsonify({'error': 'Workflow run cancellation conflict.'}), 409")); + assert.ok(cancel.includes("return jsonify({'error': exc.public_message}), 409")); assert.ok(cancel.includes("return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202")); assert.ok(!cancel.includes("'code'"), 'The cancel route started sending a code; read it in cancelWorkflowRun.'); diff --git a/functional_tests/test_workflow_cancellation.py b/functional_tests/test_workflow_cancellation.py index 033432271..8620f43d9 100644 --- a/functional_tests/test_workflow_cancellation.py +++ b/functional_tests/test_workflow_cancellation.py @@ -1,8 +1,9 @@ # test_workflow_cancellation.py """ Functional test for active workflow cancellation. -Version: 0.261.029 +Version: 0.261.305 Implemented in: 0.250.062 +Deletion/runtime conflict responses implemented in: 0.261.305 This test ensures personal and group workflow cancellation requests persist by scope and run id, stop the runner at a cooperative boundary, mark unfinished @@ -27,6 +28,7 @@ from m365_interaction import M365SignInRequired RUNNER_FILE = APP_ROOT / "functions_workflow_runner.py" ROUTES_FILE = APP_ROOT / "route_backend_workflows.py" +RUNTIME_STORE_FILE = APP_ROOT / "functions_workflow_runtime_store.py" BACKGROUND_TASKS_FILE = APP_ROOT / "background_tasks.py" CONFIG_FILE = APP_ROOT / "config.py" ACTIVITY_FILE = APP_ROOT / "functions_workflow_activity.py" @@ -84,16 +86,21 @@ def _load_nodes(path, function_names=(), class_names=(), assignment_names=(), na return resolved_namespace -def _load_cancellation_route_helpers(): +def _load_cancellation_route_helpers(namespace=None): + runtime_store = _load_nodes(RUNTIME_STORE_FILE, class_names=("WorkflowRuntimeConflict",)) return _load_nodes( ROUTES_FILE, function_names=("_request_workflow_run_cancellation",), class_names=("WorkflowCancellationConflictError",), + assignment_names=("WORKFLOW_DELETE_PENDING_MESSAGE", "WORKFLOW_RUNTIME_MISSING_MESSAGE"), namespace={ "datetime": datetime, "timezone": timezone, "M365_ACTIVE_STATES": M365_ACTIVE_STATES, + "WorkflowRuntimeConflict": runtime_store["WorkflowRuntimeConflict"], "_normalize_identifier": lambda value: str(value or "").strip(), + "_workflow_definition_response": lambda workflow: workflow, + **(namespace or {}), }, ) @@ -269,6 +276,67 @@ def test_early_cancellation_is_saved_and_mismatched_runs_are_rejected(): raise AssertionError("Expected a workflow run belonging to another workflow to be rejected.") +def test_deleting_workflow_refuses_cancellation_with_retry_guidance(): + """A workflow left mid-delete must explain that deleting again finishes cleanup.""" + assert_app_version_at_least("0.261.305") + helpers = _load_cancellation_route_helpers() + request_cancellation = helpers["_request_workflow_run_cancellation"] + conflict_type = helpers["WorkflowCancellationConflictError"] + + for workflow in ( + {"id": "wf-1", "active_run_id": "run-1", "status": "deleting", "deleting": True}, + {"id": "wf-1", "active_run_id": "run-1", "status": "Deleting"}, + ): + try: + request_cancellation( + workflow, + run_id="run-1", + requested_by="user-1", + get_run=lambda run_id: (_ for _ in ()).throw(AssertionError("No run read expected.")), + save_run=lambda run_record: run_record, + update_runtime_fields=lambda updates: updates, + ) + except conflict_type as exc: + assert exc.public_message == helpers["WORKFLOW_DELETE_PENDING_MESSAGE"] + else: + raise AssertionError("Expected cancellation of a deleting workflow to be refused.") + + +def test_durable_runtime_conflicts_become_cancellation_conflicts(): + """A tombstoned runtime control must return a 409 message instead of escaping as a 500.""" + assert_app_version_at_least("0.261.305") + probe = _load_cancellation_route_helpers() + runtime_conflict = probe["WorkflowRuntimeConflict"] + workflow = {"id": "wf-1", "active_run_id": "run-1", "status": "running"} + durable_run = {"id": "run-1", "workflow_id": "wf-1", "status": "running", "durable_execution": True} + + for conflict, expected_message in ( + (runtime_conflict("not_found", "Workflow runtime control was deleted."), probe["WORKFLOW_RUNTIME_MISSING_MESSAGE"]), + (runtime_conflict("terminal", "Workflow run is already finished."), "Workflow run is already finished."), + ): + def cancel_durable_workflow_run(*args, _conflict=conflict, **kwargs): + raise _conflict + + helpers = _load_cancellation_route_helpers({ + "cancel_durable_workflow_run": cancel_durable_workflow_run, + "WorkflowRuntimeConflict": runtime_conflict, + }) + try: + helpers["_request_workflow_run_cancellation"]( + workflow, + run_id="run-1", + requested_by="user-1", + get_run=lambda run_id: dict(durable_run), + save_run=lambda run_record: run_record, + update_runtime_fields=lambda updates: updates, + ) + except helpers["WorkflowCancellationConflictError"] as exc: + assert exc.public_message == expected_message + assert exc.__cause__ is conflict + else: + raise AssertionError("Expected a durable runtime conflict to become a cancellation conflict.") + + def test_runner_cancellation_helpers_cover_personal_and_group_items(): """Persisted requests stop new work and terminalize unfinished items in both scopes.""" helpers, saved_items = _build_runner_storage_helpers() diff --git a/functional_tests/test_workflow_delete_recovery_routes.py b/functional_tests/test_workflow_delete_recovery_routes.py new file mode 100644 index 000000000..8cfe0f0c8 --- /dev/null +++ b/functional_tests/test_workflow_delete_recovery_routes.py @@ -0,0 +1,232 @@ +# test_workflow_delete_recovery_routes.py +""" +Functional tests for retryable workflow deletion and cancellation conflicts. +Version: 0.261.305 +Implemented in: 0.261.305 + +Execute the real personal/group route bodies and cancellation helper over +scoped storage doubles. Route authentication policies are verified separately +by functional_tests/route_tests and group fixture parity tests. +""" + +import ast +import logging +from copy import deepcopy +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from azure.core.exceptions import AzureError +from flask import Blueprint, Flask, jsonify + +from functions_workflow_result_store import ( + AnalysisWorkUnitConflictError, + WorkflowResultIntegrityError, + WorkflowResultStorageUnavailableError, +) +from test_workflow_cancellation import ( + ROUTES_FILE, + RUNTIME_STORE_FILE, + _load_cancellation_route_helpers, + _load_nodes, +) + + +@pytest.fixture +def recovery_routes(): + workflow = { + "id": "workflow-one", "user_id": "owner", "name": "Review", + "active_run_id": "run-one", "status": "running", + } + run = { + "id": "run-one", "workflow_id": workflow["id"], + "status": "running", "durable_execution": True, + } + get_workflow = Mock(side_effect=lambda scope_id, workflow_id: deepcopy(workflow)) + delete_workflow = Mock(return_value=True) + get_run = Mock(side_effect=lambda scope_id, run_id: deepcopy(run)) + save_run = Mock(side_effect=lambda scope_id, record: record) + update_workflow = Mock(side_effect=lambda scope_id, workflow_id, updates: {**workflow, **updates}) + cancel_durable = Mock(return_value={"workflow": workflow, "run": run}) + deletion_log = Mock() + events = Mock() + runtime = _load_nodes( + RUNTIME_STORE_FILE, class_names=("WorkflowRuntimeConflict", "RuntimeUnavailable"), + ) + helpers = _load_cancellation_route_helpers(namespace={ + **runtime, "cancel_durable_workflow_run": cancel_durable, + }) + namespace = _load_nodes( + ROUTES_FILE, + function_names=("_workflow_delete_failure_response",), + assignment_names=("WORKFLOW_DELETE_FAILURE_ERRORS", "WORKFLOW_DELETE_FAILED_MESSAGE"), + namespace={ + **helpers, + "AzureError": AzureError, + "AnalysisWorkUnitConflictError": AnalysisWorkUnitConflictError, + "WorkflowResultIntegrityError": WorkflowResultIntegrityError, + "WorkflowResultStorageUnavailableError": WorkflowResultStorageUnavailableError, + "jsonify": jsonify, + "logging": logging, + "log_event": events, + "sanitize_log_message": lambda value: str(value).replace("\n", " ").replace("\r", " "), + "get_current_user_id": lambda: "owner", + "_resolve_active_group_for_workflow_management": Mock(return_value=("group-one", {})), + "_resolve_group_workflow_request_group": Mock(return_value=("group-one", {})), + "get_personal_workflow": get_workflow, + "get_group_workflow": get_workflow, + "delete_personal_workflow": delete_workflow, + "delete_group_workflow": delete_workflow, + "get_personal_workflow_run": get_run, + "get_group_workflow_run": get_run, + "save_personal_workflow_run": save_run, + "save_group_workflow_run": save_run, + "update_personal_workflow_runtime_fields": update_workflow, + "update_group_workflow_runtime_fields": update_workflow, + "log_workflow_deletion": deletion_log, + }, + ) + routes = ( + ("delete_user_workflow", "/api/user/workflows/", "DELETE"), + ("delete_group_workflow_route", "/api/group/workflows/", "DELETE"), + ("cancel_active_user_workflow_run", "/api/user/workflows//cancel", "POST"), + ("cancel_user_workflow_run", "/api/user/workflows//runs//cancel", "POST"), + ("cancel_active_group_workflow_run", "/api/group/workflows//cancel", "POST"), + ("cancel_group_workflow_run", "/api/group/workflows//runs//cancel", "POST"), + ) + names = {name for name, _, _ in routes} + nodes = [] + for node in ast.walk(ast.parse(ROUTES_FILE.read_text(encoding="utf-8"))): + if isinstance(node, ast.FunctionDef) and node.name in names: + node.decorator_list = [] + nodes.append(node) + assert {node.name for node in nodes} == names + exec(compile(ast.Module(body=nodes, type_ignores=[]), str(ROUTES_FILE), "exec"), namespace) + + app = Flask("workflow-delete-recovery") + app.config["TESTING"] = True + blueprint = Blueprint("backend_workflows", __name__) + for name, path, method in routes: + blueprint.add_url_rule(path, view_func=namespace[name], methods=[method]) + app.register_blueprint(blueprint) + return SimpleNamespace( + client=app.test_client(), workflow=workflow, run=run, helpers=namespace, + get_workflow=get_workflow, delete_workflow=delete_workflow, + get_run=get_run, save_run=save_run, update_workflow=update_workflow, + cancel_durable=cancel_durable, deletion_log=deletion_log, events=events, + ) + + +@pytest.mark.parametrize("scope, scope_id", [("user", "owner"), ("group", "group-one")]) +@pytest.mark.parametrize("failure", ["integrity", "analysis", "azure", "storage", "runtime", "conflict"]) +def test_delete_failures_are_safe_json_and_retryable(recovery_routes, scope, scope_id, failure): + harness = recovery_routes + failures = { + "integrity": WorkflowResultIntegrityError("PRIVATE PROVIDER DETAILS"), + "analysis": AnalysisWorkUnitConflictError("analysis_work_deleted"), + "azure": AzureError("PRIVATE PROVIDER DETAILS"), + "storage": WorkflowResultStorageUnavailableError("PRIVATE PROVIDER DETAILS"), + "runtime": harness.helpers["RuntimeUnavailable"](), + "conflict": harness.helpers["WorkflowRuntimeConflict"]("not_found"), + } + harness.delete_workflow.side_effect = [failures[failure], True] + path = f"/api/{scope}/workflows/workflow-one" + failed = harness.client.delete(path) + assert failed.status_code == 500 + assert failed.is_json + assert failed.json == {"error": harness.helpers["WORKFLOW_DELETE_FAILED_MESSAGE"]} + assert "PRIVATE" not in failed.get_data(as_text=True) + harness.delete_workflow.assert_called_with(scope_id, "workflow-one") + harness.deletion_log.assert_not_called() + harness.events.assert_called_once() + fields = harness.events.call_args.kwargs + assert fields["extra"]["error_type"] == type(failures[failure]).__name__ + assert fields["exceptionTraceback"] is True + assert fields["extra"]["workspace_type"] == ("personal" if scope == "user" else "group") + + harness.workflow.update({"deleting": True, "status": "deleting"}) + retried = harness.client.delete(path) + assert retried.status_code == 200 + assert retried.json == {"success": True} + harness.deletion_log.assert_called_once() + + +@pytest.mark.parametrize("scope", ["user", "group"]) +@pytest.mark.parametrize("missing_at", ["lookup", "delete"]) +def test_missing_workflows_remain_404(recovery_routes, scope, missing_at): + harness = recovery_routes + if missing_at == "lookup": + harness.get_workflow.side_effect = None + harness.get_workflow.return_value = None + else: + harness.delete_workflow.return_value = False + response = harness.client.delete(f"/api/{scope}/workflows/workflow-one") + assert response.status_code == 404 + assert response.json == {"error": "Workflow not found."} + harness.events.assert_not_called() + harness.deletion_log.assert_not_called() + if missing_at == "lookup": + harness.delete_workflow.assert_not_called() + + +@pytest.mark.parametrize("scope", ["user", "group"]) +@pytest.mark.parametrize("explicit_run", [False, True]) +@pytest.mark.parametrize("failure", ["deleting", "status_only", "no_active", "missing_runtime", "terminal"]) +def test_all_cancel_endpoints_return_json_conflicts(recovery_routes, scope, explicit_run, failure): + harness = recovery_routes + if failure in ("deleting", "status_only", "no_active"): + harness.workflow["status"] = "deleting" + if failure == "deleting": + harness.workflow["deleting"] = True + if failure == "no_active": + harness.workflow["active_run_id"] = "" + expected = harness.helpers["WORKFLOW_DELETE_PENDING_MESSAGE"] + elif failure == "missing_runtime": + harness.cancel_durable.side_effect = harness.helpers["WorkflowRuntimeConflict"]("not_found") + expected = harness.helpers["WORKFLOW_RUNTIME_MISSING_MESSAGE"] + else: + expected = "The run is already finished." + harness.cancel_durable.side_effect = harness.helpers["WorkflowRuntimeConflict"]("run_terminal", expected) + suffix = "/runs/run-one/cancel" if explicit_run else "/cancel" + response = harness.client.post(f"/api/{scope}/workflows/workflow-one{suffix}") + assert response.status_code == 409 + assert response.is_json + assert response.json == {"error": expected} + harness.save_run.assert_not_called() + harness.update_workflow.assert_not_called() + if failure in ("deleting", "status_only", "no_active"): + harness.get_run.assert_not_called() + harness.cancel_durable.assert_not_called() + + +@pytest.mark.parametrize("scope, scope_id", [("user", "owner"), ("group", "group-one")]) +@pytest.mark.parametrize("explicit_run", [False, True]) +def test_cancel_success_preserves_existing_status_and_scoped_reads(recovery_routes, scope, scope_id, explicit_run): + harness = recovery_routes + suffix = "/runs/run-one/cancel" if explicit_run else "/cancel" + response = harness.client.post(f"/api/{scope}/workflows/workflow-one{suffix}") + assert response.status_code == 202 + assert response.json["success"] is True + harness.get_workflow.assert_called_once_with(scope_id, "workflow-one") + harness.get_run.assert_called_once_with(scope_id, "run-one") + harness.cancel_durable.assert_called_once_with(harness.workflow, "run-one", actor_user_id="owner") + + +@pytest.mark.parametrize("scope", ["user", "group"]) +def test_cancel_never_accepts_a_run_from_another_workflow(recovery_routes, scope): + harness = recovery_routes + harness.run["workflow_id"] = "other-workflow" + response = harness.client.post(f"/api/{scope}/workflows/workflow-one/runs/run-one/cancel") + assert response.status_code == 404 + harness.cancel_durable.assert_not_called() + harness.save_run.assert_not_called() + harness.update_workflow.assert_not_called() + + +def test_group_delete_checks_management_access_before_storage(recovery_routes): + harness = recovery_routes + harness.helpers["_resolve_active_group_for_workflow_management"].side_effect = PermissionError("Forbidden.") + response = harness.client.delete("/api/group/workflows/workflow-one") + assert response.status_code == 403 + harness.get_workflow.assert_not_called() + harness.delete_workflow.assert_not_called() diff --git a/functional_tests/test_workflow_result_store.py b/functional_tests/test_workflow_result_store.py index 093bdbb98..565a857c1 100644 --- a/functional_tests/test_workflow_result_store.py +++ b/functional_tests/test_workflow_result_store.py @@ -1,8 +1,9 @@ # test_workflow_result_store.py """ Functional tests for durable, scoped workflow task-result persistence. -Version: 0.261.106 +Version: 0.261.305 Implemented in: 0.261.106 +HNS directory cleanup implemented in: 0.261.305 JSON-copying Cosmos and byte-copying Blob fakes exercise immutable writes, backend selection, bounded range reads, identity/integrity checks, quota @@ -184,12 +185,17 @@ def download_blob(self, *, offset=None, length=None, etag=None, match_condition= data = data[offset:offset + length] return FakeDownload(data, self.owner, ranged=offset is not None) - def delete_blob(self, *, delete_snapshots, etag=None, match_condition=None): + def delete_blob(self, *, delete_snapshots=None, etag=None, match_condition=None): if self.owner.delete_error is not None: raise self.owner.delete_error self._check_etag(etag, match_condition) - self._row() - if delete_snapshots != "include": + row = self._row() + if row.get("directory"): + # HNS refuses a non-recursive delete of a directory that still has children. + if any(container == self.key[0] and name != self.key[1] and name.startswith(self.key[1].rstrip("/") + "/") + for container, name in self.owner.records): + raise ResourceExistsError(message="Fake provider directory is not empty.") + elif delete_snapshots != "include": raise AssertionError("Private result snapshots should be cleaned with their blob.") self.owner.deletes.append(self.key) del self.owner.records[self.key] @@ -204,7 +210,11 @@ def list_blobs(self, *, name_starts_with, include): self.owner.lists.append({"container": self.name, "prefix": name_starts_with, "include": include}) if self.owner.list_error is not None: raise self.owner.list_error - for container, name in list(self.owner.records): + keys = list(self.owner.records) + if self.owner.hierarchical: + # HNS flat listings are name-ordered, so each directory precedes its files. + keys.sort(key=lambda key: key[1]) + for container, name in keys: if container == self.name and name.startswith(name_starts_with): yield FakeBlob(self.owner, container, name).get_blob_properties() @@ -212,6 +222,7 @@ def list_blobs(self, *, name_starts_with, include): class FakeBlobService: def __init__(self): self.records = {} + self.hierarchical = False self.uploads = [] self.downloads = [] self.property_reads = [] @@ -228,6 +239,28 @@ def __init__(self): def get_blob_client(self, *, container, blob): return FakeBlob(self, container, blob) + def add_hierarchical_directories(self, container="personal-chat"): + """Mirror HNS: every path segment of every stored blob becomes a zero-byte directory.""" + self.hierarchical = True + for blob_container, name in list(self.records): + if blob_container != container: + continue + parts = name.split("/") + for depth in range(1, len(parts)): + self.add_directory("/".join(parts[:depth]), container=container) + + def add_directory(self, name, *, container="personal-chat", metadata=None, data=b""): + key = (container, name) + if key in self.records: + return + self.etag_counter += 1 + self.records[key] = { + "data": data, + "metadata": {"hdi_isfolder": "true"} if metadata is None else metadata, + "etag": f"etag-{self.etag_counter}", + "directory": True, + } + def get_container_client(self, container): return FakeBlobContainer(self, container) @@ -847,6 +880,185 @@ def test_cleanup_refuses_unrecognized_blob_or_wrong_stored_scope(self): self.assertEqual(service.deletes, []) self.assertEqual(container.deletes, []) + def test_cleanup_on_hierarchical_namespace_removes_directories_after_files(self): + store, container, service = self.make_store(blob=True) + self.save(store) + store.save(self.workflow, self.run_id, "another-task", {"text": "second"}) + store.save(self.workflow, "other-run", self.task_id, {"text": "keep"}) + service.add_hierarchical_directories() + prefix = store_module._run_blob_prefix(store_module._scope(self.workflow, self.run_id)) + run_directory = prefix.rstrip("/") + files = {key for key, row in service.records.items() if key[1].startswith(prefix) and not row.get("directory")} + task_directories = {key for key, row in service.records.items() if key[1].startswith(prefix) and row.get("directory")} + self.assertEqual(len(files), 2) + self.assertEqual(len(task_directories), 2) + listing = [entry.name for entry in FakeBlobContainer(service, "personal-chat").list_blobs( + name_starts_with=prefix, include=["metadata"])] + service.lists.clear() + self.assertTrue(store_module._is_hierarchical_directory( + FakeBlob(service, "personal-chat", listing[0]).get_blob_properties())) + store.delete_run_results(self.workflow, self.run_id) + self.assertFalse(any(key[1].startswith(prefix) for key in service.records)) + self.assertEqual(set(service.deletes), files | task_directories) + self.assertEqual(set(service.deletes[:2]), files) + self.assertIn(("personal-chat", run_directory), service.records) + self.assertTrue(any(key[1].startswith(store_module._run_blob_prefix( + store_module._scope(self.workflow, "other-run"))) for key in service.records)) + self.assertFalse(any(key[0] == self.run_id for key in container.records)) + store.delete_run_results(self.workflow, self.run_id) + + def test_cleanup_hierarchical_directory_validation_is_scope_exact(self): + digest = "a" * 64 + cases = ( + (store_module._scope(self.workflow, self.run_id), digest, True), + (store_module._scope({**self.workflow, "group_id": "group-one"}, self.run_id), digest, True), + (store_module._scope(self.workflow, self.run_id), f"{digest}/", True), + (store_module._scope(self.workflow, self.run_id), "analyze", False), + (store_module._scope(self.workflow, self.run_id), f"{digest}/{digest}", False), + (store_module._chat_scope("owner-one", "conversation-one"), digest, True), + (store_module._chat_scope("owner-one", "conversation-one", "message-one"), digest, False), + (store_module._orchestration_scope("owner-one", "conversation-one"), digest, True), + (store_module._orchestration_scope("owner-one", "conversation-one"), f"{digest}/{digest}", True), + (store_module._orchestration_scope("owner-one", "conversation-one"), f"{digest}/{digest}/{digest}", False), + (store_module._orchestration_scope("owner-one", "conversation-one", "run-one"), digest, True), + (store_module._orchestration_scope("owner-one", "conversation-one", "run-one"), f"{digest}/{digest}", False), + ) + for scope, suffix, allowed in cases: + with self.subTest(scope_type=scope["scope_type"], keys=sorted(scope), suffix=suffix): + store, _, service = self.make_store(blob=True) + if scope["scope_type"] == "chat": + prefix = store_module._chat_blob_prefix(scope) + elif scope["scope_type"] == "orchestration": + prefix = store_module._orchestration_blob_prefix(scope) + else: + prefix = store_module._run_blob_prefix(scope) + service.hierarchical = True + parts = suffix.rstrip("/").split("/") + for depth in range(1, len(parts) + 1): + directory_name = prefix + "/".join(parts[:depth]) + if depth == len(parts) and suffix.endswith("/"): + directory_name += "/" + service.add_directory(directory_name) + if allowed: + store._delete_scoped_blobs(scope) + self.assertFalse(any(key[1].startswith(prefix) for key in service.records)) + else: + with self.assertRaises(store_module.WorkflowResultIntegrityError): + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, []) + + def test_cleanup_refuses_directory_markers_with_content_or_changed_type(self): + scope = store_module._scope(self.workflow, self.run_id) + prefix = store_module._run_blob_prefix(scope) + for variant in ("payload", "metadata_case", "became_file"): + with self.subTest(variant=variant): + store, _, service = self.make_store(blob=True) + service.hierarchical = True + name = prefix + "b" * 64 + if variant == "payload": + service.add_directory(name, data=b"hidden") + else: + service.add_directory(name, metadata={"HDI_ISFOLDER": "TRUE"} if variant == "metadata_case" else None) + if variant == "metadata_case": + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, [("personal-chat", name)]) + continue + original_list = FakeBlobContainer.list_blobs + + def list_then_replace(container_self, **kwargs): + yield from original_list(container_self, **kwargs) + if variant == "became_file": + service.records[("personal-chat", name)] = {"data": b"", "metadata": {}, "etag": "replaced"} + + with patch.object(FakeBlobContainer, "list_blobs", list_then_replace): + with self.assertRaises(store_module.WorkflowResultIntegrityError): + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, []) + self.assertIn(("personal-chat", name), service.records) + + def test_cleanup_requires_explicit_directory_identity_and_size(self): + scope = store_module._scope(self.workflow, self.run_id) + prefix = store_module._run_blob_prefix(scope) + for variant in ("unmarked", "slash_only", "missing_size", "resource_type"): + with self.subTest(variant=variant): + store, _, service = self.make_store(blob=True) + name = prefix + "a" * 64 + ("/" if variant == "slash_only" else "") + service.add_directory(name, metadata={}) + original_properties = FakeBlob.get_blob_properties + + def properties(blob): + result = original_properties(blob) + if variant == "missing_size": + result.metadata = {"hdi_isfolder": "true"} + del result.size + elif variant == "resource_type": + result.resource_type = "directory" + return result + + with patch.object(FakeBlob, "get_blob_properties", properties): + if variant == "resource_type": + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, [("personal-chat", name)]) + else: + with self.assertRaises(store_module.WorkflowResultIntegrityError): + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, []) + + def test_cleanup_directory_etag_change_is_not_ignored(self): + scope = store_module._scope(self.workflow, self.run_id) + store, _, service = self.make_store(blob=True) + name = store_module._run_blob_prefix(scope) + "a" * 64 + service.add_directory(name) + original_properties = FakeBlob.get_blob_properties + reads = [] + + def properties(blob): + result = original_properties(blob) + reads.append(blob.key) + if len(reads) == 2: + blob._row()["etag"] += "-changed" + return result + + with patch.object(FakeBlob, "get_blob_properties", properties): + with self.assertRaises(ResourceModifiedError): + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, []) + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, [("personal-chat", name)]) + + def test_cleanup_does_not_recursively_delete_new_directory_children(self): + scope = store_module._scope(self.workflow, self.run_id) + store, _, service = self.make_store(blob=True) + name = store_module._run_blob_prefix(scope) + "a" * 64 + service.add_directory(name) + child_key = ("personal-chat", name + "/unexpected.txt") + original_list = FakeBlobContainer.list_blobs + + def list_then_add_child(container, **kwargs): + yield from original_list(container, **kwargs) + service.records[child_key] = {"data": b"keep", "metadata": {}, "etag": "child"} + + with patch.object(FakeBlobContainer, "list_blobs", list_then_add_child): + with self.assertRaises(ResourceExistsError): + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, []) + self.assertIn(child_key, service.records) + + def test_cleanup_tolerates_a_directory_removed_after_listing(self): + scope = store_module._scope(self.workflow, self.run_id) + store, _, service = self.make_store(blob=True) + name = store_module._run_blob_prefix(scope) + "a" * 64 + service.add_directory(name) + original_list = FakeBlobContainer.list_blobs + + def list_then_remove_directory(container, **kwargs): + yield from original_list(container, **kwargs) + del service.records[("personal-chat", name)] + + with patch.object(FakeBlobContainer, "list_blobs", list_then_remove_directory): + store._delete_scoped_blobs(scope) + self.assertEqual(service.deletes, []) + def test_cleanup_removes_incomplete_cosmos_chunks(self): store, container, _ = self.make_store(chunk_size_bytes=16) container.fail_create_at = 3