diff --git a/application/single_app/action_credential_contracts.py b/application/single_app/action_credential_contracts.py new file mode 100644 index 000000000..be4c29787 --- /dev/null +++ b/application/single_app/action_credential_contracts.py @@ -0,0 +1,173 @@ +# action_credential_contracts.py +"""Bootstrap-free personal action credential contracts and connector adapters.""" + +from copy import deepcopy +import hashlib +import json +from urllib.parse import urlsplit + +from functions_action_manifest import get_action_origin, resolve_action_type +from functions_yamcs_operations import ( + normalize_yamcs_auth_method, + normalize_yamcs_server_url, + yamcs_basic_auth_conflicts_with_auth_method, +) + + +class ActionCredentialError(RuntimeError): + """A stable, data-free credential failure.""" + + def __init__(self, code="credentials_unavailable", status_code=409): + self.code = code + self.status_code = status_code + super().__init__("Action credentials are unavailable. Refresh the credential request.") + + +class ActionCredentialsRequired(ActionCredentialError): + """A connector has stopped before model interpretation of an authentication failure.""" + + def __init__(self, request): + super().__init__("action_credentials_required") + self.credential_request = request + + @property + def event(self): + return {"type": "action_credentials_required", "credential_request": self.credential_request} + + +def find_credential_stop(error): + seen = set() + while error is not None and id(error) not in seen: + if isinstance(error, ActionCredentialsRequired): + return error + seen.add(id(error)) + error = error.__cause__ or error.__context__ + return None + + +class YamcsCredentialAdapter: + slots = {"service": {"username_password", "api_key", "bearer_token"}, "proxy": {"username_password"}} + + @staticmethod + def validate(manifest, slots): + auth = manifest.get("auth") or {} + fields = manifest.get("additionalFields") or {} + service = slots.get("service") + if service and any(( + manifest.get("identity_id"), fields.get("identity_id"), auth.get("key"), + auth.get("identity"), auth.get("tenantId"), fields.get("api_key"), + fields.get("password"), fields.get("access_token"), fields.get("token"), + )): + raise ValueError("Personal service credentials cannot include a shared credential or identity.") + if "proxy" in slots and any(fields.get(key) for key in ( + "basic_auth_identity_id", "basic_auth_username", "basic_auth_password", + )): + raise ValueError("Personal proxy credentials cannot include a shared credential or identity.") + method = service["auth_type"] if service else normalize_yamcs_auth_method(fields, auth.get("type")) + if ("proxy" in slots or fields.get("enable_basic_auth")) and yamcs_basic_auth_conflicts_with_auth_method(method): + raise ValueError("Yamcs proxy Basic authentication requires service none or API key authentication.") + + @staticmethod + def destination(manifest): + fields = manifest.get("additionalFields") or {} + endpoint = normalize_yamcs_server_url(manifest.get("endpoint") or fields.get("server_url")) + target = urlsplit(endpoint) + if target.scheme not in {"http", "https"} or not target.hostname or target.username or target.password: + raise ValueError("Yamcs requires a valid server destination without URL credentials.") + return { + "endpoint": endpoint, "instance": fields.get("instance") or fields.get("yamcs_instance"), + "tls_verify": fields.get("tls_verify", True), + "service_auth": normalize_yamcs_auth_method(fields, (manifest.get("auth") or {}).get("type")), + "proxy_enabled": bool(fields.get("enable_basic_auth")), + } + + @staticmethod + def inject(manifest, slot, auth): + fields = manifest.setdefault("additionalFields", {}) + if slot == "proxy": + fields.update( + enable_basic_auth=True, basic_auth_username=auth["username"], + basic_auth_password=auth["password"], + ) + else: + method = auth["auth_type"] + fields["auth_method"] = method + manifest["auth"] = { + "type": "username_password" if method == "username_password" else "key", + "key": auth["password"] if method == "username_password" else auth["secret"], + "identity": auth.get("username", ""), + } + + @staticmethod + def rejected_slots(manifest, error): + """Use provider response status only, never exception text.""" + response = getattr(error, "response", None) + status = getattr(response, "status_code", None) + # yamcs-client HttpError retains its numeric HTTP status on status_code. + if status is None: + status = getattr(error, "status_code", None) + slots = credential_slots(manifest) + if status == 407: + return ["proxy"] if "proxy" in slots else [] + if status != 401: + return [] + if len(slots) == 1: + return list(slots) + # An API-key service behind a Basic reverse proxy cannot disambiguate a 401. + return list(slots) + + +ADAPTERS = {"yamcs": YamcsCredentialAdapter()} + + +def credential_slots(manifest): + definition = manifest.get("credential_requirements") + if definition is None: + return {} + if ( + type(definition) is not dict or set(definition) != {"version", "slots"} + or type(definition["version"]) is not int or definition["version"] != 1 + or type(definition["slots"]) is not dict + ): + raise ValueError("Unsupported action credential requirements.") + if not definition["slots"]: + return {} + adapter = ADAPTERS.get(resolve_action_type(manifest).lower()) + if adapter is None: + raise ValueError("Personal credentials are not supported for this action type.") + for slot, requirement in definition["slots"].items(): + if ( + slot not in adapter.slots or type(requirement) is not dict + or set(requirement) != {"source", "auth_type"} or requirement["source"] != "personal" + or requirement["auth_type"] not in adapter.slots[slot] + ): + raise ValueError("Unsupported action credential slot.") + return deepcopy(definition["slots"]) + + +def validate_credential_requirements(manifest): + slots = credential_slots(manifest) + if slots: + adapter = ADAPTERS[resolve_action_type(manifest).lower()] + adapter.validate(manifest, slots) + adapter.destination(manifest) + return slots + + +def credential_fingerprint(manifest): + origin = get_action_origin(manifest) + if origin is None or not origin.action_id: + raise ActionCredentialError("trusted_origin_required") + slots = validate_credential_requirements(manifest) + payload = { + "origin": [origin.scope_type, origin.scope_id, origin.action_id], + "slots": slots, + "destination": ADAPTERS[resolve_action_type(manifest).lower()].destination(manifest), + } + return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def destination_label(manifest): + adapter = ADAPTERS[resolve_action_type(manifest).lower()] + target = urlsplit(adapter.destination(manifest)["endpoint"]) + return f"{target.hostname}:{target.port}" if target.port else target.hostname diff --git a/application/single_app/action_credential_deadline.py b/application/single_app/action_credential_deadline.py new file mode 100644 index 000000000..c91ce44ba --- /dev/null +++ b/application/single_app/action_credential_deadline.py @@ -0,0 +1,38 @@ +# action_credential_deadline.py +"""Account for human credential waits without changing retained deadline guards.""" + +from datetime import datetime, timedelta, timezone +from collections.abc import Mapping +import math + + +MAX_HUMAN_WAIT_SECONDS = 86400 + + +def execution_deadline(value, *, now=None): + getter = value.get if isinstance(value, Mapping) else lambda key, default=None: getattr(value, key, default) + deadline = datetime.fromisoformat(getter("execution_deadline_at")) + if deadline.tzinfo is None: + raise ValueError("An aware execution deadline is required.") + seconds = getter("credential_wait_seconds", 0) + if type(seconds) not in (int, float) or not math.isfinite(seconds) or not 0 <= seconds <= MAX_HUMAN_WAIT_SECONDS: + raise ValueError("Invalid credential wait accounting.") + started = getter("credential_wait_started_at") + if started and getter("status") == "waiting": + waits = getter("pending_results", {}) or {} + if any(wait.get("kind") == "action_credentials" for wait in waits.values()): + moment = datetime.fromisoformat(started) + if moment.tzinfo is None: + raise ValueError("An aware credential wait timestamp is required.") + seconds = min(MAX_HUMAN_WAIT_SECONDS, seconds + max(0, ((now or datetime.now(timezone.utc)) - moment).total_seconds())) + return deadline + timedelta(seconds=seconds) + + +def finish_human_wait(record, *, now): + """Only the owned continuation CAS admits this server-recorded clock adjustment.""" + deadline = execution_deadline(record, now=now) + original = datetime.fromisoformat(record["execution_deadline_at"]) + return { + "credential_wait_seconds": (deadline - original).total_seconds(), + "credential_wait_started_at": None, + } diff --git a/application/single_app/agent_delegation_runtime.py b/application/single_app/agent_delegation_runtime.py index 742b9c507..1053cdaf3 100644 --- a/application/single_app/agent_delegation_runtime.py +++ b/application/single_app/agent_delegation_runtime.py @@ -350,7 +350,9 @@ async def execute_target(target, task, context, frame, *, scope=None): [metadata.model_dump(mode="json") for metadata in kernel.get_full_list_of_function_metadata()], ) _check_cancelled(frame) - result = await _invoke_local(agent, messages) + from functions_action_credential_continuation import invoke_credential_agent + + result = await invoke_credential_agent(agent, messages) if journal is not None: journal.require_valid() text = str(result) if result is not None else "" @@ -480,6 +482,12 @@ async def call_agent(action_id, task, context=""): error = "The configured agent is unavailable or you are not permitted to use it." raise AgentDelegationError(error) from exc except Exception as exc: + from action_credential_contracts import find_credential_stop + + credential_stop = find_credential_stop(exc) + if credential_stop: + error = "The delegated agent needs personal action credentials." + raise credential_stop from None # Consent failures retain their dedicated exception type and safe message. if type(exc).__name__ == "FoundryAgentUserAuthenticationRequired": error = "The called agent requires sign-in or consent for Foundry." @@ -531,6 +539,7 @@ def __init__(self, agent, reference, *, identity, settings, cancel_requested=Non self._stream_retry_override = None self._drain_consumer_id = str(uuid.uuid4()) self.completed_delegation_count = 0 + self.credential_checkpoint_id = f"checkpoint_{uuid.uuid4().hex}" def __getattr__(self, name): return getattr(self.agent, name) @@ -553,6 +562,9 @@ def drain_completed_delegations(self): return {"citations": _delegation_records_citations(records), "usage": _delegation_records_usage(records)} async def _run(self, messages, *, streaming=False, **kwargs): + from functions_action_credentials import _continuation, credential_continuation + from functions_action_credential_continuation import invoke_credential_agent, stream_credential_agent + if self.prevent_replay and self._invocations and self.budget.attempts: raise AgentDelegationError("This agent turn cannot be replayed after a delegated action.") self._invocations += 1 @@ -564,6 +576,10 @@ async def _run(self, messages, *, streaming=False, **kwargs): invocation_retry_state = None retry_override = self._stream_retry_override selected = self.agent + descriptor = _continuation.get() or { + "mode": "chat", "checkpoint_id": self.credential_checkpoint_id, + "agent_reference": deepcopy(self.reference), + } try: with agent_execution(frame) as frame: frame.budget.raise_authentication_requirement(frame.identity) @@ -591,12 +607,12 @@ async def _run(self, messages, *, streaming=False, **kwargs): if retry_override: invocation_retry_state = retry_override[1](selected, retry_override[0]) if streaming: - with agent_execution(frame): + with agent_execution(frame), credential_continuation(descriptor): # Stream wrappers such as the M365 continuation journal set # ContextVars on the first pull and reset them on the last, # so create, pull, and close the stream in one Context. stream_context = contextvars.copy_context() - stream = stream_context.run(selected.invoke_stream, messages=messages, **kwargs) + stream = stream_context.run(stream_credential_agent, selected, messages, **kwargs) while True: with agent_execution(frame): try: @@ -611,8 +627,8 @@ async def _run(self, messages, *, streaming=False, **kwargs): # token across the yield; the inner stream keeps stream_context. yield result else: - with agent_execution(frame): - result = await await_agent_operation(_invoke_local(selected, messages), frame) + with agent_execution(frame), credential_continuation(descriptor): + result = await await_agent_operation(invoke_credential_agent(selected, messages, **kwargs), frame) self.last_usage = _observed_usage(result, kernel or getattr(selected, "kernel", None)) yield result finally: diff --git a/application/single_app/app.py b/application/single_app/app.py index d682baf09..d516a1748 100644 --- a/application/single_app/app.py +++ b/application/single_app/app.py @@ -73,6 +73,10 @@ register_route_backend_v2_admin, ) from route_backend_orchestration import register_route_backend_orchestration +from route_backend_action_credentials import register_route_backend_action_credentials +from functions_action_credential_state import configure_action_credential_state +from functions_action_credential_runtime import configure_action_credential_runtime +from config import cosmos_action_credential_state_container from route_backend_search import * from route_backend_conversations import * from route_backend_documents import * @@ -1618,6 +1622,9 @@ def finalize_m365_json_request(response): # ------------------- API Workspace Identity Routes ------ register_route_blueprint('backend_workspace_identities', register_route_backend_workspace_identities, login_required_blueprint) +configure_action_credential_state(cosmos_action_credential_state_container) +configure_action_credential_runtime() +register_route_blueprint('backend_action_credentials', register_route_backend_action_credentials, user_required_blueprint) # ------------------- API User Agreement Routes ---------- register_route_blueprint('backend_user_agreement', register_route_backend_user_agreement, user_required_blueprint) diff --git a/application/single_app/config.py b/application/single_app/config.py index 4e30a7601..276949d40 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.336" +VERSION = "0.261.337" IS_DEVELOPMENT = is_development_env_enabled() # Opt-out for deployments where App Service Easy Auth is active but the platform @@ -883,6 +883,13 @@ def _create_container_if_not_exists_with_conflict_recovery(*args, **kwargs): ) cosmos_personal_workspace_identities_container_name = "personal_workspace_identities" +cosmos_action_credential_state_container_name = "action_credential_state" +cosmos_action_credential_state_container = cosmos_database.create_container_if_not_exists( + id=cosmos_action_credential_state_container_name, + partition_key=PartitionKey(path="/user_id"), + default_ttl=-1, +) + cosmos_personal_workspace_identities_container = cosmos_database.create_container_if_not_exists( id=cosmos_personal_workspace_identities_container_name, partition_key=PartitionKey(path="/user_id") diff --git a/application/single_app/functions_action_credential_continuation.py b/application/single_app/functions_action_credential_continuation.py new file mode 100644 index 000000000..3fb962e59 --- /dev/null +++ b/application/single_app/functions_action_credential_continuation.py @@ -0,0 +1,331 @@ +# functions_action_credential_continuation.py +"""Durable SK tool-boundary continuation, including nested local agents. + +Only paused calls are resumed. Completed tool messages remain in the real SK +history; uncertain calls are never restarted. This is not a chat-message retry. +""" + +from contextlib import aclosing, contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +import inspect +from copy import deepcopy +from typing import Callable + +from semantic_kernel.agents import ChatHistoryAgentThread +from semantic_kernel.contents import ChatHistory, ChatMessageContent, FunctionCallContent, FunctionResultContent +from semantic_kernel.filters import FilterTypes + +from action_credential_contracts import ActionCredentialError, find_credential_stop +from agent_execution_context import current_agent_execution +from functions_action_manifest import get_action_origin +import functions_action_credential_state as state +from functions_action_credentials import _continuation + + +_journal = ContextVar("action_credential_journal", default=None) +_call_id = ContextVar("action_credential_call_id", default=None) +WAIT_MARKER = "simplechat_action_credentials_waiting" +_services = None + + +@dataclass(frozen=True) +class CredentialContinuationServices: + collect_sources: Callable + authorize_source: Callable + resolve_agent: Callable + + +def configure_action_credential_continuation_services(services): + global _services + _services = services + + +def continuation_services(): + if _services is None: + raise ActionCredentialError("credential_runtime_unavailable", 503) + return _services + + +def journal_active(): + return _journal.get() is not None + + +async def _function_boundary(context, next): + journal = _journal.get() + if journal is None: + return await next(context) + if journal.failure: + raise journal.failure + if journal.pending: + journal.deferred.add(_call_id.get()) + raise journal.pending + try: + await next(context) + except Exception as error: + pending = find_credential_stop(error) + if pending: + journal.pending = pending + journal.deferred.add(_call_id.get()) + elif isinstance(error, ActionCredentialError): + journal.failure = error + raise + return None + + +async def _auto_boundary(context, next): + journal = _journal.get() + token = _call_id.set(context.function_call_content.id) + try: + await next(context) + finally: + _call_id.reset(token) + if journal and journal.failure: + context.terminate = True + if journal and journal.pending: + context.terminate = True + if context.function_call_content.id in journal.deferred: + context.function_result.metadata = { + **(context.function_result.metadata or {}), WAIT_MARKER: True, + } + context.function_result.value = {WAIT_MARKER: True, "message": "This tool call has not completed."} + + +def install_credential_filters(kernel): + for kind, attribute, callback in ( + (FilterTypes.FUNCTION_INVOCATION, "function_invocation_filters", _function_boundary), + (FilterTypes.AUTO_FUNCTION_INVOCATION, "auto_function_invocation_filters", _auto_boundary), + ): + if not any(existing is callback for _identity, existing in getattr(kernel, attribute)): + kernel.add_filter(kind, callback) + + +class CredentialJournal: + def __init__(self, kernel, key, signature, user_id, conversation_id): + descriptor = _continuation.get() + if not descriptor or not descriptor.get("checkpoint_id"): + raise ActionCredentialError("credential_continuation_unavailable") + self.kernel, self.key, self.signature = kernel, key, signature + self.user_id, self.conversation_id = user_id, conversation_id + self.checkpoint_id = descriptor["checkpoint_id"] + self.pending, self.deferred, self.failure = None, set(), None + self.history = None + self.loaded = None + self.sources = [] + install_credential_filters(kernel) + + @contextmanager + def scope(self): + parent = _journal.get() + token = _journal.set(self) + try: + yield + finally: + _journal.reset(token) + if parent is not None: + parent.sources.extend(deepcopy(self.sources)) + + def collect_sources(self): + for source in continuation_services().collect_sources(): + if source not in self.sources: + self.sources.append(source) + + def load(self): + # New invocations have no retained work. Shared/unused actions must not + # acquire a dependency on private storage before a credential stop. + if not (_continuation.get() or {}).get("resume"): + return None + try: + checkpoint = state.read(self.user_id, self.checkpoint_id, record_type="credential_checkpoint") + except LookupError: + return None + if checkpoint["conversation_id"] != self.conversation_id: + raise ActionCredentialError("credential_checkpoint_changed") + entry = checkpoint["value"].get(self.key) + if not entry: + return None + if entry["signature"] != self.signature: + raise ActionCredentialError("credential_execution_changed") + for value in checkpoint["value"].values(): + for source in value.get("sources") or []: + continuation_services().authorize_source(source, user_id=self.user_id, purpose="model") + self.sources = deepcopy(entry.get("sources") or []) + self.loaded = entry + return ChatHistory.restore_chat_history(entry["history"]) + + def save(self, history, *, completed=False, response=None): + self.collect_sources() + for _attempt in range(5): + try: + checkpoint = state.read(self.user_id, self.checkpoint_id, record_type="credential_checkpoint") + except LookupError: + checkpoint = None + entries = dict(checkpoint["value"]) if checkpoint else {} + entries[self.key] = { + "signature": self.signature, "history": history.serialize(), "completed": completed, + "response": response, "sources": deepcopy(self.sources), + } + try: + state.checkpoint( + self.user_id, self.checkpoint_id, self.conversation_id, entries, + expected_etag=checkpoint["_etag"] if checkpoint else None, + ) + return + except ActionCredentialError as error: + if error.code != "credential_request_changed": + raise + raise ActionCredentialError("credential_checkpoint_changed") + + async def resume_calls(self, history, *, arguments=None, function_behavior=None): + calls = { + item.id: item for message in history.messages for item in message.items + if isinstance(item, FunctionCallContent) + } + for index, message in enumerate(list(history.messages)): + paused = [ + item for item in message.items + if isinstance(item, FunctionResultContent) and item.metadata.get(WAIT_MARKER) is True + ] + if not paused: + continue + if len(paused) != 1 or paused[0].id not in calls: + raise ActionCredentialError("credential_checkpoint_changed") + working = ChatHistory(messages=list(history.messages)) + await self.kernel.invoke_function_call( + calls[paused[0].id], working, arguments=arguments, function_behavior=function_behavior, + ) + if self.failure: + raise self.failure + history.messages[index] = working.messages[-1] + self.save(history) + if self.pending: + raise self.pending + + +def _signature(agent): + frame = current_agent_execution() + configuration = None + if frame and frame.caller.get("scope_type") and frame.caller.get("id"): + # Fresh canonical configuration includes knowledge, model and attached + # action selections; a slim catalogue reference is not a change guard. + configuration = continuation_services().resolve_agent(frame.caller, user_id=frame.identity.user_id) + manifests = [] + for plugin in agent.kernel.plugins.values(): + for function in plugin.functions.values(): + instance = getattr(getattr(function, "method", None), "__self__", None) + manifest = getattr(instance, "manifest", None) + if manifest: + origin = get_action_origin(manifest) + manifests.append({ + "manifest": manifest, + "origin": (origin.scope_type, origin.scope_id, origin.action_id) if origin else None, + }) + services = [] + for service_id, service in sorted(getattr(agent.kernel, "services", {}).items()): + client = getattr(service, "client", None) + services.append({ + "id": service_id, "model": getattr(service, "ai_model_id", None), + "endpoint": str(getattr(client, "base_url", "")), + "api_version": getattr(client, "_api_version", None), + }) + return state.key_for( + agent.name, agent.instructions, getattr(agent, "deployment_name", None), + getattr(agent, "azure_endpoint", None), getattr(agent, "api_version", None), + configuration, manifests, services, + sorted(item.fully_qualified_name for item in agent.kernel.get_full_list_of_function_metadata()), + ) + + +def _journal_for(agent, messages): + frame = current_agent_execution() + descriptor = _continuation.get() + if ( + not frame or not descriptor or not state.is_configured() + or not all(hasattr(agent, field) for field in ("kernel", "name", "instructions", "arguments", "function_choice_behavior")) + ): + return None + key = "root" if frame.depth == 0 and descriptor.get("mode") == "chat" else state.key_for( + agent.name, [ + message.to_dict() if hasattr(message, "to_dict") else str(message) + for message in (messages if isinstance(messages, list) else [messages]) + ], + ) + return CredentialJournal( + agent.kernel, key, _signature(agent), frame.identity.user_id, frame.identity.conversation_id, + ) + + +async def _prepare_agent(journal, agent, messages, kwargs): + history = journal.load() + if history: + await journal.resume_calls( + history, arguments=agent.arguments, function_behavior=agent.function_choice_behavior, + ) + thread = ChatHistoryAgentThread(history) + return thread, {"messages": None, "thread": thread, **kwargs} + thread = kwargs.pop("thread", None) or ChatHistoryAgentThread() + return thread, {"messages": messages, "thread": thread, **kwargs} + + +async def invoke_credential_agent(agent, messages, **kwargs): + journal = _journal_for(agent, messages) + if journal is None: + return await _invoke_agent(agent, messages=messages, **kwargs) + with journal.scope(): + loaded = journal.load() + if loaded is not None and journal.loaded.get("completed"): + return ChatMessageContent(role="assistant", content=journal.loaded["response"] or "") + thread, kwargs = await _prepare_agent(journal, agent, messages, kwargs) + result = await _invoke_agent(agent, **kwargs) + if journal.failure: + raise journal.failure + journal.collect_sources() + history = ChatHistory(messages=[message async for message in thread.get_messages()]) + if journal.pending or journal.loaded: + journal.save(history, completed=journal.pending is None, response=str(result) if result is not None else "") + if journal.pending: + raise journal.pending + return result + + +async def _invoke_agent(agent, **kwargs): + result = agent.invoke(**kwargs) + if inspect.isawaitable(result): + result = await result + if hasattr(result, "__aiter__"): + last = None + async with aclosing(result) as stream: + async for item in stream: + last = item + result = last + return result + + +async def stream_credential_agent(agent, messages, **kwargs): + journal = _journal_for(agent, messages) + if journal is None: + async with aclosing(agent.invoke_stream(messages=messages, **kwargs)) as stream: + async for result in stream: + yield result + return + with journal.scope(): + loaded = journal.load() + if loaded is not None and journal.loaded.get("completed"): + yield ChatMessageContent(role="assistant", content=journal.loaded["response"] or "") + return + thread, kwargs = await _prepare_agent(journal, agent, messages, kwargs) + async with aclosing(agent.invoke_stream(**kwargs)) as stream: + async for result in stream: + if journal.pending is None and journal.failure is None: + yield result + if journal.failure: + raise journal.failure + journal.collect_sources() + history = ChatHistory(messages=[message async for message in thread.get_messages()]) + if journal.pending or journal.loaded: + journal.save( + history, completed=journal.pending is None, + response=history.messages[-1].content if journal.pending is None and history.messages else None, + ) + if journal.pending: + raise journal.pending diff --git a/application/single_app/functions_action_credential_runtime.py b/application/single_app/functions_action_credential_runtime.py new file mode 100644 index 000000000..797405ed7 --- /dev/null +++ b/application/single_app/functions_action_credential_runtime.py @@ -0,0 +1,52 @@ +# functions_action_credential_runtime.py +"""Web/scheduler composition of credential services after runtime initialization.""" + +from action_credential_contracts import ActionCredentialError +import functions_action_credentials as credentials +import functions_action_credential_continuation as continuation + + +def configure_action_credential_runtime(): + # Owners are imported only by initialized web/scheduler composition. Lower + # credential modules never import these owners, even through deferred imports. + import functions_action_catalog + import functions_agent_delegation + import functions_authentication + import functions_settings + import functions_workspace_identities as identities + import route_backend_chats + from content_screening import access + + def resolve_manifest(user_id, action_ref): + user = functions_authentication.get_current_user_info() or {} + return functions_action_catalog.resolve_action_manifest( + user_id, action_ref, settings=functions_settings.get_settings(), user_roles=user.get("roles") or [], + ) + + def resolve_agent(reference, *, user_id): + return functions_agent_delegation.resolve_delegation_agent( + reference, user_id=user_id, settings=functions_settings.get_settings(), + ) + + def update_identity(*args, **kwargs): + try: + return identities.update_workspace_identity_conditional(*args, **kwargs) + except identities.WorkspaceIdentityConflict: + raise ActionCredentialError("identity_changed") from None + + credentials.configure_action_credential_services(credentials.CredentialServices( + current_user_id=functions_authentication.get_current_user_id_or_none, + authorize_conversation=route_backend_chats._authorize_personal_conversation_access, + resolve_manifest=resolve_manifest, + read_identity=identities.get_workspace_identity, + list_identities=identities.list_workspace_identities, + identity_supports_usage=identities.identity_supports_usage, + read_identity_auth=identities.get_workspace_identity_auth, + create_identity=identities.create_workspace_identity, + update_identity=update_identity, + )) + continuation.configure_action_credential_continuation_services(continuation.CredentialContinuationServices( + collect_sources=access.current_request_source_provenance, + authorize_source=access.assert_document_available, + resolve_agent=resolve_agent, + )) diff --git a/application/single_app/functions_action_credential_state.py b/application/single_app/functions_action_credential_state.py new file mode 100644 index 000000000..a948673b8 --- /dev/null +++ b/application/single_app/functions_action_credential_state.py @@ -0,0 +1,198 @@ +# functions_action_credential_state.py +"""Private, conditional credential bindings and expiring interaction checkpoints. + +Storage is supplied by application bootstrap. Neither secrets nor principals are +accepted by this owner. Requests are transient and are never backup resources. +""" + +from copy import deepcopy +import hashlib +import json +import time +import uuid + +from azure.core import MatchConditions +from azure.cosmos.exceptions import CosmosHttpResponseError, CosmosResourceNotFoundError + +from action_credential_contracts import ActionCredentialError +from cosmos_document_utils import normalize_cosmos_document + + +_container = None +REQUEST_TTL = 86400 +MAX_CHECKPOINT_BYTES = 1500000 + + +def configure_action_credential_state(container): + global _container + _container = container + + +def is_configured(): + return _container is not None + + +def container(): + if _container is None: + raise ActionCredentialError("credential_storage_unavailable", 503) + return _container + + +def key_for(*values): + return hashlib.sha256(json.dumps(values, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def read(user_id, item_id, *, record_type=None): + try: + item = normalize_cosmos_document(container().read_item(item_id, partition_key=user_id)) + except CosmosResourceNotFoundError: + raise LookupError("Credential request not found.") from None + if item.get("user_id") != user_id or (record_type and item.get("record_type") != record_type): + raise LookupError("Credential request not found.") + if item.get("expires_at") is not None and item["expires_at"] <= time.time(): + raise ActionCredentialError("credential_request_expired", 410) + return item + + +def create(user_id, body): + body = {**deepcopy(body), "user_id": user_id} + return normalize_cosmos_document(container().create_item(body=body)) + + +def replace(user_id, item, updates, expected_etag): + if item.get("user_id") != user_id or not expected_etag or item.get("_etag") != expected_etag: + raise ActionCredentialError("credential_request_changed") + body = {**deepcopy(item), **deepcopy(updates)} + if body.get("id") != item["id"] or body.get("user_id") != user_id: + raise ActionCredentialError("credential_request_changed") + try: + response = container().replace_item( + item["id"], body=body, etag=expected_etag, match_condition=MatchConditions.IfNotModified, + ) + except CosmosHttpResponseError as error: + if error.status_code in {409, 412}: + raise ActionCredentialError("credential_request_changed") from None + raise + return normalize_cosmos_document(response) + + +def request_for(user_id, conversation_id, action_ref, fingerprint, requirements, *, reason, continuation): + item_id = "request_" + key_for(conversation_id, action_ref, fingerprint, continuation, reason, requirements) + try: + existing = read(user_id, item_id, record_type="credential_request") + if existing["status"] == "continuing": + return replace(user_id, existing, { + "status": "pending", "submission_etag": None, + }, existing["_etag"]) + return existing + except LookupError: + pass + now = time.time() + body = { + "id": item_id, "record_type": "credential_request", "conversation_id": conversation_id, + "action_ref": action_ref, "fingerprint": fingerprint, "requirements": requirements, + "reason": reason, "status": "pending", "continuation": deepcopy(continuation), + "created_at": now, "expires_at": now + REQUEST_TTL, "ttl": REQUEST_TTL, + } + try: + return create(user_id, body) + except CosmosHttpResponseError as error: + if error.status_code != 409: + raise + return read(user_id, item_id, record_type="credential_request") + + +def pending(user_id, conversation_id): + return [ + dict(item) for item in container().query_items( + query="SELECT * FROM c WHERE c.user_id = @user AND c.conversation_id = @conversation " + "AND c.record_type = 'credential_request' AND c.status IN ('pending', 'ready', 'saving') " + "AND c.expires_at > @now", + parameters=[ + {"name": "@user", "value": user_id}, {"name": "@conversation", "value": conversation_id}, + {"name": "@now", "value": time.time()}, + ], partition_key=user_id, + ) + ] + + +def binding(user_id, action_ref, slot): + try: + return read(user_id, "binding_" + key_for(action_ref, slot), record_type="credential_binding") + except LookupError: + return None + + +def save_binding(user_id, action_ref, slot, identity_id, identity_etag, fingerprint): + prior = binding(user_id, action_ref, slot) + fields = { + "action_ref": action_ref, "slot": slot, "identity_id": identity_id, + "identity_etag": identity_etag, "fingerprint": fingerprint, "confirmed": True, + } + if prior: + if all(prior.get(key) == value for key, value in fields.items()): + return prior + return replace(user_id, prior, fields, prior["_etag"]) + return create(user_id, { + "id": "binding_" + key_for(action_ref, slot), "record_type": "credential_binding", **fields, + }) + + +def invalidate_identity(user_id, identity_id): + """Invalidate private bindings without revealing action names to identity management.""" + rows = list(container().query_items( + query="SELECT * FROM c WHERE c.user_id = @user AND c.record_type = 'credential_binding' " + "AND c.identity_id = @identity", + parameters=[{"name": "@user", "value": user_id}, {"name": "@identity", "value": identity_id}], + partition_key=user_id, + )) + for row in rows: + current = read(user_id, row["id"], record_type="credential_binding") + replace(user_id, current, {"confirmed": False}, current["_etag"]) + + +def checkpoint(user_id, checkpoint_id, conversation_id, value, *, expected_etag=None): + if len(json.dumps(value, separators=(",", ":")).encode()) > MAX_CHECKPOINT_BYTES: + raise ActionCredentialError("credential_checkpoint_too_large") + try: + current = read(user_id, checkpoint_id, record_type="credential_checkpoint") + except LookupError: + return create(user_id, { + "id": checkpoint_id, "record_type": "credential_checkpoint", "conversation_id": conversation_id, + "value": deepcopy(value), "ttl": REQUEST_TTL, "expires_at": time.time() + REQUEST_TTL, + }) + if current.get("conversation_id") != conversation_id: + raise ActionCredentialError("credential_checkpoint_changed") + return replace(user_id, current, {"value": deepcopy(value)}, expected_etag or current["_etag"]) + + +def claim_request(user_id, item, expected_etag): + if item["status"] != "ready": + raise ActionCredentialError("credential_request_not_ready") + return replace(user_id, item, { + "status": "continuing", "continuation_claim": uuid.uuid4().hex, + }, expected_etag) + + +def delete_private_state(user_id, *, conversation_id=None): + """The authorized lifecycle owner supplies the actor; queries never span partitions.""" + if not is_configured(): + return + parameters = [{"name": "@user", "value": user_id}] + query = "SELECT * FROM c WHERE c.user_id = @user" + if conversation_id is not None: + query += " AND c.conversation_id = @conversation" + parameters.append({"name": "@conversation", "value": conversation_id}) + for row in container().query_items(query=query, parameters=parameters, partition_key=user_id): + if row.get("user_id") != user_id or ( + conversation_id is not None and row.get("conversation_id") != conversation_id + ): + raise ActionCredentialError("credential_request_changed") + try: + container().delete_item( + row["id"], partition_key=user_id, etag=row["_etag"], + match_condition=MatchConditions.IfNotModified, + ) + except CosmosResourceNotFoundError: + # Concurrent deletion already achieved this idempotent lifecycle operation. + continue diff --git a/application/single_app/functions_action_credentials.py b/application/single_app/functions_action_credentials.py new file mode 100644 index 000000000..3b4a5db77 --- /dev/null +++ b/application/single_app/functions_action_credentials.py @@ -0,0 +1,309 @@ +# functions_action_credentials.py +"""Authorize, acquire and inject the invoking actor's private action credentials.""" + +from contextlib import contextmanager +from contextvars import ContextVar +from copy import deepcopy +from dataclasses import dataclass +import time +from typing import Callable + +from flask import g, has_request_context + +from action_credential_contracts import ( + ADAPTERS, ActionCredentialError, ActionCredentialsRequired, credential_fingerprint, + credential_slots, destination_label, +) +from agent_execution_context import current_agent_execution +from functions_action_manifest import get_action_origin, resolve_action_type +import functions_action_credential_state as state + + +_continuation = ContextVar("action_credential_continuation", default=None) +_services = None + + +@dataclass(frozen=True) +class CredentialServices: + current_user_id: Callable + authorize_conversation: Callable + resolve_manifest: Callable + read_identity: Callable + list_identities: Callable + identity_supports_usage: Callable + read_identity_auth: Callable + create_identity: Callable + update_identity: Callable + + +def configure_action_credential_services(services): + global _services + _services = services + + +def credential_services(): + if _services is None: + raise ActionCredentialError("credential_runtime_unavailable", 503) + return _services + + +@contextmanager +def credential_continuation(value): + token = _continuation.set(deepcopy(value)) + try: + yield + finally: + _continuation.reset(token) + + +def actor_context(): + frame = current_agent_execution() + if frame: + user_id, conversation_id = frame.identity.user_id, frame.identity.conversation_id + elif has_request_context(): + user_id, conversation_id = credential_services().current_user_id(), getattr(g, "conversation_id", None) + else: + raise ActionCredentialError("credentials_need_user_action") + if not user_id or not conversation_id: + raise ActionCredentialError("credentials_need_user_action") + authorize_conversation(user_id, conversation_id) + return user_id, conversation_id, frame + + +def authorize_conversation(user_id, conversation_id): + return credential_services().authorize_conversation(user_id, conversation_id) + + +def authorized_manifest(user_id, action_ref): + return credential_services().resolve_manifest(user_id, action_ref) + + +def manifest_reference(manifest): + origin = get_action_origin(manifest) + if origin is None or not origin.action_id: + raise ActionCredentialError("trusted_origin_required") + return f"{origin.scope_type}:{origin.scope_id}:{origin.action_id}" + + +def identity_metadata(user_id, identity_id, auth_type): + services = credential_services() + identity = services.read_identity("personal", user_id, identity_id) + if not services.identity_supports_usage(identity, "action", source_type="action", auth_types={auth_type}): + raise ActionCredentialError("identity_incompatible", 400) + return identity + + +def safe_identities(user_id, requirements): + services = credential_services() + types = {item["auth_type"] for item in requirements} + return [ + {"id": item["id"], "name": item.get("name") or "Personal identity", + "auth_type": (item.get("auth") or {}).get("auth_type"), "etag": item.get("_etag")} + for item in services.list_identities("personal", user_id) + if services.identity_supports_usage(item, "action", source_type="action", auth_types=types) + ] + + +def safe_request(item, manifest): + user_id = item["user_id"] + requirements = [ + {"slot": slot, "auth_type": auth_type, + "label": "Proxy sign-in" if slot == "proxy" else "YAMCS sign-in"} + for slot, auth_type in item["requirements"].items() + ] + result = { + "id": item["id"], "etag": item["_etag"], "conversation_id": item["conversation_id"], + "action_name": manifest.get("displayName") or manifest.get("name") or "YAMCS", + "destination_label": destination_label(manifest), "reason": item["reason"], + "requirements": requirements, "identities": safe_identities(user_id, requirements), + "status": "pending" if item["status"] == "saving" else item["status"], + } + if item["status"] == "saving": + result["etag"] = item["submission_etag"] + for key in ("run_id", "turn_id"): + if item.get("continuation", {}).get(key): + result[key] = item["continuation"][key] + return result + + +def authorize_request(user_id, request_id): + item = state.read(user_id, request_id, record_type="credential_request") + authorize_conversation(user_id, item["conversation_id"]) + manifest = authorized_manifest(user_id, item["action_ref"]) + if credential_fingerprint(manifest) != item["fingerprint"]: + raise ActionCredentialError("action_destination_changed") + return item, manifest + + +def require_credentials(manifest, slots, *, reason="setup_required"): + user_id, conversation_id, frame = actor_context() + continuation = _continuation.get() + if not continuation: + raise ActionCredentialError("credentials_need_user_action") + current = authorized_manifest(user_id, manifest_reference(manifest)) + if credential_fingerprint(current) != credential_fingerprint(manifest): + raise ActionCredentialError("action_destination_changed") + item = state.request_for( + user_id, conversation_id, manifest_reference(manifest), credential_fingerprint(manifest), + {slot: credential_slots(manifest)[slot]["auth_type"] for slot in slots}, + reason=reason, continuation=continuation, + ) + if item["status"] in {"cancelled", "completed"}: + raise ActionCredentialError("credential_request_closed") + error = ActionCredentialsRequired(safe_request(item, current)) + # The SK journal must persist its real tool history before the stop reaches + # the outer worker; publishing through the polling budget here would cancel it early. + raise error from None + + +def resolve_personal_credentials(manifest, *, rejected_slots=()): + slots = credential_slots(manifest) + if not slots: + return deepcopy(manifest) + user_id, _conversation_id, _frame = actor_context() + reference = manifest_reference(manifest) + current = authorized_manifest(user_id, reference) + fingerprint = credential_fingerprint(manifest) + if credential_fingerprint(current) != fingerprint: + raise ActionCredentialError("action_destination_changed") + if rejected_slots: + require_credentials(manifest, rejected_slots, reason="credentials_rejected") + missing, identities = [], {} + reason = "setup_required" + for slot, requirement in slots.items(): + binding = state.binding(user_id, reference, slot) + if not binding: + missing.append(slot) + continue + if binding.get("fingerprint") != fingerprint or not binding.get("confirmed"): + missing.append(slot) + reason = "binding_changed" + continue + try: + identity = identity_metadata(user_id, binding["identity_id"], requirement["auth_type"]) + except LookupError: + missing.append(slot) + reason = "identity_removed" + continue + if identity.get("_etag") != binding.get("identity_etag"): + missing.append(slot) + reason = "identity_changed" + continue + identities[slot] = identity + if missing: + require_credentials(manifest, missing, reason=reason) + services = credential_services() + prepared = deepcopy(manifest) + for slot, identity in identities.items(): + auth = services.read_identity_auth("personal", user_id, identity["id"]) + latest = identity_metadata(user_id, identity["id"], slots[slot]["auth_type"]) + if latest.get("_etag") != identity.get("_etag"): + raise ActionCredentialError("identity_changed") + auth_type = slots[slot]["auth_type"] + required = ("username", "password") if auth_type == "username_password" else ("secret",) + if auth.get("auth_type") != auth_type or any(not auth.get(key) for key in required): + # Secret storage failures are not evidence of provider rejection. + raise ActionCredentialError("credential_storage_unavailable", 503) + ADAPTERS[resolve_action_type(manifest).lower()].inject(prepared, slot, auth) + return prepared + + +def _validate_submission(item, slots): + if type(slots) is not list or not slots or len(slots) > 2: + raise ActionCredentialError("invalid_credentials", 400) + seen = set() + for entry in slots: + if type(entry) is not dict or set(entry) - { + "slot", "identity_id", "name", "credentials", "expected_identity_etag", + }: + raise ActionCredentialError("invalid_credentials", 400) + slot = entry.get("slot") + if type(slot) is not str or slot not in item["requirements"] or slot in seen: + raise ActionCredentialError("invalid_credentials", 400) + if entry.get("identity_id") is not None and ( + type(entry["identity_id"]) is not str or not entry["identity_id"] or len(entry["identity_id"]) > 255 + ): + raise ActionCredentialError("invalid_credentials", 400) + if entry.get("expected_identity_etag") is not None and type(entry["expected_identity_etag"]) is not str: + raise ActionCredentialError("invalid_credentials", 400) + seen.add(slot) + credentials = entry.get("credentials") + if not entry.get("identity_id") and not credentials: + raise ActionCredentialError("invalid_credentials", 400) + if credentials is not None: + auth_type = item["requirements"][slot] + fields = {"auth_type", "username", "password"} if auth_type == "username_password" else {"auth_type", "secret"} + if ( + type(credentials) is not dict or set(credentials) != fields + or credentials["auth_type"] != auth_type + or any(type(credentials[key]) is not str or not credentials[key] or len(credentials[key]) > 8192 + for key in fields) + ): + raise ActionCredentialError("invalid_credentials", 400) + if entry.get("identity_id") and not entry.get("expected_identity_etag"): + raise ActionCredentialError("identity_etag_required", 400) + if not entry.get("identity_id") and ( + type(entry.get("name")) is not str or not entry["name"].strip() or len(entry["name"]) > 255 + ): + raise ActionCredentialError("identity_name_required", 400) + if seen != set(item["requirements"]): + raise ActionCredentialError("incomplete_credentials", 400) + + +def submit_credentials(user_id, request_id, expected_etag, slots): + item, manifest = authorize_request(user_id, request_id) + if item["status"] == "ready" and item.get("submission_etag") == expected_etag: + return safe_request(item, manifest) + _validate_submission(item, slots) + selection = sorted([ + {key: entry[key] for key in ("slot", "identity_id", "name", "expected_identity_etag") if key in entry} + for entry in slots + ], key=lambda entry: entry["slot"]) + if item["status"] == "saving": + if item.get("submission_etag") != expected_etag or item.get("saving_until", 0) > time.time(): + raise ActionCredentialError("credential_request_changed") + def targets(value): + return [{key: field for key, field in row.items() if key != "expected_identity_etag"} for row in value] + + if targets(item.get("submission_selection") or []) != targets(selection): + raise ActionCredentialError("credential_request_changed") + # A failed response/write can be recovered using the same deterministic identity IDs. + expected_etag = item["_etag"] + elif item["status"] != "pending": + raise ActionCredentialError("credential_request_closed") + submission_etag = item.get("submission_etag") or expected_etag + item = state.replace(user_id, item, { + "status": "saving", "submission_etag": submission_etag, "saving_until": time.time() + 60, + "submission_selection": selection, + }, expected_etag) + services = credential_services() + for entry in slots: + slot = entry["slot"] + auth_type = item["requirements"][slot] + identity_id = entry.get("identity_id") + operation_id = state.key_for(request_id, submission_etag, slot) + if identity_id: + identity = identity_metadata(user_id, identity_id, auth_type) + if entry.get("credentials"): + identity = services.update_identity( + "personal", user_id, identity_id, {"credentials": entry["credentials"]}, user_id, + entry["expected_identity_etag"], operation_id=operation_id, + ) + else: + identity_id = "action_" + operation_id + try: + identity = services.read_identity("personal", user_id, identity_id) + except LookupError: + services.create_identity( + "personal", user_id, { + "name": entry["name"], "provider": "generic", "usage_contexts": ["action"], + "supported_source_types": ["action"], "credentials": entry["credentials"], + }, user_id, stage_secrets=True, identity_id=identity_id, + ) + identity = services.read_identity("personal", user_id, identity_id) + identity_metadata(user_id, identity_id, auth_type) + # Access/destination can change during secret storage; never commit a binding to the old target. + authorize_request(user_id, request_id) + state.save_binding(user_id, item["action_ref"], slot, identity_id, identity["_etag"], item["fingerprint"]) + item = state.replace(user_id, item, {"status": "ready"}, item["_etag"]) + return safe_request(item, manifest) diff --git a/application/single_app/functions_legacy_action_management.py b/application/single_app/functions_legacy_action_management.py index de034fce9..188c64576 100644 --- a/application/single_app/functions_legacy_action_management.py +++ b/application/single_app/functions_legacy_action_management.py @@ -197,6 +197,10 @@ def prepare_scoped_action(action, scope_type, scope_id): if not key.startswith("_") and key not in _MANAGEMENT_FIELDS } payload = bind_action_origin(payload, scope_type, scope_id) + # Pure contract validation runs before any identity or Key Vault write. + from action_credential_contracts import validate_credential_requirements + + validate_credential_requirements(payload) if is_retired_mcp_stdio(payload): raise McpStdioRemovedError() if payload["type"] == "mcp": diff --git a/application/single_app/functions_orchestration_actions.py b/application/single_app/functions_orchestration_actions.py index 9fd825c78..815ddb951 100644 --- a/application/single_app/functions_orchestration_actions.py +++ b/application/single_app/functions_orchestration_actions.py @@ -539,6 +539,10 @@ async def invoke_action( from semantic_kernel_plugins.logged_plugin_loader import create_logged_plugin_loader from agent_delegation_runtime import await_agent_operation from functions_settings import get_settings + from action_credential_contracts import find_credential_stop + from functions_action_credential_continuation import CredentialJournal + from functions_action_credentials import _continuation + from functions_action_credential_state import key_for invocation_capture = require_invocation_capture(invocation_capture) visual_request = visual_request if isinstance(visual_request, dict) else {} @@ -675,6 +679,10 @@ async def guard_function(invocation, next): failure = step_error(exc) raise failure from exc except Exception as exc: + credential_stop = find_credential_stop(exc) + if credential_stop is not None: + failure = credential_stop + raise credential_stop from None # Contain arbitrary plugin exceptions before SK puts them in model context. log_event( '[ORCHESTRATION_ADAPTERS] Action function failed.', @@ -795,9 +803,30 @@ async def stop_after_failure(invocation, next): ) else: execution_settings.function_choice_behavior = function_choice - replies = await await_agent_operation( - service.get_chat_message_contents(history, execution_settings, kernel=kernel), frame, - ) + descriptor = _continuation.get() + credential_journal = None + if descriptor: + credential_journal = CredentialJournal( + kernel, "action", key_for(manifest), + user_id, getattr(context, "conversation_id", None), + ) + if credential_journal: + with credential_journal.scope(): + saved_history = credential_journal.load() + if saved_history is not None: + history = saved_history + await credential_journal.resume_calls(history, function_behavior=function_choice) + replies = await await_agent_operation( + service.get_chat_message_contents(history, execution_settings, kernel=kernel), frame, + ) + if credential_journal.pending or credential_journal.loaded: + credential_journal.save(history, completed=credential_journal.pending is None) + if credential_journal.pending: + raise credential_journal.pending + else: + replies = await await_agent_operation( + service.get_chat_message_contents(history, execution_settings, kernel=kernel), frame, + ) if cancel_requested(): raise AgentExecutionCancelled('Action execution was cancelled.') if failure is not None: diff --git a/application/single_app/functions_orchestration_adapters.py b/application/single_app/functions_orchestration_adapters.py index 024f51554..8c44f18ee 100644 --- a/application/single_app/functions_orchestration_adapters.py +++ b/application/single_app/functions_orchestration_adapters.py @@ -1900,12 +1900,36 @@ def _serialize(value): return citations +def _credential_step_continuation(step, context): + from functions_action_credential_state import key_for + + descriptor = { + "mode": "orchestration", "run_id": _ctx(context, "run_id"), + "step_id": step["step_id"], "turn_id": _ctx(context, "turn_id"), + "checkpoint_id": "checkpoint_" + key_for(_ctx(context, "run_id"), step["step_id"]), + } + wait = (_ctx(context, "pending_results", {}) or {}).get(step["step_id"]) + if isinstance(wait, dict) and wait.get("kind") == "action_credentials": + descriptor["resume"] = True + return descriptor + + +def _credential_wait_result(error, emit): + _emit(emit, error.event) + return build_step_result( + status="waiting", summary="Waiting for personal action credentials.", + wait={"kind": "action_credentials", "request_id": error.credential_request["id"]}, + ) + + def run_action_invoke(step, context, *, settings, user_id, emit, cancel_requested): if _is_cancelled(cancel_requested): return _cancelled_result('Action execution was cancelled.') # Action dependencies initialize SK/Azure; keep them out of the adapter import path. import asyncio from agent_execution_context import AgentExecutionCancelled + from action_credential_contracts import ActionCredentialsRequired + from functions_action_credentials import credential_continuation arguments = _arguments(step) action_ref = _text(arguments.get('action_ref')) @@ -1953,6 +1977,7 @@ def run_action_invoke(step, context, *, settings, user_id, emit, cancel_requeste # Microsoft 365 email and event times are shown in the turn's browser time zone. with ( + credential_continuation(_credential_step_continuation(step, context)), orchestration_action_card_events(_m365_card_callback(step, context, emit)), m365_display_time_zone(_ctx(context, 'time_zone')), operation_step_scope(step, context, user_id, named_inputs), @@ -1964,6 +1989,8 @@ def run_action_invoke(step, context, *, settings, user_id, emit, cancel_requeste )) except (AgentExecutionCancelled, OrchestrationInvocationCancelledError): return _cancelled_result('Action execution was cancelled.') + except ActionCredentialsRequired as error: + return _credential_wait_result(error, emit) except ( OrchestrationInvocationServiceError, OrchestrationInvocationControlError, OrchestrationInvocationDeniedError, OrchestrationInvocationHeldError, @@ -2092,6 +2119,8 @@ def run_agent_invoke(step, context, *, settings, user_id, emit, cancel_requested # Flask state, and all steps share the run's root delegation budget. import asyncio from agent_execution_context import AgentExecutionCancelled, DelegationBudget + from action_credential_contracts import ActionCredentialsRequired + from functions_action_credentials import credential_continuation try: from functions_agent_delegation import agent_reference @@ -2147,6 +2176,7 @@ def scope(_target): from functions_m365_citations import m365_display_time_zone with ( + credential_continuation(_credential_step_continuation(step, context)), orchestration_action_card_events(_m365_card_callback(step, context, emit)), m365_display_time_zone(_ctx(context, 'time_zone')), operation_step_scope(step, context, user_id, named_inputs) as journal, @@ -2166,6 +2196,8 @@ async def invoke(): operation_receipts = deepcopy(journal.receipts) if journal is not None else [] except (AgentExecutionCancelled, OrchestrationInvocationCancelledError): return _cancelled_result('Agent execution was cancelled.') + except ActionCredentialsRequired as error: + return _credential_wait_result(error, emit) except ( OrchestrationInvocationServiceError, OrchestrationInvocationControlError, OrchestrationInvocationDeniedError, OrchestrationInvocationHeldError, diff --git a/application/single_app/functions_orchestration_continuation.py b/application/single_app/functions_orchestration_continuation.py index a8b0195b9..082b83ae6 100644 --- a/application/single_app/functions_orchestration_continuation.py +++ b/application/single_app/functions_orchestration_continuation.py @@ -188,6 +188,7 @@ def _continuation_guards(record, authorize, message_container): def claim_run_continuation( run_id, user_id, conversation_id, *, authorize, message_container, mode="execute", + credential_request_id=None, ): """CAS-claim one existing attempt; a live or no-longer-due run returns None. @@ -198,6 +199,24 @@ def claim_run_continuation( if type(mode) is not str or mode not in _MODES or message_container is None: raise RecoveryError(code="invalid_request", status_code=400) record = _read_owned(run_id, user_id, conversation_id, authorize) + credential_waits = [ + row for row in record.get("execution_steps") or [] + if row.get("status") == "waiting" and (row.get("wait") or {}).get("kind") == "action_credentials" + ] + if credential_waits and mode == "execute": + # Scheduler/recovery reads never consume an interactive credential wait. + if not credential_request_id: + return None + from functions_action_credentials import authorize_request + + request, _manifest = authorize_request(user_id, credential_request_id) + if ( + request["status"] != "continuing" + or request["continuation"].get("run_id") != run_id + or request["conversation_id"] != conversation_id + or not any(row["wait"]["request_id"] == credential_request_id for row in credential_waits) + ): + raise CheckpointError("context_unavailable") if _lease_live(record): return None status = record.get("status") @@ -268,6 +287,10 @@ def claim_run_continuation( } if mode != "delivery": updates["status"] = "running" + if credential_waits and mode == "execute": + from action_credential_deadline import finish_human_wait + + updates.update(finish_human_wait(record, now=_now())) if authorize() is False: raise CheckpointError("context_unavailable") try: @@ -498,6 +521,7 @@ def __init__(self, record, context, settings, lease): elif self.binding != self.record.get("execution_binding"): raise CheckpointError("recovery_changed") context.execution_deadline_at = self.record["execution_deadline_at"] + context.credential_wait_seconds = self.record.get("credential_wait_seconds", 0) self.store = checkpoint_store(self.record, lease.read, token=lease.token, claim_id=lease.claim_id) source = self.record if self.initializing else reconcile_checkpoints(self.record, lease.read) self.records = _execution_steps(source) diff --git a/application/single_app/functions_orchestration_execution.py b/application/single_app/functions_orchestration_execution.py index d326be308..fefba2731 100644 --- a/application/single_app/functions_orchestration_execution.py +++ b/application/single_app/functions_orchestration_execution.py @@ -975,7 +975,9 @@ def _initialize_delivery(self, settings, services): or not self.record.get("execution_binding") ): raise HarnessExecutionError("checkpoint_unavailable") - deadline = datetime.fromisoformat(self.record["execution_deadline_at"]) + from action_credential_deadline import execution_deadline + + deadline = execution_deadline(self.record) if deadline.tzinfo is None: raise HarnessExecutionError("checkpoint_invalid") if self.record["status"] == "completed": @@ -1071,7 +1073,9 @@ def _finalize_delivery_failure(self, failure_code): if not current.get("started_at") or not current.get("execution_deadline_at"): raise HarnessExecutionError("checkpoint_unavailable") try: - deadline = datetime.fromisoformat(current["execution_deadline_at"]) + from action_credential_deadline import execution_deadline + + deadline = execution_deadline(current) except (ValueError, TypeError) as exc: raise HarnessExecutionError("checkpoint_invalid") from exc if deadline.tzinfo is None: @@ -1101,6 +1105,11 @@ def _send(self, frame): _log_failure("Execution progress delivery stopped.", self.record, exc) def _progress(self, event): + if type(event) is dict and event.get("type") == "action_credentials_required": + from functions_orchestration_events import serialize_sse + + self._send(serialize_sse(event)) + return if type(event) is dict and event.get("type") == "m365_pending_action_reference": self._pending_action_created(event) return @@ -1627,6 +1636,10 @@ def _finalize(self, result, error, *, refreshes=0): "assistant_message_created_at": timestamp, "message_saved": False, "finalization_status": "pending", } + if status == "waiting" and any( + wait.get("kind") == "action_credentials" for wait in updates["pending_results"].values() + ): + updates["credential_wait_started_at"] = current.get("credential_wait_started_at") or _now_iso() if not self._delivery_only: updates["harness_step_token_usage"] = _usage( result.get("token_usage", self.context.token_usage if self.context is not None else {}), diff --git a/application/single_app/functions_orchestration_executor.py b/application/single_app/functions_orchestration_executor.py index 1323d4c3b..3fae8f906 100644 --- a/application/single_app/functions_orchestration_executor.py +++ b/application/single_app/functions_orchestration_executor.py @@ -44,6 +44,7 @@ from dataclasses import replace from datetime import datetime, timezone from werkzeug.utils import secure_filename +from action_credential_deadline import execution_deadline from content_screening.access import assert_current_request_sources_available from content_screening.contracts import ScreeningError @@ -1002,6 +1003,10 @@ def _run_dependency_step( if step['role'] == 'render': return _validate_render_step_result(step, result) task = result.get('task_result') + if result.get("status") == STEP_STATUS_WAITING and (result.get("wait") or {}).get("kind") == "action_credentials": + from functions_orchestration_result_contracts import TaskResult + + task = TaskResult(context.result_producer(step), step["role"], "pending", ()) if result['status'] in (STEP_STATUS_FAILED, STEP_STATUS_CANCELLED): if result['status'] == STEP_STATUS_FAILED and task is not None: validate_task_diagnostics(step, context, task) @@ -1202,7 +1207,7 @@ def resume_native_dependency_step( if task.status != 'pending' or type(wait) is not dict or not wait: raise ResultContractError('result_wait_required') if context.execution_deadline_at is not None: - deadline = datetime.fromisoformat(context.execution_deadline_at) + deadline = execution_deadline(context) if deadline.tzinfo is None: raise ResultContractError('result_deadline_invalid') if datetime.now(timezone.utc) >= deadline: @@ -1225,6 +1230,35 @@ def resume_waiting_dependency_step( wait = context.pending_results.get(step['step_id']) if type(wait) is not dict: raise ResultContractError('result_wait_required') + if wait.get("kind") == "action_credentials": + from functions_action_credentials import authorize_request, credential_continuation + from functions_orchestration_adapters import _credential_step_continuation + + if ( + set(wait) != {"kind", "request_id"} or user_id != context.user_id + or step.get("capability_id") not in {"action_invoke", "agent_invoke"} + ): + raise ResultContractError("result_wait_invalid") + request, _manifest = authorize_request(user_id, wait["request_id"]) + descriptor = _credential_step_continuation(step, context) + if any( + request["continuation"].get(key) != descriptor.get(key) for key in ("run_id", "step_id", "checkpoint_id") + ): + raise ResultContractError("result_wait_invalid") + if request["status"] in {"pending", "ready", "saving"}: + # A sibling's explicit Continue never consumes this credential interaction. + return build_step_result( + status=STEP_STATUS_WAITING, task_result=context.task_results[step["step_id"]], + wait=wait, summary="Waiting for personal action credentials.", + ) + if request["status"] != "continuing": + raise ResultContractError("result_wait_invalid") + # The adapter's SK journal resumes the paused tool; completed sibling + # steps continue to come from ContinuationCheckpoints. + return _run_dependency_step( + step, context, settings=settings, user_id=user_id, emit=emit, cancel_probe=cancel_requested, + resolver=_dependency_adapter, input_fingerprint=input_fingerprint, + ) if wait.get('kind') == 'native_tabular_compute': return resume_native_dependency_step( step, context, settings=settings, user_id=user_id, input_fingerprint=input_fingerprint, @@ -1272,7 +1306,7 @@ def resume_waiting_dependency_step( return build_step_result(status=STEP_STATUS_CANCELLED, failure=build_failure('user_cancelled')) if type(context.execution_deadline_at) is not str: raise ResultContractError('result_deadline_invalid') - deadline = datetime.fromisoformat(context.execution_deadline_at) + deadline = execution_deadline(context) if deadline.tzinfo is None: raise ResultContractError('result_deadline_invalid') if datetime.now(timezone.utc) >= deadline: @@ -1341,7 +1375,7 @@ def _execute_dependency_plan( context.durable_checkpoints = True checkpoints = checkpoints(context) if callable(checkpoints) else checkpoints checkpoints.initialize() - deadline = datetime.fromisoformat(context.execution_deadline_at) + deadline = execution_deadline(context) if deadline.tzinfo is None: raise ResultContractError('result_deadline_invalid') _persist(persist, 'run', { diff --git a/application/single_app/functions_orchestration_operations.py b/application/single_app/functions_orchestration_operations.py index d691bab05..672c97a6a 100644 --- a/application/single_app/functions_orchestration_operations.py +++ b/application/single_app/functions_orchestration_operations.py @@ -385,7 +385,20 @@ async def function_filter(self, invocation, next): return try: await next(invocation) - except Exception: + except Exception as error: + from action_credential_contracts import find_credential_stop + safe_pause = False + if find_credential_stop(error) is not None: + # Connector classes are loaded only at invocation, not during bootstrap. + from semantic_kernel_plugins.yamcs_plugin import YamcsPlugin + from semantic_kernel_plugins.agent_plugin import AgentPlugin + + safe_pause = isinstance(plugin, (YamcsPlugin, AgentPlugin)) + if safe_pause: + # The YAMCS adapter only reads; a suspended local-agent call + # retains its completed child tools in the credential journal. + self.finish(key, {"status": "credential_wait"}, refused=True) + raise self.failure = OperationRecoveryError() raise value = invocation.result.value if invocation.result is not None else None diff --git a/application/single_app/functions_orchestration_output_store.py b/application/single_app/functions_orchestration_output_store.py index 7de5d5f02..cef21b72c 100644 --- a/application/single_app/functions_orchestration_output_store.py +++ b/application/single_app/functions_orchestration_output_store.py @@ -656,7 +656,9 @@ def current_run(self, record, *, live=False, stopping=False): if parse_time(record["deadline_at"]) <= self.clock(): raise OutputUnavailableError("output_deadline_exceeded") persisted_deadline = run.get("execution_deadline_at") - if persisted_deadline and parse_time(persisted_deadline) < parse_time(record["deadline_at"]): + from action_credential_deadline import execution_deadline + + if persisted_deadline and execution_deadline(run) < parse_time(record["deadline_at"]): raise OutputUnavailableError("output_deadline_invalid") return run diff --git a/application/single_app/functions_orchestration_rendering.py b/application/single_app/functions_orchestration_rendering.py index fe09abda6..b29fec649 100644 --- a/application/single_app/functions_orchestration_rendering.py +++ b/application/single_app/functions_orchestration_rendering.py @@ -18,6 +18,7 @@ from copy import deepcopy from dataclasses import asdict from datetime import timedelta +from action_credential_deadline import execution_deadline from azure.core.exceptions import AzureError, ResourceNotFoundError from azure.cosmos.exceptions import CosmosResourceNotFoundError @@ -1482,13 +1483,14 @@ def resume_render_file( ): raise OutputUnavailableError("output_binding_invalid") if parse_time(getattr(context, "execution_deadline_at", None)).isoformat() != record["deadline_at"]: - raise OutputError("output_deadline_invalid") + if execution_deadline(context).isoformat() != record["deadline_at"]: + raise OutputError("output_deadline_invalid") run = service._read_metadata(service.store.current_run, record) if record["state"] != "completed" and ( run.get("cancellation_requested_at") or run.get("status") in {"cancelled", "canceled"} ): raise OutputUnavailableError("output_cancelled") - if run.get("execution_deadline_at") and parse_time(run["execution_deadline_at"]) < parse_time( + if run.get("execution_deadline_at") and execution_deadline(run) < parse_time( record["deadline_at"], ): raise OutputError("output_deadline_invalid") @@ -1572,7 +1574,7 @@ def execute_render_file( export_request=export_request, file_name=file_name, approved_work_id=getattr(context, "approved_work_id", None) or producer.run_id, - deadline_at=context.execution_deadline_at, + deadline_at=execution_deadline(context).isoformat(), ) output = service.render_attempt( output["output_id"], stop=cancel_requested, observe=observe_render, diff --git a/application/single_app/functions_orchestration_scheduler.py b/application/single_app/functions_orchestration_scheduler.py index d5c171d98..407763a77 100644 --- a/application/single_app/functions_orchestration_scheduler.py +++ b/application/single_app/functions_orchestration_scheduler.py @@ -310,7 +310,9 @@ def terminal_condition(self, record): if not record.get("execution_deadline_at"): raise CheckpointError("checkpoint_unavailable") try: - deadline = _aware_time(datetime.fromisoformat(record["execution_deadline_at"])) + from action_credential_deadline import execution_deadline + + deadline = _aware_time(execution_deadline(record, now=self.resources.clock())) except (TypeError, ValueError) as exc: raise CheckpointError("checkpoint_invalid") from exc states = {step["step_id"]: step["status"] for step in record.get("execution_steps") or []} diff --git a/application/single_app/functions_orchestration_services.py b/application/single_app/functions_orchestration_services.py index bf1cbaa2b..4e371b09c 100644 --- a/application/single_app/functions_orchestration_services.py +++ b/application/single_app/functions_orchestration_services.py @@ -313,6 +313,7 @@ def checked_checkpoint(step_id): context.composition_profiles = composition_profiles() context.composition_profile_validator = validate_composition_profile context.execution_deadline_at = record.get("execution_deadline_at") + context.credential_wait_seconds = record.get("credential_wait_seconds", 0) context.rendering_service = self.rendering # Each attempt is approved separately and owns its files, so a retry renders a new # output rather than presenting the superseded attempt's work identity. diff --git a/application/single_app/functions_orchestration_workflow_runs.py b/application/single_app/functions_orchestration_workflow_runs.py index 06c08a911..396a4a655 100644 --- a/application/single_app/functions_orchestration_workflow_runs.py +++ b/application/single_app/functions_orchestration_workflow_runs.py @@ -41,6 +41,7 @@ import uuid from copy import deepcopy from datetime import datetime, timezone +from action_credential_deadline import execution_deadline from azure.core import MatchConditions from azure.core.exceptions import AzureError @@ -1095,7 +1096,9 @@ def _wait_deadline(context, settings, step_id, now): from functions_workflow_limits import get_chat_orchestration_workflow_run_wait_max_seconds return workflow_run_wait_deadline( - now=now, execution_deadline_at=getattr(context, 'execution_deadline_at', None), + now=now, execution_deadline_at=( + execution_deadline(context).isoformat() if getattr(context, 'execution_deadline_at', None) else None + ), cap_seconds=get_chat_orchestration_workflow_run_wait_max_seconds(settings), step_timeout_seconds=positive_setting_int(settings, 'chat_orchestration_step_timeout_seconds', 120), dependents=waited_run_dependents(getattr(context, 'plan_steps', None) or [], step_id), diff --git a/application/single_app/functions_workspace_identities.py b/application/single_app/functions_workspace_identities.py index ee0cee459..d02882dd2 100644 --- a/application/single_app/functions_workspace_identities.py +++ b/application/single_app/functions_workspace_identities.py @@ -28,6 +28,7 @@ ui_trigger_word, ) from functions_settings import get_settings +from cosmos_document_utils import normalize_cosmos_document WORKSPACE_IDENTITY_SCOPE_PERSONAL = "personal" @@ -435,12 +436,15 @@ def create_workspace_identity( created_by: str, *, stage_secrets: bool = False, + identity_id: Optional[str] = None, ) -> Dict[str, Any]: scope_type = _validate_scope(scope_type) - identity_id = str(uuid.uuid4()) + deterministic_identity_id = identity_id is not None + identity_id = identity_id or str(uuid.uuid4()) staged_names: Optional[List[str]] = [] if stage_secrets else None normalized_payload = _normalize_identity_payload( - scope_type, scope_id, payload or {}, identity_id, staged_names=staged_names + scope_type, scope_id, payload or {}, identity_id, + fresh_names=stage_secrets and deterministic_identity_id, staged_names=staged_names, ) scope_field = _scope_field(scope_type) now_iso = _now_iso() @@ -457,20 +461,29 @@ def create_workspace_identity( **normalized_payload, } try: - _get_identities_container(scope_type).create_item(body=identity) + stored = _get_identities_container(scope_type).create_item(body=identity) except Exception: - # The create failed after secrets may have been stored; drop them so a - # failed create never leaves an orphan behind in Key Vault. + # A lost write response is not proof that staged secrets are unused. + # Confirm the owned point record before cleanup or an idempotent return. + try: + current = get_workspace_identity(scope_type, scope_id, identity_id) + except LookupError: + current = None + except Exception: + raise + if current is not None and all(current.get(key) == value for key, value in identity.items()): + return normalize_cosmos_document(current) if staged_names: keyvault_identity_discard_staged_helper(staged_names, scope_id, scope=_keyvault_scope(scope_type)) raise - return identity + return normalize_cosmos_document(stored) def update_workspace_identity(scope_type: str, scope_id: str, identity_id: str, payload: Dict[str, Any], updated_by: str) -> Dict[str, Any]: identity = get_workspace_identity(scope_type, scope_id, identity_id) normalized_payload = _normalize_identity_payload(scope_type, scope_id, payload or {}, identity_id, existing_identity=identity) identity.update(normalized_payload) + identity.pop("credential_operation_id", None) identity["updated_by"] = updated_by identity["updated_at"] = _now_iso() _get_identities_container(scope_type).upsert_item(identity) @@ -480,9 +493,26 @@ def update_workspace_identity(scope_type: str, scope_id: str, identity_id: str, def delete_workspace_identity(scope_type: str, scope_id: str, identity_id: str, deleted_by: str) -> Dict[str, Any]: identity = get_workspace_identity(scope_type, scope_id, identity_id) _get_identities_container(scope_type).delete_item(item=identity["id"], partition_key=scope_id) + _invalidate_personal_action_bindings(scope_type, scope_id, identity_id) return {"identity_id": identity_id, "deleted_by": deleted_by} +def _invalidate_personal_action_bindings(scope_type, scope_id, identity_id): + # Initialized identity mutations run after the application supplies private state storage. + # Resolution also checks the identity point read, so a cleanup failure cannot reuse it. + if scope_type == WORKSPACE_IDENTITY_SCOPE_PERSONAL: + import functions_action_credential_state + + if functions_action_credential_state.is_configured(): + try: + functions_action_credential_state.invalidate_identity(scope_id, identity_id) + except Exception as error: + # The identity deletion already committed; resolver point reads + # refuse it even if best-effort binding cleanup is unavailable. + log_event("[IDENTITIES] Personal action binding cleanup failed.", + extra={"error_type": type(error).__name__}, level=logging.WARNING) + + class WorkspaceIdentityConflict(Exception): """A conditional identity write failed because the stored record changed. @@ -500,6 +530,8 @@ def update_workspace_identity_conditional( payload: Dict[str, Any], updated_by: str, expected_etag: str, + *, + operation_id: Optional[str] = None, ) -> Dict[str, Any]: """Update an identity only if its stored ``_etag`` still matches ``expected_etag``. @@ -514,6 +546,8 @@ def update_workspace_identity_conditional( :class:`LookupError` (404) and is never recreated. """ identity = get_workspace_identity(scope_type, scope_id, identity_id) + if operation_id and identity.get("credential_operation_id") == operation_id: + return normalize_cosmos_document(identity) current_etag = identity.get("_etag") if not expected_etag or expected_etag != current_etag: raise WorkspaceIdentityConflict("Workspace identity changed before the update") @@ -532,6 +566,10 @@ def update_workspace_identity_conditional( updated.update(normalized_payload) updated["updated_by"] = updated_by updated["updated_at"] = _now_iso() + if operation_id: + updated["credential_operation_id"] = operation_id + else: + updated.pop("credential_operation_id", None) container = _get_identities_container(scope_type) try: @@ -541,20 +579,30 @@ def update_workspace_identity_conditional( etag=expected_etag, match_condition=MatchConditions.IfNotModified, ) - except CosmosResourceNotFoundError: - keyvault_identity_discard_staged_helper(staged_names, scope_id, scope=_keyvault_scope(scope_type)) - raise LookupError("Workspace identity not found") - except CosmosAccessConditionFailedError: - keyvault_identity_discard_staged_helper(staged_names, scope_id, scope=_keyvault_scope(scope_type)) - raise WorkspaceIdentityConflict("Workspace identity changed before the update") - except Exception: - keyvault_identity_discard_staged_helper(staged_names, scope_id, scope=_keyvault_scope(scope_type)) - raise + except Exception as error: + try: + current = get_workspace_identity(scope_type, scope_id, identity_id) + except LookupError: + current = None + except Exception: + # Preserve possibly live staged references when confirmation is unavailable. + raise error + if current is not None and all( + current.get(key) == value for key, value in updated.items() if not key.startswith("_") + ): + stored = current + else: + keyvault_identity_discard_staged_helper(staged_names, scope_id, scope=_keyvault_scope(scope_type)) + if isinstance(error, CosmosResourceNotFoundError): + raise LookupError("Workspace identity not found") from None + if isinstance(error, CosmosAccessConditionFailedError) or getattr(error, "status_code", None) == 412: + raise WorkspaceIdentityConflict("Workspace identity changed before the update") from None + raise keyvault_identity_cleanup_helper( previous_auth, normalized_payload.get("auth") or {}, scope_id, scope=_keyvault_scope(scope_type) ) - return stored if isinstance(stored, dict) else updated + return normalize_cosmos_document(stored) def delete_workspace_identity_conditional( @@ -592,6 +640,7 @@ def delete_workspace_identity_conditional( keyvault_identity_delete_helper( dict(identity.get("auth") or {}), scope_id, scope=_keyvault_scope(scope_type) ) + _invalidate_personal_action_bindings(scope_type, scope_id, identity_id) return {"identity_id": identity_id, "deleted_by": deleted_by} diff --git a/application/single_app/json_schema_validation.py b/application/single_app/json_schema_validation.py index fe7d7c891..9ddf6003f 100644 --- a/application/single_app/json_schema_validation.py +++ b/application/single_app/json_schema_validation.py @@ -13,6 +13,7 @@ is_retired_mcp_stdio, resolve_action_type, ) +from action_credential_contracts import validate_credential_requirements from functions_mcp_operations import normalize_mcp_transport, validate_mcp_endpoint_for_transport from functions_agent_delegation import ( AGENT_ACTION_VALIDATION_ERROR, @@ -308,8 +309,9 @@ def validate_plugin(plugin): return MCP_STDIO_REMOVED_MESSAGE try: resolve_action_type(plugin) + validate_credential_requirements(plugin) except ValueError: - return 'Invalid action type.' + return 'Invalid action type or credential requirements.' try: plugin_copy = apply_plugin_validation_defaults(plugin) except ValueError: diff --git a/application/single_app/route_backend_action_credentials.py b/application/single_app/route_backend_action_credentials.py new file mode 100644 index 000000000..ff8181c20 --- /dev/null +++ b/application/single_app/route_backend_action_credentials.py @@ -0,0 +1,264 @@ +# route_backend_action_credentials.py +"""Credential-only interaction API. General workspace authoring stays unchanged.""" + +from functools import wraps +import hmac +import secrets +import uuid +from datetime import datetime, timezone + +from flask import Response, jsonify, request, session, stream_with_context + +from action_credential_contracts import ActionCredentialError, ActionCredentialsRequired +from functions_authentication import get_current_user_id, login_required, user_required +from functions_appinsights import log_event +from functions_action_credentials import ( + authorize_conversation, authorize_request, credential_continuation, safe_request, submit_credentials, +) +import functions_action_credential_state as state +from swagger_wrapper import swagger_route, get_auth_security + + +def _csrf_token(): + if not session.get("action_credentials_csrf_token"): + session["action_credentials_csrf_token"] = secrets.token_urlsafe(32) + return session["action_credentials_csrf_token"] + + +def _body(fields): + expected = session.get("action_credentials_csrf_token") or "" + supplied = request.headers.get("X-CSRFToken") or "" + if not expected or not hmac.compare_digest(expected, supplied): + raise ActionCredentialError("credential_csrf_invalid", 403) + if request.content_length is not None and request.content_length > 32768: + raise ActionCredentialError("invalid_request", 400) + value = request.get_json(silent=True) + if type(value) is not dict or set(value) != fields: + raise ActionCredentialError("invalid_request", 400) + if type(value.get("expected_etag")) is not str or not value["expected_etag"]: + raise ActionCredentialError("invalid_request", 400) + return value + + +def _credential_boundary(function): + @wraps(function) + def guarded(*args, **kwargs): + try: + response = function(*args, **kwargs) + except (PermissionError, LookupError): + response = jsonify({"error": "Credential request not found.", "code": "not_found"}), 404 + except ActionCredentialError as error: + response = jsonify({"error": "Refresh the credential request and try again.", "code": error.code}), error.status_code + except Exception as error: + # No payloads, provider text, or exception tracebacks at this secret-bearing boundary. + log_event("[PLUGINS] Credential interaction failed.", extra={"error_type": type(error).__name__}) + response = jsonify({"error": "Action credentials could not be saved or resumed.", "code": "credentials_unavailable"}), 503 + from flask import make_response + + result = make_response(response) + result.headers["Cache-Control"] = "no-store" + result.headers["Pragma"] = "no-cache" + return result + return guarded + + +def _finish_request(user_id, request_id, claim, status): + current = state.read(user_id, request_id, record_type="credential_request") + if current["status"] == "continuing" and current.get("continuation_claim") == claim: + state.replace(user_id, current, {"status": status}, current["_etag"]) + + +class _CredentialExecution: + """Complete interaction state on the durable worker, not the browser socket.""" + + def __init__(self, execution, item, user_id): + self.execution, self.item, self.user_id = execution, item, user_id + + def execute(self, **kwargs): + try: + yield from self.execution.execute(**kwargs) + except Exception: + _finish_request(self.user_id, self.item["id"], self.item["continuation_claim"], "failed") + raise + else: + _finish_request(self.user_id, self.item["id"], self.item["continuation_claim"], "completed") + + def close(self): + return self.execution.close() + + +def _orchestration_continue(item, user_id): + from agent_execution_context import capture_execution_identity + from config import cosmos_messages_container + from functions_settings import get_settings + from functions_orchestration_continuation import ContinuationCheckpoints, claim_run_continuation + from functions_orchestration_execution import prepare_harness_execution + from route_backend_orchestration import ( + _conversation_context_for_run, _request_identity, _stream_execution, + ) + + conversation_id, run_id = item["conversation_id"], item["continuation"]["run_id"] + owned = claim_run_continuation( + run_id, user_id, conversation_id, + authorize=lambda: authorize_conversation(user_id, conversation_id), + message_container=cosmos_messages_container, credential_request_id=item["id"], + ) + if owned is None: + raise ActionCredentialError("credential_execution_busy") + record, lease = owned + try: + settings = get_settings() + snapshot = _conversation_context_for_run(record, user_id, settings) + record = {**record, "conversation_context": snapshot} + # Harness initialization reacquires current capability/source permissions. + execution = prepare_harness_execution( + record, settings=settings, identity_context=_request_identity(user_id), + execution_identity=capture_execution_identity(user_id, conversation_id), + lease=lease, checkpoint_factory=ContinuationCheckpoints, + ) + except BaseException: + lease.close() + raise + return _stream_execution( + _CredentialExecution(execution, item, user_id), + run_id=run_id, conversation_id=conversation_id, settings=settings, + ) + + +def _chat_continue(item, user_id): + from agent_execution_context import AgentExecutionFrame, DelegationBudget, agent_execution, capture_execution_identity + from agent_delegation_runtime import _build_local_agent, close_agent_resources + from functions_agent_delegation import resolve_delegation_agent + from functions_settings import get_settings + from functions_action_credential_continuation import invoke_credential_agent + from route_backend_chats import _persist_screened_assistant + from functions_orchestration_events import serialize_sse + import asyncio + + descriptor = {**item["continuation"], "resume": True} + checkpoint = state.read(user_id, descriptor["checkpoint_id"], record_type="credential_checkpoint") + if checkpoint["conversation_id"] != item["conversation_id"] or "root" not in checkpoint["value"]: + raise ActionCredentialError("credential_checkpoint_changed") + identity = capture_execution_identity(user_id, item["conversation_id"]) + settings = get_settings() + target = resolve_delegation_agent(descriptor["agent_reference"], user_id=user_id, settings=settings) + frame = AgentExecutionFrame(identity, target, DelegationBudget()) + # Reconstruct a new invocation-local kernel under the current actor's bridge. + with identity.bridge(target), agent_execution(frame): + kernel, agent = _build_local_agent(target, settings) + + async def execute(): + with identity.bridge(target), agent_execution(frame), credential_continuation(descriptor): + try: + return await invoke_credential_agent(agent, None) + finally: + await close_agent_resources(kernel) + + @stream_with_context + def frames(): + try: + result = asyncio.run(execute()) + authorize_conversation(user_id, item["conversation_id"]) + message = { + "id": uuid.uuid4().hex, "conversation_id": item["conversation_id"], "user_id": user_id, + "role": "assistant", "content": str(result), + "created_at": datetime.now(timezone.utc).isoformat(), + "metadata": {"action_credential_continuation": True}, + } + saved = _persist_screened_assistant(message, user_id, settings=settings) + _finish_request(user_id, item["id"], item["continuation_claim"], "completed") + yield serialize_sse({"content": saved["content"]}) + yield serialize_sse({ + "done": True, "conversation_id": item["conversation_id"], "message_id": saved["id"], + "reload_messages": True, + }) + except ActionCredentialsRequired as error: + _finish_request(user_id, item["id"], item["continuation_claim"], "superseded") + yield serialize_sse(error.event) + except Exception as error: + _finish_request(user_id, item["id"], item["continuation_claim"], "failed") + log_event("[PLUGINS] Credential continuation stopped.", extra={"error_type": type(error).__name__}) + yield serialize_sse({"error": "Saved work could not be continued. Refresh this conversation."}) + + return Response(frames(), mimetype="text/event-stream", headers={"Cache-Control": "no-store"}) + + +def register_route_backend_action_credentials(bp): + @bp.route("/api/action-credentials/pending", methods=["GET"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + @_credential_boundary + def action_credentials_pending(): + user_id = get_current_user_id() + conversation_id = request.args.get("conversation_id") + authorize_conversation(user_id, conversation_id) + requests = [] + for row in state.pending(user_id, conversation_id): + try: + item, manifest = authorize_request(user_id, row["id"]) + requests.append(safe_request(item, manifest)) + except (PermissionError, LookupError, ActionCredentialError): + continue + return jsonify({"requests": requests, "csrf_token": _csrf_token()}) + + @bp.route("/api/action-credentials/", methods=["GET"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + @_credential_boundary + def action_credentials_request(request_id): + item, manifest = authorize_request(get_current_user_id(), request_id) + return jsonify({"credential_request": safe_request(item, manifest), "csrf_token": _csrf_token()}) + + @bp.route("/api/action-credentials//submit", methods=["POST"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + @_credential_boundary + def action_credentials_submit(request_id): + body = _body({"expected_etag", "slots"}) + dto = submit_credentials(get_current_user_id(), request_id, body["expected_etag"], body["slots"]) + return jsonify({"credential_request": dto}) + + @bp.route("/api/action-credentials//continue", methods=["POST"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + @_credential_boundary + def action_credentials_continue(request_id): + body = _body({"expected_etag"}) + user_id = get_current_user_id() + item, _manifest = authorize_request(user_id, request_id) + descriptor = item.get("continuation") or {} + if descriptor.get("mode") not in {"chat", "orchestration"}: + raise ActionCredentialError("credential_continuation_unavailable") + claimed = state.claim_request(user_id, item, body["expected_etag"]) + try: + return _orchestration_continue(claimed, user_id) if descriptor["mode"] == "orchestration" else _chat_continue(claimed, user_id) + except Exception: + # Neither response generator has started: an admission/preparation + # failure cannot have executed the blocked operation. + _finish_request(user_id, claimed["id"], claimed["continuation_claim"], "ready") + raise + + @bp.route("/api/action-credentials//cancel", methods=["POST"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + @_credential_boundary + def action_credentials_cancel(request_id): + body = _body({"expected_etag"}) + user_id = get_current_user_id() + item, manifest = authorize_request(user_id, request_id) + if item["status"] not in {"pending", "ready"}: + raise ActionCredentialError("credential_request_closed") + item = state.replace(user_id, item, {"status": "cancelled"}, body["expected_etag"]) + if item["continuation"].get("run_id"): + from functions_orchestration_recovery import request_cancellation + + request_cancellation( + item["continuation"]["run_id"], user_id, item["conversation_id"], + lambda: authorize_conversation(user_id, item["conversation_id"]), + ) + return jsonify({"credential_request": safe_request(item, manifest)}) diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index 80ecd96e3..e7619c8d5 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -14,6 +14,7 @@ ) from content_screening.contracts import DocumentHeldError, ScreeningConflictError, ScreeningError, document_is_available from agent_execution_context import AgentExecutionCancelled, DelegationBudget +from action_credential_contracts import ActionCredentialsRequired from agent_delegation_runtime import AgentExecution, delegation_citations, delegation_usage, prepare_agent_execution from semantic_kernel import Kernel from semantic_kernel.agents import ChatCompletionAgent @@ -20757,7 +20758,7 @@ def try_fallback_chain(steps): try: result = step['func']() return step['on_success'](result) - except (M365ApprovalRequired, M365SignInRequired): + except (M365ApprovalRequired, M365SignInRequired, ActionCredentialsRequired): raise except Exception as e: log_event( @@ -22018,6 +22019,8 @@ def gpt_error(e): 'thoughts_enabled': thought_tracker.enabled })), 200 + except ActionCredentialsRequired as error: + return jsonify(error.event), 409 except M365ApprovalRequired as error: return jsonify(record_m365_pending( error, @@ -25578,6 +25581,10 @@ def finalize_cancelled_agent_stream_response(): ) continue raise + except ActionCredentialsRequired as error: + plugin_logger_cb.deregister_callbacks(callback_key) + yield f"data: {json.dumps(error.event)}\n\n" + return except (M365ApprovalRequired, M365SignInRequired): plugin_logger_cb.deregister_callbacks(callback_key) raise diff --git a/application/single_app/route_backend_conversations.py b/application/single_app/route_backend_conversations.py index 15ff7fd45..564f39ad9 100644 --- a/application/single_app/route_backend_conversations.py +++ b/application/single_app/route_backend_conversations.py @@ -2149,6 +2149,10 @@ def delete_conversation(conversation_id): ) try: + # Ephemeral credential history is never archived with ordinary messages. + from functions_action_credential_state import delete_private_state + + delete_private_state(user_id, conversation_id=conversation_id) cosmos_conversations_container.delete_item( item=conversation_id, partition_key=conversation_id @@ -2282,6 +2286,9 @@ def delete_multiple_conversations(): # Delete the conversation stage = 'conversation_delete' + from functions_action_credential_state import delete_private_state + + delete_private_state(user_id, conversation_id=conversation_id) cosmos_conversations_container.delete_item( item=conversation_id, partition_key=conversation_id diff --git a/application/single_app/semantic_kernel_loader.py b/application/single_app/semantic_kernel_loader.py index 8b8d3adb4..793257a31 100644 --- a/application/single_app/semantic_kernel_loader.py +++ b/application/single_app/semantic_kernel_loader.py @@ -2615,7 +2615,10 @@ def hydrate_workspace_identity_in_plugin(plugin_manifest): return plugin_manifest if is_retired_mcp_stdio(plugin_manifest): raise McpStdioRemovedError() - if not get_action_identity_reference_id(plugin_manifest): + if not get_action_identity_reference_id(plugin_manifest) and not ( + isinstance(plugin_manifest.get("additionalFields"), dict) + and plugin_manifest["additionalFields"].get("basic_auth_identity_id") + ): return plugin_manifest mcp_action = is_mcp_action(plugin_manifest) diff --git a/application/single_app/semantic_kernel_plugins/plugin_health_checker.py b/application/single_app/semantic_kernel_plugins/plugin_health_checker.py index 39e77a053..d3dbbe10e 100644 --- a/application/single_app/semantic_kernel_plugins/plugin_health_checker.py +++ b/application/single_app/semantic_kernel_plugins/plugin_health_checker.py @@ -346,6 +346,12 @@ def validate_plugin_manifest(manifest: Dict[str, Any], plugin_type: str) -> Tupl errors.append(f"Tableau additionalFields.{field_name} must be between {minimum} and {maximum}") elif plugin_type == YAMCS_PLUGIN_TYPE: + from action_credential_contracts import validate_credential_requirements + + try: + personal_slots = validate_credential_requirements(manifest) + except ValueError: + return False, ["Invalid personal action credential requirements."] auth = manifest.get('auth', {}) if isinstance(manifest.get('auth'), dict) else {} auth_type = str(auth.get('type') or 'username_password').strip() raw_additional_fields = manifest.get('additionalFields', {}) @@ -359,6 +365,8 @@ def validate_plugin_manifest(manifest: Dict[str, Any], plugin_type: str) -> Tupl identity_id = str(manifest.get('identity_id') or '').strip() raw_auth_method = str(raw_additional_fields.get('auth_method') or '').strip().lower() auth_method = additional_fields.get('auth_method') + if "service" in personal_slots: + auth_method = personal_slots["service"]["auth_type"] if parsed_endpoint.scheme not in {'http', 'https'} or not parsed_endpoint.netloc: errors.append("Yamcs plugin requires an http(s) Yamcs server URL") @@ -368,7 +376,9 @@ def validate_plugin_manifest(manifest: Dict[str, Any], plugin_type: str) -> Tupl errors.append("Yamcs plugin supports auth.type values 'NoAuth', 'key', 'identity', or 'username_password'") if raw_auth_method and raw_auth_method not in YAMCS_SUPPORTED_AUTH_METHODS: errors.append("Yamcs plugin supports additionalFields.auth_method values 'username_password', 'api_key', 'bearer_token', or 'none'") - if auth_type == 'identity' and not auth.get('identity') and not identity_id: + if "service" in personal_slots: + pass + elif auth_type == 'identity' and not auth.get('identity') and not identity_id: errors.append("Yamcs reusable identity auth requires auth.identity or identity_id") elif auth_method == YAMCS_AUTH_METHOD_USERNAME_PASSWORD: if auth_type == 'username_password' and (not auth.get('identity') or not auth.get('key')): @@ -385,7 +395,7 @@ def validate_plugin_manifest(manifest: Dict[str, Any], plugin_type: str) -> Tupl errors.append(YAMCS_BASIC_AUTH_CONFLICT_MESSAGE) # A referenced identity supplies both values at runtime, so only unreferenced # configurations must carry an inline username and password. - if not basic_auth_identity_id: + if not basic_auth_identity_id and "proxy" not in personal_slots: if not additional_fields.get('basic_auth_username'): errors.append("Yamcs HTTP Basic authentication requires additionalFields.basic_auth_username") if not additional_fields.get('basic_auth_password'): diff --git a/application/single_app/semantic_kernel_plugins/yamcs_plugin.py b/application/single_app/semantic_kernel_plugins/yamcs_plugin.py index 8fecb0b56..ab65900ff 100644 --- a/application/single_app/semantic_kernel_plugins/yamcs_plugin.py +++ b/application/single_app/semantic_kernel_plugins/yamcs_plugin.py @@ -10,6 +10,9 @@ from typing import Any, Callable, Dict, Iterable, List, Optional from semantic_kernel.functions import kernel_function +from action_credential_contracts import ( + ADAPTERS, ActionCredentialError, credential_slots, validate_credential_requirements, +) from functions_appinsights import log_event from functions_debug import debug_print @@ -228,6 +231,7 @@ def get_functions(self) -> List[str]: ] def _validate_configuration(self) -> None: + personal = validate_credential_requirements(self.manifest) if not self.server_url: raise ValueError("Yamcs action requires a server URL.") if not self.instance: @@ -241,6 +245,8 @@ def _validate_configuration(self) -> None: "Yamcs action supports auth methods username_password, api_key, bearer_token, or none." ) self._validate_basic_auth_configuration() + if "service" in personal: + return if self.auth_type == "identity": if not (self._auth.get("identity") or self.manifest.get("identity_id")): raise ValueError("Yamcs reusable identity auth requires auth.identity or identity_id.") @@ -260,6 +266,8 @@ def _validate_basic_auth_configuration(self) -> None: return if yamcs_basic_auth_conflicts_with_auth_method(self.auth_method): raise ValueError(YAMCS_BASIC_AUTH_CONFLICT_MESSAGE) + if "proxy" in credential_slots(self.manifest): + return if not self.basic_auth_username: raise ValueError( "Yamcs HTTP Basic authentication requires additionalFields.basic_auth_username." @@ -276,7 +284,10 @@ def _build_credentials(self): return self._build_basic_auth_credentials() if self.enable_basic_auth else None try: - from yamcs.client import APIKeyCredentials, Credentials + try: + from yamcs.client import Credentials + except ImportError: + from yamcs.core.auth import Credentials except ImportError as exc: raise ImportError( "Yamcs client library is not installed. Install yamcs-client to use Yamcs actions." @@ -284,6 +295,19 @@ def _build_credentials(self): auth_key = str(self._auth.get("key") or "") if self.auth_method == YAMCS_AUTH_METHOD_API_KEY: + try: + from yamcs.client import APIKeyCredentials + except ImportError: + # yamcs-client 1.8.8 exposes core Credentials, but no API-key + # holder. Its Context calls before_request for every request. + class APIKeyCredentials(Credentials): + def __init__(self, api_key): + super().__init__() + self.api_key = api_key + + def before_request(self, session, auth_url): + session.headers.update({"x-api-key": self.api_key}) + return APIKeyCredentials(auth_key) if self.auth_method == YAMCS_AUTH_METHOD_BEARER_TOKEN: return Credentials(access_token=auth_key) @@ -296,7 +320,10 @@ def _build_basic_auth_credentials(self): running an older yamcs-client keeps working for every non-proxy auth method. """ try: - from yamcs.client import BasicAuthCredentials + try: + from yamcs.client import BasicAuthCredentials + except ImportError: + from yamcs.core.auth import BasicAuthCredentials except ImportError as exc: raise ImportError( "Yamcs HTTP Basic authentication requires yamcs-client 1.8.8 or newer. " @@ -319,9 +346,24 @@ def _connect(self): f"tls_verify={self.tls_verify} timeout={self.timeout} " f"basic_auth_enabled={self.enable_basic_auth}" ) + credentials = self._build_credentials() + if credentials is not None and getattr(self, "_credential_status_transport", False): + plugin = self + delegate_credentials = credentials + + class BoundCredentials: + def login(self, session, auth_root, on_token_update): + session.request = plugin._with_timeout(session.request) + try: + return delegate_credentials.login(session, auth_root, on_token_update) + except Exception: + session.close() + raise + + credentials = BoundCredentials() client = YamcsClient( self.server_url, - credentials=self._build_credentials(), + credentials=credentials, tls_verify=self.tls_verify, user_agent="SimpleChat", ) @@ -353,7 +395,14 @@ def _with_timeout(self, request_callable: Callable) -> Callable: def request_with_timeout(*args, **kwargs): kwargs.setdefault("timeout", configured_timeout) - return request_callable(*args, **kwargs) + response = request_callable(*args, **kwargs) + if getattr(self, "_credential_status_transport", False) and response.status_code in {401, 403, 407}: + # yamcs-client discards HTTP status on several error paths. + # Preserve it before the SDK parses provider bodies/exceptions. + error = ActionCredentialError("yamcs_authentication_status") + error.status_code = response.status_code + raise error + return response return request_with_timeout @@ -459,18 +508,55 @@ def _run(self, operation_name: str, operation: Callable[[Any], Dict[str, Any]]) """Open a Yamcs client, run a read-only operation, and normalize failures.""" client = None try: - client = self._connect() + if credential_slots(self.manifest): + # Resolve only when a function is actually invoked, never while agents load. + from functions_action_credentials import resolve_personal_credentials + + prepared = resolve_personal_credentials(self.manifest) + prepared.pop("credential_requirements", None) + runtime = YamcsPlugin(prepared) + runtime._credential_status_transport = True + client = runtime._connect() + else: + client = self._connect() result = operation(client) debug_print( f"[YAMCS_PLUGIN] {operation_name} succeeded instance={self.instance} " f"row_count={result.get('row_count')} truncated={result.get('truncated')}" ) return result + except ActionCredentialError as error: + if error.code != "yamcs_authentication_status": + raise + from functions_action_credentials import resolve_personal_credentials + + rejected = ADAPTERS["yamcs"].rejected_slots(self.manifest, error) + if rejected: + resolve_personal_credentials(self.manifest, rejected_slots=rejected) + return self._error_response("Yamcs access was denied.", error_type="permission") except ValueError as exc: + if credential_slots(self.manifest): + return self._error_response("Yamcs configuration or response is invalid.", error_type="validation") return self._error_response(str(exc), error_type="validation") except ImportError as exc: + if credential_slots(self.manifest): + return self._error_response("The Yamcs client dependency is unavailable.", error_type="dependency") return self._error_response(str(exc), error_type="dependency") except Exception as exc: + if credential_slots(self.manifest): + from functions_action_credentials import resolve_personal_credentials + + rejected = ADAPTERS["yamcs"].rejected_slots(self.manifest, exc) + if rejected: + resolve_personal_credentials(self.manifest, rejected_slots=rejected) + status = getattr(getattr(exc, "response", None), "status_code", None) + if status is None: + status = getattr(exc, "status_code", None) + # Provider bodies/exception messages can contain submitted credentials. + return self._error_response( + "Yamcs access was denied." if status == 403 else "Yamcs is unavailable. Please retry later.", + error_type="permission" if status == 403 else "service", operation=operation_name, + ) message = self._safe_error_message(exc, f"Yamcs {operation_name} failed.") debug_print( f"[YAMCS_PLUGIN] {operation_name} failed server_url={self.server_url} " @@ -491,10 +577,20 @@ def _run(self, operation_name: str, operation: Callable[[Any], Dict[str, Any]]) finally: if client is not None: try: - client.close() + close_client = getattr(client, "close", None) + if callable(close_client): + close_client() debug_print("[YAMCS_PLUGIN] Yamcs connection closed.") except Exception: pass + finally: + session = getattr(getattr(client, "ctx", None), "session", None) + close_session = getattr(session, "close", None) + if callable(close_session): + try: + close_session() + except Exception: + pass def _first_word(self, statement: str) -> str: first_word_match = self._FIRST_WORD_PATTERN.match(statement or "") diff --git a/application/single_app/semantic_kernel_plugins/yamcs_plugin_factory.py b/application/single_app/semantic_kernel_plugins/yamcs_plugin_factory.py index 0c2adb269..90624d40f 100644 --- a/application/single_app/semantic_kernel_plugins/yamcs_plugin_factory.py +++ b/application/single_app/semantic_kernel_plugins/yamcs_plugin_factory.py @@ -2,6 +2,8 @@ """Factory for creating Yamcs Semantic Kernel plugins from action manifests.""" from typing import Any, Dict +from functions_action_manifest import copy_action_manifest +from action_credential_contracts import credential_slots from functions_yamcs_operations import ( YAMCS_AUTH_METHOD_API_KEY, @@ -24,7 +26,7 @@ def create_from_config(cls, config: Dict[str, Any]) -> YamcsPlugin: @classmethod def normalize_manifest(cls, config: Dict[str, Any]) -> Dict[str, Any]: - manifest = dict(config or {}) + manifest = copy_action_manifest(config or {}) auth = manifest.get("auth") if isinstance(manifest.get("auth"), dict) else {} auth = dict(auth) additional_fields = manifest.get("additionalFields") if isinstance(manifest.get("additionalFields"), dict) else {} @@ -34,7 +36,10 @@ def normalize_manifest(cls, config: Dict[str, Any]) -> Dict[str, Any]: auth["type"] = auth_type additional_fields = normalize_yamcs_additional_fields(additional_fields, auth_type=auth_type) - if auth_type == "username_password": + personal = credential_slots(manifest) + if "service" in personal: + additional_fields["auth_method"] = personal["service"]["auth_type"] + elif auth_type == "username_password": additional_fields["auth_method"] = YAMCS_AUTH_METHOD_USERNAME_PASSWORD elif auth_type == "NoAuth": additional_fields["auth_method"] = YAMCS_AUTH_METHOD_NONE diff --git a/application/single_app/simplechat_scheduler.py b/application/single_app/simplechat_scheduler.py index 13ec5d06b..886b1ae64 100644 --- a/application/single_app/simplechat_scheduler.py +++ b/application/single_app/simplechat_scheduler.py @@ -8,7 +8,9 @@ from background_tasks import run_scheduler_forever import functions_redis_client -from config import get_redis_cache_infrastructure_endpoint, initialize_clients +from config import cosmos_action_credential_state_container, get_redis_cache_infrastructure_endpoint, initialize_clients +from functions_action_credential_state import configure_action_credential_state +from functions_action_credential_runtime import configure_action_credential_runtime from functions_appinsights import setup_appinsights_logging from functions_settings import configure_application_cache, get_settings @@ -24,6 +26,8 @@ def initialize_scheduler_runtime(): redis_client_factory=functions_redis_client.create_redis_client, ) initialize_clients(settings) + configure_action_credential_state(cosmos_action_credential_state_container) + configure_action_credential_runtime() setup_appinsights_logging(settings) logging.basicConfig(level=logging.DEBUG) print('SimpleChat scheduler runtime initialized.') diff --git a/application/single_app/static/js/chat/chat-action-credentials.js b/application/single_app/static/js/chat/chat-action-credentials.js new file mode 100644 index 000000000..884376f13 --- /dev/null +++ b/application/single_app/static/js/chat/chat-action-credentials.js @@ -0,0 +1,424 @@ +// chat-action-credentials.js +// Bootstrap task UI; credential inputs belong only to this form, never a chat draft. +(() => { + 'use strict'; + + const cards = new Map(); + const authTypes = new Set(['username_password', 'api_key', 'bearer_token']); + let conversationId = null; + let epoch = 0; + let sequence = 0; + let csrfToken = null; + let continuationConsumer = null; + let pendingController = null; + + function node(tag, className = '', text = '') { + const element = document.createElement(tag); + element.className = className; + element.textContent = text; + return element; + } + + function active(id, capturedEpoch = epoch) { + return capturedEpoch === epoch && id === conversationId + && id === window.currentConversationId; + } + + function clearInputs(card) { + card.querySelectorAll('input').forEach(input => { input.value = ''; }); + } + + function removeCard(id) { + const state = cards.get(id); + if (!state) return; + clearInputs(state.card); + state.controller?.abort(); + state.card.remove(); + cards.delete(id); + } + + function clear() { + epoch += 1; + pendingController?.abort(); + pendingController = null; + Array.from(cards.keys()).forEach(removeCard); + document.getElementById('chat-action-credentials')?.remove(); + conversationId = null; + csrfToken = null; + } + + function safeRequest(value) { + if (!value || typeof value.id !== 'string' || typeof value.etag !== 'string' + || typeof value.conversation_id !== 'string' || !Array.isArray(value.requirements) + || !Array.isArray(value.identities)) return null; + const requirements = value.requirements.filter(requirement => + requirement && ['service', 'proxy'].includes(requirement.slot) && authTypes.has(requirement.auth_type) + && (requirement.slot !== 'proxy' || requirement.auth_type === 'username_password')); + if (requirements.length !== value.requirements.length + || new Set(requirements.map(item => item.slot)).size !== requirements.length) return null; + return { + id: value.id, + etag: value.etag, + conversation_id: value.conversation_id, + action_name: typeof value.action_name === 'string' ? value.action_name : 'Action', + destination_label: typeof value.destination_label === 'string' ? value.destination_label : 'Configured service', + reason: typeof value.reason === 'string' ? value.reason : 'setup_required', + status: value.status, + requirements: requirements.map(item => ({ + slot: item.slot, auth_type: item.auth_type, + label: typeof item.label === 'string' ? item.label : '', + })), + identities: value.identities.filter(identity => + identity && typeof identity.id === 'string' && typeof identity.name === 'string' && authTypes.has(identity.auth_type)) + .map(identity => ({ + id: identity.id, name: identity.name, auth_type: identity.auth_type, + etag: typeof identity.etag === 'string' ? identity.etag : null, + })), + }; + } + + function safeError(status) { + if (status === 401 || status === 403) return 'Your session or access has changed. Sign in again or reopen the original conversation.'; + if (status === 409 || status === 410 || status === 412) return 'This request has changed or expired. Refresh the request before continuing.'; + return 'Unable to complete this request. Check your connection and try again.'; + } + + async function request(path, options = {}) { + const capturedEpoch = epoch; + const headers = { 'Content-Type': 'application/json', ...(options.headers || {}) }; + if (csrfToken) headers['X-CSRFToken'] = csrfToken; + const response = await fetch(path, { + ...options, headers, credentials: 'same-origin', cache: 'no-store', + body: options.body ? JSON.stringify(options.body) : undefined, + }); + if (!response.ok) throw new Error(safeError(response.status)); + const result = await response.json(); + if (capturedEpoch === epoch && typeof result.csrf_token === 'string') csrfToken = result.csrf_token; + return result; + } + + function endpoint(id, action = '') { + return `/api/action-credentials/${encodeURIComponent(id)}${action ? `/${action}` : ''}`; + } + + function root() { + const chatbox = document.getElementById('chatbox'); + if (!chatbox) return null; + let container = document.getElementById('chat-action-credentials'); + if (!container) { + container = node('section', 'my-3'); + container.id = 'chat-action-credentials'; + container.setAttribute('aria-label', 'Private action credential requests'); + chatbox.appendChild(container); + } + return container; + } + + function field(parent, labelText, type, required = true) { + const group = node('div', 'mb-3'); + const input = node('input', 'form-control'); + input.id = `action-credential-field-${++sequence}`; + input.type = type; + input.required = required; + input.autocomplete = type === 'password' ? 'new-password' : 'off'; + input.maxLength = type === 'password' ? 8192 : 256; + const label = node('label', 'form-label', labelText); + label.htmlFor = input.id; + group.append(label, input); + parent.appendChild(group); + return input; + } + + function render(value) { + const dto = safeRequest(value); + if (!dto || !active(dto.conversation_id)) return; + if (!['pending', 'ready'].includes(dto.status)) { + removeCard(dto.id); + return; + } + const previous = cards.get(dto.id); + if (previous?.dto.etag === dto.etag && previous.card.isConnected) return; + removeCard(dto.id); + const container = root(); + if (!container) return; + const capturedEpoch = epoch; + const card = node('article', 'card text-break mb-3'); + card.dataset.credentialRequestId = dto.id; + const body = node('div', 'card-body'); + const title = node('h3', 'h5', `Connect to ${dto.action_name}`); + title.id = `action-credential-heading-${++sequence}`; + card.setAttribute('aria-labelledby', title.id); + const reasonLabels = { + credentials_rejected: 'Your saved credentials were rejected. Review or replace your personal identity.', + credentials_expired: 'Your saved credentials have expired. Provide a current personal identity.', + binding_changed: 'The action connection changed. Confirm credentials for the destination below.', + identity_removed: 'The previously selected identity is no longer available.', + }; + const reason = node('p', 'mb-2', reasonLabels[dto.reason] || 'This action needs your personal credentials before it can continue.'); + const destination = node('p', 'fw-semibold mb-2', `Destination: ${dto.destination_label}`); + const privacy = node('p', 'small text-body-secondary', 'Credentials are saved in your personal identity, not in chat messages or the shared action.'); + const form = node('form'); + form.autocomplete = 'off'; + const sections = []; + if (dto.status !== 'ready') { + dto.requirements.forEach(requirement => { + const section = node('fieldset', 'border-0 p-0 mb-3'); + section.appendChild(node('legend', 'h6', requirement.label || (requirement.slot === 'proxy' ? 'Proxy sign-in' : 'Yamcs sign-in'))); + const select = node('select', 'form-select mb-3'); + select.id = `action-credential-identity-${++sequence}`; + const selectLabel = node('label', 'form-label', 'Personal identity'); + selectLabel.htmlFor = select.id; + const createOption = node('option', '', 'Create a new personal identity'); + createOption.value = ''; + select.appendChild(createOption); + dto.identities.filter(identity => identity.auth_type === requirement.auth_type).forEach(identity => { + const option = node('option', '', identity.name); + option.value = identity.id; + select.appendChild(option); + }); + const inputs = node('div'); + const name = field(inputs, 'Identity name', 'text'); + const username = requirement.auth_type === 'username_password' ? field(inputs, 'Username', 'text') : null; + const secret = field(inputs, requirement.auth_type === 'username_password' ? 'Password' + : (requirement.auth_type === 'api_key' ? 'API key' : 'Bearer token'), 'password'); + const rotateGroup = node('div', 'form-check mb-3 d-none'); + const rotate = node('input', 'form-check-input'); + rotate.type = 'checkbox'; + rotate.id = `action-credential-rotate-${++sequence}`; + const rotateLabel = node('label', 'form-check-label', 'Replace credentials in this personal identity'); + rotateLabel.htmlFor = rotate.id; + rotateGroup.append(rotate, rotateLabel); + function updateInputs() { + clearInputs(inputs); + const identity = dto.identities.find(item => item.id === select.value); + rotateGroup.classList.toggle('d-none', !identity?.etag); + const enteringCredentials = !select.value || rotate.checked; + inputs.classList.toggle('d-none', !enteringCredentials); + name.parentElement.classList.toggle('d-none', Boolean(select.value)); + name.required = !select.value; + if (username) username.required = enteringCredentials; + secret.required = enteringCredentials; + } + select.addEventListener('change', () => { + rotate.checked = false; + updateInputs(); + }); + rotate.addEventListener('change', updateInputs); + section.append(selectLabel, select, rotateGroup, inputs); + form.appendChild(section); + sections.push({ requirement, select, name, username, secret, rotate }); + }); + } + const status = node('p', 'small mb-2 d-none'); + status.setAttribute('role', 'status'); + status.setAttribute('aria-live', 'polite'); + const error = node('div', 'alert alert-danger d-none'); + error.setAttribute('role', 'alert'); + error.tabIndex = -1; + const controls = node('div', 'd-flex flex-wrap gap-2'); + const save = node('button', 'btn btn-primary', 'Save and continue'); + save.type = 'submit'; + const cancel = node('button', 'btn btn-outline-secondary', 'Cancel'); + cancel.type = 'button'; + const refresh = node('button', 'btn btn-outline-secondary d-none', 'Refresh request'); + refresh.type = 'button'; + controls.append(save, cancel, refresh); + form.append(status, error, controls); + body.append(title, reason, destination, privacy, form); + card.appendChild(body); + container.appendChild(card); + const state = { dto, card, controller: new AbortController() }; + cards.set(dto.id, state); + + function busy(value, message = '') { + form.querySelectorAll('input, select, button').forEach(input => { input.disabled = value; }); + card.setAttribute('aria-busy', String(value)); + status.textContent = message; + status.classList.toggle('d-none', !message); + error.classList.add('d-none'); + } + + function failed(message) { + if (!active(dto.conversation_id, capturedEpoch) || !card.isConnected) return; + busy(false); + clearInputs(card); + error.textContent = message; + error.classList.remove('d-none'); + refresh.classList.remove('d-none'); + error.focus(); + } + + refresh.addEventListener('click', () => { + removeCard(dto.id); + void fetchRequest(dto.id); + }); + cancel.addEventListener('click', async () => { + clearInputs(card); + busy(true, 'Cancelling…'); + try { + await request(endpoint(dto.id, 'cancel'), { + method: 'POST', body: { expected_etag: state.dto.etag }, signal: state.controller.signal, + }); + removeCard(dto.id); + } catch { + failed('Unable to cancel this request. Refresh the request and try again.'); + } + }); + form.addEventListener('submit', async event => { + event.preventDefault(); + if ((state.dto.status !== 'ready' && !form.reportValidity()) || !active(dto.conversation_id, capturedEpoch)) return; + if (typeof continuationConsumer !== 'function') { + failed('The chat connection is not ready. Reload the conversation before continuing.'); + return; + } + const slots = sections.map(({ requirement, select, name, username, secret, rotate }) => { + const slot = { slot: requirement.slot }; + if (select.value) { + slot.identity_id = select.value; + if (!rotate.checked) return slot; + slot.expected_identity_etag = dto.identities.find(identity => identity.id === select.value)?.etag; + } else { + slot.name = name.value.trim(); + } + slot.credentials = username + ? { auth_type: requirement.auth_type, username: username.value.trim(), password: secret.value } + : { auth_type: requirement.auth_type, secret: secret.value }; + return slot; + }); + clearInputs(card); + busy(true, state.dto.status === 'ready' ? 'Continuing…' : 'Saving your personal identity…'); + try { + if (state.dto.status !== 'ready') { + const result = await request(endpoint(dto.id, 'submit'), { + method: 'POST', body: { expected_etag: state.dto.etag, slots }, signal: state.controller.signal, + }); + const updated = safeRequest(result.credential_request); + if (!updated || updated.id !== dto.id || updated.conversation_id !== dto.conversation_id || updated.status !== 'ready') { + throw new Error('Credentials could not be confirmed. Refresh the request before continuing.'); + } + state.dto = updated; + form.querySelectorAll('fieldset').forEach(section => { + section.classList.add('d-none'); + section.querySelectorAll('input').forEach(input => { input.required = false; }); + }); + } + clearInputs(card); + slots.length = 0; + if (!active(dto.conversation_id, capturedEpoch)) return; + status.textContent = 'Continuing the blocked action…'; + await continuationConsumer(async signal => { + const response = await fetch(endpoint(dto.id, 'continue'), { + method: 'POST', headers: { 'Content-Type': 'application/json', 'X-CSRFToken': csrfToken || '' }, + credentials: 'same-origin', cache: 'no-store', + body: JSON.stringify({ expected_etag: state.dto.etag }), signal, + }); + if (!response.ok) throw new Error(safeError(response.status)); + if (!active(dto.conversation_id, capturedEpoch)) { + await response.body?.cancel(); + throw new Error('The active conversation changed.'); + } + return response; + }, dto.conversation_id); + if (active(dto.conversation_id, capturedEpoch) && cards.get(dto.id) === state) removeCard(dto.id); + } catch (failure) { + failed(failure.message === safeError(409) ? failure.message + : 'Unable to save or continue. Refresh the request to check its current status, then try again.'); + } finally { + clearInputs(card); + slots.length = 0; + } + }); + } + + async function fetchRequest(id) { + const capturedEpoch = epoch; + const loading = node('p', 'small', 'Loading private credential request…'); + loading.setAttribute('role', 'status'); + root()?.appendChild(loading); + try { + const result = await request(endpoint(id)); + if (capturedEpoch === epoch) render(result.credential_request); + } catch { + if (capturedEpoch !== epoch) return; + const container = root(); + if (container) { + const error = node('div', 'alert alert-warning', 'Unable to load the private credential request. Reopen this conversation to try again.'); + error.setAttribute('role', 'alert'); + const retry = node('button', 'btn btn-sm btn-outline-secondary ms-2', 'Retry'); + retry.type = 'button'; + retry.addEventListener('click', () => { + error.remove(); + void fetchRequest(id); + }); + error.appendChild(retry); + container.appendChild(error); + } + } finally { + loading.remove(); + } + } + + async function refreshConversation(id) { + if (id !== window.currentConversationId) return; + if (conversationId !== id) { + clear(); + conversationId = id; + } + if (!id) return; + pendingController?.abort(); + pendingController = new AbortController(); + const controller = pendingController; + const capturedEpoch = epoch; + try { + const result = await request(`/api/action-credentials/pending?conversation_id=${encodeURIComponent(id)}`, { + signal: controller.signal, + }); + if (!active(id, capturedEpoch)) return; + const requestIds = new Set((result.requests || []).map(item => item.id)); + Array.from(cards.keys()).filter(requestId => !requestIds.has(requestId)).forEach(removeCard); + (result.requests || []).forEach(render); + } catch { + if (!active(id, capturedEpoch) || controller.signal.aborted) return; + const warning = node('div', 'alert alert-warning', 'Private action credential requests could not be refreshed.'); + warning.setAttribute('role', 'alert'); + const retry = node('button', 'btn btn-sm btn-outline-secondary ms-2', 'Retry'); + retry.type = 'button'; + retry.addEventListener('click', () => { + warning.remove(); + void refreshConversation(id); + }); + warning.appendChild(retry); + root()?.appendChild(warning); + } + } + + function handleEvent(payload) { + const dto = safeRequest(payload?.credential_request); + if (!dto || dto.conversation_id !== window.currentConversationId) return false; + if (conversationId !== dto.conversation_id) { + clear(); + conversationId = dto.conversation_id; + } + // Read again for the session CSRF token and the current ETag, not the SSE revision. + void fetchRequest(dto.id); + return true; + } + + window.SimpleChatActionCredentials = { + handleEvent, refreshConversation, clear, + setContinuationConsumer(consumer) { continuationConsumer = consumer; }, + }; + window.addEventListener('chat:conversation-context-changed', clear); + window.addEventListener('pagehide', clear); + window.addEventListener('beforeunload', clear); + window.addEventListener('simplechat:signout', clear); + document.addEventListener('click', event => { + const link = event.target.closest?.('a[href]'); + if (link && /logout|signout|sign-out/i.test(link.getAttribute('href'))) clear(); + }, true); + new MutationObserver(() => { + cards.forEach((state, id) => { + if (!state.card.isConnected) removeCard(id); + }); + }).observe(document.documentElement, { childList: true, subtree: true }); +})(); diff --git a/application/single_app/static/js/chat/chat-messages.js b/application/single_app/static/js/chat/chat-messages.js index 254213b1f..4382630a6 100644 --- a/application/single_app/static/js/chat/chat-messages.js +++ b/application/single_app/static/js/chat/chat-messages.js @@ -2311,6 +2311,7 @@ export function loadMessages(conversationId) { .finally(() => { if (!ownsConversation()) return; void window.SimpleChatM365PendingActions?.refreshConversation(conversationId); + void window.SimpleChatActionCredentials?.refreshConversation(conversationId); // Check if there's a search highlight to apply if (window.searchHighlight && window.searchHighlight.term) { const elapsed = Date.now() - window.searchHighlight.timestamp; diff --git a/application/single_app/static/js/chat/chat-streaming.js b/application/single_app/static/js/chat/chat-streaming.js index c38f2212a..65be5b60a 100644 --- a/application/single_app/static/js/chat/chat-streaming.js +++ b/application/single_app/static/js/chat/chat-streaming.js @@ -27,6 +27,32 @@ let currentStreamContext = null; const MAX_STREAM_CLIENT_ERROR_LENGTH = 500; const USER_MESSAGE_PERSISTED_EVENT_TYPE = 'user_message_persisted'; +window.SimpleChatActionCredentials?.setContinuationConsumer(continueActionCredentialStream); + +export function continueActionCredentialStream(requestFactory, conversationId) { + if (!isConversationCurrentlyActive(conversationId)) { + return Promise.reject(new Error('The active conversation changed.')); + } + return new Promise((resolve, reject) => { + let completed = false; + consumeStreamingResponse(requestFactory, createStreamingPlaceholder('Continuing'), null, { + allowRecovery: false, + recoveryConversationId: conversationId, + onDone: data => { + completed = true; + resolve(data); + }, + onError: () => { + completed = true; + reject(new Error('Unable to continue the blocked action.')); + }, + onFinally: () => { + if (!completed) reject(new Error('The continuation was interrupted.')); + }, + }); + }); +} + window.addEventListener('m365-chat-resumed', event => { const conversationId = event.detail?.conversationId; if (!isConversationCurrentlyActive(conversationId)) { @@ -780,6 +806,27 @@ function consumeStreamingResponse(requestFactory, tempAiMessageId, tempUserMessa ); } + function pauseForActionCredentials(data) { + if (data?.type !== 'action_credentials_required') return false; + streamCompleted = true; + stopThoughtPolling(); + clearStreamingThoughtSession(tempAiMessageId); + removeStreamingStopButton(tempAiMessageId); + clearCurrentStreamController(abortController); + finalizePendingUserMessageMetadata(); + enablePersistedUserMessageActions(); + if (accumulatedContent) { + updateStreamingMessage(tempAiMessageId, accumulatedContent); + getStreamingMessageElement(tempAiMessageId)?.querySelector('.streaming-cursor')?.remove(); + } else { + removeStreamingPlaceholder(tempAiMessageId); + } + window.SimpleChatActionCredentials?.handleEvent(data); + if (typeof onDone === 'function') onDone({ status: 'waiting_for_credentials' }); + if (typeof onFinally === 'function') onFinally(); + return true; + } + function pauseForMicrosoft365SignIn(data) { if (!isM365SignInRequired(data)) { return false; @@ -816,6 +863,7 @@ function consumeStreamingResponse(requestFactory, tempAiMessageId, tempUserMessa if (response.headers?.get('Content-Type')?.includes('application/json')) { return response.json().then(data => { capturePendingActions(data); + if (pauseForActionCredentials(data)) return; if (pauseForMicrosoft365SignIn(data)) { recoverPendingActions(); return; @@ -863,6 +911,7 @@ function consumeStreamingResponse(requestFactory, tempAiMessageId, tempUserMessa function processStreamData(data) { eventCount += 1; lastChunkAt = Date.now(); + if (pauseForActionCredentials(data)) return true; if (Array.isArray(data.reasoning_adjustments) || Array.isArray(data.metadata?.reasoning_adjustments)) { reasoningAdjustments = getMessageReasoningAdjustments(data, reasoningAdjustments); renderMessageReasoningAdjustments(getStreamingMessageElement(tempAiMessageId), reasoningAdjustments); @@ -1088,6 +1137,10 @@ function consumeStreamingResponse(requestFactory, tempAiMessageId, tempUserMessa try { const data = JSON.parse(jsonStr); + const eventName = eventBlock.split('\n').find(line => line.startsWith('event:'))?.slice(6).trim(); + if (eventName === 'action_credentials_required' && !data.type) { + data.type = eventName; + } return processStreamData(data); } catch (error) { console.error('Error parsing SSE data:', error); @@ -1279,6 +1332,7 @@ function consumeStreamingResponse(requestFactory, tempAiMessageId, tempUserMessa if (pauseForMicrosoft365SignIn(getStreamErrorPayload(error))) { return; } + if (pauseForActionCredentials(getStreamErrorPayload(error))) return; stopThoughtPolling(); console.error('Streaming request error:', error); diff --git a/application/single_app/static/js/plugin_modal_stepper.js b/application/single_app/static/js/plugin_modal_stepper.js index f17bed257..f2744567a 100644 --- a/application/single_app/static/js/plugin_modal_stepper.js +++ b/application/single_app/static/js/plugin_modal_stepper.js @@ -763,6 +763,11 @@ export class PluginModalStepper { document.getElementById('tableau-auth-method').addEventListener('change', () => this.toggleTableauAuthFields()); document.getElementById('tableau-identity-select').addEventListener('change', () => this.handleActionIdentityChange('tableau')); document.getElementById('yamcs-auth-method').addEventListener('change', () => this.toggleYamcsAuthFields()); + ['service', 'proxy'].forEach(slot => { + document.getElementById(`yamcs-${slot}-credential-source`)?.addEventListener('change', () => { + this.applyYamcsCredentialSources(true); + }); + }); document.getElementById('yamcs-identity-select').addEventListener('change', () => this.handleActionIdentityChange('yamcs')); document.getElementById('yamcs-enable-basic-auth').addEventListener('change', () => this.toggleYamcsBasicAuthFields()); document.getElementById('yamcs-basic-auth-identity-select').addEventListener('change', () => this.handleActionIdentityChange('yamcsBasicAuth')); @@ -2975,9 +2980,11 @@ export class PluginModalStepper { const apiKeyGroup = document.getElementById('yamcs-api-key-group'); const bearerTokenGroup = document.getElementById('yamcs-bearer-token-group'); + const personalService = this.isYamcsPersonalSource('service'); + document.getElementById('yamcs-action-identity-group')?.classList.toggle('d-none', personalService || !this.getActionIdentitiesForKind('yamcs').length); if (authMethodSelect) { - authMethodSelect.disabled = Boolean(selectedIdentity); - if (selectedIdentity) { + authMethodSelect.disabled = Boolean(selectedIdentity) && !personalService; + if (selectedIdentity && !personalService) { authMethodSelect.value = this.getYamcsIdentityAuthMethod(selectedIdentity); } } @@ -2990,7 +2997,7 @@ export class PluginModalStepper { this.toggleYamcsBasicAuthFields(); - if (selectedIdentity) { + if (selectedIdentity || personalService) { return; } @@ -3006,7 +3013,7 @@ export class PluginModalStepper { getYamcsAuthMethodForConflictCheck() { const selectedIdentity = this.getSelectedActionIdentity('yamcs'); - if (selectedIdentity) { + if (selectedIdentity && !this.isYamcsPersonalSource('service')) { return this.getYamcsIdentityAuthMethod(selectedIdentity); } return document.getElementById('yamcs-auth-method')?.value || YAMCS_AUTH_METHOD_USERNAME_PASSWORD; @@ -3020,27 +3027,65 @@ export class PluginModalStepper { return document.getElementById('yamcs-enable-basic-auth')?.checked === true; } + isYamcsPersonalSource(slot) { + return document.getElementById(`yamcs-${slot}-credential-source`)?.value === 'personal'; + } + + applyYamcsCredentialSources(clearCompeting = false) { + if (clearCompeting && this.isYamcsPersonalSource('service')) { + this.setSelectedActionIdentity('yamcs', ''); + ['yamcs-username', 'yamcs-password', 'yamcs-api-key', 'yamcs-bearer-token'].forEach(id => { + document.getElementById(id).value = ''; + }); + } + if (clearCompeting && this.isYamcsPersonalSource('proxy')) { + this.setSelectedActionIdentity('yamcsBasicAuth', ''); + ['yamcs-basic-auth-username', 'yamcs-basic-auth-password'].forEach(id => { + document.getElementById(id).value = ''; + }); + } + this.toggleYamcsAuthFields(); + } + + getYamcsCredentialRequirements() { + const slots = {}; + if (this.isYamcsPersonalSource('service')) { + slots.service = { source: 'personal', auth_type: this.getYamcsAuthMethodForConflictCheck() }; + } + if (this.isYamcsBasicAuthEnabled() && this.isYamcsPersonalSource('proxy')) { + slots.proxy = { source: 'personal', auth_type: 'username_password' }; + } + return { version: 1, slots }; + } + toggleYamcsBasicAuthFields() { const fields = document.getElementById('yamcs-basic-auth-fields'); const conflictAlert = document.getElementById('yamcs-basic-auth-conflict'); const usernameInput = document.getElementById('yamcs-basic-auth-username'); const passwordInput = document.getElementById('yamcs-basic-auth-password'); const enabled = this.isYamcsBasicAuthEnabled(); + const personalProxy = this.isYamcsPersonalSource('proxy'); + const hasPersonalCredentials = this.isYamcsPersonalSource('service') || (enabled && personalProxy); + const testButton = document.getElementById('yamcs-test-connection-btn'); + if (testButton) testButton.disabled = hasPersonalCredentials; + document.getElementById('yamcs-personal-test-note')?.classList.toggle('d-none', !hasPersonalCredentials); fields?.classList.toggle('d-none', !enabled); conflictAlert?.classList.toggle('d-none', !(enabled && this.yamcsBasicAuthConflicts())); + document.getElementById('yamcs-basic-auth-identity-group')?.classList.toggle('d-none', personalProxy || !this.getActionIdentitiesForKind('yamcsBasicAuth').length); + usernameInput?.closest('.row')?.classList.toggle('d-none', personalProxy); // A reusable identity supplies both values, so the inline inputs become read-only // mirrors of the stored credential rather than a second place to edit it. const selectedIdentity = this.getSelectedActionIdentity('yamcsBasicAuth'); if (usernameInput) { - usernameInput.disabled = Boolean(selectedIdentity); - if (selectedIdentity) { + usernameInput.disabled = Boolean(selectedIdentity) || personalProxy; + if (selectedIdentity && !personalProxy) { usernameInput.value = selectedIdentity.credentials?.username || ''; } } if (passwordInput) { - passwordInput.disabled = Boolean(selectedIdentity); + passwordInput.disabled = Boolean(selectedIdentity) || personalProxy; if (selectedIdentity) { passwordInput.value = ''; } @@ -3048,8 +3093,14 @@ export class PluginModalStepper { } populateYamcsForm(plugin) { + ['yamcs-username', 'yamcs-password', 'yamcs-api-key', 'yamcs-bearer-token'].forEach(id => { + document.getElementById(id).value = ''; + }); const additionalFields = plugin.additionalFields || plugin.additional_fields || {}; const auth = plugin.auth || {}; + const slots = plugin.credential_requirements?.slots || {}; + document.getElementById('yamcs-service-credential-source').value = slots.service?.source === 'personal' ? 'personal' : 'shared'; + document.getElementById('yamcs-proxy-credential-source').value = slots.proxy?.source === 'personal' ? 'personal' : 'shared'; const serverUrl = this.normalizeYamcsServerUrl(plugin.endpoint || additionalFields.server_url || ''); document.getElementById('yamcs-server-url').value = serverUrl; @@ -3059,7 +3110,7 @@ export class PluginModalStepper { document.getElementById('yamcs-timeout').value = additionalFields.timeout || 30; document.getElementById('yamcs-tls-verify').checked = additionalFields.tls_verify !== false; document.getElementById('yamcs-enable-archive-sql').checked = additionalFields.enable_archive_sql === true; - document.getElementById('yamcs-enable-basic-auth').checked = additionalFields.enable_basic_auth === true; + document.getElementById('yamcs-enable-basic-auth').checked = additionalFields.enable_basic_auth === true || slots.proxy?.source === 'personal'; document.getElementById('yamcs-basic-auth-username').value = additionalFields.basic_auth_username || ''; document.getElementById('yamcs-basic-auth-password').value = additionalFields.basic_auth_password || ''; @@ -3079,17 +3130,20 @@ export class PluginModalStepper { } } - document.getElementById('yamcs-auth-method').value = authMethod; + document.getElementById('yamcs-auth-method').value = slots.service?.auth_type || authMethod; this.setSelectedActionIdentity('yamcs', plugin.identity_id || ''); this.setSelectedActionIdentity('yamcsBasicAuth', additionalFields.basic_auth_identity_id || ''); this.handleActionIdentityChange('yamcs'); this.handleActionIdentityChange('yamcsBasicAuth'); + this.applyYamcsCredentialSources(true); } getYamcsConfiguration() { const serverUrl = this.normalizeYamcsServerUrl(document.getElementById('yamcs-server-url')?.value || ''); - const selectedIdentity = this.getSelectedActionIdentity('yamcs'); - const basicAuthIdentity = this.getSelectedActionIdentity('yamcsBasicAuth'); + const personalService = this.isYamcsPersonalSource('service'); + const personalProxy = this.isYamcsPersonalSource('proxy'); + const selectedIdentity = personalService ? null : this.getSelectedActionIdentity('yamcs'); + const basicAuthIdentity = personalProxy ? null : this.getSelectedActionIdentity('yamcsBasicAuth'); const enableBasicAuth = this.isYamcsBasicAuthEnabled(); const authMethod = selectedIdentity ? this.getYamcsIdentityAuthMethod(selectedIdentity) @@ -3103,18 +3157,18 @@ export class PluginModalStepper { read_only: true, enable_archive_sql: document.getElementById('yamcs-enable-archive-sql')?.checked === true, enable_basic_auth: enableBasicAuth, - // Only an identity selection blanks the inline credential. Turning the toggle off + // Identity selection or personal source explicitly clears inline credentials. Turning the toggle off // must keep the stored values, otherwise saving would drop the Key Vault reference // and leave its secret orphaned. Runtime and validation already ignore these fields // while the toggle is off. An untouched password field still holds the Key Vault // placeholder, which the save helper resolves back to the existing reference. - basic_auth_identity_id: basicAuthIdentity + basic_auth_identity_id: basicAuthIdentity && !personalProxy ? (basicAuthIdentity.id || basicAuthIdentity.identity_id || '') : '', - basic_auth_username: basicAuthIdentity + basic_auth_username: basicAuthIdentity || personalProxy ? '' : (document.getElementById('yamcs-basic-auth-username')?.value.trim() || ''), - basic_auth_password: basicAuthIdentity + basic_auth_password: basicAuthIdentity || personalProxy ? '' : (document.getElementById('yamcs-basic-auth-password')?.value || ''), max_rows: parseInt(document.getElementById('yamcs-max-rows')?.value, 10) || 500, @@ -3123,7 +3177,10 @@ export class PluginModalStepper { const auth = {}; let identityId = ''; - if (selectedIdentity) { + if (personalService) { + Object.assign(auth, { type: 'NoAuth', key: '', identity: '' }); + additionalFields.identity_auth_type = ''; + } else if (selectedIdentity) { identityId = selectedIdentity.id || selectedIdentity.identity_id || ''; auth.type = 'identity'; auth.identity = identityId; @@ -4760,15 +4817,20 @@ export class PluginModalStepper { this.showError('Select a supported Yamcs authentication method.'); return false; } - if (!selectedIdentity && authMethod === YAMCS_AUTH_METHOD_USERNAME_PASSWORD && (!username || !password)) { + const personalService = this.isYamcsPersonalSource('service'); + if (personalService && authMethod === YAMCS_AUTH_METHOD_NONE) { + this.showError('Personal Yamcs credentials require an authenticated service method. Use shared configuration for No Authentication.'); + return false; + } + if (!personalService && !selectedIdentity && authMethod === YAMCS_AUTH_METHOD_USERNAME_PASSWORD && (!username || !password)) { this.showError('Yamcs username and password are required for username/password authentication.'); return false; } - if (!selectedIdentity && authMethod === YAMCS_AUTH_METHOD_API_KEY && !apiKey) { + if (!personalService && !selectedIdentity && authMethod === YAMCS_AUTH_METHOD_API_KEY && !apiKey) { this.showError('Yamcs API key is required for API key authentication.'); return false; } - if (!selectedIdentity && authMethod === YAMCS_AUTH_METHOD_BEARER_TOKEN && !bearerToken) { + if (!personalService && !selectedIdentity && authMethod === YAMCS_AUTH_METHOD_BEARER_TOKEN && !bearerToken) { this.showError('Yamcs bearer token is required for bearer token authentication.'); return false; } @@ -4780,7 +4842,7 @@ export class PluginModalStepper { this.showError('HTTP Basic authentication cannot be combined with username/password or access token authentication. Choose "No Authentication" or "API Key".'); return false; } - if (!basicAuthIdentity && (!basicAuthUsername || !basicAuthPassword)) { + if (!this.isYamcsPersonalSource('proxy') && !basicAuthIdentity && (!basicAuthUsername || !basicAuthPassword)) { this.showError('A proxy username and password are required for HTTP Basic authentication.'); return false; } @@ -5655,6 +5717,13 @@ export class PluginModalStepper { const alertDiv = document.getElementById('yamcs-test-connection-alert'); if (!btn || !resultDiv || !alertDiv) return; + if (Object.keys(this.getYamcsCredentialRequirements().slots).length) { + resultDiv.classList.remove('d-none'); + alertDiv.className = 'alert alert-info mb-0 py-2 px-3 small'; + alertDiv.textContent = 'Personal credentials are checked when each user connects in chat. Save this action without shared credentials; this test does not acquire or replace a user identity.'; + return; + } + const serverUrl = this.normalizeYamcsServerUrl(document.getElementById('yamcs-server-url')?.value || ''); const instance = document.getElementById('yamcs-instance')?.value?.trim() || ''; const selectedIdentity = this.getSelectedActionIdentity('yamcs'); @@ -7332,6 +7401,11 @@ export class PluginModalStepper { if (identityId) { formData.identity_id = identityId; } + if (isYamcsVisible) { + // Empty references are intentional clears, not requests to preserve old secrets. + formData.identity_id = identityId || ''; + formData.credential_requirements = this.getYamcsCredentialRequirements(); + } if (this.isEditMode && this.originalPlugin?.id) { formData.id = this.originalPlugin.id; } @@ -7615,6 +7689,9 @@ export class PluginModalStepper { } return this.formatTableauAuthMethod(document.getElementById('tableau-auth-method')?.value || TABLEAU_AUTH_METHOD_PAT); } else if (this.isYamcsType()) { + if (this.isYamcsPersonalSource('service')) { + return `Personal identity for each user (${this.formatYamcsAuthMethod(this.getYamcsAuthMethodForConflictCheck())})`; + } if (this.getSelectedActionIdentity('yamcs')) { return 'Reusable Identity'; } @@ -7940,7 +8017,9 @@ export class PluginModalStepper { document.getElementById('summary-yamcs-instance').textContent = document.getElementById('yamcs-instance')?.value.trim() || '-'; document.getElementById('summary-yamcs-processor').textContent = document.getElementById('yamcs-processor')?.value.trim() || YAMCS_DEFAULT_PROCESSOR; - document.getElementById('summary-yamcs-auth-method').textContent = selectedIdentity + document.getElementById('summary-yamcs-auth-method').textContent = this.isYamcsPersonalSource('service') + ? `Personal identity for each user (${this.formatYamcsAuthMethod(authMethod)})` + : selectedIdentity ? `Reusable Identity (${this.formatYamcsAuthMethod(authMethod)})` : this.formatYamcsAuthMethod(authMethod); document.getElementById('summary-yamcs-tls-verify').textContent = document.getElementById('yamcs-tls-verify')?.checked === false ? 'Disabled' : 'Enabled'; @@ -7955,6 +8034,9 @@ export class PluginModalStepper { if (!this.isYamcsBasicAuthEnabled()) { return 'Disabled'; } + if (this.isYamcsPersonalSource('proxy')) { + return 'Personal proxy sign-in for each user'; + } const basicAuthIdentity = this.getSelectedActionIdentity('yamcsBasicAuth'); if (basicAuthIdentity) { return `Enabled (reusable identity: ${basicAuthIdentity.name || 'Workspace identity'})`; @@ -8639,6 +8721,9 @@ export class PluginModalStepper { element.dataset[key] = value; } }; + safeSetValue('yamcs-service-credential-source', 'shared'); + safeSetValue('yamcs-proxy-credential-source', 'shared'); + this.populateYamcsForm({ endpoint: '', auth: { type: 'username_password' }, additionalFields: {} }); // Step 2 fields - Basic Info safeSetValue('plugin-name'); diff --git a/application/single_app/static/json/schemas/plugin.schema.json b/application/single_app/static/json/schemas/plugin.schema.json index 402b8e95c..307f5d997 100644 --- a/application/single_app/static/json/schemas/plugin.schema.json +++ b/application/single_app/static/json/schemas/plugin.schema.json @@ -56,6 +56,38 @@ "type": "string", "description": "Workspace identity reference used by this action. Credentials are resolved server-side within the action scope." }, + "credential_requirements": { + "type": "object", + "additionalProperties": false, + "required": ["version", "slots"], + "properties": { + "version": {"const": 1}, + "slots": { + "type": "object", + "additionalProperties": false, + "properties": { + "service": { + "type": "object", + "additionalProperties": false, + "required": ["source", "auth_type"], + "properties": { + "source": {"const": "personal"}, + "auth_type": {"enum": ["username_password", "api_key", "bearer_token"]} + } + }, + "proxy": { + "type": "object", + "additionalProperties": false, + "required": ["source", "auth_type"], + "properties": { + "source": {"const": "personal"}, + "auth_type": {"const": "username_password"} + } + } + } + } + } + }, "endpoint": { "type": "string" }, @@ -80,17 +112,7 @@ } }, "additionalProperties": false, - "allOf": [ - { "if": { "properties": { "type": { "const": "NoAuth" } } }, "then": { "required": ["type"] } }, - { "if": { "properties": { "type": { "const": "key" } } }, "then": { "required": ["type", "key"] } }, - { "if": { "properties": { "type": { "const": "identity" } } }, "then": { "required": ["type", "identity"] } }, - { "if": { "properties": { "type": { "const": "user" } } }, "then": { "required": ["type"] } }, - { "if": { "properties": { "type": { "const": "servicePrincipal" } } }, "then": { "required": ["type", "tenantId", "identity", "key"] } }, - { "if": { "properties": { "type": { "const": "connection_string" } } }, "then": { "required": ["type", "key"] } }, - { "if": { "properties": { "type": { "const": "basic" } } }, "then": { "required": ["type", "key", "identity"] } }, - { "if": { "properties": { "type": { "const": "username_password" } } }, "then": { "required": ["type", "key", "identity"] } }, - { "required": ["type"] } - ] + "required": ["type"] }, "metadata": { "type": "object", @@ -103,6 +125,29 @@ "additionalProperties": true } }, + "allOf": [{ + "if": { + "required": ["credential_requirements"], + "properties": { + "credential_requirements": { + "required": ["slots"], + "properties": {"slots": {"required": ["service"]}} + } + } + }, + "then": {"properties": {"type": {"const": "yamcs"}}}, + "else": {"properties": {"auth": {"allOf": [ + { "if": { "properties": { "type": { "const": "NoAuth" } } }, "then": { "required": ["type"] } }, + { "if": { "properties": { "type": { "const": "key" } } }, "then": { "required": ["type", "key"] } }, + { "if": { "properties": { "type": { "const": "identity" } } }, "then": { "required": ["type", "identity"] } }, + { "if": { "properties": { "type": { "const": "user" } } }, "then": { "required": ["type"] } }, + { "if": { "properties": { "type": { "const": "servicePrincipal" } } }, "then": { "required": ["type", "tenantId", "identity", "key"] } }, + { "if": { "properties": { "type": { "const": "connection_string" } } }, "then": { "required": ["type", "key"] } }, + { "if": { "properties": { "type": { "const": "basic" } } }, "then": { "required": ["type", "key", "identity"] } }, + { "if": { "properties": { "type": { "const": "username_password" } } }, "then": { "required": ["type", "key", "identity"] } }, + { "required": ["type"] } + ]}}} + }], "required": ["name", "type", "description", "endpoint", "auth", "metadata", "additionalFields"], "title": "Plugin" } diff --git a/application/single_app/templates/_plugin_modal.html b/application/single_app/templates/_plugin_modal.html index 731e988b1..43bda45d3 100644 --- a/application/single_app/templates/_plugin_modal.html +++ b/application/single_app/templates/_plugin_modal.html @@ -566,6 +566,14 @@
API Information
Authentication
+
+ + +
Personal credentials are requested privately in chat when this action needs access. They are not saved in the shared action.
+
@@ -614,6 +622,14 @@
API Information
+
+ + +
The proxy sign-in is separate from Yamcs service authentication.
+
+

+ Save this action, then use it in chat to configure and check your personal identity. + A shared connection test cannot check each user's credentials. +