Credential Identity & Multi-Account Foundation (Issue #4755)

Credential Identity & Multi-Account Foundation (Issue #4755)

Context

Agents are identity-blind. When gmail_read_email runs, neither the LLM nor the tool knows whose inbox it's operating on. One ADEN_API_KEY can back N accounts of the same provider (e.g., 10 Gmail accounts), but today the system can only surface one — the last one synced silently overwrites all others.

This plan traces the 5-tuple relationship (Agent Definition → Agent Instance → Agent Tool → Auth Provider → Auth User Identity) through every layer of the stack, identifies exactly where things break, and prescribes targeted fixes.

Motivating Scenarios

Scenario A — Executive Assistant Agent: A company deploys an agent that manages calendars for 5 executives. Each executive has connected their Google account through Aden. The agent's job is to check each person's availability and schedule meetings. Today: the agent can only see ONE person's calendar (whichever synced last). The other 4 accounts are silently lost in the index collision. The agent schedules meetings on the wrong person's calendar with no indication anything is wrong.

Scenario B — Multi-Channel Support Agent: A support team agent is connected to 3 Slack workspaces (Engineering, Sales, Support), a shared Gmail inbox, and a personal Gmail for the team lead. Today: the agent sees one Slack workspace, one Gmail. It cannot tell which Slack workspace it's posting to or whose Gmail it's reading. It might reply to a customer email from the team lead's personal inbox.

Scenario C — Compliance & Audit: An enterprise client requires audit logs showing which account was accessed, when, and by which agent. Today: the system logs credentials.get("google") — no record of which of the 10 Google accounts was used. Impossible to audit.

Scenario D — Single-Account Agent (backward compat): A simple agent uses one Gmail account and one Slack bot. Nothing should change. credentials.get("google") returns the same token it always did. Zero migration, zero configuration changes.

The 5-Tuple Model

Every credential interaction involves five entities. Understanding how they relate (and where the relationships break) is the key to the fix.

Agent Definition ──→ Agent Instance ──→ Agent Tool ──→ Auth Provider ──→ Auth User Identity
  "I need Gmail"    "Here's your       "Give me a      "Here's one      "Whose token
                     Gmail tool"        token"           token"           is this?"
                                                                          ← MISSING

1. Agent Definition (what tools are needed)

Files: exports/{name}/agent.py, nodes/__init__.py, mcp_servers.json

An exported agent declares NodeSpec.tools = ["gmail_read_email", "gmail_send_email"]. The mcp_servers.json points to the tools MCP server. The agent definition has NO credential awareness — it names tools, not credentials. This is intentional: the same agent definition can run against different credential sets in different environments (dev vs. prod, tenant A vs. tenant B).

Business logic: Agent definitions are portable templates. A "Gmail Triage" agent built by one team can be deployed to 50 different customers, each with their own Google accounts. The agent definition never hard-codes credential IDs.

Status: Fine. No changes needed.

2. Agent Instance (runtime wiring)

Files: runner.py, tool_registry.py, mcp_client.py

AgentRunner.__init__() does three things in sequence:

1. validate_agent_credentials(graph.nodes) — checks presence + health
2. ToolRegistry.load_mcp_config() → MCPClient spawns subprocess
3. _setup() → create_agent_runtime() with discovered tools

The ToolRegistry bridges parent ↔ MCP subprocess:

• CONTEXT_PARAMS = {"workspace_id", "agent_id", "session_id", "data_dir"} — stripped from LLM schema, injected at call time via make_mcp_executor closure
• set_session_context() — set once at startup
• set_execution_context() — per-execution via contextvars

The MCP subprocess inherits os.environ at spawn time via merged_env = {**os.environ, **(config.env or {})} in mcp_client.py:157.

Business logic: The agent instance is where "portable template" meets "specific deployment." An instance knows which Aden API key to use, which workspace it belongs to, which tools are available. The CONTEXT_PARAMS mechanism is how the framework passes deployment-specific context into tools without the LLM knowing or caring. This is the natural extension point for account routing in the future.

Scenario: Two customers both deploy the same "Email Triage" agent. Customer A has 2 Google accounts; Customer B has 5. Each customer's AgentRunner validates against their own Aden key, discovers different sets of credentials, and wires them into the same agent graph. The agent definition is identical.

Status: Works for single-account. The CONTEXT_PARAMS pattern is the right mechanism for future multi-account routing (adding account param).

3. Agent Tool (credential consumption)

Files: tools/src/aden_tools/tools/*/, tools/mcp_server.py

Every tool follows the same pattern:

def register_gmail_tools(mcp, credentials=None):
    def _get_token():
        if credentials is not None:
            return credentials.get("google")   # ← single token, identity unknown
        return os.getenv("GOOGLE_ACCESS_TOKEN")

    @mcp.tool()
    def gmail_read_email(message_id: str):
        token = _get_token()
        ...

The credentials object is CredentialStoreAdapter, created once at MCP server startup via CredentialStoreAdapter.default(). All tool closures capture this single shared instance.

Business logic: Tools are the consumer endpoint — they need a valid access token to call external APIs. They don't care about Aden, sync, or storage. They just need _get_token() to return the right token. Today, "right" is undefined because there's no way to say "the token for alice@company.com, not bob@company.com."

Where it breaks — Scenario A revisited: The executive assistant agent calls gmail_read_email() intending to read Alice's inbox. _get_token() returns credentials.get("google") which resolves to... Bob's token (he synced last). The agent reads Bob's emails, thinks they're Alice's, and schedules meetings accordingly. No error is raised. No indication anything is wrong. The agent is confidently operating on the wrong person's data.

Where it breaks — Scenario B revisited: The support agent calls slack_post_message(channel="support-tickets"). It uses a Slack token from the Engineering workspace (last synced). The message goes to a channel that doesn't exist in Engineering, returns an error, and the agent retries in a loop with no understanding of why it's failing.

4. Auth Provider (credential storage & resolution)

Files: store.py, aden/storage.py, aden/provider.py, aden/client.py

Resolution chain:

credentials.get("google")
→ CredentialStoreAdapter.get("google")
→ CredentialStore.get("google")
→ AdenCachedStorage.load("google")
→ _provider_index.get("google") → "google_def456"  (last write wins)
→ _load_by_id("google_def456")
→ Returns ONE CredentialObject

The index collision bug (storage.py:303):

def _index_provider(self, credential):
    provider_name = integration_type_key.value.get_secret_value()
    self._provider_index[provider_name] = credential.id   # ← OVERWRITES

Business logic: The storage layer is responsible for mapping human-readable provider names ("google") to internal hash-based credential IDs ("google_abc123"). This mapping is essential because Aden generates unique hash IDs per connected account, but tools reference providers by name. The _provider_index is this mapping.

Why it's a dict[str, str] today: The original design assumed 1:1 between provider name and credential. "One Google account per API key." This was valid for simple deployments but breaks fundamentally when an Aden API key backs multiple accounts of the same provider.

The collision mechanics: When sync_all() runs, it iterates over all active integrations from Aden. For a user with 3 Gmail accounts:

1. Sync google_abc123 (alice@co.com) → _provider_index["google"] = "google_abc123"
2. Sync google_def456 (bob@co.com) → _provider_index["google"] = "google_def456" ← Alice lost
3. Sync google_ghi789 (carol@co.com) → _provider_index["google"] = "google_ghi789" ← Bob lost

All three .enc files exist on disk. Only Carol's is reachable by name. Alice's and Bob's tokens are orphaned — encrypted, on disk, but invisible to the resolution chain.

Why the disk layer is fine: EncryptedFileStorage uses the hash ID as filename: google_abc123.enc, google_def456.enc. No collision. The problem is purely in the in-memory index that maps names to IDs.

5. Auth User Identity (THE MISSING PIECE)

Files: models.py (no identity model), aden/provider.py (metadata discarded), health_check.py (identity parsed then discarded), validation.py (details ignored)

Business logic: Identity answers "whose account is this?" Every external service provides identity data in its API responses — Gmail returns emailAddress, GitHub returns login, Slack returns team + user. This data already flows through the system during health checks and Aden syncs. It's parsed, briefly held in local variables, and then discarded. No model captures it. No property exposes it. No downstream consumer reads it.

Identity data exists at two sources but is discarded:

The waste: Every agent startup already makes these health check API calls. The identity data is RIGHT THERE in the response body. We parse it for validation logic, then throw it away. Zero additional API calls needed — we just need to keep what we already have.

What identity enables downstream:

• LLM knows whose inbox it's reading (system prompt awareness)
• Tools can route to specific accounts (future account parameter)
• Audit logs can record which identity was accessed
• Users can see which accounts are connected in TUI/dashboard
• Agents can reason about cross-account operations ("forward from alice to bob")

What Changes — Layer by Layer

Step 1: CredentialIdentity model on CredentialObject

File: core/framework/credentials/models.py

Business logic: Every credential needs a structured way to answer "who does this belong to?" Different providers express identity differently:

The CredentialIdentity model normalizes these into four universal fields: email, username, workspace, account_id. The label property picks the best human-readable identifier for display (email preferred, then username, etc.).

Why a computed property, not a stored field: Identity is derived from _identity_* keys that already exist in the credential's key vault. Storing it as a separate field would create a sync problem (what if keys update but the field doesn't?). A computed property always reflects current state.

Scenarios this enables:

• Display: cred.identity.label → "alice@company.com" (for system prompts, TUI, logs)
• Comparison: cred.identity.email == "alice@company.com" (for account routing)
• Serialization: cred.identity.to_dict() → {"email": "alice@company.com"} (for MCP tool responses)
• Existence check: cred.identity.is_known → True (skip accounts with no identity)
• Provider type: cred.provider_type → "google" (from _integration_type key)

Key design decision: set_identity(**fields) persists as _identity_* keys using the existing set_key() method. This means identity survives serialization/deserialization through EncryptedFileStorage without any schema migration. Old credentials without identity keys simply return CredentialIdentity() with all None fields and label == "unknown".

class CredentialIdentity(BaseModel):
    email: str | None = None
    username: str | None = None
    workspace: str | None = None
    account_id: str | None = None

    @property
    def label(self) -> str:
        return self.email or self.username or self.workspace or self.account_id or "unknown"

    @property
    def is_known(self) -> bool:
        return bool(self.email or self.username or self.workspace or self.account_id)

    def to_dict(self) -> dict[str, str]:
        return {k: v for k, v in self.model_dump().items() if v is not None}

On CredentialObject:

@property
def identity(self) -> CredentialIdentity:
    fields = {}
    for key_name, key_obj in self.keys.items():
        if key_name.startswith("_identity_"):
            field = key_name[len("_identity_"):]
            fields[field] = key_obj.value.get_secret_value()
    return CredentialIdentity(**{k: v for k, v in fields.items()
                                 if k in CredentialIdentity.model_fields})

@property
def provider_type(self) -> str | None:
    key = self.keys.get("_integration_type")
    return key.value.get_secret_value() if key else None

def set_identity(self, **fields: str) -> None:
    for field_name, value in fields.items():
        if value:
            self.set_key(f"_identity_{field_name}", value)

Step 2: Fix storage multi-account index

File: core/framework/credentials/aden/storage.py

Business logic: The core bug. When a user connects multiple accounts of the same provider type through Aden, all but the last one becomes unreachable. This affects every multi-account deployment silently — no error, no warning, just missing accounts.

_provider_index: dict[str, str] → dict[str, list[str]]

Before (broken):

sync google_abc123 (alice)  → index["google"] = "google_abc123"
sync google_def456 (bob)    → index["google"] = "google_def456"  ← alice lost
load("google")              → returns bob's token

After (fixed):

sync google_abc123 (alice)  → index["google"] = ["google_abc123"]
sync google_def456 (bob)    → index["google"] = ["google_abc123", "google_def456"]
load("google")              → returns alice's token (first = backward compat)
load_all_for_provider("google") → returns [alice, bob]

Backward compatibility contract: Every existing tool calls credentials.get("google") and expects a single token string back. This MUST continue to work. load("google") returns the first credential in the list — same behavior as before for single-account deployments, deterministic (first-synced-first-served) for multi-account.

Scenarios:

• Single account (most common today): index["google"] = ["google_abc123"]. load("google") returns the only entry. Identical behavior to before.

• Two accounts, same provider: index["google"] = ["google_abc123", "google_def456"]. load("google") returns first. load_all_for_provider("google") returns both. Existing tools see no change; new APIs can enumerate.

• Mixed providers: index["google"] = ["google_abc123"], index["slack"] = ["slack_xyz"]. Each provider resolves independently.

• Credential removed from Aden: On next sync_all(), rebuild_provider_index() rebuilds from disk. The removed credential's .enc file is gone, so it drops from the index naturally.

• exists() check: Validation calls exists("google") to check if credentials are available before running health checks. Must return True if ANY Google account exists, not just the last-synced one.

# _index_provider — append, don't overwrite
def _index_provider(self, credential):
    ...
    if provider_name not in self._provider_index:
        self._provider_index[provider_name] = []
    if credential.id not in self._provider_index[provider_name]:
        self._provider_index[provider_name].append(credential.id)

# load — first match (backward compat)
def load(self, credential_id):
    resolved_ids = self._provider_index.get(credential_id)
    if resolved_ids:
        for rid in resolved_ids:
            if rid != credential_id:
                result = self._load_by_id(rid)
                if result is not None:
                    return result
    return self._load_by_id(credential_id)

# NEW: enumerate all accounts
def load_all_for_provider(self, provider_name: str) -> list[CredentialObject]:
    results = []
    for cid in self._provider_index.get(provider_name, []):
        cred = self._load_by_id(cid)
        if cred:
            results.append(cred)
    return results

Step 3: Preserve Aden metadata as identity

File: core/framework/credentials/aden/provider.py

Business logic: When a user connects a Google account through Aden's OAuth flow, the Aden server stores metadata about the connected account — most importantly, the email address. This metadata comes back in every API response as metadata: {"email": "alice@company.com"}. Today, this metadata is present in AdenCredentialResponse.metadata (the from_dict() parser already handles it) but is never written into the CredentialObject's key vault. It's silently dropped.

Why Aden metadata is the primary identity source: Aden captures identity at the moment of OAuth authorization — the user explicitly grants access, and the Aden server records who they are. This is more authoritative than health checks because:

1. It's captured at consent time, not at validation time
2. It works even if the health check endpoint is down
3. It's available immediately on first sync, before any health check runs

When metadata arrives: Two code paths create/update credentials from Aden responses:

1. _aden_response_to_credential() — first-time sync. The credential doesn't exist locally yet. We're building it from scratch. Metadata should be written as _identity_* keys in the initial key dict.

2. _update_credential_from_aden() — token refresh. The credential already exists. The access token is updated. Metadata should be written/overwritten as _identity_* keys on the existing credential object.

Scenario — first sync: User connects alice@company.com through Aden. Aden returns {access_token: "...", metadata: {email: "alice@company.com"}}. The credential is created with _identity_email = "alice@company.com". Later, cred.identity.email returns "alice@company.com".

Scenario — token refresh: Alice's token expires. Aden refreshes it and returns updated metadata. _update_credential_from_aden() updates the access token AND refreshes _identity_email. If Alice changed her email (e.g., name change), the identity stays current.

Scenario — no metadata: Some Aden integrations may not return metadata (e.g., a simple API key integration). The loop for meta_key, meta_value in (metadata or {}).items() safely does nothing. The credential has no _identity_* keys, and cred.identity returns CredentialIdentity() with label == "unknown".

# In _aden_response_to_credential, after building keys dict:
for meta_key, meta_value in (aden_response.metadata or {}).items():
    if meta_value and isinstance(meta_value, str):
        keys[f"_identity_{meta_key}"] = CredentialKey(
            name=f"_identity_{meta_key}",
            value=SecretStr(meta_value),
        )

# In _update_credential_from_aden, after updating access_token:
for meta_key, meta_value in (aden_response.metadata or {}).items():
    if meta_value and isinstance(meta_value, str):
        credential.keys[f"_identity_{meta_key}"] = CredentialKey(
            name=f"_identity_{meta_key}",
            value=SecretStr(meta_value),
        )

Step 4: Extract identity from health checks

File: tools/src/aden_tools/credentials/health_check.py

Business logic: Health checks are the second identity source. Every agent startup runs validate_agent_credentials() which calls provider-specific health check endpoints. These endpoints return identity data as a side effect of validation:

Why health checks matter as an identity source:

1. Fallback when Aden metadata is missing: Not all Aden integrations return metadata. The health check always hits the actual service, so identity is always available on success.

2. Ground truth verification: Aden metadata is captured at OAuth time. If the user's email changed since then, the health check returns the CURRENT identity.

3. Non-Aden credentials: When credentials are configured via environment variables (no Aden), health checks are the ONLY identity source. A dev sets GOOGLE_ACCESS_TOKEN manually — the health check reveals whose token it is.

4. Zero additional cost: The health check API call is already happening. We just need to parse the response body that's currently discarded after the status code check.

Design — _extract_identity() hook: The base OAuthBearerHealthChecker gets a new virtual method _extract_identity(data: dict) -> dict[str, str] that subclasses override. The check() method calls it when the response is 200 OK:

class OAuthBearerHealthChecker:
    def _extract_identity(self, data: dict) -> dict[str, str]:
        """Override to extract identity fields from successful response."""
        return {}

    def check(self, access_token: str) -> HealthCheckResult:
        ...
        if response.status_code == 200:
            identity = {}
            try:
                data = response.json()
                identity = self._extract_identity(data)
            except Exception:
                pass  # Identity extraction is best-effort
            return HealthCheckResult(
                valid=True,
                message=f"{self.service_name} credentials valid",
                details={"identity": identity} if identity else {},
            )

Why details["identity"]: The existing HealthCheckResult has a details: dict field that's used ad-hoc by different checkers. By putting identity under a standardized "identity" key, Step 5 can generically extract it without knowing which checker ran. Existing details fields (username, team, bot_id) continue to exist alongside — no breaking changes.

Standalone checkers (Slack, GitHub, Discord) don't extend OAuthBearerHealthChecker. They already parse identity data into their details dict. For these, we simply add an "identity" key with the structured fields alongside existing keys.

Scenario — Gmail health check enriches a credential without Aden metadata: A dev sets GOOGLE_ACCESS_TOKEN as an env var. The credential has no _identity_* keys. On startup, the Gmail health check calls /users/me/profile, gets {emailAddress: "dev@gmail.com"}, returns details={"identity": {"email": "dev@gmail.com"}}. Step 5 persists this. Now cred.identity.email works even without Aden.

Scenario — health check fails: Token is expired or revoked. Response is 401. No identity extracted (identity extraction only runs on 200). The health check returns valid=False. Step 5 skips persistence. The credential's existing identity (if any, from Aden metadata) remains unchanged.

Scenario — identity extraction throws: The response body is malformed or missing expected fields. The try/except in check() catches it. Health check still returns valid=True (the token worked). Identity is just not extracted. Best-effort, never blocks validation.

Step 5: Persist identity during validation

File: core/framework/credentials/validation.py

Business logic: Steps 3 and 4 produce identity data. Step 5 is the bridge that takes identity from health check results and persists it to the credential store. This runs during validate_agent_credentials(), which is called at every agent startup.

Why persist during validation: Validation is the natural lifecycle hook because:

1. It runs on every agent startup (guaranteed execution)
2. It already has access to the credential store
3. It already runs health checks (identity is available in the result)
4. It runs BEFORE the agent executes (identity is available for system prompt injection)

Flow:

Agent startup
→ validate_agent_credentials()
  → for each credential:
    → check_credential_health(token) → HealthCheckResult
    → if result.valid AND result.details["identity"] exists:
      → cred_obj = store.get_credential(cred_id)
      → cred_obj.set_identity(**identity_data)
      → store.save_credential(cred_obj)  ← persisted to disk

Scenario — identity from health check augments Aden metadata: Aden provides metadata.email = "alice@company.com" (stored as _identity_email in Step 3). The Slack health check returns identity: {workspace: "Acme Corp", username: "hive-bot"}. Step 5 adds _identity_workspace and _identity_username to the Slack credential. Now both credentials have rich identity data from their respective sources.

Scenario — identity update on restart: Between agent runs, the GitHub user renamed from old-username to new-username. On next startup, the health check returns identity: {username: "new-username"}. Step 5 calls set_identity(username="new-username"), which overwrites _identity_username. The credential now reflects the current identity.

Scenario — multiple accounts of same provider: With the index fix (Step 2), validate_agent_credentials() iterates over all credentials. Each Google account gets its own health check. Each health check returns a different emailAddress. Each identity is persisted to the correct CredentialObject. Account A gets _identity_email = "alice@co.com", Account B gets _identity_email = "bob@co.com".

Error handling: Identity persistence is best-effort. If get_credential() fails or save_credential() fails, the exception is caught and swallowed. The agent still starts. The credential still works. It just won't have identity data for that account. This is acceptable because identity is informational, not functional.

if result.valid:
    identity_data = result.details.get("identity")
    if identity_data and isinstance(identity_data, dict):
        try:
            cred_obj = store.get_credential(cred_id, refresh_if_needed=False)
            if cred_obj:
                cred_obj.set_identity(**identity_data)
                store.save_credential(cred_obj)
        except Exception:
            pass  # Identity persistence is best-effort

Step 6: Account listing & identity APIs

Files: core/framework/credentials/store.py, tools/src/aden_tools/credentials/store_adapter.py

Business logic: Steps 1-5 populate identity data. Step 6 exposes it through clean APIs. Two layers need new methods:

1. CredentialStore (framework layer) — knows about CredentialObject and storage
2. CredentialStoreAdapter (tool boundary) — wraps the store with CredentialSpec-aware APIs, sits in the MCP subprocess, consumed by tools

Why two layers: The store is a framework concept (core/). The adapter is a tools concept (tools/). Tools never import from core directly. The adapter bridges the gap, translating between credential IDs and spec names, handling the "is this credential configured and available?" logic.

APIs added to CredentialStore:

• list_accounts(provider_name) — returns all accounts for a provider type with their identities. Delegates to storage.load_all_for_provider() (Step 2). Returns a list of dicts, not raw CredentialObjects, to avoid leaking secrets upstream.

• get_credential_by_identity(provider_name, label) — finds a specific account by matching cred.identity.label against the provided label. This is the resolution mechanism for future multi-account routing: "give me the token for alice@co.com."

APIs added to CredentialStoreAdapter:

• get_identity(name) — returns the identity dict for a named credential spec. Used by tools that want to know whose token they're using for logging/display.

• list_accounts(provider_name) — delegates to store. Used by the get_account_info MCP tool (Step 8).

• get_all_account_info() — iterates over all configured credential specs, collects all accounts across all providers. Used to build the system prompt (Step 7). Deduplicates by provider name to avoid listing the same provider's accounts twice when multiple specs map to the same provider.

• get_by_identity(provider_name, label) — resolves a specific account's token by identity label. Used by future multi-account routing (Step 9). Returns a raw token string, not a CredentialObject.

Scenario — system prompt building: At agent startup, the runner calls adapter.get_all_account_info(). The adapter iterates over specs: {"gmail": CredentialSpec(credential_id="google"), "gcal": CredentialSpec(credential_id="google"), "slack": CredentialSpec(...)}. It deduplicates by provider: google and slack. For google, list_accounts("google") returns 2 accounts. For slack, 1 account. Result: 3 account entries for the system prompt.

Scenario — identity-based routing (future): The LLM calls gmail_read_email(account="alice@co.com"). The tool calls credentials.get_by_identity("google", "alice@co.com"). The adapter delegates to store.get_credential_by_identity("google", "alice@co.com") which scans all Google credentials, finds the one where identity.label == "alice@co.com", and returns its access token. The right inbox is read.

# CredentialStore
def list_accounts(self, provider_name: str) -> list[dict[str, Any]]:
    if hasattr(self._storage, 'load_all_for_provider'):
        creds = self._storage.load_all_for_provider(provider_name)
    else:
        cred = self.get_credential(provider_name)
        creds = [cred] if cred else []
    return [
        {"credential_id": c.id, "provider": provider_name,
         "identity": c.identity.to_dict(), "label": c.identity.label}
        for c in creds
    ]

def get_credential_by_identity(self, provider_name: str, label: str) -> CredentialObject | None:
    if hasattr(self._storage, 'load_all_for_provider'):
        for cred in self._storage.load_all_for_provider(provider_name):
            if cred.identity.label == label:
                return cred
    return None

# CredentialStoreAdapter
def get_all_account_info(self) -> list[dict[str, Any]]:
    accounts = []
    seen: set[str] = set()
    for name, spec in self._specs.items():
        provider = spec.credential_id or name
        if provider in seen or not self.is_available(name):
            continue
        seen.add(provider)
        accounts.extend(self._store.list_accounts(provider))
    return accounts

def get_by_identity(self, provider_name: str, label: str) -> str | None:
    cred = self._store.get_credential_by_identity(provider_name, label)
    return cred.get_default_key() if cred else None

Step 7: Surface identity to LLM via system prompt

Files: prompt_composer.py, executor.py, event_loop_node.py, node.py, runner.py

Business logic: The LLM needs to know what accounts are connected so it can:

1. Communicate clearly to the user: "I checked alice@company.com's inbox and found 3 unread messages" vs. "I checked the inbox and found 3 unread messages"

2. Disambiguate operations: When asked "check my emails," the LLM can respond "You have 2 Google accounts connected: alice@company.com and bob@company.com. Which would you like me to check?" (requires Step 9 routing, but awareness comes first)

3. Prevent hallucination: Without account info, the LLM might invent account names or assume capabilities it doesn't have. With the accounts prompt, it knows exactly what's available.

4. Cross-account reasoning: "Forward the email from alice's inbox to bob's inbox" requires knowing both accounts exist and which is which.

Where it sits in the three-layer prompt:

Layer 1 — Identity: "You are a thorough email management agent."
         Accounts:  "Connected accounts:
                     - google: alice@company.com (email: alice@company.com)
                     - google: bob@company.com (email: bob@company.com)
                     - slack: Acme Corp (workspace: Acme Corp, username: hive-bot)"
Layer 2 — Narrative: "We've triaged 15 emails so far..."
Layer 3 — Focus:     "Your current task: categorize remaining unread emails"

Accounts sit between identity (static personality) and narrative (dynamic state) because connected accounts are semi-static — they don't change during a session but are deployment-specific (different from the agent definition).

Injection path through the framework:

AgentRunner._setup()
  → CredentialStoreAdapter.get_all_account_info()
  → build_accounts_prompt(accounts)           ← new function in prompt_composer.py
  → GraphExecutor(accounts_prompt=...)        ← new init param
  → NodeContext(accounts_prompt=...)          ← new field
  → compose_system_prompt(..., accounts_prompt=...)  ← new param

Why it flows through NodeContext: For the first node in a graph (or an isolated EventLoopNode), the system prompt is built in EventLoopNode.execute(), not through the continuous transition path. NodeContext.accounts_prompt carries the data to both paths:

• Continuous transition: compose_system_prompt() in the executor uses self.accounts_prompt directly
• First node / isolated node: EventLoopNode.execute() reads ctx.accounts_prompt and appends it to the system prompt

Scenario — no credentials: An agent with no external integrations (pure LLM reasoning, no tools). get_all_account_info() returns []. build_accounts_prompt([]) returns "". The accounts block is omitted from the system prompt. Zero impact.

Scenario — single account: One Google account. System prompt shows "Connected accounts:\n- google: alice@company.com (email: alice@company.com)". The LLM knows who it's operating as.

Scenario — unknown identity: A credential exists but has no _identity_* keys (maybe Aden didn't provide metadata and health checks haven't run yet). identity.label returns "unknown". The prompt shows "- google: unknown". Better than nothing — the LLM knows Google is connected, just not whose account.

def build_accounts_prompt(accounts: list[dict[str, Any]]) -> str:
    if not accounts:
        return ""
    lines = ["Connected accounts:"]
    for acct in accounts:
        provider = acct.get("provider", "unknown")
        label = acct.get("label", "unknown")
        identity = acct.get("identity", {})
        detail_parts = [f"{k}: {v}" for k, v in identity.items() if v]
        detail = f" ({', '.join(detail_parts)})" if detail_parts else ""
        lines.append(f"- {provider}: {label}{detail}")
    return "\n".join(lines)

Step 8: get_account_info MCP tool

New directory: tools/src/aden_tools/tools/account_info_tool/

Business logic: Step 7 gives the LLM passive awareness (system prompt). Step 8 gives the LLM active introspection — it can call get_account_info() to query connected accounts at runtime, even mid-conversation.

Why both passive and active: The system prompt provides context at conversation start. But in long-running agents with many tools, the system prompt may get compacted (truncated during context management). The MCP tool ensures the LLM can always re-discover account info even after compaction.

Use cases:

• User asks "what accounts are connected?": LLM calls get_account_info(), formats the response for the user.

• LLM needs to decide which account to use: Before sending an email, the LLM calls get_account_info(provider="google") to see which Gmail accounts are available, then asks the user which one to send from.

• Dynamic account discovery: In a long-running session, accounts might be added/revoked (Aden dashboard). The tool provides current state vs. the stale system prompt.

• Debugging/transparency: The user can ask "which Slack workspace are you connected to?" and get a precise answer.

API design:

@mcp.tool()
def get_account_info(provider: str = "") -> dict:
    """List connected accounts and their identities.

    Call with no arguments to see all connected accounts.
    Call with provider="google" to filter by provider type.

    Returns account IDs, provider types, and identity labels
    (email, username, workspace) for each connected account.
    """
    if credentials is None:
        return {"accounts": [], "message": "No credential store configured"}
    if provider:
        accounts = credentials.list_accounts(provider)
    else:
        accounts = credentials.get_all_account_info()
    return {"accounts": accounts, "count": len(accounts)}

Response example:

{
  "accounts": [
    {"credential_id": "google_abc123", "provider": "google",
     "identity": {"email": "alice@company.com"}, "label": "alice@company.com"},
    {"credential_id": "google_def456", "provider": "google",
     "identity": {"email": "bob@company.com"}, "label": "bob@company.com"},
    {"credential_id": "slack_xyz", "provider": "slack",
     "identity": {"workspace": "Acme Corp", "username": "hive-bot"},
     "label": "Acme Corp"}
  ],
  "count": 3
}

Register in tools/src/aden_tools/tools/__init__.py alongside existing tools.

Step 9: Multi-account routing extension point (design only, no code)

Business logic: Steps 1-8 build the foundation. Step 9 designs (but does not implement) the per-tool-call account selection mechanism. This is the endgame: when the LLM calls gmail_read_email(account="alice@co.com"), the right token is used.

Why design-only in this PR: Multi-account routing requires changes to every tool's _get_token() function and introduces the account parameter across all tool signatures. This is a significant surface area change that should be a separate PR with its own testing. The foundation from Steps 1-8 makes it a straightforward addition.

How it will work — the full flow:

1. LLM discovers accounts: Via system prompt (Step 7) or get_account_info tool (Step 8), the LLM knows alice@company.com and bob@company.com are connected.

2. **User

…(truncated)