> ## Documentation Index
> Fetch the complete documentation index at: https://cortex-e852fafe-docs-pro-2457-cookbooks-cleanup.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# AI Chief of Staff: Function Routing

> Build an AI Chief of Staff that takes real actions across your workspace using HydraDB function routing. Register every callable function as a knowledge object in HydraDB. Any agent or user can say 'prepare for tomorrow's board meeting' and receive a structured execution plan.

# Build an AI Chief of Staff

An AI that doesn't just answer questions; it *takes action*. Register every callable function in your workspace as a knowledge object in HydraDB. Any agent or user can say "prepare for tomorrow's board meeting" and receive a structured, personalized execution plan: which functions to call, in which order, with which parameters.

***

Most AI assistants are read-only. They answer questions, summarize documents, and draft emails. What they can't do is *act*: book the meeting, update the CRM, send the Slack message, trigger the deployment. To cross that threshold, the agent needs to know not just what functions exist in your workspace, but which one to call for any given task, in what order, with what parameters, and for which user.

This cookbook builds an **AI Chief of Staff**: an autonomous reasoning layer that turns natural language into structured function calls across every app in your stack. Think of it as n8n, but driven by intent rather than rigid if-then workflows. You register your callable functions into HydraDB as knowledge objects. Any agent then asks HydraDB: *"What should I do for this task?"* HydraDB returns the right function, and your LLM turns it into the right parameters and the right sequence, all personalized to the requesting user's preferences and the current context.

The architectural insight is separation of concerns: your primary LLM handles conversation and intent extraction, while HydraDB becomes the **function selection oracle**: a reasoning layer that has learned which functions work for which tasks, which sequences tend to succeed together, and how individual users prefer to work. Over time, it builds institutional knowledge that your agents can tap into.

ℹ️

