Fix workflow visibility closure and durable effect receipts

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,13 +429,20 @@ 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(
             run_migration(
                 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,

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,

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,13 +359,29 @@ 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(
             execution_id=self.execution_id,
             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,