Add personal credentials for shared Yamcs actions

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

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,
+    }

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:

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)

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")