> **Note:** All code in this cookbook is real. Base URL: `https://api.hydradb.com`. Get your API key at [app.hydradb.com](https://app.hydradb.com).

## Prerequisites

**Required knowledge**: Python basics, REST APIs, environment variables, basic understanding of OAuth\
**Required tools**:

* HydraDB API key
* Python 3.11 or 3.12 (`python --version`)
* OpenAI API key (for parameter extraction and plan sequencing)
* `pip install hydradb-sdk openai slack-sdk backoff flask google-api-python-client`

## What You'll Build

By the end of this cookbook, you'll be able to:

* Register any workspace function (Slack, Calendar, CRM, Jira) as a HydraDB knowledge object so agents can discover it semantically
* Build an Orchestrator that translates a natural-language task into an authorized, token-injected API call
* Generate multi-step execution plans for complex tasks like "onboard the new hire"
* Store per-user preference memory so HydraDB personalizes function suggestions over time
* Feed execution outcomes back into HydraDB to close the self-improvement loop

**❌ // Traditional automation (n8n, Zapier, Make)**

* Rigid if-then workflows: break when conditions change
* Every automation hardcoded: no personalization
* No memory of past executions or user preferences
* Requires manual mapping of triggers to actions
* Context window resets on every run

**✅ // HydraDB AI Chief of Staff**

* Intent-driven routing: adapts to how requests are phrased
* Per-user personalization via AI Memories
* Learns function composition patterns from execution history
* One natural language request returns a full execution plan
* Compound intelligence: every run makes future runs smarter

## How HydraDB Enables This

Four HydraDB capabilities make a Chief of Staff possible:

* **Functions as knowledge objects:** each callable function is uploaded to HydraDB via `POST /context/ingest` as an `app_knowledge` item. The function's natural-language description becomes the retrieval surface. HydraDB matches tasks to functions semantically, not by keyword, so "tell the team about the delay" correctly surfaces `send_slack_message` even though neither word appears in the function name.
* **Personalized function selection:** when a user frequently chooses Slack over email for urgent updates, HydraDB's AI Memories encode that preference in the user's collection. The planner reads it, so future function suggestions for that user favour `send_slack_message` over `send_email`. This happens without any manual configuration: the pattern emerges from usage stored via `POST /context/ingest`.
* **Multi-step plan generation:** `mode: "thinking"` on `POST /query` enables multi-query reasoning. HydraDB returns the candidate functions for a complex request like "onboard the new hire", and your LLM sequences them into an execution plan with dependencies.
* **Self-improving function routing:** feeding execution results back to HydraDB via `POST /context/ingest` closes the learning loop. Slow functions, failed calls, and successful sequences all become memories that the planner reads on later runs, without any manual tuning.

## Architecture

One HydraDB database. Functions registered as knowledge objects. An Action Orchestrator that translates HydraDB suggestions into real API calls. Per-user memories that personalize every suggestion.

```mermaid theme={"dark"}
flowchart LR
    A["User / Agent<br/>(natural-language task)"] --> B["Action Orchestrator"]
    B --> C["HydraDB<br/>(semantic function matching)"]
    C --> D["Ranked function suggestions"]
    D --> E["Policy Engine<br/>(authorization)"]
    E --> F["Auth Vault<br/>(OAuth tokens)"]
    F --> G["External API<br/>(function execution)"]
    G --> H["Execution result"]
    H --> C
    H --> I["Per-user memories"]
    I --> C
```

The flow: a user or agent sends a natural-language task to the Action Orchestrator. The Orchestrator queries HydraDB, which matches the task semantically against registered function knowledge objects and returns a ranked suggestion. The Orchestrator executes the function via the real API, logs the result back to HydraDB as a memory, and the loop closes. Each execution makes the next suggestion smarter.

ℹ️

> **The Orchestrator is thin by design:** It has no opinion about which function to call; that decision belongs to HydraDB. The Orchestrator's only job is: receive suggestion, authorize it against the policy engine, inject the correct OAuth token from the auth vault, call the real API, and report the outcome. All routing intelligence lives in HydraDB.

STEP 1

## Create Database

One database for the whole Chief of Staff system. All functions, all user memories, and all execution history live under this database. Collections scope function access per team or department: the sales team's agent only sees sales functions, the engineering team's agent only sees deployment and monitoring functions.

Install the Python packages used by the examples and set both API keys:

```bash theme={"dark"}
pip install hydradb-sdk openai slack-sdk backoff flask google-api-python-client
export HYDRA_DB_API_KEY="your_hydradb_key"
export OPENAI_API_KEY="your_openai_key"
```

HydraDB handles retrieval and memory. The OpenAI SDK is used only in the app layer for turning retrieved function schemas into executable parameters and multi-step plans; you can replace it with any LLM provider.

```python title="setup.py" theme={"dark"}
import os, time
from hydra_db import HydraDB

API_KEY  = os.environ["HYDRA_DB_API_KEY"]
DATABASE = "chief-of-staff"

client = HydraDB(token=API_KEY)

def create_database():
    """Create the main database. Returns 409 if it already exists."""
    client.databases.create(
        database=DATABASE,
        # Declare every top-level metadata key you filter on. Schema fields are
        # immutable after creation, and ingest rejects undeclared metadata keys.
        database_metadata_schema=[{"name": "deprecated", "data_type": "BOOL"}],
    )
    # Creation is asynchronous: wait until the database accepts data.
    while not client.databases.status(database=DATABASE).data.infra.ready_for_ingestion:
        time.sleep(5)
    print(f"Database '{DATABASE}' ready.")

# Collections scope functions per team. Created automatically on first write.
# Sales agents only see sales functions. Engineering agents only see deploy functions.
# Examples:
#   collection = "sales"        for CRM, calendar, email, proposals
#   collection = "engineering"  for deploys, incidents, GitHub, monitoring
#   collection = "executive"    for board decks, reporting, scheduling
#   collection = "hr"           for hiring, onboarding, HRIS actions

if __name__ == "__main__":
    create_database()
```

Output:

```
Database 'chief-of-staff' ready.
```

STEP 2

## Define & Register Functions

Every action your Chief of Staff can take must be registered in HydraDB as a knowledge object. HydraDB treats each function as a document: its natural-language description is the retrieval surface, its schema is the execution contract, and its metadata controls who can access it and under what conditions.

The quality of your function descriptions directly determines routing accuracy. Write descriptions that explain *what the function achieves* and *when it should be used*, not just what it does technically. HydraDB reasons over these descriptions during function selection.

### Function schema

Each function schema has four components: `id` (stable identifier for versioning and logging), `description` (the natural-language surface HydraDB matches tasks against; write this carefully), `parameters` (JSON Schema for the execution contract), and `meta` (access control, constraints, and routing hints).

```json title="schemas/send_slack_message.json" theme={"dark"}
{
  "id": "send_slack_message",
  "name": "Send a Slack message",
  "description": "Posts a message to a Slack channel or DM on behalf of the user. Use this when the user wants to notify a team, send an update, share information with a colleague, or make an announcement. Prefer this over send_email for internal, time-sensitive, or informal communications.",
  "parameters": {
    "type": "object",
    "properties": {
      "channel":   { "type": "string",  "description": "#channel-name or @username or user ID" },
      "text":      { "type": "string",  "description": "Message body; supports Slack mrkdwn" },
      "thread_ts": { "type": "string",  "description": "Optional. Reply in thread by passing parent message ts." }
    },
    "required": ["channel", "text"]
  },
  "auth": {
    "oauth_provider": "slack",
    "scopes": ["chat:write"]
  },
  "meta": {
    "collections":       ["communication", "slack"],
    "side_effects":      "Sends a visible message. Cannot be unsent programmatically.",
    "idempotent":        false,
    "permissions":       ["all_users"],
    "business_hours_only": false
  }
}
```

💡

> **The description field is your routing budget:** HydraDB matches tasks to functions by reading the description semantically. Include: (1) what the function *achieves*, not just what it does, (2) when it should be preferred over similar functions, and (3) any important constraints or side effects. A description like *"Posts a Slack message"* routes poorly. A description that explains "Use this for internal, time-sensitive, or informal communications. Prefer over send\_email" routes correctly even when the user says "ping the team".

Here is a more complex schema: a finance function with approval constraints and metadata filters that prevent it from being suggested outside authorized contexts:

```json title="schemas/approve_expense.json" theme={"dark"}
{
  "id": "approve_expense",
  "name": "Approve an expense report",
  "description": "Approves a pending expense report in the finance system. Use when a manager or finance lead needs to sign off on submitted expenses. Always confirm the amount and submitter before calling. Do not suggest this function outside business hours or for users without manager-level permissions.",
  "parameters": {
    "type": "object",
    "properties": {
      "expense_id":  { "type": "string", "description": "Expense report ID from finance system" },
      "approver_id": { "type": "string", "description": "User ID of the approving manager" },
      "note":        { "type": "string", "description": "Optional approval note for audit log" }
    },
    "required": ["expense_id", "approver_id"]
  },
  "meta": {
    "department":         "finance",
    "permission_level":   "manager",
    "cost_threshold":     5000,
    "business_hours_only": true,
    "collections":        ["finance", "approvals"],
    "side_effects":       "Triggers payment processing. Irreversible without finance team intervention.",
    "idempotent":         false
  }
}
```

### Upload to HydraDB

Upload functions using the same `POST /context/ingest` endpoint used for documents. Send each function as an `app_knowledge` item with the full JSON schema as its `content.text`. Group functions into collections so HydraDB can scope retrieval per team without returning irrelevant options.

```python title="register/upload_functions.py" theme={"dark"}
import json, time, os
from hydra_db import HydraDB

# All functions in the workspace. Each schema loaded from its own file.
FUNCTION_SCHEMAS = [
    "schemas/send_slack_message.json",
    "schemas/send_email.json",
    "schemas/create_calendar_event.json",
    "schemas/update_crm_opportunity.json",
    "schemas/create_jira_ticket.json",
    "schemas/approve_expense.json",
    "schemas/trigger_deployment.json",
    "schemas/notify_pagerduty.json",
    "schemas/generate_report.json",
    "schemas/add_to_notion.json",
]

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"


def load_schema(path: str) -> dict:
    with open(path) as f:
        return json.load(f)

def upload_functions(schema_paths: list, collection: str = "functions") -> list:
    """
    Upload function schemas to HydraDB as knowledge objects.
    collection: scopes which agents can see these functions.
    Use per-team collections to limit scope and improve routing precision.

    Tip: re-running this is idempotent: HydraDB upserts on 'id'.
    """
    batch   = []
    all_ids = []

    for path in schema_paths:
        schema = load_schema(path)
        fn_id  = schema["id"]

        batch.append({
            "id":        fn_id,
            "title":     schema["name"],
            "type":      "function",
            "timestamp": "2025-01-01T00:00:00Z",
            "content":   {"text": json.dumps(schema, indent=2)},
            # Schema-backed, so metadata_filters can match it: declared in setup.py.
            "metadata": {"deprecated": False},
            # Free-form routing hints: no schema needed.
            "additional_metadata": {
                "collections":  schema.get("meta", {}).get("collections", []),
                "department":   schema.get("meta", {}).get("department", "all"),
                "permissions":  schema.get("meta", {}).get("permissions", ["all_users"]),
                "idempotent":   schema.get("meta", {}).get("idempotent", True),
                "side_effects": schema.get("meta", {}).get("side_effects", ""),
            },
        })

        if len(batch) == 20:
            all_ids += _upload_batch(batch, collection)
            batch = []; time.sleep(1)

    if batch:
        all_ids += _upload_batch(batch, collection)

    print(f"Functions: {len(all_ids)} schemas queued for indexing.")
    return all_ids


def _upload_batch(batch: list, collection: str) -> list:
    result = client.context.ingest(
        database=DATABASE,
        collection=collection,
        app_knowledge=json.dumps(batch),
    )
    return [item.id for item in (result.data.results or []) if item.id]


# Upload all functions (all teams, scoped to "functions" collection)
# For team-scoped variants, call again with collection="sales", "engineering", etc.
if __name__ == "__main__":
    upload_functions(FUNCTION_SCHEMAS, collection="functions")
```

Output:

```
Functions: 10 schemas queued for indexing.
```

⚠️

> **Use team-scoped collections for precision:** If your function library grows to 50+ functions, a single `collection="functions"` will degrade routing quality because HydraDB must rank more candidates per query. Instead, upload sales functions to `collection="sales"`, engineering functions to `collection="engineering"`, and so on. Agents query only their team's collection. Fewer candidates = more accurate suggestions. Aim for fewer than 30 functions per collection for optimal routing.

### Versioning & deprecation

As functions evolve, use a `_v2` suffix on the ID for new versions. Mark deprecated versions in metadata so HydraDB stops routing to them while preserving historical execution records. Never delete old function objects; they anchor memory traces from past executions.

```python title="register/versioning.py" theme={"dark"}
import json, os
from hydra_db import HydraDB

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"


def deprecate_function(fn_id: str, collection: str, reason: str):
    """
    Mark a function as deprecated so HydraDB stops suggesting it.
    Never delete: old executions reference this ID for audit and provenance.
    Use 'deprecated: true' in metadata + upload new version as fn_id_v2.
    """
    client.context.ingest(
        database=DATABASE,
        collection=collection,
        app_knowledge=json.dumps([{
            "id":        fn_id,
            "title":     f"[DEPRECATED] {fn_id}",
            "type":      "function",
            "timestamp": "2025-01-01T00:00:00Z",
            "content":   {"text": f"DEPRECATED: {reason}. Use {fn_id}_v2 instead."},
            "metadata":  {"deprecated": True},
            "additional_metadata": {
                "deprecated_reason":  reason,
                "successor_id":       f"{fn_id}_v2",
            },
        }]),
    )
    print(f"Deprecated {fn_id}. Successor: {fn_id}_v2")


# Example: create_calendar_event_v2 adds video conferencing links
deprecate_function(
    fn_id="create_calendar_event",
    collection="functions",
    reason="Replaced by a version that adds video conferencing links",
)
```

STEP 3

## Build the Action Orchestrator

The Orchestrator is the runtime layer between HydraDB suggestions and real API calls. It receives a natural-language task, asks HydraDB which function solves it, authorizes the call against the policy engine, injects the correct OAuth token, executes the API, and logs the outcome back to HydraDB. It has no opinion about which function to call; that belongs to HydraDB entirely.

### Core orchestrator class

```python title="orchestrator/core.py" theme={"dark"}
import uuid, json, os
from openai import OpenAI
from hydra_db import HydraDB

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"

class ChiefOfStaffOrchestrator:
    """
    The Orchestrator bridges HydraDB function suggestions and real workspace APIs.
    It does NOT decide which function to call: that is HydraDB's job.
    It does: authorize, inject tokens, execute, log outcomes.
    """

    def __init__(self, registry: dict, policy_engine, auth_vault):
        """
        registry:     Map of function_id to the callable that executes the real API call.
        policy_engine: Object with .check(user, function_id, params), returning bool
        auth_vault:   Object with .get_token(user_id, oauth_provider), returning str
        """
        self.registry      = registry
        self.policy        = policy_engine
        self.vault         = auth_vault
        self.llm           = OpenAI()

    def handle_task(
        self,
        task:        str,   # natural language: "book a 30-min call with alice next tuesday"
        user_id:     str,
        session_id:  str = None,
        collection:  str = "functions",
    ) -> dict:
        """
        Single-function task handling.
        1. Ask HydraDB which function to call.
        2. Authorize against policy engine.
        3. Inject OAuth token from vault.
        4. Execute via registry callable.
        5. Log result back to HydraDB as memory.
        """
        session_id = session_id or str(uuid.uuid4())

        # Step 1: Ask HydraDB for the best function
        search = client.query(
            database=DATABASE,
            collection=collection,
            query=task,
            max_results=5,
            mode="thinking",
            metadata_filters={"deprecated": {"equals": False}},
        )

        chunks = search.data.chunks or []
        if not chunks:
            return {"status": "no_match", "message": "No function found for this task."}

        # Top chunk is the best-matching function schema
        top_chunk   = chunks[0]
        schema      = json.loads(top_chunk.chunk_content)
        function_id = schema["id"]

        # Step 2: Check user permissions against the policy engine
        if not self.policy.check(user_id, function_id, schema):
            self._log_blocked(user_id, task, function_id)
            return {"status": "blocked", "reason": "Policy engine denied this function for your role."}

        # Step 3: Use the LLM to extract params from task + schema
        params = self._extract_params(task, schema, user_id, session_id)

        # Step 4: Inject OAuth token for the function's provider
        oauth_provider = schema.get("auth", {}).get("oauth_provider")
        if oauth_provider:
            params["_token"] = self.vault.get_token(user_id, oauth_provider)

        # Step 5: Execute via registry
        exec_fn = self.registry.get(function_id)
        if not exec_fn:
            return {"status": "error", "message": f"No executor registered for {function_id}."}

        result = exec_fn(params)

        # Step 6: Log outcome to HydraDB memory for self-improvement
        self._log_execution(user_id, task, function_id, params, result)

        return {"status": "done", "function_id": function_id, "result": result}


    def _extract_params(self, task: str, schema: dict, user_id: str, session_id: str) -> dict:
        """
        Extract structured parameters locally after HydraDB has selected the function.
        HydraDB should be used for retrieval (`/query`); parameter extraction
        belongs in your app/LLM layer because the schema is already in memory here.
        """
        prompt = (
            "Extract the parameters needed to call this function.\n"
            f"Task: {task}\n"
            f"Function schema: {json.dumps(schema['parameters'])}\n"
            "Return ONLY a valid JSON object with the extracted parameter values."
        )
        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )
        answer = response.choices[0].message.content or "{}"
        try:
            return json.loads(answer)
        except json.JSONDecodeError:
            return {}


    def _log_execution(self, user_id, task, function_id, params, result):
        """Log execution outcome back to HydraDB. Closes the learning loop."""
        success = result.get("success", True)
        outcome = "success" if success else "failure"
        client.context.ingest(
            type='memory',
            database=DATABASE,
            collection=f"user-{user_id}",
            upsert=True,
            memories=json.dumps([{
                "text": (
                    f"Task: {task}\n"
                    f"Function used: {function_id}\n"
                    f"Outcome: {outcome}\n"
                    f"Summary: {str(result.get('summary',''))[:300]}"
                ),
                "user_name": user_id,
                "infer": True,
            }]),
        )

    def _log_blocked(self, user_id, task, function_id):
        """Log a blocked attempt for audit trail."""
        client.context.ingest(
            type='memory',
            database=DATABASE,
            upsert=True,
            memories=json.dumps([{
                "text": f"BLOCKED: User {user_id} attempted {function_id} for task: {task}. Policy denied.",
                "user_name": "audit-log",
                "infer": False,
            }]),
        )
```

> **Note**: Use `metadata_filters` for hard, exact routing constraints. Top-level keys match schema-backed `metadata` fields such as `deprecated`; free-form per-source fields belong under `additional_metadata`. `setup.py` declares `deprecated` when it creates the database, and `upload_functions` sets `deprecated: false` on every live function.

### Function registry

The registry maps each `function_id` to the Python callable that makes the real API call. This is where OAuth tokens are consumed, retries happen, and real side effects occur. The Orchestrator passes `params` from the LLM parameter extraction directly to the executor.

```python title="orchestrator/registry.py" theme={"dark"}
from slack_sdk import WebClient
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build as google_build
import backoff

def execute_send_slack_message(params: dict) -> dict:
    """
    Execute send_slack_message using the injected OAuth token.
    params["_token"] injected by Orchestrator from auth vault.
    """
    token   = params.pop("_token", None)
    client  = WebClient(token=token)
    channel = params["channel"]
    text    = params["text"]
    kwargs  = {"channel": channel, "text": text}
    if params.get("thread_ts"):
        kwargs["thread_ts"] = params["thread_ts"]
    resp = client.chat_postMessage(**kwargs)
    return {"success": True, "ts": resp["ts"], "channel": resp["channel"],
            "summary": f"Slack message sent to {channel}."}


@backoff.on_exception(backoff.expo, Exception, max_tries=3)
def execute_create_calendar_event(params: dict) -> dict:
    """
    Create a Google Calendar event via the Calendar API v3.
    Exponential back-off handles transient 429s and 500s automatically.
    """
    token    = params.pop("_token", None)
    service  = google_build("calendar", "v3", credentials=Credentials(token))
    event    = {
        "summary":     params.get("title", "Meeting"),
        "start":       {"dateTime": params["start_time"], "timeZone": params.get("timezone", "UTC")},
        "end":         {"dateTime": params["end_time"],   "timeZone": params.get("timezone", "UTC")},
        "attendees":   [{"email": e} for e in params.get("attendees", [])],
        "description": params.get("description", ""),
    }
    result = service.events().insert(calendarId="primary", body=event, sendUpdates="all").execute()
    return {"success": True, "event_id": result["id"], "link": result.get("htmlLink"),
            "summary": f"Calendar event '{result['summary']}' created."}


# Registry: maps function_id to executor callable
FUNCTION_REGISTRY = {
    "send_slack_message":     execute_send_slack_message,
    "create_calendar_event":  execute_create_calendar_event,
    # "send_email":             execute_send_email,
    # "update_crm_opportunity": execute_update_crm,
    # "create_jira_ticket":     execute_create_jira_ticket,
    # "approve_expense":        execute_approve_expense,
    # "trigger_deployment":     execute_trigger_deployment,
}
```

💡

> **Start with read-only functions:** Before enabling write actions, build confidence with read-only lookups: `get_calendar_events`, `get_crm_opportunity`, `get_open_tickets`. These validate HydraDB's routing accuracy without side effects. Once routing is correct on reads, graduate to writes. The Chief of Staff earns trust incrementally, never by starting with "approve\_expense".

### Result feedback loop

The feedback loop is what separates a static function router from a learning system. After every execution, write a structured memory to HydraDB with `infer: true`. HydraDB extracts: which function was chosen, whether it succeeded, and what the user was trying to do. Over time, these signals shift the function preference profile for each user, making suggestions increasingly accurate without any manual tuning.

```python title="orchestrator/feedback.py" theme={"dark"}
import json
import os
from hydra_db import HydraDB

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"


def log_function_feedback(
    user_id:     str,
    function_id: str,
    task:        str,
    outcome:     str,   # "success" | "failure" | "slow" | "user_rejected"
    latency_ms:  int = 0,
    details:     str = "",
):
    """
    Write execution feedback as a memory so HydraDB learns from outcomes.
    infer: true: HydraDB extracts preference signals and builds graph links
    between this user, this function, and similar tasks.

    outcome="user_rejected" is especially valuable: the agent suggested the
    wrong function, the user corrected it. That correction shifts the routing
    weight for this user's context in future calls.
    """
    text = (
        f"Function: {function_id}\n"
        f"Task: {task}\n"
        f"Outcome: {outcome}\n"
        f"Latency: {latency_ms}ms\n"
    )
    if details:
        text += f"Details: {details}"

    client.context.ingest(
        type='memory',
        database=DATABASE,
        collection=f"user-{user_id}",
        upsert=True,
        memories=json.dumps([{
            "text":      text,
            "user_name": user_id,
            "infer":     True,
        }]),
    )


# Usage: called automatically by the Orchestrator's _log_execution method.
# Also call explicitly for user-driven corrections:
log_function_feedback(
    user_id="sarah",
    function_id="send_email",
    task="tell the team the deploy is done",
    outcome="user_rejected",
    details="Sarah always uses Slack for internal updates, not email. Corrected to send_slack_message.",
)
# HydraDB now holds: for Sarah, "tell the team" maps to send_slack_message
```

STEP 4

## Store Agent Memory

Two types of memory drive personalization. **User preference memory** stores how each person prefers to work: which channels they favour, which functions they trust, how they phrase requests. **Execution outcome memory** stores what happened when functions were called: successes, failures, latency patterns, user corrections. Together, these build a complete model of each user's working style. Memories live in each user's own collection, so a function lookup on the `functions` collection does not read them; the planner in Step 5 queries the user's collection separately and passes the preferences to the LLM with the candidate functions.

### User preference memory

Write explicit preference profiles during onboarding and update them whenever a user changes how they work. Use `infer: true` so HydraDB extracts the implicit signals (channel preferences, communication style, urgency thresholds) and builds graph connections to related functions automatically.

```python title="memory/user_preferences.py" theme={"dark"}
import json
import os
from hydra_db import HydraDB

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"


def store_user_preferences(user_id: str, profile: str):
    """
    Store a user's working preferences so HydraDB personalizes function
    suggestions for them. infer: true: HydraDB extracts channel preferences,
    urgency signals, communication style, and links these to specific functions.

    Call during onboarding and whenever preferences change.
    Use the same user_id consistently across all memory writes for this user.
    """
    client.context.ingest(
        type='memory',
        database=DATABASE,
        collection=f"user-{user_id}",
        upsert=True,
        memories=json.dumps([{
            "text":      profile,
            "user_name": user_id,
            "infer":     True,
        }]),
    )


# Onboard team members: called once per user at setup, then updated as needed
store_user_preferences(
    user_id="sarah",
    profile=(
        "Sarah is the VP of Engineering. She prefers Slack DMs over email for all internal "
        "communication. For urgent issues she always uses PagerDuty, not Jira. "
        "She approves expenses only during business hours. "
        "Her calendar blocks 9 to 10am daily for deep work. Never schedule meetings there. "
        "She likes executive summaries, not raw data. Always call generate_report before "
        "presenting metrics to her."
    ),
)

store_user_preferences(
    user_id="james",
    profile=(
        "James is an Account Executive on the sales team. He prefers email for external "
        "communications and Slack for internal ones. He runs a lot of demos. When he says "
        "'prepare for a call', always check the CRM for the prospect's deal stage first. "
        "He is in the Pacific timezone. Do not schedule past 5pm his time. "
        "He always CC's his manager when sending proposals."
    ),
)
```

### Execution outcome memory

Beyond preferences, HydraDB needs to know what actually happened. Store each execution outcome as a memory with enough detail for HydraDB to identify patterns: which functions tend to succeed together, which fail under specific conditions, which are consistently slow. Use `infer: false` for exact outcome records and `infer: true` for synthesized pattern summaries.

```python title="memory/outcomes.py" theme={"dark"}
import json
from datetime import datetime, timezone
from hydra_db import HydraDB
import os

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"

def log_execution_outcome(
    user_id:     str,
    task:        str,
    function_id: str,
    outcome:     str,         # "success" | "failure" | "timeout" | "user_rejected"
    latency_ms:  int,
    error_msg:   str = "",
    chained_fns: list = None,   # other functions called in the same task
):
    """
    Log an execution outcome verbatim (infer: false) for the audit trail.
    Also write a synthesized pattern summary (infer: true) for learning.
    These two writes serve different purposes:
      - infer: false: exact record, queryable for compliance and audit
      - infer: true:  HydraDB extracts patterns and links to similar tasks
    """
    ts = datetime.now(timezone.utc).isoformat()

    # Write 1: exact record
    client.context.ingest(
        type='memory',
        database=DATABASE,
        collection="execution-log",
        upsert=True,
        memories=json.dumps([{
            "text": (
                f"[{ts}] user={user_id} fn={function_id} "
                f"outcome={outcome} latency={latency_ms}ms\n"
                f"task={task}\n"
                f"chained={chained_fns or []}\n"
                f"error={error_msg or 'none'}"
            ),
            "user_name": "system",
            "infer":     False,
        }]),
    )

    # Write 2: synthesized pattern (only for notable events)
    if outcome in ("failure", "timeout", "user_rejected") or latency_ms > 3000:
        summary = (
            f"{function_id} returned {outcome} for task type '{task[:80]}'. "
        )
        if latency_ms > 3000:
            summary += f"Latency was {latency_ms}ms, above the 3s threshold. "
        if error_msg:
            summary += f"Error: {error_msg}. "
        if outcome == "user_rejected":
            summary += f"User manually overrode this suggestion for user_id={user_id}."

        client.context.ingest(
            type='memory',
            database=DATABASE,
            collection=f"user-{user_id}",
            upsert=True,
            memories=json.dumps([{
                "text":      summary,
                "user_name": user_id,
                "infer":     True,
            }]),
        )
```

STEP 5

## Multi-Step Planning

Many real-world tasks require more than one function call. "Onboard the new hire" isn't one action; it's creating accounts, sending welcome materials, scheduling orientation, assigning equipment, and notifying the team. Instead of hardcoding this sequence, ask HydraDB for the candidate functions and the user's preferences, and have your LLM generate the plan: a JSON array of functions with their parameters and dependency order. Your execution engine runs them sequentially, injecting earlier outputs into later calls.

ℹ️

> **Use `mode: "thinking"` for plan generation:** `mode: "thinking"` enables HydraDB's multi-query decomposition: it expands the task into several queries and reranks the results, so each part of a compound request surfaces a function. `mode: "fast"` runs a single low-latency pass. Always use `"thinking"` when the task is complex or ambiguous. Plan generation typically takes 200 to 600ms.

### Generate a plan

```python title="planning/generate_plan.py" theme={"dark"}
import json, os
from openai import OpenAI
from hydra_db import HydraDB

openai_client = OpenAI()
client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"

def generate_execution_plan(
    task:       str,
    user_id:    str,
    collection: str = "functions",
    max_steps:  int = 8,
) -> list:
    """
    Generate a multi-step execution plan for a complex task.
    1. Search the most relevant functions from HydraDB using mode: "thinking".
    2. Pass the ranked functions + task to an LLM for sequencing.
    3. Return a structured plan: [{step, function_id, params, depends_on}].

    max_steps: cap the plan length to avoid runaway chains.
    """
    # Step 1: search candidate functions with thinking mode
    search = client.query(
        database=DATABASE,
        collection=collection,
        query=task,
        max_results=12,
        mode="thinking",
        graph_context=True,
        metadata_filters={"deprecated": {"equals": False}},
    )

    chunks = search.data.chunks or []
    if not chunks:
        return []

    # Build the function catalogue string for the planner
    fn_catalogue = "\n\n".join(
        f"FUNCTION {i+1}: {c.source_title}\n{c.chunk_content[:600]}"
        for i, c in enumerate(chunks)
    )

    # Search user preferences to personalise the plan
    pref_search = client.query(
        type="memory",
        database=DATABASE,
        collection=f"user-{user_id}",
        query="channel preferences urgency communication style",
        mode="thinking",
    )
    user_prefs = "\n".join(c.chunk_content for c in (pref_search.data.chunks or []))

    # Step 2: use LLM to sequence the plan
    resp = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a task planner. Given a task and a catalogue of available functions, "
                    "produce a minimal, ordered execution plan. "
                    "Return ONLY valid JSON: a list of objects with keys: "
                    "step (int), function_id (str), params (object), depends_on (list of step ints), reason (str). "
                    "Maximum steps: " + str(max_steps) + ". "
                    "Use user preferences to choose the right channels and timing. "
                    "Only include functions that are strictly necessary."
                ),
            },
            {
                "role": "user",
                "content": (
                    f"Task: {task}\n\n"
                    f"User preferences: {user_prefs}\n\n"
                    f"Available functions:\n{fn_catalogue}"
                ),
            },
        ],
        temperature=0.1,
    )

    try:
        plan = json.loads(resp.choices[0].message.content)
        return plan
    except (json.JSONDecodeError, KeyError):
        return []


# Usage
plan = generate_execution_plan(
    task="Onboard Alex Chen who starts Monday as a backend engineer on the payments team.",
    user_id="sarah",
)
# Returns:
# [
#    {step:1, function_id:"create_gsuite_account",    params:{...}, depends_on:[], reason:"..."},
#    {step:2, function_id:"create_jira_account",      params:{...}, depends_on:[1], reason:"..."},
#    {step:3, function_id:"create_github_access",     params:{...}, depends_on:[1], reason:"..."},
#    {step:4, function_id:"send_slack_message",        params:{channel:"#engineering",...}, depends_on:[1,2,3], reason:"..."},
#    {step:5, function_id:"create_calendar_event",    params:{title:"Orientation...",...}, depends_on:[1], reason:"..."},
#    {step:6, function_id:"send_email",               params:{to:"alex.chen@company.com",...}, depends_on:[1,2,3,4], reason:"..."},
#  ]
```

### Execute with dependency resolution

Run the plan in topological order: steps with no dependencies run first, then steps whose dependencies are complete. Pass outputs from earlier steps into later ones: the new hire's `account_id` from step 1 flows into step 2's Jira onboarding, their `email` flows into step 6's welcome message.

```python title="planning/executor.py" theme={"dark"}
def execute_plan(
    plan:        list,    # [{step, function_id, params, depends_on, reason}]
    user_id:     str,
    orchestrator,
) -> dict:
    """
    Execute a plan in dependency order.
    Outputs from completed steps are available to downstream steps
    via the results dict keyed by step number.
    Stops on first failure unless the step is marked optional.
    """
    results    = {}   # step_number to execution result
    completed  = set()
    failed     = set()

    def _ready(step: dict) -> bool:
        """A step is ready when all its dependencies have completed successfully."""
        return all(d in completed for d in step.get("depends_on", []))

    remaining = list(plan)
    max_iterations = len(plan) * 2   # guard against circular deps
    iterations = 0

    while remaining and iterations < max_iterations:
        iterations += 1
        progress = False

        for step in list(remaining):
            if not _ready(step): continue

            step_num    = step["step"]
            function_id = step["function_id"]
            params      = dict(step.get("params", {}))

            # Inject outputs from dependency steps into params
            for dep_step in step.get("depends_on", []):
                dep_result = results.get(dep_step, {}).get("result", {})
                params.update({
                    k: v for k, v in dep_result.items()
                    if k in params.get("_inject_from_deps", {})
                })

            print(f"  Step {step_num}: {function_id}: {step.get('reason','')[:60]}")

            result = orchestrator.registry.get(function_id, lambda p: {"success": False, "error": "not registered"})(params)
            results[step_num] = {"function_id": function_id, "result": result}

            if result.get("success", False):
                completed.add(step_num)
            else:
                failed.add(step_num)
                print(f"  Step {step_num} FAILED: {result.get('error','unknown error')}")
                if not step.get("optional", False):
                    return {"status": "failed", "failed_step": step_num, "results": results}

            remaining.remove(step)
            progress = True

        if not progress:
            break   # circular dependency or all remaining steps blocked

    return {"status": "complete", "completed": list(completed), "failed": list(failed), "results": results}
```

### Rollback & compensation

For destructive or irreversible actions, register a compensation function alongside the main one. If step *n* fails after steps 1 to *n-1* have completed, the compensation chain runs in reverse order to undo what it can. Not all actions have meaningful rollbacks; a sent Slack message cannot be unsent. Mark those as `compensatable: false` in their schema metadata.

```python title="planning/rollback.py" theme={"dark"}
# Compensation registry: maps function_id to compensation callable
# Compensation functions undo the side effects of their paired function.
# Not every function has a compensation; mark those as non-compensatable.
COMPENSATION_REGISTRY = {
    "create_gsuite_account":   lambda r: delete_gsuite_account(r.get("account_id")),
    "create_jira_account":    lambda r: delete_jira_account(r.get("jira_id")),
    "create_github_access":   lambda r: revoke_github_access(r.get("username")),
    "create_calendar_event":  lambda r: delete_calendar_event(r.get("event_id")),
    # "send_slack_message":    None,  # cannot be meaningfully rolled back
    # "send_email":            None,  # cannot be unsent
}

def rollback_plan(results: dict, failed_step: int):
    """
    Compensate for completed steps in reverse order after a failure.
    Skips steps without a compensation function.
    Logs each compensation attempt to HydraDB's execution-log collection.
    """
    completed_steps = sorted(
        [s for s in results if s < failed_step], reverse=True
    )
    for step_num in completed_steps:
        entry       = results[step_num]
        function_id = entry["function_id"]
        result      = entry["result"]
        compensate  = COMPENSATION_REGISTRY.get(function_id)

        if compensate:
            try:
                compensate(result)
                print(f"  Rolled back step {step_num}: {function_id}")
                log_execution_outcome(
                    user_id="system", task=f"rollback {function_id}",
                    function_id=f"rollback_{function_id}",
                    outcome="success", latency_ms=0,
                )
            except Exception as e:
                print(f"  Rollback FAILED for step {step_num}: {e}")
        else:
            print(f"  Step {step_num} ({function_id}): no compensation registered. Manual review needed.")
```

STEP 6

## Event & Trigger Model

The Chief of Staff should react to three types of input: direct commands from users, scheduled jobs, and system events from external services. All three converge on the same `Orchestrator.handle_task()` call (or the planner, for multi-step tasks); only the source of the natural-language task differs.

**Flow:** Direct command, scheduled job, or system event, then Orchestrator, then Execution

### Slack slash-command (direct commands)

Expose a Slack slash-command that forwards the user's natural-language instruction directly to the Orchestrator. The Slack user ID maps to the `user_id` used for memory search, so the planner reads this user's preferences and personalizes multi-step plans accordingly.

```python title="triggers/slack_command.py" theme={"dark"}
import os, threading
from flask import Flask, request, jsonify
from slack_sdk import WebClient

flask_app  = Flask(__name__)
slack_bot  = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
orchestrator = ChiefOfStaffOrchestrator(
    registry=FUNCTION_REGISTRY,
    policy_engine=PolicyEngine(),
    auth_vault=AuthVault(),
)

# Maps Slack user IDs to internal user IDs (use CRM or HRIS lookup in production)
def slack_to_user_id(slack_uid: str) -> str:
    return f"slack-{slack_uid}"   # simplest; use CRM lookup in production

@flask_app.route("/slack/ai", methods=["POST"])
def handle_slack_ai_command():
    """
    Handles /ai slash-command from Slack.
    /ai book a 30-min call with alice next tuesday
    /ai send the deploy-complete message to #engineering
    /ai prepare for board meeting next Tuesday
    """
    slack_uid  = request.form["user_id"]
    task       = request.form.get("text", "").strip()
    channel    = request.form["channel_id"]
    user_id    = slack_to_user_id(slack_uid)

    if not task:
        return jsonify({"text": "Please describe what you'd like me to do."})

    def run_task():
        # Detect multi-step task and route accordingly
        MULTI_STEP_SIGNALS = ["prepare", "onboard", "set up", "ship", "launch"]
        is_multi = any(s in task.lower() for s in MULTI_STEP_SIGNALS)

        if is_multi:
            plan   = generate_execution_plan(task, user_id)
            result = execute_plan(plan, user_id, orchestrator)
            steps  = result.get("completed", [])
            reply  = f"Done. Completed {len(steps)} step(s): {', '.join(str(s) for s in steps)}."
        else:
            result = orchestrator.handle_task(task, user_id)
            reply  = result.get("result", {}).get("summary", "Done.")

        slack_bot.chat_postEphemeral(channel=channel, user=slack_uid, text=reply)

    # Run the work in the background and ack now: Slack times out the slash-command after 3s
    threading.Thread(target=run_task, daemon=True).start()
    return jsonify({"response_type": "ephemeral", "text": f"_Working on it: “{task[:80]}…”_"})

if __name__ == "__main__":
    flask_app.run(host="0.0.0.0", port=8080)
```

### Scheduled jobs

For recurring tasks (daily standup summaries, weekly metric reports, Monday morning briefings), use a cron-triggered cloud function. The task description is static, but HydraDB's function selection and personalization still apply because the `user_id` carries the recipient's memory and preferences. The CEO's Monday briefing looks different from the CTO's even though both come from the same cron.

```python title="triggers/scheduled_jobs.py" theme={"dark"}
# Cron: 0 8 * * 1  (Monday 8am UTC): weekly executive briefing
def weekly_executive_briefing():
    """
    Generate and deliver a personalized weekly briefing to each exec.
    HydraDB uses each exec's memory profile to select the right
    report type, channel, and level of detail automatically.
    """
    execs = [
        {"user_id": "sarah",   "task": "Generate and send this week's engineering team briefing."},
        {"user_id": "james",   "task": "Generate and send this week's sales pipeline summary."},
        {"user_id": "priya",   "task": "Generate and send this week's customer health report."},
    ]
    for exec_user in execs:
        try:
            plan   = generate_execution_plan(exec_user["task"], exec_user["user_id"])
            result = execute_plan(plan, exec_user["user_id"], orchestrator)
            print(f"Briefing sent to {exec_user['user_id']}: {result['status']}")
        except Exception as e:
            print(f"Briefing failed for {exec_user['user_id']}: {e}")


# Cron: 0 9 * * *  (Daily 9am): standup prompt for each team channel
def daily_standup_prompt():
    """Send the daily standup prompt to each team channel via Slack."""
    teams = [
        ("eng-lead",   "#engineering"),
        ("design-lead", "#design"),
        ("sales-lead",  "#sales"),
    ]
    for user_id, channel in teams:
        orchestrator.handle_task(
            task=f"Post the daily standup prompt to {channel}.",
            user_id=user_id,
        )

if __name__ == "__main__":
    weekly_executive_briefing()
    daily_standup_prompt()
```

### System event webhooks

Monitoring alerts, new Jira tickets, CRM stage changes, and GitHub PR events all carry natural-language context when translated into task descriptions. A thin webhook handler converts each event into an actionable task description and routes it through the Orchestrator. HydraDB decides the appropriate response based on severity, time of day, and the team's learned patterns.

```python title="triggers/webhooks.py" theme={"dark"}
@flask_app.route("/webhooks/monitoring", methods=["POST"])
def handle_monitoring_alert():
    """
    Receives a monitoring alert and routes it to the correct on-call response.
    HydraDB's memory of past incidents and the team's learned response patterns
    determine whether this triggers immediate escalation or scheduled review.
    """
    data     = request.json
    service  = data.get("service", "unknown")
    severity = data.get("severity", "low")   # "low" | "medium" | "high" | "critical"
    message  = data.get("message", "")

    # Translate the monitoring event into a natural-language task.
    # HydraDB's context graph knows which functions respond to incidents for this service.
    task = (
        f"{severity.upper()} monitoring alert for {service}: {message}. "
        f"Determine the appropriate incident response and take action."
    )

    # For critical alerts: immediately escalate, do not wait for plan generation
    if severity == "critical":
        orchestrator.handle_task(task=task, user_id="oncall-system")
    else:
        # For non-critical: generate a plan (may include "monitor and review at standup")
        plan   = generate_execution_plan(task, user_id="oncall-system")
        execute_plan(plan, user_id="oncall-system", orchestrator=orchestrator)

    return jsonify({"status": "handled"}), 200


@flask_app.route("/webhooks/crm", methods=["POST"])
def handle_crm_event():
    """
    Handles CRM stage-change events.
    When a deal moves to 'Demo Requested', the planner returns:
    check_customer_tier, find_available_demo_slots, create_demo_meeting,
    update_crm_opportunity, send_confirmation_email,
    all from one natural language task description.
    """
    data        = request.json
    deal_id     = data.get("deal_id")
    new_stage   = data.get("new_stage")
    company     = data.get("company_name", "the prospect")
    owner_id    = data.get("deal_owner_id", "sales-lead")

    # Translate the CRM event into a natural language task
    stage_tasks = {
        "Demo Requested":  f"Schedule a product demo for {company} (deal {deal_id}) and send confirmation.",
        "Proposal Sent":   f"Log a follow-up task in 3 days for {company} deal {deal_id} and set a reminder.",
        "Closed Won":      f"Celebrate the {company} win in Slack, update CRM deal {deal_id}, and trigger onboarding.",
        "Closed Lost":     f"Log loss reason for {company} deal {deal_id} and schedule a retrospective.",
    }
    task = stage_tasks.get(new_stage)
    if task:
        plan   = generate_execution_plan(task, owner_id)
        execute_plan(plan, owner_id, orchestrator)

    return jsonify({"status": "ok"}), 200
```

💡

> **The task description quality determines the response quality:** Don't pass raw webhook payloads to the Orchestrator. Always translate them into a clear natural-language task first: *"CRITICAL alert for payments-service: response time degraded to 8s. Determine appropriate incident response."* is far better than *"alert: payments-service, latency: 8000ms"*. HydraDB reasons over the task description: the more context you include, the more accurate the function selection.

STEP 7

## Security & Governance

An agent that takes real actions needs real guardrails. Four layers of protection: a policy engine that blocks unauthorized functions before they reach the execution layer, an auth vault that stores OAuth tokens and injects them per-function, an approval workflow that routes risky actions through a human gate, and a persistent audit log stored in HydraDB.

### Policy engine

The policy engine sits between HydraDB's suggestion and the execution. It checks: does this user have the required `permission_level`? Is this function restricted to specific departments? Is the current time within `business_hours_only` constraints? Is the function flagged as requiring human approval above a cost threshold?

```python title="security/policy.py" theme={"dark"}
from datetime import datetime, timezone

# User role definitions: load from your HRIS or auth system in production
USER_ROLES = {
    "sarah":  {"level": "manager", "departments": ["engineering", "all"]},
    "james":  {"level": "contributor", "departments": ["sales"]},
    "priya":  {"level": "manager", "departments": ["customer-success", "all"]},
}
PERMISSION_HIERARCHY = ["viewer", "contributor", "manager", "admin"]

class PolicyEngine:
    def check(self, user_id: str, function_id: str, schema: dict) -> bool:
        """
        Return True if the user is authorized to call this function.
        Checks: permission_level, department, business_hours, cost_threshold.
        """
        user_role = USER_ROLES.get(user_id, {"level": "viewer", "departments": []})
        meta      = schema.get("meta", {})

        # Check permission level
        required_level = meta.get("permission_level", "contributor")
        user_level     = user_role["level"]
        if PERMISSION_HIERARCHY.index(user_level) < PERMISSION_HIERARCHY.index(required_level):
            print(f"Policy blocked {user_id}: needs {required_level}, has {user_level}.")
            return False

        # Check department scope
        fn_dept       = meta.get("department", "all")
        user_depts    = user_role.get("departments", [])
        if fn_dept != "all" and fn_dept not in user_depts and "all" not in user_depts:
            print(f"Policy blocked {user_id}: {function_id} is department '{fn_dept}'.")
            return False

        # Check business hours constraint
        if meta.get("business_hours_only", False):
            now  = datetime.now(timezone.utc)
            hour = now.hour   # UTC: adjust for user timezone in production
            if not (9 <= hour <= 17) or now.weekday() >= 5:
                print(f"Policy blocked {function_id}: business hours only.")
                return False

        return True

    def requires_approval(self, user_id: str, schema: dict, params: dict) -> bool:
        """
        Return True if this call needs human approval before execution.
        Triggered when: cost exceeds threshold, function has 'requires_approval: true',
        or user is requesting an action outside their normal patterns.
        """
        meta      = schema.get("meta", {})
        threshold = meta.get("cost_threshold", None)
        amount    = params.get("amount", 0)
        if threshold and amount > threshold:
            return True
        return meta.get("requires_approval", False)
```

### Approval workflow

When `policy.requires_approval()` returns `True`, route the action through a Slack approval message before executing. Store the pending action in HydraDB so the Orchestrator can resume it after approval without losing context.

````python title="security/approval.py" theme={"dark"}
import json, os, uuid
from hydra_db import HydraDB

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"


def request_approval(
    user_id:     str,
    function_id: str,
    params:      dict,
    task:        str,
    approver_id: str,   # Slack user ID of the approver (their manager, finance, etc.)
) -> str:
    """
    Send an approval request via Slack and store the pending action in HydraDB.
    Returns an approval_id that the approver sends back to resume execution.
    """
    approval_id = str(uuid.uuid4())[:8]

    # Store the pending action in HydraDB so it can be resumed after approval
    client.context.ingest(
        type='memory',
        database=DATABASE,
        collection="approvals",
        upsert=True,
        memories=json.dumps([{
            "id":   f"approval-{approval_id}",   # fetch or delete it by this id later
            "text": json.dumps({
                "approval_id": approval_id,
                "user_id":     user_id,
                "function_id": function_id,
                "params":      params,
                "task":        task,
                "status":      "pending",
            }),
            "user_name": "approval-system",
            "infer":     False,
        }]),
    )

    # Send the Slack approval request to the approver (slack_bot from triggers/slack_command.py)
    slack_bot.chat_postMessage(
        channel=approver_id,
        text=(
            f"*Approval required* [ID: `{approval_id}`]\n"
            f"*Requested by:* {user_id}\n"
            f"*Action:* `{function_id}`\n"
            f"*Task:* {task}\n"
            f"*Params:* ```{json.dumps(params, indent=2)}```\n\n"
            f"Reply `/approve {approval_id}` or `/reject {approval_id} [reason]`"
        ),
    )
    return approval_id
````

⚠️

> **Never skip the policy check for system-triggered events:** Monitoring alerts and CRM webhooks bypass human input; there is no user to correct a bad suggestion. Always run `policy.check()` even for automated triggers, and set `user_id="oncall-system"` or a role-specific service account with appropriately scoped permissions. Automated triggers with admin-level access are the most dangerous pattern in this architecture.

STEP 8

## Observability & Self-Improvement

Track five metrics to understand if the Chief of Staff is working. Feed failures back to HydraDB to close the improvement loop. The system gets measurably better over time, not by manual tuning, but by accumulating execution memory.

| Metric | What to measure | Target | Action if below target |
| - | - | - | - |
| Suggestion acceptance rate | % of suggested functions the user actually runs without rejecting | >85% | Improve function descriptions; add more user preference memories |
| Multi-step plan completion rate | % of generated plans that complete all steps without rollback | >90% | Add compensation functions; fix idempotency issues in executors |
| P95 end-to-end latency | Time from task submission to last function execution complete | under 3s single-step, under 15s 5-step plan | Use mode: "fast" for single-function tasks; cache function registry |
| Rollback frequency | % of plans that trigger rollback due to mid-plan failure | under 2% | Add retries with back-off; mark flaky functions as optional: true |
| Function routing accuracy | % of tasks where HydraDB's top-1 suggestion matches what the user intended | >90% | Add user\_rejected feedback memories; rewrite function descriptions |

### Feeding metrics back to HydraDB

Every execution metric is a signal HydraDB can learn from. Query the `function-performance` collection with `type: "memory"` in the planner and pass the signals to the LLM next to the user preferences: a consistent `slow_response` signal for `create_calendar_event` then moves that function to the end of plans where it won't block other steps. Routing accuracy below threshold triggers re-examination of the function description and preference memory quality.

```python title="observability/metrics.py" theme={"dark"}
from datetime import datetime, timezone
import json, os
from hydra_db import HydraDB

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])
DATABASE = "chief-of-staff"


def report_function_performance(
    function_id: str,
    signal:      str,    # "slow_response" | "frequent_failure" | "high_rejection"
    details:     dict,   # {"p95_ms": 2500, "failure_rate": 0.12, "rejection_rate": 0.22}
):
    """
    Write performance signals as memories in the function-performance collection.
    A planner that reads this collection can steer away from problematic
    functions without removing them from the registry.
    Combine with function description updates for best results.
    """
    text = (
        f"Performance signal for {function_id}: {signal}.\n"
        f"Details: {json.dumps(details)}\n"
        f"Observed: {datetime.now(timezone.utc).isoformat()}"
    )
    client.context.ingest(
        type='memory',
        database=DATABASE,
        collection="function-performance",
        upsert=True,
        memories=json.dumps([{
            "text":      text,
            "user_name": "observability-system",
            "infer":     True,
        }]),
    )


# Example: called by a weekly metrics job
report_function_performance(
    function_id="create_calendar_event",
    signal="slow_response",
    details={"p95_ms": 2800, "sample_size": 412, "period": "2025-W22"},
)

report_function_performance(
    function_id="send_email",
    signal="high_rejection",
    details={
        "rejection_rate": 0.34,
        "most_common_correction": "send_slack_message",
        "note": "Users reject email in favour of Slack for 34% of internal tasks.",
    },
)
```

💡

> **The compound effect:** Every execution memory shifts the routing for the next call. After 500 executions per user, HydraDB has a detailed model of how that person works: which functions they trust, which channels they prefer, which task types they delegate vs. handle personally. The Chief of Staff becomes measurably more useful without any manual configuration. Track suggestion acceptance rate week-over-week as your primary health metric; it should trend upward continuously as memories accumulate.

## Complete API Reference

All endpoints used in this cookbook. Base URL: `https://api.hydradb.com`\
Header: `Authorization: Bearer YOUR_API_KEY`

`POST /context/ingest` bodies are shown as JSON for readability. Send each top-level key as a multipart form field, with `app_knowledge` or `memories` as a JSON string and `type=memory` for memories, as the Python examples do.

### Create database

**`POST /databases`**: one database for the full Chief of Staff system

```json title="body" theme={"dark"}
{
  "database": "chief-of-staff",
  "database_metadata_schema": [{ "name": "deprecated", "data_type": "BOOL" }]
}
```

### Upload function schemas

**`POST /context/ingest`**: upload function schemas as app sources or documents. The body below is the `app_knowledge` form field value. This cookbook sends batches of 20, 1s apart

```json title="body: one function" theme={"dark"}
[{
  "id":        "send_slack_message",
  "title":     "Send a Slack message",
  "type":      "function",
  "timestamp": "2025-01-01T00:00:00Z",
  "content":   { "text": "{ full JSON schema as string }" },
  "metadata":  { "deprecated": false },
  "additional_metadata": {
    "collections":  ["communication", "slack"],
    "permissions":  ["all_users"],
    "idempotent":   false,
    "side_effects": "Sends a visible Slack message. Cannot be unsent."
  }
}]
```

### Search function suggestions (single-step)

**`POST /query`**: returns top-matched function knowledge objects. `graph_context: false` only takes effect in `mode: "fast"`; with `mode: "thinking"`, as in this body, the graph slice is always included.

```json title="body" theme={"dark"}
{
  "database":     "chief-of-staff",
  "collection": "functions",
  "query":         "book a 30-min call with alice next tuesday",
  "max_results":   5,
  "mode":          "thinking",
  "graph_context": false,
  "metadata_filters": { "deprecated": { "equals": false } }
}
```

### Search function candidates (multi-step planning)

**`POST /query`**: graph\_context: true surfaces composed function chains

```json title="body" theme={"dark"}
{
  "database":     "chief-of-staff",
  "collection": "functions",
  "query":         "Onboard Alex Chen who starts Monday as a backend engineer.",
  "max_results":   12,
  "mode":          "thinking",
  "graph_context": true,
  "metadata_filters": { "deprecated": { "equals": false } }
}
```

### Search user preferences (personalise suggestions)

**`POST /query`**: returns channel preferences, urgency signals, communication style

```json title="body" theme={"dark"}
{
  "database":     "chief-of-staff",
  "collection": "user-sarah",
  "type":          "memory",
  "query":         "channel preferences urgency communication timing",
  "mode":          "thinking"
}
```

### Store user preference memory

**`POST /context/ingest`**: infer: true extracts channel + style signals

```json title="body" theme={"dark"}
{
  "type":       "memory",
  "memories": [{
    "text":      "Sarah always uses Slack DMs for urgent internal updates, not email.",
    "user_name": "sarah",
    "infer":     true
  }],
  "database":     "chief-of-staff",
  "collection": "user-sarah",
  "upsert":        true
}
```

### Store execution outcome (audit log)

**`POST /context/ingest`**: infer: false for verbatim audit records

```json title="body" theme={"dark"}
{
  "type":       "memory",
  "memories": [{
    "text":      "[2025-06-10T09:12:44Z] user=sarah fn=send_slack_message outcome=success latency=180ms",
    "user_name": "system",
    "infer":     false
  }],
  "database":     "chief-of-staff",
  "collection": "execution-log",
  "upsert":        true
}
```

### Store performance signal (self-improvement)

**`POST /context/ingest`**: infer: true so HydraDB links signal to future suggestions

```json title="body" theme={"dark"}
{
  "type":       "memory",
  "memories": [{
    "text":      "create_calendar_event returned slow_response: p95=2800ms over 412 calls in W22.",
    "user_name": "observability-system",
    "infer":     true
  }],
  "database":     "chief-of-staff",
  "collection": "function-performance",
  "upsert":        true
}
```

### Parameter extraction

**Local LLM parameter extraction**: extract structured params after HydraDB returns the best function schema

```json title="LLM prompt inputs" theme={"dark"}
{
  "task": "ping #engineering that the deploy is done",
  "function_schema": {
    "id": "send_slack_message",
    "parameters": {
      "channel": "string",
      "text": "string"
    }
  },
  "expected_output": {
    "channel": "#engineering",
    "text": "the deploy is done"
  }
}
```

## Benchmarks

Tested across 3,200 task executions spanning 48 registered functions and 6 user profiles. Comparison baseline: a standard LLM agent with function-calling and no persistent memory layer, using the same function schemas as tool definitions.

| Metric | Standard LLM function-calling | HydraDB Chief of Staff | Delta |
| - | - | - | - |
| Top-1 function routing accuracy (week 1) | 71% | 78% | +10% |
| Top-1 function routing accuracy (week 8, after memory accumulation) | 72% | 93% | +29% |
| Multi-step plan completion rate (5-step plans) | 54% | 88% | +63% |
| Personalization accuracy (correct channel/timing per user) | 31% | 86% | +177% |
| Suggestion acceptance rate (no user rejection) | 68% | 91% | +34% |
| P95 function selection latency (single step, mode: fast) | N/A | under 120 ms | Sub-second |
| P95 plan generation latency (5-step, mode: thinking) | N/A | under 680 ms | Sub-second |

ℹ️

> **Benchmark methodology:** Figures are based on internal HydraDB testing. For the formal benchmark paper and methodology, see [research.hydradb.com/hydradb.pdf](https://research.hydradb.com/hydradb.pdf). Results will vary by function library size, description quality, and the volume of execution memory accumulated.

ℹ️

> The jump from 78% to 93% routing accuracy between week 1 and week 8 reflects HydraDB's memory accumulation. In week 1, function selection is purely semantic: it reads descriptions and matches tasks. By week 8, 3,200 execution outcomes have been stored as memory, and HydraDB has learned that Sarah always uses Slack over email, that the engineering team routes alerts to PagerDuty not Jira, and that "prepare for a call" for sales users means checking the CRM first. Standard LLM function-calling stays flat at 72% because it resets every session.

***

***

**Prev:** Build an AI Competitive Intelligence Agent\
**Next:** Build a Multi-Agent Research Pipeline


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.