diff --git a/scripts/migrate_jvnautosci_2591_arxiv_identity_workflow.py b/scripts/migrate_jvnautosci_2591_arxiv_identity_workflow.py index e803b3a55..5b5a19456 100644 --- a/scripts/migrate_jvnautosci_2591_arxiv_identity_workflow.py +++ b/scripts/migrate_jvnautosci_2591_arxiv_identity_workflow.py @@ -24,6 +24,16 @@ load_dotenv(_PROJECT_ROOT / ".env", override=False) from src.backend.security.access_control import override_current_actor # noqa: E402 +from src.backend.security.visibility_predicates import ( # noqa: E402 + get_specific_to_org_values, + get_specific_to_user_values, + set_specific_to_org_values, + set_specific_to_user_values, +) +from src.backend.services import concept_service # noqa: E402 +from src.backend.services.concept_service import ( # noqa: E402 + _find_raw_concept_by_exact_concept_id, +) from src.backend.workflows.vontology_loader import ( # noqa: E402 load_workflow_definition_from_vontology, ) @@ -37,6 +47,9 @@ apply_workflow_authoring_spec, preview_workflow_authoring_spec, ) +from src.backend.workflows.workflow_concept_authority_service import ( # noqa: E402 + workflow_child_visibility_covers_parent, +) WORKFLOW_ID = "#V#scholarly_article_metadata_representation_workflow" @@ -45,6 +58,7 @@ ENSURE_STATE_SUFFIX = "_ensure_arxiv_paper_concept" ENSURE_ACTION_ID = "scholarly_paper.ensure_paper_concept" DEFAULT_ACTOR_USER_ID = "#V#zhan_von_witbrock" +DEFAULT_VERIFICATION_ACTOR_USER_ID = "#V#michael_witbrock" DEFAULT_ACTOR_ORGANISATION_ID = "#V#university_of_auckland_strong_ai_lab" @@ -211,8 +225,98 @@ def _definition_identity(definition: Any) -> dict[str, Any]: ) +def _ensure_visibility_child_ids(authoring_spec: Mapping[str, Any]) -> tuple[str, ...]: + raw_steps = authoring_spec.get("steps") + steps = ( + [dict(step) for step in raw_steps if isinstance(step, Mapping)] + if isinstance(raw_steps, list) + else [] + ) + ensure_step = _step_with_suffix(steps, ENSURE_STATE_SUFFIX) + child_ids = [ + _clean_text(ensure_step.get("concept_id")) + or _clean_text(ensure_step.get("state_id")) + ] + raw_mappings = ensure_step.get("tool_output_context_mappings") + if isinstance(raw_mappings, list): + child_ids.extend( + _clean_text(mapping.get("mapping_concept_id")) + for mapping in raw_mappings + if isinstance(mapping, Mapping) + ) + return tuple(dict.fromkeys(child_id for child_id in child_ids if child_id)) + + +def repair_ensure_arxiv_visibility_closure( + authoring_spec: Mapping[str, Any], + *, + apply: bool, +) -> dict[str, Any]: + """Align the incident's three child artefacts with the root audience.""" + + workflow_doc = _find_raw_concept_by_exact_concept_id(WORKFLOW_ID) + if not isinstance(workflow_doc, Mapping): + raise ValueError(f"workflow_not_found:{WORKFLOW_ID}") + workflow_relationships = ( + dict(workflow_doc.get("relationships") or {}) + if isinstance(workflow_doc.get("relationships"), Mapping) + else {} + ) + parent_users = get_specific_to_user_values(workflow_relationships) + parent_orgs = get_specific_to_org_values(workflow_relationships) + repaired_ids: list[str] = [] + already_covered_ids: list[str] = [] + for child_id in _ensure_visibility_child_ids(authoring_spec): + child_doc = _find_raw_concept_by_exact_concept_id(child_id) + if not isinstance(child_doc, Mapping): + raise ValueError(f"workflow_child_not_found:{child_id}") + if workflow_child_visibility_covers_parent(workflow_doc, child_doc): + already_covered_ids.append(child_id) + continue + if apply: + child_relationships = ( + dict(child_doc.get("relationships") or {}) + if isinstance(child_doc.get("relationships"), Mapping) + else {} + ) + updated_relationships = set_specific_to_user_values( + child_relationships, + parent_users, + ) + updated_relationships = set_specific_to_org_values( + updated_relationships, + parent_orgs, + ) + concept_service.update_concept( + child_id, + {"relationships": updated_relationships}, + defer_side_effects=True, + ) + readback = _find_raw_concept_by_exact_concept_id(child_id) + if not workflow_child_visibility_covers_parent(workflow_doc, readback): + raise ValueError( + f"workflow_child_visibility_repair_readback_failed:{child_id}" + ) + repaired_ids.append(child_id) + return { + "schema_version": "workflow_visibility_closure_repair.v1", + "apply": apply, + "parent_user_scope": parent_users, + "parent_organisation_scope": parent_orgs, + "repair_required_ids": repaired_ids, + "already_covered_ids": already_covered_ids, + "closure_verified": bool( + apply or not repaired_ids + ), + } + + def run_migration( - *, apply: bool, actor_user_id: str, actor_organisation_id: str + *, + apply: bool, + actor_user_id: str, + actor_organisation_id: str, + verification_actor_user_id: str = DEFAULT_VERIFICATION_ACTOR_USER_ID, ) -> dict[str, Any]: with override_current_actor( user_concept_id=actor_user_id, @@ -226,6 +330,10 @@ def run_migration( candidate_spec, changed = ( rewrite_metadata_workflow_for_canonical_arxiv_identity(before_spec) ) + visibility_repair = repair_ensure_arxiv_visibility_closure( + candidate_spec, + apply=False, + ) base_hash = _clean_text(before_identity.get("definition_hash")) preview = preview_workflow_authoring_spec( WORKFLOW_ID, @@ -249,9 +357,16 @@ def run_migration( "contract_errors": validation.get("errors") or [], "contract_warnings": validation.get("warnings") or [], "diff_summary": (preview.get("preview") or {}).get("diff_summary"), + "visibility_repair": visibility_repair, } if not apply: return result + if not bool(validation.get("valid")): + raise ValueError("workflow_authoring_preview_invalid") + result["visibility_repair"] = repair_ensure_arxiv_visibility_closure( + candidate_spec, + apply=True, + ) apply_result = apply_workflow_authoring_spec( WORKFLOW_ID, @@ -283,6 +398,22 @@ def run_migration( "definition_hash" ) result["canonical_readback_verified"] = True + with override_current_actor( + user_concept_id=verification_actor_user_id, + organisation_concept_id=actor_organisation_id, + ): + verification_actor_definition = ( + load_workflow_definition_from_vontology(WORKFLOW_ID) + ) + if verification_actor_definition is None: + raise ValueError("workflow_missing_for_verification_actor") + if set(verification_actor_definition.states) != set(after.states): + raise ValueError( + "workflow_cross_actor_graph_closure_readback_mismatch" + ) + result["cross_actor_graph_closure_verified"] = True + result["verification_actor_user_id"] = verification_actor_user_id + result["verified_state_count"] = len(after.states) return result @@ -298,6 +429,10 @@ def main() -> None: "--actor-organisation-id", default=DEFAULT_ACTOR_ORGANISATION_ID, ) + parser.add_argument( + "--verification-actor-user-id", + default=DEFAULT_VERIFICATION_ACTOR_USER_ID, + ) args = parser.parse_args() print( json.dumps( @@ -305,6 +440,9 @@ def main() -> None: apply=bool(args.apply), actor_user_id=str(args.actor_user_id).strip(), actor_organisation_id=str(args.actor_organisation_id).strip(), + verification_actor_user_id=str( + args.verification_actor_user_id + ).strip(), ), indent=2, sort_keys=True, diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index a5bab45ec..ff5e5d4e8 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -18538,6 +18538,7 @@ def _workflow_create_instance(**kwargs): def _workflow_execute(**kwargs): """Launch a durable workflow instance and optionally await a terminal result.""" + from .transport import record_internal_mcp_effect_receipt from ...workflows.durable import WorkflowInstanceManager from ...workflows.durable.execution_observability import ( await_workflow_terminal_state, @@ -18685,6 +18686,25 @@ def _workflow_execute(**kwargs): else None ) if instance_id: + record_internal_mcp_effect_receipt( + { + "schema_version": "workflow_durable_submission_receipt.v1", + "success": bool(submission.success), + "status": "submitted", + "effect_status": "partial", + "changed": ( + submission.created_new + if isinstance(submission.created_new, bool) + else None + ), + "workflow_id": workflow_id.strip(), + "instance_id": instance_id, + "created_new": submission.created_new, + "durable_submission_status": submission.status, + "mutation_outcome": "partial", + "outcome_finality": "pending_durable_observation", + } + ) if await_terminal: wait_result = await_workflow_terminal_state( manager, diff --git a/src/backend/integrations/internal_mcp/transport.py b/src/backend/integrations/internal_mcp/transport.py index 8796501bc..aba708b83 100644 --- a/src/backend/integrations/internal_mcp/transport.py +++ b/src/backend/integrations/internal_mcp/transport.py @@ -78,6 +78,20 @@ "mutation_outcome", "partial_failures", ) +_INTERMEDIATE_EFFECT_RECEIPT_FIELDS = ( + "schema_version", + "success", + "status", + "effect_status", + "changed", + "workflow_id", + "instance_id", + "created_new", + "durable_submission_status", + "final_status", + "mutation_outcome", + "outcome_finality", +) def _positive_float_env(name: str, default: float) -> float: @@ -156,6 +170,7 @@ class InternalMCPExecutionScope: method_name: str deadline_monotonic: float cancellation_event: threading.Event + effect_receipt_recorder: Callable[[Mapping[str, Any]], None] | None = None @property def cancellation_requested(self) -> bool: @@ -253,6 +268,21 @@ def raise_if_internal_mcp_cancelled() -> None: ) +def record_internal_mcp_effect_receipt(receipt: Mapping[str, Any]) -> bool: + """Record bounded durable effect identity before a handler returns. + + This is intentionally an observation channel, not a way to extend the + handler deadline or rewrite a terminal turn result. Only handlers running + inside the internal MCP transport can publish a receipt. + """ + + scope = get_internal_mcp_execution_scope() + if scope is None or scope.effect_receipt_recorder is None: + return False + scope.effect_receipt_recorder(receipt) + return True + + @dataclass(frozen=True) class TransportResult: """Container for a terminal turn result and truthful transport timing.""" @@ -329,6 +359,21 @@ class _HandlerTask: admission_denied_before_start: bool = False admission_remaining_window_sec: float | None = None late_completion_recorded: bool = False + effect_receipt: dict[str, Any] | None = None + + def record_effect_receipt(self, receipt: Mapping[str, Any]) -> None: + bounded: dict[str, Any] = {} + for key in _INTERMEDIATE_EFFECT_RECEIPT_FIELDS: + if key not in receipt: + continue + value, truncated = _bounded_late_completion_value(receipt.get(key)) + if truncated: + continue + bounded[key] = value + if not bounded: + return + with self.lock: + self.effect_receipt = bounded def _invoke(self) -> Any: scope = InternalMCPExecutionScope( @@ -336,6 +381,7 @@ def _invoke(self) -> Any: method_name=self.method_name, deadline_monotonic=self.deadline_monotonic, cancellation_event=self.cancellation_event, + effect_receipt_recorder=self.record_effect_receipt, ) token = _ACTIVE_EXECUTION_SCOPE.set(scope) try: @@ -747,6 +793,7 @@ def _timeout_payload( queue_duration_ms: float, handler_elapsed_ms: float | None, late_result_policy: str = "discard_from_turn", + effect_receipt: Mapping[str, Any] | None = None, ) -> Dict[str, Any]: is_write = str(category or "").strip().lower() == "write" outcome_unknown = is_write and timeout_phase == "handler" @@ -776,7 +823,41 @@ def _timeout_payload( "outcome_finality": "terminal_for_turn", "late_result_policy": late_result_policy, } - if outcome_unknown: + durable_instance_id = ( + str((effect_receipt or {}).get("instance_id") or "").strip() + if isinstance(effect_receipt, Mapping) + else "" + ) + if outcome_unknown and durable_instance_id: + bounded_receipt = dict(effect_receipt or {}) + payload.update( + { + "error_code": "tool_timeout_after_durable_submission", + "retryable": False, + "effect_status": "partial", + "mutation_outcome": "partial", + "changed": ( + bounded_receipt.get("created_new") + if isinstance(bounded_receipt.get("created_new"), bool) + else None + ), + "workflow_id": bounded_receipt.get("workflow_id"), + "instance_id": durable_instance_id, + "created_new": bounded_receipt.get("created_new"), + "durable_submission_status": bounded_receipt.get( + "durable_submission_status" + ), + "durable_effect_receipt": bounded_receipt, + } + ) + payload["recovery_affordances"] = [ + { + "action_type": "inspect_workflow_instance", + "capability": "workflow_get_instance", + "arguments": {"instance_id": durable_instance_id}, + } + ] + elif outcome_unknown: payload["mutation_outcome"] = "unknown" payload["recovery_affordances"] = [ {"action_type": "inspect_operation_state_before_retry"} @@ -1086,6 +1167,11 @@ def execute( admission_remaining_window_sec = ( task.admission_remaining_window_sec ) + effect_receipt = ( + dict(task.effect_receipt) + if isinstance(task.effect_receipt, Mapping) + else None + ) queue_ms, handler_ms, handler_elapsed_ms = self._timing_snapshot( task, @@ -1172,6 +1258,7 @@ def execute( queue_duration_ms=queue_ms, handler_elapsed_ms=handler_elapsed_ms, late_result_policy=late_result_policy, + effect_receipt=effect_receipt, ) return TransportResult( payload=timeout_payload, diff --git a/src/backend/services/adaptive_turn_service.py b/src/backend/services/adaptive_turn_service.py index 3daf962f6..8109ff109 100644 --- a/src/backend/services/adaptive_turn_service.py +++ b/src/backend/services/adaptive_turn_service.py @@ -1782,6 +1782,7 @@ def _effect_result_target_ids(raw_payload: Any) -> list[str]: "created_concept_id", "existing_concept_id", "file_copy_concept_id", + "instance_id", "object_id", "relation_id", "source_concept_id", @@ -3232,6 +3233,43 @@ def enter_answer_only(reason: str) -> None: ), ) continue + if turn_id: + stable_launch_inputs = dict( + trusted_arguments.get("inputs") or {} + ) + # Tool results appended to the adaptive context must not + # change the identity of an otherwise identical retry. + stable_launch_inputs.pop("augmented_context", None) + idempotency_material = _json_bytes( + { + "turn_id": turn_id, + "workflow_id": str(workflow_capability.workflow_id), + "launch_inputs": stable_launch_inputs, + } + ) + trusted_arguments.update( + { + "source_event_type": "conversation_turn", + "source_event_id": turn_id, + "event_idempotency_key": ( + "conversation_turn_workflow:" + + hashlib.sha256(idempotency_material).hexdigest() + ), + } + ) + binding_diagnostics = { + **dict(binding_diagnostics or {}), + "durable_idempotency": { + "schema_version": ( + "workflow_turn_durable_idempotency.v1" + ), + "source_event_type": "conversation_turn", + "source_event_id": turn_id, + "event_idempotency_key_source": ( + "turn_workflow_inputs_fingerprint" + ), + }, + } else: trusted_arguments = _trusted_tool_payload( gateway=gateway, @@ -3818,6 +3856,11 @@ def contained( "mutation_outcome", "outcome_finality", "error_code", + "instance_id", + "workflow_id", + "durable_submission_status", + "final_status", + "created_new", ): receipt_value = raw_payload.get(receipt_key) if isinstance(receipt_value, (str, int, float, bool)): @@ -3865,6 +3908,11 @@ def contained( "mutation_outcome", "outcome_finality", "error_code", + "instance_id", + "workflow_id", + "durable_submission_status", + "final_status", + "created_new", ): receipt_value = raw_payload.get(receipt_key) if isinstance(receipt_value, (str, int, float, bool)): diff --git a/src/backend/services/turn_execution_record_service.py b/src/backend/services/turn_execution_record_service.py index c6bcf00dc..6942d58e3 100644 --- a/src/backend/services/turn_execution_record_service.py +++ b/src/backend/services/turn_execution_record_service.py @@ -5286,6 +5286,13 @@ def _summarise_tool_invocations( "timeout_phase", "effect_id", "effect_status", + "capability_kind", + "execution_method", + "represented_workflow_id", + "workflow_id", + "instance_id", + "durable_submission_status", + "final_status", ): identifier_value = _safe_str(invocation.get(identifier_key)) if identifier_value: diff --git a/src/backend/services/workflow_turn_capability_service.py b/src/backend/services/workflow_turn_capability_service.py index c38534efd..592f36058 100644 --- a/src/backend/services/workflow_turn_capability_service.py +++ b/src/backend/services/workflow_turn_capability_service.py @@ -602,9 +602,31 @@ def normalise_workflow_effect_receipt( status_key = (final_status or "").lower() timed_out = bool(receipt.get("timed_out")) created_new = receipt.get("created_new") - changed = bool(instance_id and created_new is not False) + incoming_mutation_outcome = ( + _normalise_non_empty_text(receipt.get("mutation_outcome")) or "" + ).lower() + incoming_error_code = ( + _normalise_non_empty_text(receipt.get("error_code")) or "" + ).lower() + explicit_not_started = bool( + status_key == "not_started" + or incoming_mutation_outcome == "not_started" + ) + explicit_indeterminate = bool( + incoming_mutation_outcome == "unknown" + or incoming_error_code == "tool_timeout_outcome_unknown" + ) + changed: bool | None = ( + created_new if instance_id is not None and isinstance(created_new, bool) else None + ) - if receipt.get("success") is False and instance_id is None: + if instance_id is None and explicit_not_started: + effect_status = "not_started" + changed = False + elif instance_id is None and explicit_indeterminate: + effect_status = "indeterminate" + changed = None + elif receipt.get("success") is False and instance_id is None: effect_status = "failed" changed = False elif status_key in _TERMINAL_SUCCESS_STATUSES and not timed_out: @@ -633,7 +655,15 @@ def normalise_workflow_effect_receipt( "mutation_outcome": ( "completed" if effect_status == "succeeded" - else ("partial" if changed else "not_started") + else ( + "partial" + if instance_id is not None + else ( + "unknown" + if effect_status == "indeterminate" + else "not_started" + ) + ) ), "outcome_finality": "terminal_for_turn", } @@ -651,7 +681,11 @@ def normalise_workflow_effect_receipt( else: recovery_affordances.append( { - "action_type": "inspect_launch_failure_before_retry", + "action_type": ( + "inspect_operation_state_before_retry" + if effect_status == "indeterminate" + else "inspect_launch_failure_before_retry" + ), "workflow_id": capability.workflow_id, } ) diff --git a/src/backend/workflows/workflow_concept_authority_service.py b/src/backend/workflows/workflow_concept_authority_service.py index 495f8395b..955ed2ea7 100644 --- a/src/backend/workflows/workflow_concept_authority_service.py +++ b/src/backend/workflows/workflow_concept_authority_service.py @@ -25,6 +25,12 @@ upsert_singleton_text_relation, upsert_text_for_concept, ) +from ..security.visibility_predicates import ( + get_specific_to_org_values, + get_specific_to_user_values, + set_specific_to_org_values, + set_specific_to_user_values, +) from .definitions import ( CHAT_ASSISTANT_WORKFLOW_ID, CHAT_BUTTONIFY_WORKFLOW_ID, @@ -2736,6 +2742,79 @@ def _normalise_relationships(concept_doc: Dict[str, Any] | None) -> Dict[str, An return dict(relationships) +def workflow_child_visibility_covers_parent( + parent_concept: Mapping[str, Any] | None, + child_concept: Mapping[str, Any] | None, +) -> bool: + """Return whether every actor who can see the parent can see the child. + + Visibility predicates are an OR across user and organisation scopes. The + conservative syntactic coverage check avoids guessing organisation + membership: each explicit parent audience must also appear in the + corresponding child audience. An unrestricted child covers any parent; + an unrestricted parent requires an unrestricted child. + """ + + parent_relationships = ( + parent_concept.get("relationships") + if isinstance(parent_concept, Mapping) + else {} + ) + child_relationships = ( + child_concept.get("relationships") + if isinstance(child_concept, Mapping) + else {} + ) + parent_users = set(get_specific_to_user_values(parent_relationships)) + parent_orgs = set(get_specific_to_org_values(parent_relationships)) + child_users = set(get_specific_to_user_values(child_relationships)) + child_orgs = set(get_specific_to_org_values(child_relationships)) + if not child_users and not child_orgs: + return True + if not parent_users and not parent_orgs: + return False + return parent_users.issubset(child_users) and parent_orgs.issubset(child_orgs) + + +def _inherit_workflow_visibility_for_new_child( + *, + workflow_doc: Mapping[str, Any], + child_concept_id: str, + child_doc: Mapping[str, Any] | None = None, +) -> tuple[dict[str, Any] | None, str | None]: + """Copy the containing workflow's exact visibility to a newly made child.""" + + loaded_child = dict(child_doc) if isinstance(child_doc, Mapping) else None + if loaded_child is None: + loaded_child, load_error = _load_concept(child_concept_id) + if load_error: + return None, f"visibility_lookup_failed:{load_error}" + if loaded_child is None: + return None, "visibility_child_missing" + + workflow_relationships = _normalise_relationships(dict(workflow_doc)) + child_relationships = _normalise_relationships(loaded_child) + inherited_relationships = set_specific_to_user_values( + child_relationships, + get_specific_to_user_values(workflow_relationships), + ) + inherited_relationships = set_specific_to_org_values( + inherited_relationships, + get_specific_to_org_values(workflow_relationships), + ) + if inherited_relationships != child_relationships: + try: + concept_service.update_concept( + child_concept_id, + {"relationships": inherited_relationships}, + defer_side_effects=True, + ) + except Exception as exc: # pragma: no cover - defensive + return None, f"visibility_update_failed:{exc}" + loaded_child["relationships"] = inherited_relationships + return loaded_child, None + + def _strip_relationship_aliases( relationships: Dict[str, Any], *, @@ -3478,6 +3557,20 @@ def _resolve_workflow_definition_for_validation( step_update_failed = True break if step_created: + step_doc, visibility_error = ( + _inherit_workflow_visibility_for_new_child( + workflow_doc=workflow_doc, + child_concept_id=step_concept_id, + child_doc=step_doc, + ) + ) + if visibility_error: + errors_by_workflow_id[workflow_id] = ( + "step_visibility_inheritance_failed:" + f"{step_concept_id}:{visibility_error}" + ) + step_update_failed = True + break created_step_ids.append(step_concept_id) step_relationships = _strip_relationship_aliases( @@ -3711,6 +3804,19 @@ def _resolve_workflow_definition_for_validation( step_update_failed = True break if created_mapping: + _mapping_doc, visibility_error = ( + _inherit_workflow_visibility_for_new_child( + workflow_doc=workflow_doc, + child_concept_id=mapping_spec.concept_id, + ) + ) + if visibility_error: + errors_by_workflow_id[workflow_id] = ( + "context_mapping_visibility_inheritance_failed:" + f"{mapping_spec.concept_id}:{visibility_error}" + ) + step_update_failed = True + break created_mapping_concept_ids.append(mapping_spec.concept_id) if step_update_failed: break @@ -3750,6 +3856,19 @@ def _resolve_workflow_definition_for_validation( step_update_failed = True break if created_mapping: + _mapping_doc, visibility_error = ( + _inherit_workflow_visibility_for_new_child( + workflow_doc=workflow_doc, + child_concept_id=mapping_concept_id, + ) + ) + if visibility_error: + errors_by_workflow_id[workflow_id] = ( + "mapping_visibility_inheritance_failed:" + f"{mapping_concept_id}:{visibility_error}" + ) + step_update_failed = True + break created_mapping_concept_ids.append(mapping_concept_id) if step_update_failed: break diff --git a/src/backend/workflows/workflow_studio_service.py b/src/backend/workflows/workflow_studio_service.py index 315f3d85e..03d0eeaa7 100644 --- a/src/backend/workflows/workflow_studio_service.py +++ b/src/backend/workflows/workflow_studio_service.py @@ -76,6 +76,7 @@ publish_workflow_definition_from_definition, upsert_workflow_json_policy_text, upsert_workflow_publication_lifecycle, + workflow_child_visibility_covers_parent, WORKFLOW_BACKGROUND_LAUNCH_POLICY_TEXT_PREDICATE, WORKFLOW_DISCOVERY_EXEMPLARS_TEXT_PREDICATE, WORKFLOW_LAUNCH_INPUT_CONTRACT_TEXT_PREDICATE, @@ -210,10 +211,10 @@ def _workflow_definition_loader(workflow_id: str): return None -def _actor_visible_concept_exists(concept_id: str) -> bool: +def _actor_visible_raw_concept(concept_id: str) -> dict[str, Any] | None: concept_id_clean = _clean_text(concept_id) if not concept_id_clean: - return False + return None try: from ..services.concept_service import _find_raw_concept_by_exact_concept_id @@ -232,10 +233,10 @@ def _actor_visible_concept_exists(concept_id: str) -> bool: "workflow_concept_existence_authority_unavailable" ) from exc if raw_concept is None: - return False + return None try: if can_access_concept(concept_id_clean): - return True + return dict(raw_concept) except Exception as exc: logger.warning( "workflow studio concept visibility authority unavailable for %s", @@ -250,6 +251,10 @@ def _actor_visible_concept_exists(concept_id: str) -> bool: ) +def _actor_visible_concept_exists(concept_id: str) -> bool: + return _actor_visible_raw_concept(concept_id) is not None + + def _workflow_concept_exists(workflow_id: str) -> bool: return _actor_visible_concept_exists(workflow_id) @@ -2132,15 +2137,31 @@ def _append(value: Any, *, field: str, require_concept_id: bool = True) -> None: def _preflight_explicit_authored_concept_ids( + workflow_id: str, authoring_spec: Mapping[str, Any], ) -> None: - """Fail before publication when an authored child ID is hidden or unknown.""" + """Fail before publication for hidden or audience-narrower authored children.""" - for concept_id in _collect_explicit_authored_concept_ids(authoring_spec): + concept_ids = _collect_explicit_authored_concept_ids(authoring_spec) + if not concept_ids: + return + workflow_doc = _actor_visible_raw_concept(workflow_id) + for concept_id in concept_ids: # A provably absent child may be created by canonical publication. # Existing children must be visible to the ambient actor; the helper # raises a concealment-safe authority error when they are not. - _actor_visible_concept_exists(concept_id) + child_doc = _actor_visible_raw_concept(concept_id) + if ( + workflow_doc is not None + and child_doc is not None + and not workflow_child_visibility_covers_parent( + workflow_doc, + child_doc, + ) + ): + raise WorkflowStudioAuthorityError( + "workflow_child_visibility_narrower_than_parent" + ) def apply_workflow_authoring_spec( @@ -2171,7 +2192,7 @@ def apply_workflow_authoring_spec( create_missing_workflow = bool( runtime_definition is None and _runtime_source == "new_workflow" ) - _preflight_explicit_authored_concept_ids(authoring_spec) + _preflight_explicit_authored_concept_ids(workflow_id, authoring_spec) publication = publish_workflow_definition_from_definition( definition=definition, create_missing=create_missing_workflow, diff --git a/tests/backend/test_adaptive_turn_service.py b/tests/backend/test_adaptive_turn_service.py index 69e7f4ead..865ced0d1 100644 --- a/tests/backend/test_adaptive_turn_service.py +++ b/tests/backend/test_adaptive_turn_service.py @@ -348,6 +348,9 @@ def _workflow_gateway(handler: Any) -> InternalMCPGateway: "poll_interval_seconds": (int, float), "include_step_result_envelopes": bool, "include_trace": bool, + "source_event_type": str, + "source_event_id": str, + "event_idempotency_key": str, }, allow_unknown=False, ), @@ -4418,6 +4421,11 @@ def _execute_workflow(**kwargs: Any) -> dict[str, Any]: assert seen_arguments["inputs"]["user_concept_id"] == "#V#real_user" assert seen_arguments["inputs"]["record_id"] == "#V#record" assert seen_arguments["inputs"]["authorised_record"] == "#V#request_record" + assert seen_arguments["source_event_type"] == "conversation_turn" + assert seen_arguments["source_event_id"] == "turn-represented-workflow" + assert seen_arguments["event_idempotency_key"].startswith( + "conversation_turn_workflow:" + ) invocation = next( item for item in result.tool_invocations @@ -4431,9 +4439,132 @@ def _execute_workflow(**kwargs: Any) -> dict[str, Any]: ) assert invocation["effect_status"] == "succeeded" assert invocation["changed"] is True + assert invocation["instance_id"] == "workflow-instance-1" + assert invocation["workflow_id"] == "#V#represented_test_workflow" assert result.response_text == "The represented work product was completed." +def test_represented_workflow_retry_reuses_same_turn_idempotency_key( + monkeypatch, +) -> None: + from src.backend.services.workflow_turn_capability_service import ( + WorkflowTurnCapability, + ) + + workflow_capability = WorkflowTurnCapability( + name="represented_workflow_retry_test", + workflow_id="#V#represented_retry_workflow", + display_name="Represented retry workflow", + description="Produce one durable work product.", + relevance_score=0.95, + input_schema={ + "type": "object", + "properties": { + "inputs": { + "type": "object", + "properties": {"record_id": {"type": "string"}}, + } + }, + "additionalProperties": False, + }, + ) + monkeypatch.setattr( + "src.backend.services.workflow_turn_capability_service." + "discover_turn_workflow_capabilities", + lambda *_args, **_kwargs: ( + [workflow_capability], + { + "schema_version": "workflow_turn_capability_discovery.v1", + "status": "completed", + "match_count": 1, + }, + ), + ) + idempotency_keys: list[str] = [] + + def _execute_workflow(**kwargs: Any) -> dict[str, Any]: + key = str(kwargs["event_idempotency_key"]) + idempotency_keys.append(key) + return { + "success": True, + "instance_id": "workflow-instance-reused", + "created_new": len(idempotency_keys) == 1, + "final_status": "running", + "timed_out": True, + } + + repeated_call = { + "name": workflow_capability.name, + "arguments": {"inputs": {"record_id": "#V#record"}}, + } + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_capabilities", + call_id="discover-retry-workflow", + payload={"query": "produce the durable work product"}, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="invoke-retry-workflow-1", + payload=repeated_call, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="invoke-retry-workflow-2", + payload=repeated_call, + ) + ], + ), + LLMResponse( + text_response="The durable workflow is still running.", + ), + ) + + result = execute_adaptive_turn( + gateway=_workflow_gateway(_execute_workflow), + prompt="Produce the durable work product.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#real_user@real_org", + user_concept_id="#V#real_user", + org_concept_id="#V#real_org", + turn_id="turn-represented-workflow-retry", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + assert len(idempotency_keys) == 2 + assert idempotency_keys[0] == idempotency_keys[1] + workflow_invocations = [ + invocation + for invocation in result.tool_invocations + if invocation.get("tool") == workflow_capability.name + ] + assert [item["instance_id"] for item in workflow_invocations] == [ + "workflow-instance-reused", + "workflow-instance-reused", + ] + assert [item["changed"] for item in workflow_invocations] == [True, False] + assert [item["effect_status"] for item in workflow_invocations] == [ + "partial", + "partial", + ] + + def test_unchanged_terminally_failed_effect_is_not_dispatched_twice( monkeypatch, ) -> None: diff --git a/tests/backend/test_internal_mcp_transport_deadlines.py b/tests/backend/test_internal_mcp_transport_deadlines.py index acb3cffcd..e116b413d 100644 --- a/tests/backend/test_internal_mcp_transport_deadlines.py +++ b/tests/backend/test_internal_mcp_transport_deadlines.py @@ -21,6 +21,7 @@ InternalMCPTransport, get_internal_mcp_execution_scope, internal_mcp_cancellation_requested, + record_internal_mcp_effect_receipt, raise_if_internal_mcp_cancelled, ) @@ -836,6 +837,55 @@ def test_write_timeout_is_explicitly_indeterminate_and_not_retryable() -> None: ] +def test_write_timeout_preserves_intermediate_durable_effect_receipt() -> None: + transport = InternalMCPTransport( + write_timeout_sec=0.03, + write_advisory_timeout_sec=0.01, + ) + + def _handler(): + assert record_internal_mcp_effect_receipt( + { + "schema_version": "workflow_durable_submission_receipt.v1", + "workflow_id": "#V#test_workflow", + "instance_id": "instance-durable-1", + "created_new": True, + "durable_submission_status": "submitted", + "secret": "must-not-be-projected", + } + ) + time.sleep(0.1) + + gateway = _gateway_for( + method_name="synthetic_durable_slow_write", + handler=_handler, + transport=transport, + category="write", + ) + + result = gateway.invoke("synthetic_durable_slow_write", {}) + + assert result.outcome == "timed_out" + assert result.payload["error_code"] == "tool_timeout_after_durable_submission" + assert result.payload["effect_status"] == "partial" + assert result.payload["mutation_outcome"] == "partial" + assert result.payload["changed"] is True + assert result.payload["workflow_id"] == "#V#test_workflow" + assert result.payload["instance_id"] == "instance-durable-1" + assert result.payload["durable_submission_status"] == "submitted" + assert result.payload["durable_effect_receipt"]["instance_id"] == ( + "instance-durable-1" + ) + assert "secret" not in result.payload["durable_effect_receipt"] + assert result.payload["recovery_affordances"] == [ + { + "action_type": "inspect_workflow_instance", + "capability": "workflow_get_instance", + "arguments": {"instance_id": "instance-durable-1"}, + } + ] + + def test_dispatched_write_reports_one_bounded_late_completion_observation() -> None: observations = [] observation_seen = Event() diff --git a/tests/backend/test_internal_mcp_workflow_tools.py b/tests/backend/test_internal_mcp_workflow_tools.py index 0f6f8b02b..8d130f0c4 100644 --- a/tests/backend/test_internal_mcp_workflow_tools.py +++ b/tests/backend/test_internal_mcp_workflow_tools.py @@ -1,5 +1,6 @@ from datetime import datetime, timedelta, timezone import json +import time from types import SimpleNamespace from typing import Any from unittest.mock import patch @@ -82,6 +83,7 @@ def _fake_submit_verified_workflow_instance(**kwargs): ) status = "created" if created_new else "reused" else: + created_new = True instance_id = manager.create_instance( workflow_id, user_id=user_id, @@ -109,6 +111,7 @@ def _fake_submit_verified_workflow_instance(**kwargs): "preflight": {"errors": []}, "postflight": {"errors": []}, }, + created_new=created_new, ) monkeypatch.setattr( @@ -1803,6 +1806,62 @@ def _get_instance(instance_id: str): assert boolean_fields["include_trace"]["raw_type"] == "int" +def test_workflow_execute_outer_timeout_preserves_durable_instance_receipt( + monkeypatch, +): + manager = _StubWorkflowManager() + _patch_submit_verified_instance_success(monkeypatch) + monkeypatch.setattr( + "src.backend.workflows.durable.WorkflowInstanceManager", + lambda: manager, + ) + + def _slow_await(*_args, **_kwargs): + time.sleep(0.1) + return SimpleNamespace( + instance=None, + poll_count=1, + timed_out=True, + ) + + monkeypatch.setattr( + "src.backend.workflows.durable.execution_observability." + "await_workflow_terminal_state", + _slow_await, + ) + gateway = _build_gateway() + + result = gateway.invoke( + "workflow_execute", + { + "workflow_id": "#V#meeting_invitation_testing_workflow", + "user_id": "#V#user", + "org_id": "#V#org", + "namespace": "#V#user@org", + "inputs": {"fixture_id": "fixture-timeout"}, + "await_terminal": True, + "timeout_seconds": 5, + "poll_interval_seconds": 0, + }, + deadline_monotonic=time.monotonic() + 0.03, + ) + + assert result.outcome == "timed_out" + assert result.payload["error_code"] == ( + "tool_timeout_after_durable_submission" + ) + assert result.payload["effect_status"] == "partial" + assert result.payload["mutation_outcome"] == "partial" + assert result.payload["changed"] is True + assert result.payload["workflow_id"] == ( + "#V#meeting_invitation_testing_workflow" + ) + assert result.payload["instance_id"] in manager.instances + assert result.payload["durable_effect_receipt"]["instance_id"] == ( + result.payload["instance_id"] + ) + + def test_workflow_execute_forwards_exact_required_worker_build(monkeypatch): from src.backend.workflows.durable.workflow_instance_submission_service import ( WorkflowInstanceSubmissionResult, diff --git a/tests/backend/test_turn_execution_record_service_execution_summary.py b/tests/backend/test_turn_execution_record_service_execution_summary.py index a422385b2..ce627f7d6 100644 --- a/tests/backend/test_turn_execution_record_service_execution_summary.py +++ b/tests/backend/test_turn_execution_record_service_execution_summary.py @@ -250,6 +250,43 @@ def test_partial_and_indeterminate_effects_remain_unsafe( ] +def test_partial_represented_workflow_preserves_durable_identity_in_projection() -> None: + record = _build_effect_projection_record( + "req-partial-represented-workflow", + [ + { + "tool": "represented_workflow_turn_test", + "status": "error", + "capability_kind": "represented_workflow", + "execution_method": "workflow_execute", + "represented_workflow_id": "#V#represented_test_workflow", + "workflow_id": "#V#represented_test_workflow", + "instance_id": "workflow-instance-durable-1", + "durable_submission_status": "created", + "effect_id": "effect_represented_partial", + "effect_status": "partial", + "changed": True, + "mutation_outcome": "partial", + "outcome_finality": "terminal_for_turn", + "error_code": "tool_timeout_after_durable_submission", + } + ], + ) + + serialised = record["execution"]["tool_invocations"][0] + assert serialised["status"] == "partial" + assert serialised["capability_kind"] == "represented_workflow" + assert serialised["execution_method"] == "workflow_execute" + assert serialised["represented_workflow_id"] == ( + "#V#represented_test_workflow" + ) + assert serialised["workflow_id"] == "#V#represented_test_workflow" + assert serialised["instance_id"] == "workflow-instance-durable-1" + assert serialised["durable_submission_status"] == "created" + assert record["execution"]["summary"]["invocation_count"] == 1 + assert record["completion_gate"]["safe_to_claim_completion"] is False + + def test_successful_effect_with_canonical_readback_remains_completable() -> None: record = _build_effect_projection_record( "req-successful-effect-readback", diff --git a/tests/backend/test_workflow_concept_authority_service.py b/tests/backend/test_workflow_concept_authority_service.py index 172857fed..c781beb36 100644 --- a/tests/backend/test_workflow_concept_authority_service.py +++ b/tests/backend/test_workflow_concept_authority_service.py @@ -77,6 +77,95 @@ def _mock_load_concept(concept_id: str): assert "#V#wf_untyped" in report["missing_required_type_by_workflow_id"] +def test_workflow_child_visibility_requires_parent_audience_coverage() -> None: + global_parent = {"relationships": {}} + user_parent = { + "relationships": {"#V#specific_to_user": ["#V#researcher"]} + } + org_parent = { + "relationships": { + "#V#specific_to_organisation": ["#V#research_lab"] + } + } + global_child = {"relationships": {}} + researcher_child = { + "relationships": {"#V#specific_to_user": ["#V#researcher"]} + } + other_researcher_child = { + "relationships": {"#V#specific_to_user": ["#V#other_researcher"]} + } + lab_child = { + "relationships": { + "#V#specific_to_organisation": ["#V#research_lab"] + } + } + + assert authority_service.workflow_child_visibility_covers_parent( + global_parent, global_child + ) + assert not authority_service.workflow_child_visibility_covers_parent( + global_parent, researcher_child + ) + assert authority_service.workflow_child_visibility_covers_parent( + user_parent, global_child + ) + assert authority_service.workflow_child_visibility_covers_parent( + user_parent, researcher_child + ) + assert not authority_service.workflow_child_visibility_covers_parent( + user_parent, other_researcher_child + ) + assert authority_service.workflow_child_visibility_covers_parent( + org_parent, lab_child + ) + + +def test_new_workflow_child_inherits_exact_parent_visibility(monkeypatch) -> None: + updates: list[tuple[str, dict[str, Any]]] = [] + workflow_doc = { + "concept_id": "#V#workflow", + "relationships": { + "#V#specific_to_user": ["#V#author"], + "#V#specific_to_organisation": ["#V#lab"], + "#V#hasStep": ["#V#step"], + }, + } + child_doc = { + "concept_id": "#V#step", + "relationships": { + "#V#specific_to_user": ["#V#author"], + "#V#is_an_instance_of": ["#V#workflow_step"], + }, + } + monkeypatch.setattr( + authority_service.concept_service, + "update_concept", + lambda concept_id, payload, **_kwargs: updates.append( + (concept_id, payload) + ), + ) + + updated, error = authority_service._inherit_workflow_visibility_for_new_child( + workflow_doc=workflow_doc, + child_concept_id="#V#step", + child_doc=child_doc, + ) + + assert error is None + assert updated is not None + assert updated["relationships"]["#V#specific_to_user"] == ["#V#author"] + assert updated["relationships"]["#V#specific_to_organisation"] == ["#V#lab"] + assert updated["relationships"]["#V#is_an_instance_of"] == [ + "#V#workflow_step" + ] + assert updates == [ + ( + "#V#step", + {"relationships": updated["relationships"]}, + ) + ] + + def test_workflow_authority_report_accepts_required_type_via_subtype(monkeypatch): registry = _DummyRegistry(workflow_ids=["#V#wf_durable"]) diff --git a/tests/backend/test_workflow_studio_service.py b/tests/backend/test_workflow_studio_service.py index 6914bc8aa..ab2015825 100644 --- a/tests/backend/test_workflow_studio_service.py +++ b/tests/backend/test_workflow_studio_service.py @@ -350,6 +350,48 @@ def test_apply_workflow_authoring_spec_updates_existing_workflow_without_root_cr assert publication_calls[0]["purpose"] == "Updated description" +def test_authoring_preflight_rejects_child_visibility_narrower_than_workflow( + monkeypatch, +) -> None: + workflow_id = "#V#public_workflow" + child_id = "#V#private_step" + concepts = { + workflow_id: { + "concept_id": workflow_id, + "relationships": {}, + }, + child_id: { + "concept_id": child_id, + "relationships": { + "#V#specific_to_user": ["#V#studio_author"] + }, + }, + } + monkeypatch.setattr( + mod, + "_actor_visible_raw_concept", + lambda concept_id: concepts.get(concept_id), + ) + + with pytest.raises( + mod.WorkflowStudioAuthorityError, + match="workflow_child_visibility_narrower_than_parent", + ): + mod._preflight_explicit_authored_concept_ids( + workflow_id, + { + "workflow_id": workflow_id, + "steps": [ + { + "state_id": child_id, + "concept_id": child_id, + "terminal": True, + } + ], + }, + ) + + def test_apply_workflow_authoring_spec_creates_missing_workflow_when_absent( monkeypatch, ) -> None: diff --git a/tests/backend/test_workflow_turn_capability_service.py b/tests/backend/test_workflow_turn_capability_service.py index f8c7d1b82..5cd096362 100644 --- a/tests/backend/test_workflow_turn_capability_service.py +++ b/tests/backend/test_workflow_turn_capability_service.py @@ -225,6 +225,27 @@ def test_workflow_receipt_distinguishes_completion_partial_failure_and_no_start( }, capability=capability, ) + indeterminate = normalise_workflow_effect_receipt( + { + "success": False, + "status": "timed_out", + "error_code": "tool_timeout_outcome_unknown", + "mutation_outcome": "unknown", + }, + capability=capability, + ) + durable_timeout = normalise_workflow_effect_receipt( + { + "success": False, + "status": "timed_out", + "error_code": "tool_timeout_after_durable_submission", + "mutation_outcome": "partial", + "instance_id": "instance-4", + "created_new": False, + "workflow_id": "#V#test_workflow", + }, + capability=capability, + ) assert completed["effect_status"] == "succeeded" assert completed["changed"] is True @@ -241,3 +262,15 @@ def test_workflow_receipt_distinguishes_completion_partial_failure_and_no_start( assert no_start["effect_status"] == "failed" assert no_start["changed"] is False assert no_start["mutation_outcome"] == "not_started" + assert indeterminate["effect_status"] == "indeterminate" + assert indeterminate["changed"] is None + assert indeterminate["mutation_outcome"] == "unknown" + assert indeterminate["recovery_affordances"][-1]["action_type"] == ( + "inspect_operation_state_before_retry" + ) + assert durable_timeout["effect_status"] == "partial" + assert durable_timeout["changed"] is False + assert durable_timeout["mutation_outcome"] == "partial" + assert durable_timeout["recovery_affordances"][0]["arguments"] == { + "instance_id": "instance-4" + } diff --git a/tests/test_migrate_jvnautosci_2591_arxiv_identity_workflow.py b/tests/test_migrate_jvnautosci_2591_arxiv_identity_workflow.py index a3ae14ec1..88f1bb59e 100644 --- a/tests/test_migrate_jvnautosci_2591_arxiv_identity_workflow.py +++ b/tests/test_migrate_jvnautosci_2591_arxiv_identity_workflow.py @@ -89,3 +89,52 @@ def test_rewrite_is_idempotent() -> None: assert changed is True assert changed_again is False assert second == first + + +def test_visibility_repair_aligns_ensure_step_and_mappings_with_root( + monkeypatch, +) -> None: + rewritten, _changed = ( + mod.rewrite_metadata_workflow_for_canonical_arxiv_identity(_base_spec()) + ) + child_ids = mod._ensure_visibility_child_ids(rewritten) + documents = { + mod.WORKFLOW_ID: { + "concept_id": mod.WORKFLOW_ID, + "relationships": {}, + }, + **{ + child_id: { + "concept_id": child_id, + "relationships": { + "#V#specific_to_user": [mod.DEFAULT_ACTOR_USER_ID] + }, + } + for child_id in child_ids + }, + } + monkeypatch.setattr( + mod, + "_find_raw_concept_by_exact_concept_id", + lambda concept_id: documents.get(concept_id), + ) + + def _update(concept_id, payload, **_kwargs): + documents[concept_id]["relationships"] = payload["relationships"] + + monkeypatch.setattr(mod.concept_service, "update_concept", _update) + + preview = mod.repair_ensure_arxiv_visibility_closure( + rewritten, + apply=False, + ) + applied = mod.repair_ensure_arxiv_visibility_closure( + rewritten, + apply=True, + ) + + assert set(preview["repair_required_ids"]) == set(child_ids) + assert preview["closure_verified"] is False + assert set(applied["repair_required_ids"]) == set(child_ids) + assert applied["closure_verified"] is True + assert all(documents[child_id]["relationships"] == {} for child_id in child_ids)